modules/prompt-optimizer/scripts/review_session.py

from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Sequence

__all__ = [
    "ReviewSessionResult",
    "RunIntakeResult",
    "synchronize_final_artifact_sizes",
    "write_review_session_artifacts",
    "write_run_intake",
]

SCHEMA_VERSION = "1.0"
PLUGIN_NAME = "prompt-optimizer"
WORKFLOW_NAME = "prompt-optimizer"

_REVIEW_COPY: dict[str, dict[str, Any]] = {
    "en": {
        "product_title": "Prompt Optimizer",
        "handoff_title": "Review Handoff",
        "run_id": "Run ID",
        "review_payload": "Review payload",
        "run_intake": "Run intake",
        "pending_decisions": "Pending decisions",
        "applied_decisions": "Applied decisions",
        "final_artifacts": "Final artifacts",
        "review_in_codex": "Review In Codex",
        "steps": (
            "Validate the hash-bound local review reference with `{tool}`.",
            "Render the review workbench with the returned token using `{tool}`.",
            "Save reviewer actions with `{tool}`.",
            "Apply reviewer actions with `{tool}`.",
        ),
        "handoff_notice": (
            "Persistent save/apply requires the MCP or local-server review "
            "surface. Static HTML fallback can copy or download decision JSON only."
        ),
        "columns": (
            "Type",
            "Prompt item",
            "Suggested action",
            "Source",
            "Output",
            "Status",
        ),
        "artifacts": {
            "optimized_prompt": "Optimized prompt",
            "answer_contract": "Answer contract JSON",
            "prompt_contract_review": "Prompt-contract semantic review JSON",
            "prompt_audit": "Prompt audit JSON",
            "prompt_package": "Prompt package Markdown",
            "source_domains": "Source domains",
            "source_domains_comma": "Source domains comma list",
            "readme_human": "Human README",
        },
        "package_required": [
            "# Prompt Optimizer Package",
            "## Answer Contract",
            "## Model-Led Research Lens",
            "## Prompt-Contract Semantic Review",
            "## What to Use",
        ],
        "readme_required_deep_research": [
            "# How to use these files",
            "Paste `optimized_prompt.md` into Deep Research.",
        ],
        "readme_required_direct": [
            "# How to use these files",
            "Use `optimized_prompt.md` as the instructions for generating the answer.",
        ],
        "readme_required_research_plugin": [
            "# How to use these files",
            "Use the installed OpenAI Deep Research skill with `optimized_prompt.md`.",
        ],
        "dependency_note": "Codex should run scripts/check_dependencies.py before helper scripts.",
        "data_notes": [
            "Prompt validation receives question and prompt text from the current Codex/user workflow.",
            "The deterministic script does not read source files, call model APIs, use connectors, or upload data.",
        ],
        "caveats": [
            "Angle and jurisdiction choices remain Codex/user intake decisions; this widget reviews the generated package, not the pre-draft choices.",
            "Prompt-to-question and prompt-to-contract conformance is model-led and recorded in prompt_contract_review.json.",
            "Deterministic validation checks review-record shape, exact fact anchors, source-domain sidecars, and required prompt controls.",
            "ui_decisions.json is pending until Codex, the MCP widget, or fallback review records decisions.",
        ],
        "next_actions": [
            "Call validate_prompt_optimizer_review, then render_prompt_optimizer_review when MCP is available.",
            "Repair draft_prompt.md and rerun validation if prompt_audit.json fails.",
            "Follow the selected generation route and instructions in README_HUMAN.md.",
        ],
    },
    "es": {
        "product_title": "Optimización del prompt",
        "handoff_title": "Entrega para revisión",
        "run_id": "ID de ejecución",
        "review_payload": "Datos de revisión",
        "run_intake": "Datos de ejecución",
        "pending_decisions": "Decisiones pendientes",
        "applied_decisions": "Decisiones aplicadas",
        "final_artifacts": "Artefactos finales",
        "review_in_codex": "Revisión en Codex",
        "steps": (
            "Valide la referencia local vinculada por hash con `{tool}`.",
            "Abra el área de revisión con el token devuelto usando `{tool}`.",
            "Guarde las acciones del revisor con `{tool}`.",
            "Aplique las acciones del revisor con `{tool}`.",
        ),
        "handoff_notice": (
            "El guardado y la aplicación persistentes requieren la superficie MCP "
            "o el servidor local. El modo HTML estático solo permite copiar o "
            "descargar el JSON de decisiones."
        ),
        "columns": (
            "Tipo",
            "Elemento del prompt",
            "Acción sugerida",
            "Fuente",
            "Salida",
            "Estado",
        ),
        "artifacts": {
            "optimized_prompt": "Prompt optimizado",
            "answer_contract": "JSON del contrato de respuesta",
            "prompt_contract_review": "JSON de revisión semántica del contrato del prompt",
            "prompt_audit": "JSON de auditoría del prompt",
            "prompt_package": "Paquete del prompt en Markdown",
            "source_domains": "Dominios de las fuentes",
            "source_domains_comma": "Lista de dominios separados por comas",
            "readme_human": "Guía de uso",
        },
        "package_required": [
            "# Paquete de optimización del prompt",
            "## Contrato de respuesta",
            "## Enfoque de investigación dirigido por el modelo",
            "## Revisión semántica del contrato del prompt",
            "## Cómo utilizar los archivos",
        ],
        "readme_required_deep_research": [
            "# Cómo utilizar estos archivos",
            "Pegue `optimized_prompt.md` en Deep Research.",
        ],
        "readme_required_direct": [
            "# Cómo utilizar estos archivos",
            "Use `optimized_prompt.md` como instrucciones para generar la respuesta.",
        ],
        "readme_required_research_plugin": [
            "# Cómo utilizar estos archivos",
            "Use la skill Deep Research de OpenAI instalada con `optimized_prompt.md`.",
        ],
        "dependency_note": "Codex debe ejecutar scripts/check_dependencies.py antes de los scripts auxiliares.",
        "data_notes": [
            "La validación recibe la pregunta y el prompt del flujo actual de Codex y del usuario.",
            "El script determinista no lee archivos fuente, llama a modelos, utiliza conectores ni carga datos.",
        ],
        "caveats": [
            "El enfoque y la jurisdicción siguen siendo decisiones de Codex y del usuario; esta revisión comprueba el paquete generado, no las decisiones previas al borrador.",
            "La conformidad del prompt con la pregunta y el contrato se revisa mediante el modelo y se registra en prompt_contract_review.json.",
            "La validación determinista comprueba la forma del registro, los hechos exactos, los archivos auxiliares de dominios y los controles obligatorios del prompt.",
            "ui_decisions.json permanece pendiente hasta que Codex, el widget MCP o la revisión alternativa registren las decisiones.",
        ],
        "next_actions": [
            "Ejecute validate_prompt_optimizer_review y, cuando MCP esté disponible, render_prompt_optimizer_review.",
            "Corrija draft_prompt.md y vuelva a validar si prompt_audit.json falla.",
            "Siga la ruta de generación elegida y las instrucciones de README_HUMAN.md.",
        ],
    },
}


