"""Single-stage coordinator for Authoring CDSL generation. Each task makes one request-analysis call and one full Authoring CDSL call. Only structured authoring/compile/runtime failures can request a replacement, and the replacement budget is fixed at two. """ from __future__ import annotations import json import secrets from dataclasses import dataclass from hashlib import sha256 from typing import Any, AsyncIterator from pydantic import BaseModel, ConfigDict, Field from app.cad_agent.application.authoring_compiler import AuthoringCompileError from app.cad_agent.application.authoring_contract import AuthoringDocument, validation_error_code from app.cad_agent.application.authoring_guidance import load_authoring_guidance from app.cad_agent.application.single_stage import SingleStageExecutor from app.cad_agent.domain.errors import ErrorCode from app.cad_agent.domain.state import TaskPhase, TaskState, transition from app.cad_agent.ports import AdapterUnavailable, ArtifactStore, CadRuntime, ModelGateway, TaskRepository @dataclass(frozen=True, slots=True) class ModelIdentity: provider_id: str model_id: str @dataclass(frozen=True, slots=True) class WorkflowConfig: max_repairs: int = 2 class _RequirementsTarget(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) kind: str = Field(pattern=r"^[a-z][a-z0-9_]{0,80}$") expected: dict[str, Any] = Field(default_factory=dict) verification: str = Field(default="manual", pattern=r"^(deterministic|manual)$") class RequirementsAnalysis(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) schema_version: str = Field(default="cad.requirements.v1", pattern=r"^cad\.requirements\.v1$") explicit_requirements: list[str] = Field(min_length=1, max_length=64) assumptions: list[str] = Field(default_factory=list, max_length=64) acceptance_targets: list[_RequirementsTarget] = Field(default_factory=list, max_length=128) manual_targets: list[str] = Field(default_factory=list, max_length=64) clarification_question: str | None = Field(default=None, min_length=1, max_length=500) class WorkflowCoordinator: def __init__(self, config: WorkflowConfig, repository: TaskRepository, artifacts: ArtifactStore, runtime: CadRuntime, model_gateway: ModelGateway, executor: SingleStageExecutor) -> None: self.config = config self.repository = repository self.artifacts = artifacts self.runtime = runtime self.model_gateway = model_gateway self.executor = executor def create_task(self, task_id: str, request: str, *, source_blocks: list[dict[str, Any]] | None = None, image_inputs: list[dict[str, str]] | None = None) -> TaskState: self.artifacts.initialize_task(task_id, request, source_blocks=source_blocks, image_inputs=image_inputs) return self.repository.create_task(task_id, request) def resume(self, task_id: str) -> bool: state = self.repository.get_state(task_id) if state is None or state.phase != TaskPhase.FAILED or state.retry_from_phase is None: return False resumed = transition(state, "resume") return self.repository.compare_and_swap(resumed, events=[{"event": "task_resumed", "from_phase": state.retry_from_phase.value}]) def resume_with_user_clarification(self, task_id: str, clarification: str, *, message_id: str) -> bool: state = self.repository.get_state(task_id) if state is None or state.phase != TaskPhase.WAITING_FOR_USER or not clarification.strip(): return False digest = sha256(f"{message_id}:{clarification}".encode()).hexdigest()[:16] path = self.artifacts.write_json_once(task_id, f"documents/user-clarification-{digest}.json", {"schema_version": "cad.user-clarification.v1", "message_id": message_id, "text": clarification.strip()}) resumed = transition(state, "clarification_received", clarification_path=path) return self.repository.compare_and_swap(resumed, events=[{"event": "clarification_received", "clarification_path": path}]) def waiting_for_user_terminal(self, task_id: str, state: TaskState | None = None) -> dict[str, Any]: state = state or self.repository.get_state(task_id) details = self.artifacts.read_json(task_id, state.clarification_path) if state and state.clarification_path else {} question = str((details or {}).get("question") or "CAD generation needs clarification.") return {"taskId": task_id, "lifecycle": "waiting_for_user", "message": question, "questions": [question], "userActionRequired": True} async def run(self, *, task_id: str, author: ModelIdentity) -> AsyncIterator[tuple[str, dict[str, Any]]]: while True: state = self.repository.get_state(task_id) if state is None: return if state.phase in {TaskPhase.COMPLETED, TaskPhase.FAILED, TaskPhase.CANCELLED}: yield "task_terminal", self._terminal(state) return try: self.artifacts.sync_event_ledger(task_id, self.repository.ledger_events(task_id)) if state.phase == TaskPhase.WAITING_FOR_USER: yield "task_terminal", self.waiting_for_user_terminal(task_id, state) return if state.phase == TaskPhase.ANALYZING_REQUEST: result = await self._analyze(task_id, state, author) yield result continue if state.phase == TaskPhase.AUTHORING_CDSL: result = await self._author(task_id, state, author) yield result continue if state.phase == TaskPhase.COMPILING_CDSL: result = self._compile(task_id, state) yield result continue if state.phase == TaskPhase.BUILDING: result = self._build(task_id, state) yield result continue if state.phase == TaskPhase.REPAIRING: started = transition(state, "repair_started", repair_count=state.repair_count + 1) if not self.repository.compare_and_swap(started, events=[{"event": "repair_started", "repair_count": started.repair_count}]): continue yield "repair_started", {"taskId": task_id, "repairCount": started.repair_count, "repairBudget": self.config.max_repairs} continue if state.phase == TaskPhase.PUBLISHING_BEST_EFFORT: result = self._publish(task_id, state) yield result continue return except AdapterUnavailable as error: self._fail(task_id, state, ErrorCode.AUTHOR_TRANSPORT_UNAVAILABLE, str(error), retryable=True) except OSError as error: self._fail(task_id, state, ErrorCode.STORAGE_FAILURE, str(error), retryable=True) except Exception as error: # A coordinator failure must have a durable terminal record. self._fail(task_id, state, ErrorCode.FAILED_INTERNAL, str(error), retryable=False) async def _analyze(self, task_id: str, state: TaskState, author: ModelIdentity) -> tuple[str, dict[str, Any]]: if state.requirements_path: next_state = transition(state, "analysis_written") self.repository.compare_and_swap(next_state, events=[{"event": "requirements_reused", "path": state.requirements_path}]) return "requirements_ready", {"taskId": task_id, "path": state.requirements_path, "reused": True} source = self.artifacts.read_source_requirements(task_id) try: analysis = RequirementsAnalysis.model_validate(await self._tool_call( task_id, author, "analyze_requirements", RequirementsAnalysis.model_json_schema(), "Extract explicit CAD requirements, safe assumptions, and verification targets. Do not invent strict dimensions that the request did not specify.", source, )) except (AuthoringCompileError, ValueError) as error: diagnostic_path = self.artifacts.write_json_once(task_id, "documents/requirements-analysis-diagnostic.json", { "schema_version": "cad.requirements-diagnostic.v1", "diagnostics": [{"code": "AUTHOR_SCHEMA_INVALID", "message": str(error)[:1000]}], }) failed = transition(state, "failed", diagnostics_path=diagnostic_path, error=ErrorCode.AUTHOR_SCHEMA_INVALID) self.repository.compare_and_swap(failed, events=[{"event": "requirements_analysis_failed", "diagnostics_path": diagnostic_path}]) return "task_terminal", self._terminal(failed) if analysis.clarification_question: path = self.artifacts.write_json_once(task_id, "documents/clarification-request.json", {"question": analysis.clarification_question}) waiting = transition(state, "waiting_for_user", clarification_path=path, requirements_path="") self.repository.compare_and_swap(waiting, events=[{"event": "requirements_waiting_for_user", "question": analysis.clarification_question}]) return "task_terminal", self.waiting_for_user_terminal(task_id, waiting) path = self.artifacts.write_json_once(task_id, "documents/requirements-analysis.json", analysis.model_dump(mode="json")) next_state = transition(state, "analysis_written", requirements_path=path) self.repository.compare_and_swap(next_state, events=[{"event": "requirements_analyzed", "path": path, "assumptions": analysis.assumptions}]) return "requirements_ready", {"taskId": task_id, "path": path, "assumptions": analysis.assumptions} async def _author(self, task_id: str, state: TaskState, author: ModelIdentity) -> tuple[str, dict[str, Any]]: requirements = self.artifacts.read_json(task_id, state.requirements_path) or {} previous = self.artifacts.read_json(task_id, state.authoring_path) if state.authoring_path else None diagnostics = self.artifacts.read_json(task_id, state.diagnostics_path) if state.diagnostics_path else None operation_schemas = { atomic_id: self._author_operation_contract(self.runtime.operation_contract(atomic_id)) for atomic_id in self.runtime.supported_atomic_ids() } repair_instruction = "" if state.repair_count == 0 else "Return a complete replacement document. Preserve only features confirmed in executed_feature_ids unless the diagnostic identifies that feature. Features that did not execute may be corrected. Keep local names unless the diagnostic identifies a name conflict. Never add IDs, stable selectors, snapshots, or tokens." content = json.dumps({ "requirements": self._authoring_requirements_context(requirements), "supported_operations": operation_schemas, "authoring_selector_contract": { "face_cap_source_pattern": r"^[a-z][a-z0-9_]{0,63}\.(top_planar_face|bottom_planar_face|end_face|start_face)$", "face_cap_roles": ["top_planar_face", "bottom_planar_face", "end_face", "start_face"], }, "previous_authoring": previous, "diagnostics": diagnostics, }, ensure_ascii=False) try: raw = await self._tool_call( task_id, author, "write_authoring_cdsl", AuthoringDocument.model_json_schema(), "Write only cad.author.v1. Use local body and feature names. The service creates all runtime IDs. " + repair_instruction + "\n\n" + load_authoring_guidance(), content, ) document = AuthoringDocument.model_validate(raw).model_dump(mode="json") if previous is not None and state.repair_count: self._validate_repair_document(previous, document, diagnostics, self.artifacts.read_json(task_id, state.compile_audit_path) if state.compile_audit_path else None) except (AuthoringCompileError, ValueError) as error: return self._repair_or_stop( task_id, state, validation_error_code(error), str(error), self._author_validation_details(error), ) path = self.artifacts.write_json_once(task_id, f"documents/authoring-cdsl-attempt-{state.repair_count + 1:02d}.json", document) next_state = transition(state, "authoring_written", authoring_path=path) self.repository.compare_and_swap(next_state, events=[{"event": "authoring_cdsl_written", "path": path, "repair_count": state.repair_count}]) return "authoring_cdsl_ready", {"taskId": task_id, "path": path, "repairCount": state.repair_count} def _compile(self, task_id: str, state: TaskState) -> tuple[str, dict[str, Any]]: authoring = self.artifacts.read_json(task_id, state.authoring_path) if authoring is None: raise RuntimeError("Authoring CDSL artifact is unavailable") try: compiled = self.executor.compile(task_id, authoring, repair_count=state.repair_count) except AuthoringCompileError as error: return self._repair_or_stop(task_id, state, error.code, str(error), {"path": error.path}) next_state = transition(state, "compiled", runtime_cdsl_path=compiled["runtime_path"], compile_audit_path=compiled["audit_path"]) self.repository.compare_and_swap(next_state, events=[{"event": "cdsl_compiled", "runtime_path": compiled["runtime_path"], "compile_audit_path": compiled["audit_path"]}]) return "cdsl_compiled", {"taskId": task_id, "runtimePath": compiled["runtime_path"], "compileAuditPath": compiled["audit_path"]} def _build(self, task_id: str, state: TaskState) -> tuple[str, dict[str, Any]]: authoring = self.artifacts.read_json(task_id, state.authoring_path) runtime_cdsl = self.artifacts.read_json(task_id, state.runtime_cdsl_path) audit = self.artifacts.read_json(task_id, state.compile_audit_path) if authoring is None or runtime_cdsl is None or audit is None: raise RuntimeError("Compiled CDSL artifacts are unavailable") result = self.executor.build(task_id, authoring, runtime_cdsl, audit, repair_count=state.repair_count) diagnostics = result.get("diagnostics") if isinstance(result.get("diagnostics"), list) else [] diagnostic_path = self.artifacts.write_json_once(task_id, f"documents/diagnostics-attempt-{state.repair_count + 1:02d}.json", {"diagnostics": diagnostics, "executed_feature_ids": result.get("executed_feature_ids", [])}) revision = str(result.get("revision_id") or "") if diagnostics: if state.repair_count < self.config.max_repairs: repairing = transition(state, "repair_required", active_revision=revision or state.active_revision, diagnostics_path=diagnostic_path, error=ErrorCode.ENGINE_EXECUTION_FAILED) self.repository.compare_and_swap(repairing, events=[{"event": "build_failed", "paths": result.get("paths", {}), "diagnostics": diagnostics, "revision_id": revision}]) return "build_result", {"taskId": task_id, "status": "repair_required", "paths": result.get("paths", {}), "diagnostics": diagnostics, "revisionId": revision} publishing = transition(state, "build_completed", active_revision=revision or state.active_revision, diagnostics_path=diagnostic_path, error=ErrorCode.BEST_EFFORT_COMPLETED) self.repository.compare_and_swap(publishing, events=[{"event": "published_best_effort", "paths": result.get("paths", {}), "diagnostics": diagnostics, "revision_id": revision}]) return "build_result", {"taskId": task_id, "status": "published_best_effort", "paths": result.get("paths", {}), "diagnostics": diagnostics, "revisionId": revision} publishing = transition(state, "build_completed", active_revision=revision, diagnostics_path=diagnostic_path) self.repository.compare_and_swap(publishing, events=[{"event": "build_completed", "paths": result.get("paths", {}), "revision_id": revision, "executed_feature_ids": result.get("executed_feature_ids", [])}]) return "build_result", {"taskId": task_id, "status": "completed", "paths": result.get("paths", {}), "revisionId": revision} def _publish(self, task_id: str, state: TaskState) -> tuple[str, dict[str, Any]]: if not state.active_revision: failed = transition(state, "failed", error=ErrorCode.ENGINE_EXECUTION_FAILED) self.repository.compare_and_swap(failed, events=[{"event": "no_executable_model", "message": "No executable CDSL prefix could be published."}]) return "task_terminal", self._terminal(failed) requirements = self.artifacts.read_json(task_id, state.requirements_path) or {} diagnostics = self.artifacts.read_json(task_id, state.diagnostics_path) or {} authoring = self.artifacts.read_json(task_id, state.authoring_path) or {} claim_report_path = self.artifacts.write_json_once( task_id, "documents/claim-report.json", self._claim_report(requirements), ) report = self._completion_report(state, requirements, authoring, diagnostics) path = self.artifacts.write_text_once(task_id, "completion-result.md", report) completed = transition(state, "published", completion_path=path) self.repository.compare_and_swap(completed, events=[{"event": "task_published", "revision_id": state.active_revision, "completion_path": path, "claim_report_path": claim_report_path, "best_effort": bool(diagnostics.get("diagnostics"))}]) return "task_terminal", self._terminal(completed) def _repair_or_stop(self, task_id: str, state: TaskState, code: str, message: str, details: dict[str, Any]) -> tuple[str, dict[str, Any]]: diagnostic = { "code": code, "message": message, **details, } diagnostic.setdefault("repair_hint", self._repair_hint(code, str(diagnostic.get("path") or ""))) diagnostic_path = self.artifacts.write_json_once(task_id, f"documents/diagnostics-attempt-{state.repair_count + 1:02d}.json", {"diagnostics": [diagnostic]}) error = self._error_code(code) if state.repair_count < self.config.max_repairs: repairing = transition(state, "repair_required", diagnostics_path=diagnostic_path, error=error) self.repository.compare_and_swap(repairing, events=[{"event": "compile_failed", "code": code, "message": message, "diagnostics_path": diagnostic_path}]) return "cdsl_compiled", {"taskId": task_id, "status": "repair_required", "code": code, "message": message} if state.active_revision: publishing = transition( state, "publish_best_effort", diagnostics_path=diagnostic_path, error=ErrorCode.BEST_EFFORT_COMPLETED, ) self.repository.compare_and_swap(publishing, events=[{ "event": "repair_budget_exhausted", "code": code, "message": message, "diagnostics_path": diagnostic_path, "revision_id": state.active_revision, }]) return "build_result", { "taskId": task_id, "status": "published_best_effort", "code": code, "message": message, "revisionId": state.active_revision, } failed = transition(state, "failed", diagnostics_path=diagnostic_path, error=error) self.repository.compare_and_swap(failed, events=[{"event": "compile_failed", "code": code, "message": message, "diagnostics_path": diagnostic_path}]) return "task_terminal", self._terminal(failed) async def _tool_call(self, task_id: str, author: ModelIdentity, name: str, schema: dict[str, Any], system: str, user: str) -> dict[str, Any]: response = await self.model_gateway.call_tool( messages=[{"role": "system", "content": system}, {"role": "user", "content": user}], tool={"type": "function", "function": {"name": name, "description": "Return one schema-valid object.", "parameters": schema}}, provider_id=author.provider_id, model_id=author.model_id, required_tool_name=name, ) usage = response.get("usage") if isinstance(response.get("usage"), dict) else {} self.repository.record_usage(task_id, {"role": "author", "tool": name, **usage}) calls = response.get("tool_calls") if isinstance(response.get("tool_calls"), list) else [] if len(calls) != 1 or not isinstance(calls[0], dict): raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", "provider did not return exactly one tool call") function = calls[0].get("function") if isinstance(calls[0].get("function"), dict) else {} if function.get("name") != name: raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", "provider returned an unexpected tool") try: value = json.loads(str(function.get("arguments") or "")) except json.JSONDecodeError as error: raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", "provider returned invalid JSON") from error if not isinstance(value, dict): raise AuthoringCompileError("AUTHOR_SCHEMA_INVALID", "provider did not return an object") self.repository.record_tool_audit(task_id, {"tool": name, "arguments_sha256": sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()}) return value @staticmethod def _author_operation_contract(contract: dict[str, Any]) -> dict[str, Any]: """Expose only author-owned operation facts, including selector shape.""" selector = contract.get("selector_policy") or {} fragment = contract.get("fragment_shape") or {} sketch_mode = fragment.get("sketch") return { "params_schema": contract.get("author_params_schema") or {}, "sketch": sketch_mode, "authoring_sketch_template": ( { "workplane": { "origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1], }, "profile": { "type": "circle", "diameter_mm": 10, "center_mm": [0, 0], }, } if sketch_mode == "required" else None ), "selector": { "required": fragment.get("selector_tokens") == "required", "kind": selector.get("token_kind"), "min_items": selector.get("min_items"), "max_items": selector.get("max_items"), "destination": selector.get("slot"), "source_syntax": ".", }, } @staticmethod def _authoring_requirements_context(requirements: dict[str, Any]) -> dict[str, list[str]]: """Pass requirement meaning to the author without leaking analysis field names. ``acceptance_targets.expected`` is intentionally an open-ended reporting record. It may contain descriptive keys from a user request, while Authoring CDSL has a closed protocol. Passing it through verbatim invites a model to treat analysis labels as output fields. """ return { key: [item for item in requirements.get(key, []) if isinstance(item, str)] for key in ("explicit_requirements", "assumptions", "manual_targets") } @staticmethod def _validate_repair_document( previous: dict[str, Any], replacement: dict[str, Any], diagnostics: dict[str, Any] | None, compile_audit: dict[str, Any] | None, ) -> None: """Keep successful features fixed across complete-document repairs.""" targeted = { str(item.get("feature_name") or "") for item in (diagnostics or {}).get("diagnostics") or () if isinstance(item, dict) and item.get("feature_name") } for item in (diagnostics or {}).get("diagnostics") or (): if not isinstance(item, dict): continue path = str(item.get("path") or "") if path.startswith("features."): targeted.add(path.split(".", 2)[1]) feature_ids = (compile_audit or {}).get("feature_ids") if isinstance(compile_audit, dict) else {} ids_to_names = { str(feature_id): str(name) for name, feature_id in (feature_ids or {}).items() if isinstance(name, str) and isinstance(feature_id, str) } executed = { ids_to_names[feature_id] for feature_id in (diagnostics or {}).get("executed_feature_ids") or () if isinstance(feature_id, str) and feature_id in ids_to_names } old_features = { str(feature.get("name") or ""): feature for body in previous.get("bodies") or () if isinstance(body, dict) for feature in body.get("features") or () if isinstance(feature, dict) } new_features = { str(feature.get("name") or ""): feature for body in replacement.get("bodies") or () if isinstance(body, dict) for feature in body.get("features") or () if isinstance(feature, dict) } for name in executed: if name in targeted: continue old_feature = old_features.get(name) if new_features.get(name) != old_feature: raise AuthoringCompileError( "AUTHOR_SCHEMA_INVALID", f"repair changed successful feature {name}", path=f"features.{name}", ) def _fail(self, task_id: str, state: TaskState, code: ErrorCode, message: str, *, retryable: bool) -> None: current = self.repository.get_state(task_id) if current is None or current.phase in {TaskPhase.COMPLETED, TaskPhase.FAILED, TaskPhase.CANCELLED}: return failed = transition(current, "failed", error=code) self.repository.compare_and_swap(failed, events=[{"event": "service_failure", "code": code.value, "message": message[:1000], "retryable": retryable}]) @staticmethod def _error_code(value: str) -> ErrorCode: try: return ErrorCode(value) except ValueError: return ErrorCode.AUTHOR_SCHEMA_INVALID @staticmethod def _author_validation_details(error: Exception) -> dict[str, Any]: """Preserve one exact Pydantic location for a complete-document repair.""" errors = getattr(error, "errors", None) if not callable(errors): return {"path": getattr(error, "path", "")} values = errors() if not values or not isinstance(values[0], dict): return {"path": getattr(error, "path", "")} location = values[0].get("loc") path = ".".join(str(item) for item in location) if isinstance(location, tuple) else "" return {"path": path, "schema_error": str(values[0].get("msg") or "")} @staticmethod def _repair_hint(code: str, path: str) -> str: if ".sketch" in path or "sketch" in path: return ( "Use exactly sketch.workplane {origin_mm, x_dir, normal} and " "sketch.profile. A circle is {type: circle, diameter_mm, center_mm}; " "do not use profiles, plane, support, radius_mm, or center." ) if ".selectors" in path or code.startswith("SELECTOR_"): return ( "Use one declarative selector {kind, source: '.', " "match: 'unique'}. Do not add role, query, host_face, face indexes, or tokens." ) if code == "AUTHOR_FORBIDDEN_FIELD": return "Remove the forbidden runtime identity or server-injected field. Use only local names and declarative selectors." if code == "AUTHOR_REFERENCE_INVALID": return "Reference an existing local feature name and exact output role; the compiler records the selector source as a dependency." return "Return the complete document with the named diagnostic corrected. Preserve executed features unless the diagnostic targets them." @staticmethod def _claim_report(requirements: dict[str, Any]) -> dict[str, Any]: targets = requirements.get("acceptance_targets") if isinstance(requirements.get("acceptance_targets"), list) else [] claims = [ { "target": str(target.get("kind") or "target"), "status": "pending", "verification": str(target.get("verification") or "manual"), } for target in targets if isinstance(target, dict) ] manual = requirements.get("manual_targets") if isinstance(requirements.get("manual_targets"), list) else [] claims.extend({"target": item, "status": "pending", "verification": "manual"} for item in manual if isinstance(item, str)) if not claims: claims.append({"target": "no deterministic target declared", "status": "not_applicable", "verification": "manual"}) return {"schema_version": "cad.requirement-claim-report.v1", "claims": claims} @staticmethod def _completion_report( state: TaskState, requirements: dict[str, Any], authoring: dict[str, Any], diagnostics: dict[str, Any], ) -> str: lines = ["# CAD Generation Result", "", f"Published revision: `{state.active_revision}`", "", "## Generated Model", ""] bodies = authoring.get("bodies") if isinstance(authoring.get("bodies"), list) else [] for body in bodies: if not isinstance(body, dict): continue lines.append(f"- body: {body.get('name', 'body')}") features = body.get("features") if isinstance(body.get("features"), list) else [] for feature in features: if isinstance(feature, dict): lines.append(f"- feature: {feature.get('name', 'feature')} ({feature.get('operation', 'operation')})") lines.extend(["", "## Requested Requirements", ""]) lines.extend(f"- {item}" for item in requirements.get("explicit_requirements", []) if isinstance(item, str)) lines.extend(["", "## Requirement Compliance", ""]) targets = requirements.get("acceptance_targets") if isinstance(requirements.get("acceptance_targets"), list) else [] if targets: lines.extend(f"- pending: {item.get('kind', 'target')}" for item in targets if isinstance(item, dict)) else: lines.append("- not_applicable: no deterministic acceptance target was declared") lines.extend(f"- pending: {item}" for item in requirements.get("manual_targets", []) if isinstance(item, str)) lines.extend(["", "## Assumptions", ""]) lines.extend(f"- {item}" for item in requirements.get("assumptions", []) if isinstance(item, str)) lines.extend(["", "## Execution", ""]) failures = diagnostics.get("diagnostics") if isinstance(diagnostics.get("diagnostics"), list) else [] if failures: lines.append("The executable prefix was published with unresolved operations:") lines.extend(f"- {item.get('code', 'ENGINE_EXECUTION_FAILED')}: {item.get('feature_name') or item.get('feature_id') or 'document'}: {item.get('message', item)}" for item in failures if isinstance(item, dict)) else: lines.append("The complete compiled CDSL executed successfully.") lines.extend(["", "## Planning Limitations", ""]) lines.append("Requirement compliance is reported separately from executable publication. Pending targets require deterministic measurement or user review.") lines.extend(["", "## Delivery Artifacts", ""]) root = f"revisions/{state.active_revision}" lines.extend([ f"- STEP: {root}/model.step", f"- GLB: {root}/model.glb", f"- Render bundle: {root}/renders/render-manifest.json", f"- Runtime CDSL: {root}/model.cdsl.json", f"- Build diagnostics: {state.diagnostics_path or root + '/build-diagnostics.json'}", "- Requirement compliance: documents/claim-report.json", f"- Authoring CDSL: {state.authoring_path}", f"- Compile audit: {state.compile_audit_path}", ]) lines.extend(["", "## Repair Budget", ""]) reason = "all compiled features executed" if not failures else "published the best executable prefix after unresolved diagnostics" lines.append(f"- repair calls used: {state.repair_count}/2") lines.append(f"- stop reason: {reason}") return "\n".join(lines) + "\n" @staticmethod def _terminal(state: TaskState) -> dict[str, Any]: lifecycle = "completed" if state.phase == TaskPhase.COMPLETED else "cancelled" if state.phase == TaskPhase.CANCELLED else "failed" return {"taskId": state.task_id, "lifecycle": lifecycle, "revisionId": state.active_revision, "repairCount": state.repair_count, "completionPath": state.completion_path, "diagnosticsPath": state.diagnostics_path, "code": state.last_error.value if state.last_error else "", "message": "CAD result published." if lifecycle == "completed" else "CAD generation stopped.", "userActionRequired": False}