File size: 2,046 Bytes
cb505ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | 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)
|