def _language_code(value: object | None, audit: dict[str, Any] | None = None) -> str:
    text = str(value or "auto").strip().lower().replace("_", "-")
    if text.startswith("es"):
        return "es"
    policy = audit.get("jurisdiction_policy") if isinstance(audit, dict) else None
    if isinstance(policy, dict):
        effective = str(policy.get("language") or "").lower().replace("_", "-")
        if effective.startswith("es"):
            return "es"
    return "en"


def _copy(value: object | None, audit: dict[str, Any] | None = None) -> dict[str, Any]:
    return _REVIEW_COPY[_language_code(value, audit)]


@dataclass(frozen=True)
class RunIntakeResult:
    """Run intake artifact written before prompt validation packaging."""

    run_id: str
    path: Path


@dataclass(frozen=True)
class ReviewSessionResult:
    """Review-session artifacts for one optimized prompt package."""

    run_id: str
    run_intake_path: Path
    review_payload_path: Path
    ui_decisions_path: Path
    final_artifacts_path: Path
    review_item_count: int


def _utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def _safe_slug(value: str) -> str:
    slug = re.sub(r"[^a-zA-Z0-9_.-]+", "-", value).strip("-._").lower()
    return slug or "run"


def _run_id(question_text: str) -> str:
    timestamp = re.sub(r"[^0-9]", "", _utc_now())
    words = "-".join(question_text.strip().split()[:6])
    return f"{PLUGIN_NAME}-{_safe_slug(words)}-{timestamp}"


def _write_json(path: Path, payload: dict[str, Any]) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
        encoding="utf-8",
    )
    return path


