from __future__ import annotations import asyncio import hmac import json from contextlib import asynccontextmanager from typing import Any import httpx from fastapi import FastAPI, Header, HTTPException, Query, Request, WebSocket, WebSocketDisconnect from fastapi.responses import StreamingResponse from fastapi.responses import JSONResponse from pydantic import ValidationError from .asr import StreamingASRBridge from .agent_system import UnifiedAgentSystem from .downstream import DownstreamProtocolError from .event_store import EventStore, event_to_sse from .hermes import HermesAdapter from .media import MediaJobAdapter from .models import ( AgentFocusRequest, AgentFocusResponse, ASRControl, ASRStart, OperationView, NodeView, NodeKillSwitchRequest, NodeKillSwitchResponse, ModelToolDescriptor, ModelToolReceipt, ModelToolRequest, PairNodeRequest, RouteDecision, RouteRequest, SessionEndRequest, SessionEndResponse, TextInput, TranscriptInput, TranscriptReceipt, ) from .router import Router from .remote_nodes import ( HermesProviderPool, MdnsDiscovery, PairingClient, RemoteNodeRegistry, TrustStore, ) from .service import ConductorService, IdempotencyConflict, OperationNotFound from .settings import OrchestratorSettings def create_app( settings: OrchestratorSettings | None = None, *, store: EventStore | None = None, router: Router | None = None, hermes: HermesAdapter | None = None, media: MediaJobAdapter | None = None, asr: StreamingASRBridge | None = None, nodes: RemoteNodeRegistry | None = None, agent_system: UnifiedAgentSystem | None = None, ) -> FastAPI: settings = (settings or OrchestratorSettings.from_env()).validated() store = store or EventStore( settings.event_store_path, max_replay_events=settings.max_replay_events ) router = router or Router( endpoint=settings.router_url, model=settings.router_model, enabled=settings.router_enabled, connect_timeout_s=settings.connect_timeout_s, request_timeout_s=min(settings.request_timeout_s, 2.0), ) nodes = nodes or RemoteNodeRegistry( trust_store=TrustStore(settings.node_trust_store_path), pairing=PairingClient( conductor_id=settings.conductor_id, connect_timeout_s=settings.connect_timeout_s, request_timeout_s=min(settings.request_timeout_s, 10.0), ), discovery=MdnsDiscovery(settings.mdns_discovery_timeout_s), enabled=settings.remote_nodes_enabled, discovery_enabled=settings.mdns_discovery_enabled, ) provider_pool: HermesProviderPool | None = None if hermes is None: local_hermes = HermesAdapter( settings.hermes_url, api_key=settings.hermes_api_key, connect_timeout_s=settings.connect_timeout_s, request_timeout_s=settings.request_timeout_s, stream_timeout_s=settings.stream_timeout_s, ) provider_pool = HermesProviderPool( local=local_hermes, registry=nodes, health_ttl_s=settings.provider_health_ttl_s, connect_timeout_s=settings.connect_timeout_s, request_timeout_s=settings.request_timeout_s, stream_timeout_s=settings.stream_timeout_s, ) hermes = provider_pool media = media or MediaJobAdapter( settings.manager_url, settings.media_url, connect_timeout_s=settings.connect_timeout_s, request_timeout_s=settings.request_timeout_s, stream_timeout_s=settings.stream_timeout_s, ) asr = asr or StreamingASRBridge( settings.asr_url, open_timeout_s=settings.connect_timeout_s ) agent_system = agent_system or UnifiedAgentSystem() service = ConductorService( store=store, router=router, hermes=hermes, media=media, agent_system=agent_system, ) @asynccontextmanager async def lifespan(_: FastAPI): await nodes.initialize() await service.initialize() yield await service.close() app = FastAPI( title="OMNI Conductor", version="0.1.0", docs_url="/docs", redoc_url=None, lifespan=lifespan, ) app.state.settings = settings app.state.store = store app.state.service = service app.state.nodes = nodes app.state.provider_pool = provider_pool app.state.agent_system = agent_system def authorized(headers: Any) -> bool: if not settings.api_token: return True authorization = headers.get("authorization", "") scheme, separator, credential = authorization.partition(" ") return bool( separator and scheme.casefold() == "bearer" and hmac.compare_digest(credential, settings.api_token) ) @app.middleware("http") async def conductor_auth(request: Request, call_next): if not authorized(request.headers): return JSONResponse( status_code=401, content={"detail": "Invalid conductor credentials"}, headers={"WWW-Authenticate": "Bearer"}, ) return await call_next(request) @app.exception_handler(IdempotencyConflict) async def idempotency_conflict(_: Request, exc: IdempotencyConflict): from fastapi.responses import JSONResponse return JSONResponse( status_code=409, content={"detail": str(exc), "code": "idempotency_conflict"}, ) @app.get("/health") @app.get("/healthz") async def health(deep: bool = False) -> dict[str, Any]: response: dict[str, Any] = { "status": "ready", "protocol_version": "1.0", "service": "conductor", "port": settings.port, "model_downloads": "disabled", } if deep: hermes_health, media_health = await asyncio.gather( hermes.health(), media.health() ) response["dependencies"] = { "hermes": hermes_health, **media_health, } return response @app.get("/v1/topology") async def topology() -> dict[str, Any]: return { "protocol_version": "1.0", "services": settings.topology(), "policy": { "side_effects_from_final_transcripts_only": True, "media_payloads": "artifact-references-only", "auto_download": False, "model_tool_mesh": "authenticated-local-ingress", "warm_models": ["minicpm", "qwen"], "cold_models": ["hunyuan", "minimax-h3"], }, } @app.get("/v1/nodes/discovered") async def discovered_nodes(): return { "service_type": "_omni-node._tcp.local.", "nodes": [node.model_dump(mode="json") for node in await nodes.discovered()], "paired_automatically": False, } @app.get("/v1/nodes", response_model=list[NodeView]) async def trusted_nodes() -> list[NodeView]: metrics = await provider_pool.node_metrics() if provider_pool is not None else None return await nodes.views(metrics) @app.post("/v1/nodes/pair", response_model=NodeView, status_code=201) async def pair_node(request: PairNodeRequest) -> NodeView: try: node = await nodes.pair(request) except (ValueError, httpx.HTTPError, DownstreamProtocolError) as exc: raise HTTPException(status_code=400, detail=f"Pairing failed: {type(exc).__name__}") from None return NodeView( node_id=node.node_id, name=node.name, endpoint=node.endpoint, capabilities=node.capabilities, granted_permissions=node.granted_permissions, enabled=node.enabled, ) @app.post("/v1/nodes/kill-switch", response_model=NodeKillSwitchResponse) async def remote_kill_switch( request: NodeKillSwitchRequest, ) -> NodeKillSwitchResponse: if provider_pool is not None: return await provider_pool.set_kill_switch(request.engaged) await nodes.set_kill_switch(request.engaged) return NodeKillSwitchResponse(engaged=request.engaged) @app.delete("/v1/nodes/{node_id}", status_code=204) async def unpair_node(node_id: str) -> None: removed = ( await provider_pool.unpair(node_id) if provider_pool is not None else await nodes.unpair(node_id) ) if not removed: raise HTTPException(status_code=404, detail="Trusted node not found") @app.post("/v1/route", response_model=RouteDecision) async def route(request: RouteRequest) -> RouteDecision: return await router.decide(request) @app.get("/v1/agent/models") async def agent_models() -> dict[str, Any]: return {"roles": agent_system.models.describe()} @app.post("/v1/transcripts", response_model=TranscriptReceipt, status_code=202) async def ingest_transcript( transcript: TranscriptInput, idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> TranscriptReceipt: return await service.ingest_transcript( transcript, idempotency_key=idempotency_key ) @app.post("/v1/input/text", response_model=TranscriptReceipt, status_code=202) async def ingest_text( text_input: TextInput, idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> TranscriptReceipt: return await service.ingest_transcript( TranscriptInput( session_id=text_input.session_id, turn_id=text_input.turn_id, trace_id=text_input.trace_id, text=text_input.text, final=True, language=text_input.language, thinking_oz=text_input.thinking_oz, needs_vision=text_input.needs_vision, target_node=text_input.target_node, required_capabilities=text_input.required_capabilities, requested_permissions=text_input.requested_permissions, ), idempotency_key=idempotency_key, ) @app.get("/v1/model-tools", response_model=list[ModelToolDescriptor]) async def model_tools() -> list[ModelToolDescriptor]: return [ ModelToolDescriptor(tool="reasoning.delegate", target_model="qwen", lifecycle="always"), ModelToolDescriptor(tool="internet.search", target_model="qwen", lifecycle="always"), ModelToolDescriptor(tool="image.generate", target_model="hunyuan", lifecycle="cold"), ModelToolDescriptor(tool="image.edit", target_model="hunyuan", lifecycle="cold"), ModelToolDescriptor(tool="video.generate", target_model="minimax-h3", lifecycle="cold"), ModelToolDescriptor(tool="audio.generate", target_model="minimax-h3", lifecycle="cold"), ModelToolDescriptor(tool="voice.speak", target_model="minicpm", lifecycle="always"), ] @app.post("/v1/model-tools", response_model=ModelToolReceipt, status_code=202) async def invoke_model_tool( model_tool: ModelToolRequest, idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> ModelToolReceipt: key = idempotency_key or ( f"model-tool:{model_tool.source_model}:{model_tool.session_id}:" f"{model_tool.turn_id}:{model_tool.tool}" ) return await service.invoke_model_tool(model_tool, idempotency_key=key) @app.post( "/v1/sessions/{session_id}/agent-focus", response_model=AgentFocusResponse ) async def agent_focus( session_id: str, request: AgentFocusRequest ) -> AgentFocusResponse: return await service.agent_focus( session_id, turn_id=request.turn_id, trace_id=request.trace_id, needs_vision=request.needs_vision, cancel_if_active=request.cancel_if_active, armed=request.armed, target_node=request.target_node, required_capabilities=request.required_capabilities, requested_permissions=request.requested_permissions, ) @app.post("/v1/sessions/{session_id}/operations/active/cancel") async def cancel_active_operations(session_id: str) -> dict[str, Any]: cancelled = await service.cancel_session(session_id, reason="ui_request") return { "cancelled_operation_ids": [item.operation_id for item in cancelled], "count": len(cancelled), } @app.post("/v1/sessions/{session_id}/end", response_model=SessionEndResponse) async def end_session( session_id: str, request: SessionEndRequest ) -> SessionEndResponse: return await service.end_session( session_id, turn_id=request.turn_id, trace_id=request.trace_id, ) @app.get("/v1/operations/{operation_id}", response_model=OperationView) async def get_operation(operation_id: str) -> OperationView: try: return await service.get_operation(operation_id) except OperationNotFound: raise HTTPException(status_code=404, detail="Operation not found") from None @app.post("/v1/operations/{operation_id}/cancel", response_model=OperationView) async def cancel_operation(operation_id: str) -> OperationView: try: return await service.cancel_operation(operation_id) except OperationNotFound: raise HTTPException(status_code=404, detail="Operation not found") from None @app.get("/v1/events") async def replay_events( after_seq: int = Query(default=0, ge=0), session_id: str | None = None, operation_id: str | None = None, limit: int = Query(default=1000, ge=0, le=settings.max_replay_events), ) -> dict[str, Any]: events = await store.replay( after_seq=after_seq, session_id=session_id, operation_id=operation_id, limit=limit, ) return { "protocol_version": "1.0", "events": [event.model_dump(mode="json") for event in events], "last_seq": events[-1].seq if events else after_seq, } @app.get("/v1/events/stream") async def stream_events( after_seq: int = Query(default=0, ge=0), session_id: str | None = None, operation_id: str | None = None, ) -> StreamingResponse: async def generate(): async with store.subscribe( after_seq=after_seq, session_id=session_id, operation_id=operation_id, ) as events: async for event in events: yield event_to_sse(event) return StreamingResponse( generate(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @app.websocket("/v1/events/ws") async def websocket_events(websocket: WebSocket) -> None: if not authorized(websocket.headers): await websocket.close(code=4401, reason="Invalid conductor credentials") return await websocket.accept() try: after_seq = int(websocket.query_params.get("after_seq", "0")) except ValueError: await websocket.close(code=1008, reason="after_seq must be an integer") return try: async with store.subscribe( after_seq=max(after_seq, 0), session_id=websocket.query_params.get("session_id"), operation_id=websocket.query_params.get("operation_id"), ) as events: async for event in events: await websocket.send_json(event.model_dump(mode="json")) except WebSocketDisconnect: return @app.websocket("/v1/asr/ws") async def browser_asr_bridge(websocket: WebSocket) -> None: """Browser-facing bridge: JSON start/control frames and binary PCM audio.""" if not authorized(websocket.headers): await websocket.close(code=4401, reason="Invalid conductor credentials") return await websocket.accept() try: first = await websocket.receive_text() start = ASRStart.model_validate_json(first) except (WebSocketDisconnect, ValidationError, ValueError): await websocket.close(code=1008, reason="A valid input.start frame is required") return async with asr.session(start) as asr_session: async def receive_audio() -> None: while True: message = await websocket.receive() if message["type"] == "websocket.disconnect": await asr_session.cancel() return audio = message.get("bytes") if audio is not None: await asr_session.send_audio(audio) continue text = message.get("text") if text is None: continue try: control = ASRControl.model_validate(json.loads(text)) except (ValidationError, ValueError): await websocket.send_json( {"type": "error", "detail": "Invalid ASR control frame"} ) continue if control.type == "input.commit": await asr_session.commit() else: await asr_session.cancel() return async def return_transcripts() -> None: async for transcript in asr_session.events(): await websocket.send_json(transcript.model_dump(mode="json", exclude_none=True)) if transcript.type in {"input.transcript.partial", "input.transcript.final"}: await service.ingest_transcript( TranscriptInput( session_id=start.session_id, turn_id=start.turn_id, trace_id=start.trace_id, text=transcript.text, final=transcript.type == "input.transcript.final", language=transcript.language or start.language, confidence=transcript.confidence, needs_vision=start.needs_vision, target_node=start.target_node, required_capabilities=start.required_capabilities, requested_permissions=start.requested_permissions, minicpm_session_id=start.minicpm_session_id, ) ) receive_task = asyncio.create_task(receive_audio()) transcript_task = asyncio.create_task(return_transcripts()) done, pending = await asyncio.wait( {receive_task, transcript_task}, return_when=asyncio.FIRST_COMPLETED ) for task in pending: task.cancel() for task in done | pending: try: await task except (asyncio.CancelledError, WebSocketDisconnect): pass return app