"""HTTP/SSE delivery adapter for the single-stage CAD protocol.""" from __future__ import annotations import asyncio from collections.abc import AsyncIterator from hashlib import sha256 import re import secrets from typing import Any from app.cad_agent.application.workflow import ModelIdentity from app.cad_agent.composition import CadServices, compose_cad_services from app.cad_agent.domain.errors import ErrorCode from app.cad_agent.domain.state import TaskPhase, transition from app.models.contracts import ChatMessage from app.services.library import CdslLibrary from app.services.sse import event from app.services.storage import WorkspaceStore, now_iso from app.settings import Settings def text_from_message(message: ChatMessage) -> str: return "\n".join(part.text or "" for part in message.parts if part.type == "text").strip() _EVENT_LABELS = { "requirements_ready": "需求分析", "authoring_cdsl_ready": "完整 CDSL", "cdsl_compiled": "CDSL 编译", "build_result": "CAD 构建", "repair_started": "CDSL 修复", "task_terminal": "生成任务", } def _progress(name: str, payload: dict[str, Any]) -> dict[str, Any]: lifecycle = str(payload.get("lifecycle") or "") status = "error" if lifecycle == "failed" or payload.get("status") in {"failed", "repair_required"} else "waiting" if lifecycle == "waiting_for_user" else "success" if lifecycle == "completed" else "running" return {**payload, "step": name, "label": _EVENT_LABELS.get(name, name), "status": status} class AgentService: def __init__(self, settings: Settings, store: WorkspaceStore, library: CdslLibrary) -> None: self.settings, self.store, self.library = settings, store, library self.cad: CadServices = compose_cad_services(settings) self._autonomous_runs: dict[str, asyncio.Task[None]] = {} async def resume_running_tasks(self) -> None: if not self.settings.resume_running_tasks_on_startup: return try: provider, model = self.settings.resolve_model(None, None) except ValueError: return author = ModelIdentity(provider.id, model.id) for task_id in self.cad.repository.running_task_ids(): if task_id not in self._autonomous_runs: self._autonomous_runs[task_id] = asyncio.create_task(self._consume_discarding(task_id, author), name=f"resume-cad-single-stage-{task_id}") async def _consume_discarding(self, task_id: str, author: ModelIdentity) -> None: try: async for _name, _payload in self.cad.workflow.run(task_id=task_id, author=author): await self.cad.outbox.dispatch_pending(task_id=task_id) finally: await self.cad.outbox.dispatch_pending(task_id=task_id) self._autonomous_runs.pop(task_id, None) async def cancel(self, task_id: str) -> dict[str, Any] | None: state = self.cad.repository.get_state(task_id) if state is None: return None if state.phase not in {TaskPhase.COMPLETED, TaskPhase.FAILED, TaskPhase.CANCELLED}: cancelled = transition(state, "cancelled", error=ErrorCode.CANCELLED) self.cad.repository.compare_and_swap(cancelled, events=[{"event": "task_cancelled", "active_revision": state.active_revision}]) await self.cad.outbox.dispatch_pending(task_id=task_id) running = self._autonomous_runs.pop(task_id, None) if running and not running.done(): running.cancel() return self.cad.repository.get_task_projection(task_id) async def resume_retry(self, task_id: str) -> dict[str, Any] | None: state = self.cad.repository.get_state(task_id) if state is None: return None if state.phase != TaskPhase.FAILED or state.retry_from_phase is None: raise ValueError("Only a retryable failed CAD task can be resumed") if task_id in self._autonomous_runs and not self._autonomous_runs[task_id].done(): raise ValueError("The CAD task is already running") provider, model = self.settings.resolve_model(None, None) if not self.cad.workflow.resume(task_id): raise ValueError("The CAD task no longer has a retry checkpoint") self._autonomous_runs[task_id] = asyncio.create_task(self._consume_discarding(task_id, ModelIdentity(provider.id, model.id)), name=f"resume-cad-single-stage-{task_id}") return self.cad.repository.get_task_projection(task_id) async def stream(self, messages: list[ChatMessage], conversation_id: str | None, selected_task_id: str | None, provider_id: str | None = None, model_id: str | None = None, viewer_context: list[dict[str, Any]] | None = None) -> AsyncIterator[bytes]: del viewer_context latest = next((item for item in reversed(messages) if item.role == "user"), None) if latest is None or not text_from_message(latest): yield event("cad_error", {"stage": "request", "message": "A non-empty user request is required."}) yield event("done", {}) return request = text_from_message(latest) conversation = self.store.ensure_conversation(conversation_id) selected = str(selected_task_id or conversation.get("current_task_id") or "") current = self.cad.repository.get_task_projection(selected) if selected else None resumed = False if current and current.get("lifecycle") == "waiting_for_user": resumed = self.cad.workflow.resume_with_user_clarification(selected, request, message_id=latest.id) if not resumed: yield event("cad_error", {"stage": "request", "message": "The CAD task cannot apply this clarification."}) yield event("done", {}) return if current and current.get("lifecycle") == "running": yield event("cad_error", {"stage": "request", "message": "该 CAD 任务正在生成，完成后才能继续。"}) yield event("done", {}) return task_id = selected if resumed else f"cad_{secrets.token_hex(6)}" if not resumed: try: source_blocks, image_inputs = self._task_inputs(conversation, request) self.cad.workflow.create_task(task_id, request, source_blocks=source_blocks, image_inputs=image_inputs) except ValueError as error: yield event("cad_error", {"stage": "request", "message": str(error)}) yield event("done", {}) return self.store.append_conversation_message(conversation["conversation_id"], latest.model_dump(), task_id) yield event("progress", _progress("task_started", {"taskId": task_id, "message": "CAD task started."})) try: provider, model = self.settings.resolve_model(provider_id, model_id) except ValueError as error: state = self.cad.repository.get_state(task_id) if state: failed = transition(state, "failed", error=ErrorCode.MODEL_STRUCTURED_OUTPUT_UNSUPPORTED) self.cad.repository.compare_and_swap(failed, events=[{"event": "model_configuration_invalid", "message": str(error)}]) yield event("cad_error", {"stage": "configuration", "message": str(error)}) yield event("done", {}) return author = ModelIdentity(provider.id, model.id) queue: asyncio.Queue[tuple[str, dict[str, Any]] | None] = asyncio.Queue() parts: list[dict[str, Any]] = [] async def consume() -> None: try: async for name, payload in self.cad.workflow.run(task_id=task_id, author=author): decorated = {**payload, "taskId": task_id, "eventId": f"{task_id}_{secrets.token_hex(4)}", "timestamp": now_iso()} parts.append({"type": "data-cad-progress", "id": decorated["eventId"], "data": _progress(name, decorated)}) await queue.put((name, decorated)) await self.cad.outbox.dispatch_pending(task_id=task_id) if name == "task_terminal" and decorated.get("lifecycle") == "completed": projection = self.cad.repository.get_task_projection(task_id) or {} result = self._result_payload(task_id, projection) if result is not None: await queue.put(("cad_result", result)) finally: self.store.append_conversation_message(conversation["conversation_id"], {"id": f"assistant_{secrets.token_hex(8)}", "role": "assistant", "parts": parts}, task_id) self._autonomous_runs.pop(task_id, None) await queue.put(None) self._autonomous_runs[task_id] = asyncio.create_task(consume(), name=f"cad-single-stage-{task_id}") while True: item = await queue.get() if item is None: break name, payload = item if name == "task_terminal" and payload.get("lifecycle") == "failed": yield event("cad_error", {"stage": "generation", "message": str(payload.get("message") or "CAD generation failed."), "code": payload.get("code")}) yield event(name, payload) yield event("done", {}) @staticmethod def _result_payload(task_id: str, projection: dict[str, Any]) -> dict[str, Any] | None: revision_id = str(projection.get("published_revision") or projection.get("active_revision") or "") revisions = projection.get("revisions") if isinstance(projection.get("revisions"), list) else [] revision = next((item for item in revisions if isinstance(item, dict) and item.get("revision_id") == revision_id), None) if not isinstance(revision, dict): return None required = ("cdsl_path", "step_path", "glb_path", "report_path") if not all(isinstance(revision.get(path), str) and revision[path] for path in required): return None return { "taskId": task_id, "revisionId": revision_id, "cdslPath": revision["cdsl_path"], "stepPath": revision["step_path"], "glbPath": revision["glb_path"], "reportPath": revision["report_path"], "summary": str(revision.get("summary") or "CDSL CAD model"), "referenceIds": list(revision.get("reference_ids") or []), "engine": str(revision.get("engine") or "cdsl_only"), "lifecycle": str(projection.get("lifecycle") or "completed"), } def _task_inputs(self, conversation: dict[str, Any], request: str) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: blocks = [{"text": paragraph} for paragraph in re.split(r"\n\s*\n", request) if paragraph.strip()] image_inputs: list[dict[str, str]] = [] conversation_id = str(conversation.get("conversation_id") or "") if not conversation_id: raise ValueError("Conversation has no identifier") for attachment in conversation.get("attachments") or (): if not isinstance(attachment, dict): continue if str(attachment.get("conversation_id") or "") != conversation_id: raise ValueError("Attachment does not belong to this conversation") attachment_id, relative_path, digest = str(attachment.get("id") or ""), str(attachment.get("path") or ""), str(attachment.get("sha256") or "") if not attachment_id or not relative_path or not re.fullmatch(r"[a-f0-9]{64}", digest): raise ValueError("Attachment metadata is incomplete") binary = self.store.conversation_attachment_path(conversation_id, relative_path) if not binary.is_file() or sha256(binary.read_bytes()).hexdigest() != digest: raise ValueError(f"Attachment is unavailable: {attachment.get('name') or attachment_id}") kind = str(attachment.get("kind") or "") if kind == "document": text_path = self.store.conversation_attachment_path(conversation_id, str(attachment.get("extracted_path") or "")) if not text_path.is_file(): raise ValueError(f"Attachment text is unavailable: {attachment.get('name') or attachment_id}") text = text_path.read_text(encoding="utf-8", errors="replace").strip() elif kind == "image": text = f"Visual attachment {attachment.get('name') or attachment_id} (SHA-256 {digest})." image_inputs.append({"path": str(binary), "mime": str(attachment.get("mime") or "image/*"), "sha256": digest}) else: raise ValueError(f"Unsupported attachment kind: {kind or 'unknown'}") if not text: raise ValueError(f"Attachment source is empty: {attachment.get('name') or attachment_id}") blocks.append({"text": text, "attachment": {"attachment_id": attachment_id, "name": str(attachment.get("name") or attachment_id), "kind": kind, "mime": str(attachment.get("mime") or "application/octet-stream"), "sha256": digest}}) return blocks, image_inputs