| from __future__ import annotations |
|
|
| from typing import Any, Dict, Optional |
| from src.skills.models import SkillDefinition, SkillResult |
| from src.skills.registry import SkillRegistry |
| from src.runtime.agent_runtime import AgentRuntime |
| from src.runtime.execution_context import ExecutionContext |
|
|
|
|
| class ManagerRuntime: |
| """Persistent Manager Runtime for orchestration, governance, and routing skill lifecycle.""" |
|
|
| ROLE_TO_WORKER_SKILL = { |
| "patient": "patient-support", |
| "caregiver": "caregiver-support", |
| "researcher": "diabetes-research", |
| "dietary": "diabetes-dietary-coach", |
| } |
| INTENT_TO_WORKER_SKILL = { |
| "diagnosis": "clinical-diagnosis-support", |
| "treatment": "clinical-treatment-support", |
| "monitoring": "clinical-monitoring-support", |
| "general": "clinical-general-support", |
| } |
|
|
| def __init__(self, registry: SkillRegistry): |
| self.registry = registry |
| self.agent_runtime = AgentRuntime(name="ManagerRuntime", registry=registry) |
|
|
| @property |
| def active_skill(self) -> Optional[SkillDefinition]: |
| return self.agent_runtime.active_skill |
|
|
| def load_skill(self, skill_id: str, version: Optional[str] = None) -> SkillDefinition: |
| config = self.agent_runtime.load_skill(skill_id, version=version) |
| return config.definition |
|
|
| def execute(self, context: ExecutionContext, input_override: Optional[Dict[str, Any]] = None) -> SkillResult: |
| return self.agent_runtime.execute(context=context, input_override=input_override) |
|
|
| def flush_skill(self) -> None: |
| self.agent_runtime.flush() |
|
|
| def plan_route(self, context: ExecutionContext) -> Dict[str, Any]: |
| """Return the next approved runtime step for the request context.""" |
| role = (context.user_role or "").strip().lower() or None |
| intent = (context.intent or "").strip().lower() or None |
|
|
| if role is None: |
| return { |
| "route_type": "manager_skill", |
| "manager_skill": "role-routing", |
| "worker_skill": None, |
| "role": None, |
| "intent": None, |
| } |
|
|
| if role == "clinician" and intent is None: |
| return { |
| "route_type": "manager_skill", |
| "manager_skill": "clinical-intent-routing", |
| "worker_skill": None, |
| "role": role, |
| "intent": None, |
| } |
|
|
| if role == "clinician": |
| worker_skill = self.INTENT_TO_WORKER_SKILL.get(intent, self.INTENT_TO_WORKER_SKILL["general"]) |
| else: |
| worker_skill = self.ROLE_TO_WORKER_SKILL.get(role) |
|
|
| if worker_skill is None: |
| return { |
| "route_type": "unroutable", |
| "manager_skill": None, |
| "worker_skill": None, |
| "role": role, |
| "intent": intent, |
| } |
|
|
| return { |
| "route_type": "worker_skill", |
| "manager_skill": None, |
| "worker_skill": worker_skill, |
| "role": role, |
| "intent": intent or "general", |
| } |
|
|
| def route_request(self, context: ExecutionContext) -> SkillResult: |
| """Execute the required Manager skill or return an approved worker plan.""" |
| plan = self.plan_route(context) |
| if plan["route_type"] == "manager_skill": |
| self.load_skill(plan["manager_skill"]) |
| result = self.execute(context) |
| self.flush_skill() |
| result.output["route_plan"] = plan |
| return result |
|
|
| return SkillResult( |
| skill_id="manager-orchestration", |
| skill_version="1.0.0", |
| output={"status": "routing_complete", "route_plan": plan}, |
| ) |
|
|
