from __future__ import annotations import asyncio import hashlib import json import re from contextlib import suppress from dataclasses import dataclass, field from datetime import datetime from typing import Any, Literal from .agent_system import TaskVersionConflict, UnifiedAgentSystem from .event_store import EventStore from .hermes import HermesAdapter from .media import MediaJobAdapter from .models import ( AgentFocusResponse, HermesRunRequest, MediaArtifact, MediaJobRequest, ModelToolReceipt, ModelToolRequest, OperationView, RouteDecision, RouteRequest, SessionEndResponse, TranscriptInput, TranscriptReceipt, new_id, utc_now, ) from .router import Router from .remote_nodes import AmbiguousNodeTarget TERMINAL_STATUSES = {"completed", "failed", "cancelled"} _VOICE_APPROVE = {"ja", "ja bitte", "yes", "yes please", "approve", "genehmigen", "bestätigen"} _VOICE_DENY = {"nein", "no", "deny", "ablehnen"} _LIVE_SCREEN_COMMENTARY_RE = re.compile( r"(?:bildschirm|screen|desktop).{0,80}(?:live|laufend|kontinuierlich|kommentier|beobacht)" r"|(?:live|laufend|kontinuierlich|kommentier|beobacht).{0,80}(?:bildschirm|screen|desktop)", re.IGNORECASE, ) class IdempotencyConflict(RuntimeError): pass class OperationNotFound(KeyError): pass @dataclass(slots=True) class _RouteOverride: needs_vision: bool = False target_node: str | None = None required_capabilities: set[str] = field(default_factory=set) requested_permissions: set[str] = field(default_factory=set) @dataclass(slots=True) class _Operation: operation_id: str kind: Literal["hermes", "video", "image"] session_id: str turn_id: str trace_id: str prompt: str thinking_oz: Literal[2, 4, 8] needs_vision: bool target_node: str | None required_capabilities: set[str] requested_permissions: set[str] idempotency_key: str request_hash: str input_hash: str options: dict[str, Any] task_id: str | None = None task_version: int | None = None status: str = "created" downstream_id: str | None = None endpoint: str | None = None lease_id: str | None = None created_at: datetime = field(default_factory=utc_now) updated_at: datetime = field(default_factory=utc_now) terminal_event_id: str | None = None task: asyncio.Task[None] | None = None @property def prefix(self) -> str: return "tool" if self.kind == "hermes" else "media.job" def view(self) -> OperationView: return OperationView( operation_id=self.operation_id, kind=self.kind, session_id=self.session_id, turn_id=self.turn_id, trace_id=self.trace_id, task_id=self.task_id, task_version=self.task_version, status=self.status, # type: ignore[arg-type] downstream_id=self.downstream_id, created_at=self.created_at, updated_at=self.updated_at, terminal_event_id=self.terminal_event_id, ) def _request_hash( kind: str, prompt: str, thinking_oz: int, needs_vision: bool, target_node: str | None, required_capabilities: set[str], requested_permissions: set[str], options: dict[str, Any], ) -> str: raw = json.dumps( { "kind": kind, "prompt": prompt, "thinking_oz": thinking_oz, "needs_vision": needs_vision, "target_node": target_node, "required_capabilities": sorted(required_capabilities), "requested_permissions": sorted(requested_permissions), "options": options, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(raw).hexdigest() def _input_hash(transcript: TranscriptInput) -> str: raw = json.dumps( { "session_id": transcript.session_id, "turn_id": transcript.turn_id, "text": transcript.text, "language": transcript.language, "thinking_oz": transcript.thinking_oz, "needs_vision": transcript.needs_vision, "target_node": transcript.target_node, "required_capabilities": sorted(transcript.required_capabilities), "requested_permissions": sorted(transcript.requested_permissions), "minicpm_session_id": transcript.minicpm_session_id, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(raw).hexdigest() class ConductorService: def __init__( self, *, store: EventStore, router: Router, hermes: HermesAdapter, media: MediaJobAdapter, agent_system: UnifiedAgentSystem | None = None, ) -> None: self.store = store self.router = router self.hermes = hermes self.media = media self.agent_system = agent_system or UnifiedAgentSystem() self._lock = asyncio.Lock() self._operations: dict[str, _Operation] = {} self._idempotency: dict[str, tuple[str, str]] = {} self._next_hermes: dict[str, _RouteOverride] = {} self._pending_approvals: dict[str, list[str]] = {} self._approval_idempotency: dict[str, tuple[str, TranscriptReceipt]] = {} self._model_tool_idempotency: dict[str, tuple[str, ModelToolReceipt]] = {} self._initialized = False async def initialize(self) -> None: if self._initialized: return await self.store.initialize() events = await self.store.replay(limit=self.store.max_replay_events) async with self._lock: if self._initialized: return for event in events: if event.type == "route.override.set" and event.payload.get("route") == "hermes": self._next_hermes[event.session_id] = _RouteOverride( needs_vision=bool(event.payload.get("needs_vision", False)), target_node=event.payload.get("target_node"), required_capabilities=set( event.payload.get("required_capabilities", []) ), requested_permissions=set( event.payload.get("requested_permissions", []) ), ) elif event.type in {"route.override.consumed", "route.override.cleared", "session.ended"}: self._next_hermes.pop(event.session_id, None) if event.type not in {"tool.created", "media.job.created"}: operation = self._operations.get(event.operation_id) if operation is not None: self._apply_replayed_event(operation, event.type, event.event_id, event.payload) continue payload = event.payload kind = payload.get("kind") if kind not in {"hermes", "video", "image"}: continue operation = _Operation( operation_id=event.operation_id, kind=kind, session_id=event.session_id, turn_id=event.turn_id, trace_id=event.trace_id, prompt=str(payload.get("prompt", "")), needs_vision=bool(payload.get("needs_vision", False)), target_node=payload.get("target_node"), required_capabilities=set(payload.get("required_capabilities", [])), requested_permissions=set(payload.get("requested_permissions", [])), idempotency_key=str(payload.get("idempotency_key", "")), request_hash=str(payload.get("request_hash", "")), input_hash=str(payload.get("input_hash", "")), options=dict(payload.get("options") or {}), task_id=payload.get("task_id"), task_version=payload.get("task_version"), thinking_oz=int(payload.get("thinking_oz", 4)), created_at=event.timestamp, updated_at=event.timestamp, ) self._operations[operation.operation_id] = operation if operation.idempotency_key: self._idempotency[operation.idempotency_key] = ( operation.request_hash, operation.operation_id, ) self._initialized = True # An interrupted process cannot safely resume an unknown downstream stream. # Close each recovered nonterminal operation once, preserving the invariant. for operation in list(self._operations.values()): if operation.status not in TERMINAL_STATUSES: await self._terminal( operation, "failed", {"code": "conductor_restarted", "message": "Operation was interrupted by restart"}, ) @staticmethod def _apply_replayed_event( operation: _Operation, event_type: str, event_id: str, payload: dict[str, Any] ) -> None: prefix = operation.prefix mapping = { f"{prefix}.queued": "queued", f"{prefix}.started": "started", f"{prefix}.approval.required": "approval_required", f"{prefix}.cancelling": "cancelling", f"{prefix}.completed": "completed", f"{prefix}.failed": "failed", f"{prefix}.cancelled": "cancelled", } if event_type in mapping: operation.status = mapping[event_type] if operation.status in TERMINAL_STATUSES: operation.terminal_event_id = event_id downstream_id = payload.get("downstream_id") or payload.get("run_id") or payload.get("job_id") if isinstance(downstream_id, str): operation.downstream_id = downstream_id async def ingest_transcript( self, transcript: TranscriptInput, *, idempotency_key: str | None = None, ) -> TranscriptReceipt: await self.initialize() trace_id = transcript.trace_id or new_id("trace") input_operation_id = "input_" + hashlib.sha256( f"{transcript.session_id}\0{transcript.turn_id}".encode("utf-8") ).hexdigest()[:20] event = await self.store.append( "input.transcript.final" if transcript.final else "input.transcript.partial", session_id=transcript.session_id, turn_id=transcript.turn_id, operation_id=input_operation_id, trace_id=trace_id, source="asr", payload={ "text": transcript.text, "language": transcript.language, "thinking_oz": transcript.thinking_oz, "confidence": transcript.confidence, "needs_vision": transcript.needs_vision, "target_node": transcript.target_node, "required_capabilities": sorted(transcript.required_capabilities), "requested_permissions": sorted(transcript.requested_permissions), "minicpm_session_id": transcript.minicpm_session_id, }, ) if not transcript.final: return TranscriptReceipt(final=False, event_seq=event.seq) effective_idempotency_key = idempotency_key or ( f"turn:{transcript.session_id}:{transcript.turn_id}" ) approval_replay = self._approval_idempotency.get(effective_idempotency_key) if approval_replay is not None: previous_hash, previous_receipt = approval_replay if previous_hash != _input_hash(transcript): raise IdempotencyConflict( "Idempotency key was already used with a different approval response" ) return previous_receipt.model_copy( update={"event_seq": event.seq, "replayed": True} ) existing = self._idempotency.get(effective_idempotency_key) if existing is not None: operation = self._operations[existing[1]] if operation.input_hash != _input_hash(transcript): raise IdempotencyConflict( "Idempotency key was already used with a different input" ) replay_decision = RouteDecision( route=operation.kind, confidence=1.0, needs_vision=operation.needs_vision, risk="medium" if operation.kind == "hermes" else "low", reason_code="idempotent_replay", ack="Der bestehende Vorgang wird weiterverwendet.", ) return TranscriptReceipt( final=True, event_seq=event.seq, route=replay_decision, operation_id=operation.operation_id, replayed=True, ) approval_receipt = await self._handle_pending_approval( transcript, input_event_id=event.event_id, event_seq=event.seq, trace_id=trace_id, idempotency_key=effective_idempotency_key, ) if approval_receipt is not None: return approval_receipt _, intent = await self.agent_system.observe_input( transcript.session_id, transcript.text, turn_id=transcript.turn_id, ) if intent.kind == "MODIFY_TASK": await self.cancel_session(transcript.session_id, reason="task_updated") decision = await self.router.decide( RouteRequest( text=transcript.text, needs_vision=transcript.needs_vision, language=transcript.language, ) ) target_node = transcript.target_node required_capabilities = set(transcript.required_capabilities) requested_permissions = set(transcript.requested_permissions) if decision.route != "cancel": override = await self._consume_next_hermes( session_id=transcript.session_id, turn_id=transcript.turn_id, trace_id=trace_id, causation_id=event.event_id, ) if override is not None: target_node = target_node or override.target_node required_capabilities.update(override.required_capabilities) requested_permissions.update(override.requested_permissions) if override is not None or target_node or required_capabilities or requested_permissions: decision = RouteDecision( route="hermes", confidence=1.0, needs_vision=transcript.needs_vision or bool( override and override.needs_vision ), risk="medium", reason_code="user_forced_hermes", ack="Ich übernehme das mit dem ausgewählten Agenten.", ) if intent.kind in {"NEW_TASK", "MODIFY_TASK"} and decision.route == "dialogue": decision = RouteDecision( route="hermes", confidence=intent.confidence, needs_vision=transcript.needs_vision, risk="medium", reason_code="unified_agent_task", ack="Ich kümmere mich darum.", thinking_oz=decision.thinking_oz, ) if decision.route == "hermes" and target_node is None: resolver = getattr(self.hermes, "resolve_target", None) if resolver is not None: try: target_node = await resolver(transcript.text) except AmbiguousNodeTarget: decision = RouteDecision( route="clarify", confidence=1.0, needs_vision=decision.needs_vision, risk="low", reason_code="multiple_trusted_nodes_named", ack="Welchen der genannten Computer soll ich verwenden?", ) effective_thinking_oz = transcript.thinking_oz or decision.thinking_oz decision = decision.model_copy(update={"thinking_oz": effective_thinking_oz}) route_event = await self.store.append( "route.decided", session_id=transcript.session_id, turn_id=transcript.turn_id, operation_id=input_operation_id, trace_id=trace_id, causation_id=event.event_id, source="conductor", payload=decision.model_dump(mode="json"), ) await self.store.append( "assistant.activity", session_id=transcript.session_id, turn_id=transcript.turn_id, operation_id=input_operation_id, trace_id=trace_id, causation_id=route_event.event_id, source="conductor", payload={"state": "ack", "text": decision.ack, "route": decision.route}, ) if decision.route == "cancel": await self.agent_system.apply_intent( transcript.session_id, intent, route=decision.route ) await self.cancel_session(transcript.session_id, reason="voice_request") return TranscriptReceipt(final=True, event_seq=event.seq, route=decision) task_state = await self.agent_system.apply_intent( transcript.session_id, intent, route=decision.route ) if decision.route not in {"hermes", "video", "image"}: return TranscriptReceipt(final=True, event_seq=event.seq, route=decision) operation, replayed = await self._create_operation( kind=decision.route, prompt=transcript.text, thinking_oz=effective_thinking_oz, needs_vision=decision.needs_vision, target_node=target_node, required_capabilities=required_capabilities, requested_permissions=( requested_permissions | self._infer_permissions(transcript.text, decision.needs_vision) if decision.route == "hermes" else set() ), session_id=transcript.session_id, turn_id=transcript.turn_id, trace_id=trace_id, idempotency_key=effective_idempotency_key, input_hash=_input_hash(transcript), causation_id=route_event.event_id, options={}, task_id=task_state.id if task_state else None, task_version=task_state.version if task_state else None, ) if not replayed: operation.task = asyncio.create_task( self._run_operation(operation), name=f"omni-{operation.operation_id}" ) return TranscriptReceipt( final=True, event_seq=event.seq, route=decision, operation_id=operation.operation_id, replayed=replayed, ) async def invoke_model_tool( self, request: ModelToolRequest, *, idempotency_key: str, ) -> ModelToolReceipt: """One authenticated ingress shared by MiniCPM, Qwen and connectors. Media specialists stay cold until the resulting operation acquires a lease. Voice output is handed straight back to the always-warm MiniCPM session through the existing conductor event stream. """ await self.initialize() trace_id = request.trace_id or new_id("trace") tool_map = { "reasoning.delegate": ("hermes", "qwen", "always"), "internet.search": ("hermes", "qwen", "always"), "image.generate": ("image", "hunyuan", "cold"), "image.edit": ("image", "hunyuan", "cold"), "video.generate": ("video", "minimax-h3", "cold"), "audio.generate": ("video", "minimax-h3", "cold"), "voice.speak": (None, "minicpm", "always"), } kind, target_model, lifecycle = tool_map[request.tool] tool_hash = hashlib.sha256( request.model_dump_json(exclude={"trace_id"}).encode("utf-8") ).hexdigest() previous = self._model_tool_idempotency.get(idempotency_key) if previous is not None: previous_hash, previous_receipt = previous if previous_hash != tool_hash: raise IdempotencyConflict( "Idempotency key was already used with a different model tool request" ) return previous_receipt.model_copy(update={"replayed": True}) if kind is None: event = await self.store.append( "assistant.narration", session_id=request.session_id, turn_id=request.turn_id, operation_id="voice_" + tool_hash[:20], trace_id=trace_id, source=request.source_model, payload={ "kind": "voice_output", "text": request.prompt, "target_model": target_model, "local_only": True, }, ) receipt = ModelToolReceipt( tool=request.tool, target_model=target_model, lifecycle=lifecycle, status="completed", event_seq=event.seq, ) self._model_tool_idempotency[idempotency_key] = (tool_hash, receipt) return receipt media_options = dict(request.options) if request.tool == "image.edit": media_options["mode"] = "edit" elif request.tool == "audio.generate": media_options.update({"output_modalities": ["audio"], "mode": "audio_only"}) created_event = await self.store.append( "model.tool.requested", session_id=request.session_id, turn_id=request.turn_id, operation_id="modeltool_" + tool_hash[:20], trace_id=trace_id, source=request.source_model, payload={ "tool": request.tool, "target_model": target_model, "lifecycle": lifecycle, "options": media_options, }, ) required_capabilities = ( {"internet.search"} if request.tool == "internet.search" else set() ) requested_permissions = ( {"network.search"} if request.tool == "internet.search" else set() ) task_state = None if request.task_id is not None and request.task_version is not None: task_state = await self.agent_system.tasks.current( request.task_id, request.task_version ) else: task_state = await self.agent_system.tasks.create( request.session_id, request.prompt, ) operation, replayed = await self._create_operation( kind=kind, prompt=request.prompt, thinking_oz=request.thinking_oz, needs_vision=False, target_node=request.target_node, required_capabilities=required_capabilities, requested_permissions=requested_permissions, session_id=request.session_id, turn_id=request.turn_id, trace_id=trace_id, idempotency_key=idempotency_key, input_hash=tool_hash, causation_id=created_event.event_id, options=media_options, task_id=task_state.id if task_state else None, task_version=task_state.version if task_state else None, ) if not replayed: operation.task = asyncio.create_task( self._run_operation(operation), name=f"omni-{operation.operation_id}" ) receipt = ModelToolReceipt( tool=request.tool, target_model=target_model, lifecycle=lifecycle, status="queued", event_seq=created_event.seq, operation_id=operation.operation_id, replayed=replayed, ) self._model_tool_idempotency[idempotency_key] = (tool_hash, receipt) return receipt @staticmethod def _voice_approval_choice(text: str) -> str | None: normalized = re.sub(r"[.!?]+$", "", " ".join(text.casefold().split())).strip() if normalized in _VOICE_APPROVE: return "once" if normalized in _VOICE_DENY: return "deny" return None async def _handle_pending_approval( self, transcript: TranscriptInput, *, input_event_id: str, event_seq: int, trace_id: str, idempotency_key: str, ) -> TranscriptReceipt | None: async with self._lock: pending_ids = [ operation_id for operation_id in self._pending_approvals.get(transcript.session_id, []) if operation_id in self._operations and self._operations[operation_id].status == "approval_required" ] if not pending_ids: self._pending_approvals.pop(transcript.session_id, None) return None self._pending_approvals[transcript.session_id] = pending_ids choice = self._voice_approval_choice(transcript.text) if choice is None or len(pending_ids) != 1: decision = RouteDecision( route="clarify", confidence=1.0, needs_vision=False, risk="low", reason_code=( "multiple_pending_approvals" if len(pending_ids) != 1 else "approval_answer_ambiguous" ), ack="Bitte antworte eindeutig mit Ja oder Nein.", ) await self._append_control_decision( transcript=transcript, trace_id=trace_id, input_event_id=input_event_id, operation_id=pending_ids[0], decision=decision, ) return TranscriptReceipt( final=True, event_seq=event_seq, route=decision, operation_id=pending_ids[0] ) operation = self._operations[pending_ids[0]] async with self._lock: current = self._pending_approvals.get(transcript.session_id, []) self._pending_approvals[transcript.session_id] = [ item for item in current if item != operation.operation_id ] try: if not operation.downstream_id: raise RuntimeError("Pending approval has no Hermes run ID") await self.hermes.approve(operation.downstream_id, choice) except Exception: async with self._lock: self._pending_approvals.setdefault(transcript.session_id, []).append( operation.operation_id ) raise responded = await self.store.append( "approval.responded", session_id=operation.session_id, turn_id=transcript.turn_id, operation_id=operation.operation_id, trace_id=trace_id, causation_id=input_event_id, source="conductor", payload={ "choice": choice, "scope": "once" if choice == "once" else "deny", "permanent": False, }, ) async with self._lock: if operation.status == "approval_required": operation.status = "started" operation.updated_at = responded.timestamp decision = RouteDecision( route="hermes", confidence=1.0, needs_vision=operation.needs_vision, risk="medium", reason_code="approval_once" if choice == "once" else "approval_denied", ack="Einmalig genehmigt." if choice == "once" else "Abgelehnt.", ) await self._append_control_decision( transcript=transcript, trace_id=trace_id, input_event_id=responded.event_id, operation_id=operation.operation_id, decision=decision, ) receipt = TranscriptReceipt( final=True, event_seq=event_seq, route=decision, operation_id=operation.operation_id, ) self._approval_idempotency[idempotency_key] = ( _input_hash(transcript), receipt ) return receipt async def _append_control_decision( self, *, transcript: TranscriptInput, trace_id: str, input_event_id: str, operation_id: str, decision: RouteDecision, ) -> None: route_event = await self.store.append( "route.decided", session_id=transcript.session_id, turn_id=transcript.turn_id, operation_id=operation_id, trace_id=trace_id, causation_id=input_event_id, source="conductor", payload=decision.model_dump(mode="json"), ) await self.store.append( "assistant.activity", session_id=transcript.session_id, turn_id=transcript.turn_id, operation_id=operation_id, trace_id=trace_id, causation_id=route_event.event_id, source="conductor", payload={"state": "ack", "text": decision.ack, "route": decision.route}, ) async def _consume_next_hermes( self, *, session_id: str, turn_id: str, trace_id: str, causation_id: str, ) -> _RouteOverride | None: async with self._lock: override = self._next_hermes.pop(session_id, None) if override is None: return None await self.store.append( "route.override.consumed", session_id=session_id, turn_id=turn_id, operation_id=self._context_operation_id(session_id), trace_id=trace_id, causation_id=causation_id, source="conductor", payload={ "route": "hermes", "needs_vision": override.needs_vision, "target_node": override.target_node, "required_capabilities": sorted(override.required_capabilities), "requested_permissions": sorted(override.requested_permissions), }, ) return override @staticmethod def _infer_permissions(text: str, needs_vision: bool) -> set[str]: normalized = text.casefold() permissions: set[str] = set() if needs_vision: permissions.add("desktop.observe") if any(word in normalized for word in ("screenshot", "bildschirm", "screen", "anzeige")): permissions.add("desktop.observe") if any(word in normalized for word in ("klick", "öffne", "tippe", "browser", "desktop", "maus", "cursor")): permissions.add("desktop.control") permissions.add("desktop.observe") if any(word in normalized for word in ("datei", "ordner", "file", "folder")): permissions.add("files.read") if any(word in normalized for word in ("schreib", "änder", "speicher", "write", "edit", "save")): permissions.add("files.write") if any(word in normalized for word in ("code", "terminal", "powershell", "script", "skript", "debug")): permissions.add("tool.execute") if any(word in normalized for word in ("webcam", "kamera", "camera")): permissions.add("webcam.read") if any(word in normalized for word in ("internet", "websuche", "search the web", "online suchen")): permissions.add("network.search") return permissions @staticmethod def _context_operation_id(session_id: str) -> str: return "context_" + hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:20] async def agent_focus( self, session_id: str, *, turn_id: str, trace_id: str | None, needs_vision: bool, cancel_if_active: bool, armed: bool, target_node: str | None, required_capabilities: set[str], requested_permissions: set[str], ) -> AgentFocusResponse: await self.initialize() active = [ operation.operation_id for operation in self._operations.values() if operation.session_id == session_id and operation.status not in TERMINAL_STATUSES ] if active and cancel_if_active: cancelled = await asyncio.gather( *(self.cancel_operation(operation_id, reason="agent_focus_toggle") for operation_id in active) ) return AgentFocusResponse( mode="cancelled", cancelled_operation_ids=[item.operation_id for item in cancelled], ) effective_trace = trace_id or new_id("trace") if not armed: async with self._lock: self._next_hermes.pop(session_id, None) await self.store.append( "route.override.cleared", session_id=session_id, turn_id=turn_id, operation_id=self._context_operation_id(session_id), trace_id=effective_trace, source="conductor", payload={"route": "hermes"}, ) return AgentFocusResponse(mode="cleared") async with self._lock: self._next_hermes[session_id] = _RouteOverride( needs_vision=needs_vision, target_node=target_node, required_capabilities=set(required_capabilities), requested_permissions=set(requested_permissions), ) await self.store.append( "route.override.set", session_id=session_id, turn_id=turn_id, operation_id=self._context_operation_id(session_id), trace_id=effective_trace, source="conductor", payload={ "route": "hermes", "needs_vision": needs_vision, "target_node": target_node, "required_capabilities": sorted(required_capabilities), "requested_permissions": sorted(requested_permissions), "one_shot": True, }, ) return AgentFocusResponse( mode="armed", next_route="hermes", target_node=target_node ) async def end_session( self, session_id: str, *, turn_id: str, trace_id: str | None, ) -> SessionEndResponse: cancelled = await self.cancel_session(session_id, reason="session_ended") effective_trace = trace_id or new_id("trace") async with self._lock: self._next_hermes.pop(session_id, None) await self.store.append( "session.ended", session_id=session_id, turn_id=turn_id, operation_id=self._context_operation_id(session_id), trace_id=effective_trace, source="conductor", payload={"cancelled_operation_ids": [item.operation_id for item in cancelled]}, ) return SessionEndResponse( cancelled_operation_ids=[item.operation_id for item in cancelled] ) async def _create_operation( self, *, kind: Literal["hermes", "video", "image"], prompt: str, thinking_oz: Literal[2, 4, 8], needs_vision: bool, target_node: str | None, required_capabilities: set[str], requested_permissions: set[str], session_id: str, turn_id: str, trace_id: str, idempotency_key: str, input_hash: str, causation_id: str | None, options: dict[str, Any], task_id: str | None = None, task_version: int | None = None, ) -> tuple[_Operation, bool]: digest = _request_hash( kind, prompt, thinking_oz, needs_vision, target_node, required_capabilities, requested_permissions, options, ) async with self._lock: existing = self._idempotency.get(idempotency_key) if existing is not None: existing_hash, operation_id = existing if existing_hash != digest: raise IdempotencyConflict( "Idempotency key was already used with a different request" ) return self._operations[operation_id], True operation = _Operation( operation_id=new_id("op"), kind=kind, session_id=session_id, turn_id=turn_id, trace_id=trace_id, prompt=prompt, thinking_oz=thinking_oz, needs_vision=needs_vision, target_node=target_node, required_capabilities=set(required_capabilities), requested_permissions=set(requested_permissions), idempotency_key=idempotency_key, request_hash=digest, input_hash=input_hash, options=dict(options), task_id=task_id, task_version=task_version, ) self._operations[operation.operation_id] = operation self._idempotency[idempotency_key] = (digest, operation.operation_id) try: created = await self.store.append( f"{operation.prefix}.created", session_id=session_id, turn_id=turn_id, operation_id=operation.operation_id, trace_id=trace_id, causation_id=causation_id, source="conductor", payload={ "kind": kind, "prompt": prompt, "thinking_oz": thinking_oz, "needs_vision": needs_vision, "target_node": target_node, "required_capabilities": sorted(required_capabilities), "requested_permissions": sorted(requested_permissions), "idempotency_key": idempotency_key, "request_hash": digest, "input_hash": input_hash, "options": options, "task_id": task_id, "task_version": task_version, }, ) await self.store.append( f"{operation.prefix}.queued", session_id=session_id, turn_id=turn_id, operation_id=operation.operation_id, trace_id=trace_id, causation_id=created.event_id, source="conductor", payload={"kind": kind}, ) operation.status = "queued" operation.updated_at = utc_now() except Exception: self._operations.pop(operation.operation_id, None) self._idempotency.pop(idempotency_key, None) raise return operation, False async def _run_operation(self, operation: _Operation) -> None: try: if not await self._operation_is_current(operation): return if operation.kind == "hermes": await self._run_hermes(operation) else: await self._run_media(operation) except asyncio.CancelledError: await self._terminal(operation, "cancelled", {"code": "cancelled"}) raise except Exception as exc: await self._terminal( operation, "failed", {"code": "downstream_error", "error_type": type(exc).__name__}, ) async def _run_hermes(self, operation: _Operation) -> None: await self._transition(operation, "started", {"kind": "hermes"}) screen_commentary_requested = bool( _LIVE_SCREEN_COMMENTARY_RE.search(operation.prompt) ) run = await self.hermes.create_run( HermesRunRequest( prompt=operation.prompt, session_id=operation.session_id, turn_id=operation.turn_id, trace_id=operation.trace_id, needs_vision=operation.needs_vision, context={ "node_visible_activity": True, "artifact_transport": "references_only", "thinking_oz": operation.thinking_oz, "conversation_continues_during_run": True, "narrate_progress": True, "screen_commentary": { "requested": screen_commentary_requested, "observe_permission": "desktop.observe" in operation.requested_permissions, "mode": "fresh_tool_frames_only", "instruction": ( "When explicitly requested, capture fresh screen frames at useful " "intervals, describe only what the latest frame supports, emit concise " "progress events, and stop observing when the run is cancelled." ), }, }, target_node=operation.target_node, required_capabilities=operation.required_capabilities, requested_permissions=operation.requested_permissions, ), idempotency_key=operation.idempotency_key, ) operation.downstream_id = run.run_id await self._emit_progress( operation, { "run_id": run.run_id, "provider_id": run.provider_id, "stage": "accepted", }, ) await self._emit( operation, "node.activity", { "node_id": run.provider_id or "local", "state": "active", "visible": True, "permissions": sorted(operation.requested_permissions), }, source="hermes", ) async for sse in self.hermes.events(run.run_id): if not await self._operation_is_current(operation): return payload = sse.json() event_type = str(payload.get("type") or payload.get("event") or sse.event) if event_type == "run.completed": if "output" in payload and "result" not in payload: payload = {**payload, "result": payload["output"]} await self._terminal(operation, "completed", payload) return if event_type == "run.failed": await self._terminal(operation, "failed", payload) return if event_type == "run.cancelled": await self._terminal(operation, "cancelled", payload) return if event_type in { "approval.request", "approval.required", "tool.approval.required", }: await self._transition(operation, "approval_required", payload) async with self._lock: pending = self._pending_approvals.setdefault( operation.session_id, [] ) if operation.operation_id not in pending: pending.append(operation.operation_id) elif event_type.startswith( ("desktop.", "node.", "files.", "webcam.", "permission.") ): if event_type in {"desktop.screenshot", "webcam.frame"}: if any(key in payload for key in ("base64", "data", "bytes")): raise ValueError("Remote visual frames must be artifact references") payload = { **payload, "artifact": MediaArtifact.model_validate( payload.get("artifact") ).model_dump(mode="json"), } await self._emit(operation, event_type, payload, source="hermes") else: await self._emit_progress(operation, payload) await self._terminal( operation, "failed", {"code": "stream_ended_without_terminal_event"}, ) async def _run_media(self, operation: _Operation) -> None: await self._transition(operation, "started", {"kind": operation.kind}) lease = await self.media.acquire(operation.kind, idempotency_key=operation.idempotency_key) operation.lease_id = lease.lease_id operation.endpoint = lease.endpoint try: job = await self.media.create_job( MediaJobRequest( kind=operation.kind, prompt=operation.prompt, session_id=operation.session_id, turn_id=operation.turn_id, trace_id=operation.trace_id, options=operation.options, ), lease=lease, idempotency_key=operation.idempotency_key, ) operation.downstream_id = job.job_id await self._emit_progress(operation, {"job_id": job.job_id, "stage": "accepted"}) async for sse in self.media.events(job.job_id, endpoint=lease.endpoint): if not await self._operation_is_current(operation): return payload = sse.json() event_type = str(payload.get("type") or sse.event) if event_type in {"job.completed", "media.job.completed", "completed"}: artifacts = [ MediaArtifact.model_validate(item).model_dump(mode="json") for item in payload.get("artifacts", []) ] payload = {**payload, "artifacts": artifacts} await self._terminal(operation, "completed", payload) return if event_type in {"job.failed", "media.job.failed", "failed", "error"}: await self._terminal(operation, "failed", payload) return if event_type in {"job.cancelled", "media.job.cancelled", "cancelled"}: await self._terminal(operation, "cancelled", payload) return await self._emit_progress(operation, payload) await self._terminal( operation, "failed", {"code": "stream_ended_without_terminal_event"}, ) finally: with suppress(Exception): await self.media.release(lease.lease_id) async def _transition( self, operation: _Operation, status: str, payload: dict[str, Any] ) -> None: async with self._lock: if operation.status in TERMINAL_STATUSES: return event = await self.store.append( f"{operation.prefix}.{status.replace('_', '.')}", session_id=operation.session_id, turn_id=operation.turn_id, operation_id=operation.operation_id, trace_id=operation.trace_id, source="conductor", payload={**payload, "downstream_id": operation.downstream_id}, ) operation.status = status operation.updated_at = event.timestamp async def _operation_is_current(self, operation: _Operation) -> bool: if operation.task_id is None or operation.task_version is None: return True try: await self.agent_system.tasks.current( operation.task_id, operation.task_version ) return True except TaskVersionConflict: if operation.downstream_id: with suppress(Exception): if operation.kind == "hermes": await self.hermes.stop(operation.downstream_id) else: await self.media.cancel( operation.downstream_id, endpoint=operation.endpoint ) await self._terminal( operation, "cancelled", {"code": "stale_task_version"}, ) return False async def _terminal( self, operation: _Operation, status: str, payload: dict[str, Any] ) -> bool: if status not in TERMINAL_STATUSES: raise ValueError("Terminal status required") async with self._lock: if operation.terminal_event_id is not None or operation.status in TERMINAL_STATUSES: return False event = await self.store.append( f"{operation.prefix}.{status}", session_id=operation.session_id, turn_id=operation.turn_id, operation_id=operation.operation_id, trace_id=operation.trace_id, source="conductor", payload={**payload, "downstream_id": operation.downstream_id}, ) operation.status = status operation.updated_at = event.timestamp operation.terminal_event_id = event.event_id pending = self._pending_approvals.get(operation.session_id, []) self._pending_approvals[operation.session_id] = [ item for item in pending if item != operation.operation_id ] return True async def _emit_progress(self, operation: _Operation, payload: dict[str, Any]) -> None: await self._emit(operation, f"{operation.prefix}.progress", payload, source="conductor") async def _emit( self, operation: _Operation, event_type: str, payload: dict[str, Any], *, source: str ) -> None: async with self._lock: if operation.status in TERMINAL_STATUSES: return await self.store.append( event_type, session_id=operation.session_id, turn_id=operation.turn_id, operation_id=operation.operation_id, trace_id=operation.trace_id, source=source, payload=payload, ) async def get_operation(self, operation_id: str) -> OperationView: await self.initialize() operation = self._operations.get(operation_id) if operation is None: raise OperationNotFound(operation_id) return operation.view() async def cancel_operation(self, operation_id: str, *, reason: str = "requested") -> OperationView: await self.initialize() operation = self._operations.get(operation_id) if operation is None: raise OperationNotFound(operation_id) if operation.status in TERMINAL_STATUSES: return operation.view() await self._emit(operation, "cancel.requested", {"reason": reason}, source="conductor") await self._transition(operation, "cancelling", {"reason": reason}) cancel_error: str | None = None try: if operation.downstream_id: if operation.kind == "hermes": await self.hermes.stop(operation.downstream_id) else: await self.media.cancel(operation.downstream_id, endpoint=operation.endpoint) except Exception as exc: cancel_error = type(exc).__name__ finally: task = operation.task if task is not None and task is not asyncio.current_task() and not task.done(): task.cancel() with suppress(asyncio.CancelledError): await task await self._terminal(operation, "cancelled", {"reason": reason}) await self.store.append( "cancel.failed" if cancel_error else "cancel.completed", session_id=operation.session_id, turn_id=operation.turn_id, operation_id=operation.operation_id, trace_id=operation.trace_id, source="conductor", causation_id=operation.terminal_event_id, payload={ "reason": reason, "operation_status": operation.status, "error_type": cancel_error, }, ) return operation.view() async def cancel_session(self, session_id: str, *, reason: str) -> list[OperationView]: await self.initialize() candidates = [ operation.operation_id for operation in self._operations.values() if operation.session_id == session_id and operation.status not in TERMINAL_STATUSES ] if not candidates: return [] return list( await asyncio.gather( *(self.cancel_operation(operation_id, reason=reason) for operation_id in candidates) ) ) async def wait(self, operation_id: str, *, timeout_s: float = 5.0) -> OperationView: operation = self._operations.get(operation_id) if operation is None: raise OperationNotFound(operation_id) task = operation.task if task is not None: with suppress(asyncio.CancelledError): await asyncio.wait_for(asyncio.shield(task), timeout=timeout_s) return operation.view() async def close(self) -> None: for operation in list(self._operations.values()): if operation.task is not None and not operation.task.done(): operation.task.cancel() for operation in list(self._operations.values()): if operation.task is not None: with suppress(asyncio.CancelledError): await operation.task await self.router.close() await self.hermes.close() await self.media.close()