| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field, asdict |
| from typing import Any, Dict, List, Optional |
| import uuid |
| import datetime |
|
|
|
|
| @dataclass |
| class ExecutionContext: |
| """Request-scoped context that survives skill flushing and runtime transitions.""" |
|
|
| request_id: str = field(default_factory=lambda: f"req-{uuid.uuid4().hex[:8]}") |
| trace_id: str = field(default_factory=lambda: f"trc-{uuid.uuid4().hex[:8]}") |
| session_id: str = field(default_factory=lambda: f"ses-{uuid.uuid4().hex[:8]}") |
| patient_id: Optional[str] = None |
| authorization_subject: Optional[str] = None |
|
|
| user_query: str = "" |
| user_role: Optional[str] = None |
| intent: Optional[str] = None |
|
|
| patient_context: Dict[str, Any] = field(default_factory=dict) |
| guideline_context: List[Any] = field(default_factory=list) |
| retrieved_sources: List[Dict[str, Any]] = field(default_factory=list) |
|
|
| worker_result: Optional[Dict[str, Any]] = None |
| response_artifact: Optional[Dict[str, Any]] = None |
|
|
| gate_results: Dict[str, Any] = field(default_factory=dict) |
| recovery_attempts: int = 0 |
|
|
| termination_reason: Optional[str] = None |
| skill_trace: List[Dict[str, Any]] = field(default_factory=list) |
| metadata: Dict[str, Any] = field(default_factory=dict) |
|
|
| def record_trace(self, event: str, details: Dict[str, Any]) -> None: |
| trace_entry = { |
| "event": event, |
| "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), |
| **details, |
| } |
| self.skill_trace.append(trace_entry) |
|
|
| def update_state(self, **kwargs: Any) -> None: |
| for key, value in kwargs.items(): |
| if hasattr(self, key): |
| setattr(self, key, value) |
| else: |
| self.metadata[key] = value |
|
|
| def clear_worker_context(self) -> None: |
| """Release ephemeral worker results while retaining persistent session state.""" |
| self.worker_result = None |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return asdict(self) |
|
|