def synchronize_final_artifact_sizes(final_artifacts_path: Path) -> None:
    """Refresh declared byte sizes after downstream artifacts reach final form."""

    payload = json.loads(final_artifacts_path.read_text(encoding="utf-8"))
    outputs = payload.get("outputs")
    if not isinstance(outputs, list):
        raise ValueError("final_artifacts.json outputs must be a list")
    output_dir = final_artifacts_path.parent.resolve()
    for output in outputs:
        if not isinstance(output, dict) or "size_bytes" not in output:
            continue
        relative = output.get("path")
        if not isinstance(relative, str) or not relative.strip():
            raise ValueError("final artifact output path must be a non-empty string")
        artifact_path = (output_dir / relative).resolve()
        if not artifact_path.is_relative_to(output_dir) or not artifact_path.is_file():
            raise ValueError(f"final artifact output is missing: {relative}")
        output["size_bytes"] = artifact_path.stat().st_size
    _write_json(final_artifacts_path, payload)


def _write_review_handoff_card(
    output_dir: Path,
    *,
    run_id: str,
    validate_tool: str,
    render_tool: str,
    save_tool: str,
    apply_tool: str,
    language: str,
    audit: dict[str, Any],
) -> Path:
    copy = _copy(language, audit)
    steps = copy["steps"]
    path = output_dir / "review_handoff.md"
    lines = [
        f"# {copy['product_title']} · {copy['handoff_title']}",
        "",
        f"- {copy['run_id']}: `{run_id}`",
        f"- {copy['review_payload']}: `review_payload.json`",
        f"- {copy['run_intake']}: `run_intake.json`",
        f"- {copy['pending_decisions']}: `ui_decisions.json`",
        f"- {copy['applied_decisions']}: `applied_decisions.json`",
        f"- {copy['final_artifacts']}: `final_artifacts.json`",
        "",
        f"## {copy['review_in_codex']}",
        f"1. {steps[0].format(tool=validate_tool)}",
        f"2. {steps[1].format(tool=render_tool)}",
        f"3. {steps[2].format(tool=save_tool)}",
        f"4. {steps[3].format(tool=apply_tool)}",
        "",
        copy["handoff_notice"],
    ]
    if _language_code(language, audit) == "es":
        lines.insert(1, "<!-- Review Handoff -->")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return path


def _review_handoff_output_record(
    path: Path, language: str, audit: dict[str, Any]
) -> dict[str, Any]:
    copy = _copy(language, audit)
    required_text = [
        "Review Handoff",
        "review_payload.json",
        "ui_decisions.json",
        "applied_decisions.json",
        "final_artifacts.json",
    ]
    if _language_code(language, audit) == "es":
        required_text[1:1] = [copy["handoff_title"], copy["review_in_codex"]]
    return {
        "path": path.name,
        "kind": "md",
        "status": "written",
        "required_text": required_text,
        "qa_checks": ["nonempty_text", "required_text"],
    }


def _local_output_refs(final_artifacts_path: Path) -> list[str]:
    refs = [
        "run_intake.json",
        "review_payload.json",
        "ui_decisions.json",
        "final_artifacts.json",
    ]
    payload = json.loads(final_artifacts_path.read_text(encoding="utf-8"))
    outputs = payload.get("outputs")
    if isinstance(outputs, list):
        for output in outputs:
            if not isinstance(output, dict):
                continue
            path_value = output.get("path")
            if (
                isinstance(path_value, str)
                and path_value.strip()
                and "://" not in path_value
            ):
                refs.append(path_value.strip())
    return list(dict.fromkeys(refs))


def _append_execution_trace(
    run_intake_path: Path,
    final_artifacts_path: Path,
    *,
    command: Sequence[str],
) -> None:
    from vera_assurance.serialization import build_review_execution_step

    payload = json.loads(run_intake_path.read_text(encoding="utf-8"))
    step = build_review_execution_step(payload, WORKFLOW_NAME, command)
    step["outputs"] = _local_output_refs(final_artifacts_path)
    payload["execution_trace"] = [step]
    _write_json(run_intake_path, payload)


def _as_output_ref(path: str | Path | None, output_dir: Path) -> str | None:
    if path is None:
        return None
    candidate = Path(path)
    try:
        return candidate.relative_to(output_dir).as_posix()
    except ValueError:
        return candidate.as_posix()


