cdsl-cad/backend/app/cad_agent/application/workflow.py

"""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": "<local_feature_name>.<output_role>",
            },
        }

    @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: '<feature>.<output_role>', "
                "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}