def _run_path_reference(
    path: Path,
    client_engagement: dict[str, Any] | None,
) -> str:
    """Return an absolute unmanaged path or a portable managed-run reference."""

    if client_engagement is None:
        return path.as_posix()
    run_root_value = client_engagement.get("run_root")
    if not isinstance(run_root_value, str) or not run_root_value.strip():
        raise ValueError("Managed Prompt Optimizer context has no run_root.")
    run_root = Path(run_root_value).expanduser().resolve(strict=True)
    resolved = path.expanduser().resolve(strict=True)
    try:
        relative = resolved.relative_to(run_root)
    except ValueError as exc:
        raise ValueError("Prompt Optimizer path is outside the current run.") from exc
    if not relative.parts:
        raise ValueError("Prompt Optimizer path must identify a run artifact.")
    return relative.as_posix()


def _clean_text(value: Any) -> str:
    return " ".join(str(value or "").strip().split())


def _base_item(
    item_id: str,
    item_type: str,
    title: str,
    *,
    allowed_actions: Sequence[str],
    recommended_action: str,
    source_path: str | None = None,
    output_path: str | None = None,
    evidence: Sequence[dict[str, Any]] = (),
    data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return {
        "id": item_id,
        "item_type": item_type,
        "title": title,
        "source_path": source_path,
        "output_path": output_path,
        "allowed_actions": list(allowed_actions),
        "recommended_action": recommended_action,
        "evidence": list(evidence),
        "data": data or {},
        "status": "needs_review",
    }


def _review_columns(language: str, audit: dict[str, Any]) -> list[dict[str, str]]:
    labels = _copy(language, audit)["columns"]
    fields = (
        "item_type",
        "title",
        "recommended_action",
        "source_path",
        "output_path",
        "status",
    )
    return [
        {"field": field, "label": str(label)}
        for field, label in zip(fields, labels, strict=True)
    ]


def _audit_items(audit: dict[str, Any]) -> list[dict[str, Any]]:
    failed = audit.get("failed_checks", [])
    if not isinstance(failed, list):
        return []
    return [
        _base_item(
            f"audit-check-{index}",
            "audit_check",
            str(check),
            output_path="prompt_audit.json",
            allowed_actions=("accept", "reject", "edit", "mark_unclear", "skip"),
            recommended_action="reject",
            evidence=[
                {
                    "kind": "prompt_audit_check",
                    "status": "fail",
                    "check": check,
                    "missing_fact_anchors": audit.get("missing_fact_anchors"),
                    "missing_explicit_questions": audit.get(
                        "missing_explicit_questions"
                    ),
                }
            ],
            data={"check": check, "audit_ref": "prompt_audit.json"},
        )
        for index, check in enumerate(failed, start=1)
    ]


def _artifact_items(
    paths: dict[str, Path],
    output_dir: Path,
    language: str,
    audit: dict[str, Any],
) -> list[dict[str, Any]]:
    artifact_copy = _copy(language, audit)["artifacts"]
    labels = {
        "optimized_prompt": ("prompt_artifact", artifact_copy["optimized_prompt"]),
        "answer_contract": (
            "review_artifact",
            artifact_copy["answer_contract"],
        ),
        "prompt_contract_review": (
            "review_artifact",
            artifact_copy["prompt_contract_review"],
        ),
        "prompt_audit": ("review_artifact", artifact_copy["prompt_audit"]),
        "prompt_package": ("review_artifact", artifact_copy["prompt_package"]),
        "source_domains": ("source_domain_artifact", artifact_copy["source_domains"]),
        "source_domains_comma": (
            "source_domain_artifact",
            artifact_copy["source_domains_comma"],
        ),
        "readme_human": ("review_artifact", artifact_copy["readme_human"]),
    }
    items: list[dict[str, Any]] = []
    for index, (field, (item_type, title)) in enumerate(labels.items(), start=1):
        path_value = paths.get(field)
        if not path_value:
            continue
        path_ref = _as_output_ref(path_value, output_dir)
        exists = Path(path_value).exists()
        items.append(
            _base_item(
                f"artifact-{index}",
                item_type,
                title,
                output_path=path_ref,
                allowed_actions=("accept", "edit", "mark_unclear", "skip"),
                recommended_action="accept" if exists else "mark_unclear",
                evidence=[
                    {
                        "kind": "artifact_status",
                        "field": field,
                        "path": path_ref,
                        "exists": exists,
                    }
                ],
                data={"field": field, "path": path_ref, "exists": exists},
            )
        )
    return items


def _source_artifacts(
    paths: dict[str, Path],
    output_dir: Path,
    *,
    run_intake_path: Path,
) -> dict[str, str]:
    """Inventory the deterministic files that back the review payload."""

    artifacts: dict[str, str] = {}
    run_intake_ref = _as_output_ref(run_intake_path, output_dir)
    if run_intake_path.is_file() and run_intake_ref is not None:
        artifacts["run_intake"] = run_intake_ref
    for field, path in paths.items():
        path_ref = _as_output_ref(path, output_dir)
        if Path(path).is_file() and path_ref is not None:
            artifacts[field] = path_ref
    return artifacts


def _output_records(
    output_dir: Path, language: str, audit: dict[str, Any]
) -> list[dict[str, Any]]:
    review_files = {
        "run_intake.json",
        "review_payload.json",
        "ui_decisions.json",
        "final_artifacts.json",
    }
    temporary_files = {
        "draft_prompt.md",
        "draft_source_domains.txt",
    }
    copy = _copy(language, audit)
    answer_contract = audit.get("answer_contract")
    deep_research = (
        isinstance(answer_contract, dict)
        and answer_contract.get("generation_route") == "chatgpt_deep_research"
    )
    readme_required_key = (
        "readme_required_deep_research" if deep_research else "readme_required_direct"
    )
    if (
        isinstance(answer_contract, dict)
        and answer_contract.get("generation_route") == "deep_research_plugin"
    ):
        readme_required_key = "readme_required_research_plugin"
    required_text_by_path = {
        "prompt_package.md": copy["package_required"],
        "README_HUMAN.md": copy[readme_required_key],
    }
    outputs: list[dict[str, Any]] = []
    for path in sorted(output_dir.rglob("*")):
        if (
            not path.is_file()
            or path.name in review_files
            or path.name in temporary_files
        ):
            continue
        relative = path.relative_to(output_dir).as_posix()
        output = {
            "path": relative,
            "size_bytes": path.stat().st_size,
            "kind": path.suffix.lower().lstrip(".") or "file",
            "status": "written",
        }
        required_text = required_text_by_path.get(relative)
        if required_text:
            output["required_text"] = required_text
            output["qa_checks"] = ["nonempty_text", "required_text"]
        outputs.append(output)
    return outputs


def write_run_intake(
    output_dir: Path,
    *,
    question_text: str,
    prompt_text: str,
    language: str,
    source_domains: Sequence[str],
    answer_contract: dict[str, Any],
    prompt_contract_review: dict[str, Any] | None = None,
    input_paths: Sequence[Path] = (),
    client_engagement: dict[str, Any] | None = None,
    client_run_id: str | None = None,
) -> RunIntakeResult:
    """Write run intake before deterministic prompt validation."""

    context_run_id = (
        str(client_engagement["run_id"]) if client_engagement is not None else None
    )
    if client_run_id is not None and context_run_id not in {None, client_run_id}:
        raise ValueError("Prompt Optimizer run ID does not match its client context.")
    run_id = context_run_id or client_run_id or _run_id(question_text)
    copy = _copy(language)
    input_refs = [_run_path_reference(path, client_engagement) for path in input_paths]
    output_ref = _run_path_reference(output_dir, client_engagement)
    payload = {
        "schema_version": SCHEMA_VERSION,
        "plugin": PLUGIN_NAME,
        "workflow": WORKFLOW_NAME,
        "run_id": run_id,
        **(
            {"path_reference": "run_root_relative"}
            if client_engagement is not None
            else {}
        ),
        "created_at": _utc_now(),
        "language": language,
        "input_paths": input_refs,
        "output_dir": output_ref,
        "inferred_task": "prompt_optimizer_review_payload",
        "assumptions": {
            "question_character_count": len(question_text),
            "prompt_character_count": len(prompt_text),
            "source_domain_count": len(source_domains),
            "language": language,
            "generation_route": answer_contract.get("generation_route"),
            "document_type": answer_contract.get("document_type"),
            "validation_profile": answer_contract.get("validation_profile"),
            "prompt_contract_review_status": (
                prompt_contract_review.get("overall_status")
                if isinstance(prompt_contract_review, dict)
                else "not_supplied"
            ),
        },
        "unresolved_questions": [],
        "dependency_check": {
            "status": "not_run_by_script",
            "note": copy["dependency_note"],
        },
        "data_posture": {
            "local_files_read": input_refs,
            "external_connectors_used": [],
            "upload_paths_used": [],
            "remote_sql_execution_used": False,
            "hosted_notebook_execution_used": False,
            "notes": copy["data_notes"],
        },
        "status": "ready_for_prompt_validation",
    }
    return RunIntakeResult(
        run_id=run_id,
        path=_write_json(output_dir / "run_intake.json", payload),
    )


def write_review_session_artifacts(
    output_dir: Path,
    *,
    run_id: str,
    run_intake_path: Path,
    question_text: str,
    audit: dict[str, Any],
    paths: dict[str, Path],
) -> ReviewSessionResult:
    """Write review payload, pending decisions, and final artifact inventory."""

    language = str(audit.get("language") or "auto")
    copy = _copy(language, audit)
    items: list[dict[str, Any]] = []
    items.extend(_audit_items(audit))
    items.extend(_artifact_items(paths, output_dir, language, audit))

    review_payload = {
        "schema_version": SCHEMA_VERSION,
        "plugin": PLUGIN_NAME,
        "workflow": WORKFLOW_NAME,
        "run_id": run_id,
        "created_at": _utc_now(),
        "language": language,
        "source_paths": [],
        "review_type": "prompt_optimizer_review",
        "items": items,
        "item_count": len(items),
        "columns": _review_columns(language, audit),
        "source_artifacts": _source_artifacts(
            paths,
            output_dir,
            run_intake_path=run_intake_path,
        ),
        "allowed_actions": [
            "accept",
            "reject",
            "edit",
            "mark_unclear",
            "request_more_documents",
            "skip",
        ],
        "status": "ready_for_review",
        "summary": {
            "audit_status": audit.get("status"),
            "failed_check_count": len(audit.get("failed_checks", []) or []),
            "source_domain_count": len(audit.get("source_domains", []) or []),
            "requires_phased_workflow": audit.get("requires_phased_workflow"),
            "topic_flags": audit.get("topic_flags", []),
        },
    }
    review_payload_path = _write_json(
        output_dir / "review_payload.json",
        review_payload,
    )

    review_payload_sha256 = hashlib.sha256(review_payload_path.read_bytes()).hexdigest()
    ui_decisions_path = _write_json(
        output_dir / "ui_decisions.json",
        {
            "schema_version": SCHEMA_VERSION,
            "plugin": PLUGIN_NAME,
            "workflow": WORKFLOW_NAME,
            "run_id": run_id,
            "decided_at": None,
            "decision_source": "not_collected",
            "review_payload_path": review_payload_path.name,
            "review_payload_sha256": review_payload_sha256,
            "decisions": [],
            "decision_count": 0,
            "status": "pending_review",
        },
    )

    review_handoff_path = _write_review_handoff_card(
        output_dir,
        run_id=run_id,
        validate_tool="validate_prompt_optimizer_review",
        render_tool="render_prompt_optimizer_review",
        save_tool="save_prompt_optimizer_decisions",
        apply_tool="apply_prompt_optimizer_decisions",
        language=language,
        audit=audit,
    )
    outputs = _output_records(output_dir, language, audit)
    outputs = [
        output
        for output in outputs
        if not (
            isinstance(output, dict) and output.get("path") == review_handoff_path.name
        )
    ]
    outputs.append(_review_handoff_output_record(review_handoff_path, language, audit))

    final_artifacts_path = _write_json(
        output_dir / "final_artifacts.json",
        {
            "schema_version": SCHEMA_VERSION,
            "plugin": PLUGIN_NAME,
            "workflow": WORKFLOW_NAME,
            "run_id": run_id,
            "completed_at": _utc_now(),
            "review_payload_sha256": review_payload_sha256,
            "outputs": outputs,
            "caveats": copy["caveats"],
            "next_actions": copy["next_actions"],
            "status": "written_pending_review",
        },
    )
    _append_execution_trace(
        run_intake_path,
        final_artifacts_path,
        command=["python", "plugins/prompt-optimizer/scripts/validate_prompt.py"],
    )

    return ReviewSessionResult(
        run_id=run_id,
        run_intake_path=run_intake_path,
        review_payload_path=review_payload_path,
        ui_decisions_path=ui_decisions_path,
        final_artifacts_path=final_artifacts_path,
        review_item_count=len(items),
    )