modules/concordato-plan-review/scripts/apply_review_edits.py

from __future__ import annotations

import sys as _bootstrap_sys  # isort: skip

_bootstrap_sys.dont_write_bytecode = True
_bootstrap_sys.pycache_prefix = (
    r"Z:\__concordato_no_bytecode__"
    if _bootstrap_sys.platform == "win32"
    else "/dev/null/concordato-plan-review"
)

import os as _bootstrap_os  # isort: skip

_BOOTSTRAP_PATH = _bootstrap_os.path.join(
    _bootstrap_os.path.dirname(_bootstrap_os.path.abspath(__file__)),
    "implementation_bootstrap.py",
)
_BOOTSTRAP_ENTRY = _bootstrap_os.lstat(_BOOTSTRAP_PATH)
if _BOOTSTRAP_ENTRY.st_mode & 0o170000 != 0o100000 or _BOOTSTRAP_ENTRY.st_nlink != 1:
    raise RuntimeError("Concordato implementation bootstrap is not a real file.")
with open(_BOOTSTRAP_PATH, "rb") as _bootstrap_handle:
    _BOOTSTRAP_BEFORE = _bootstrap_os.fstat(_bootstrap_handle.fileno())
    _BOOTSTRAP_BYTES = _bootstrap_handle.read()
    _BOOTSTRAP_AFTER = _bootstrap_os.fstat(_bootstrap_handle.fileno())
_BOOTSTRAP_IDENTITY = (
    _BOOTSTRAP_ENTRY.st_dev,
    _BOOTSTRAP_ENTRY.st_ino,
    _BOOTSTRAP_ENTRY.st_size,
    _BOOTSTRAP_ENTRY.st_mtime_ns,
)
if (
    _BOOTSTRAP_IDENTITY
    != (
        _BOOTSTRAP_BEFORE.st_dev,
        _BOOTSTRAP_BEFORE.st_ino,
        _BOOTSTRAP_BEFORE.st_size,
        _BOOTSTRAP_BEFORE.st_mtime_ns,
    )
    or _BOOTSTRAP_IDENTITY
    != (
        _BOOTSTRAP_AFTER.st_dev,
        _BOOTSTRAP_AFTER.st_ino,
        _BOOTSTRAP_AFTER.st_size,
        _BOOTSTRAP_AFTER.st_mtime_ns,
    )
    or len(_BOOTSTRAP_BYTES) != _BOOTSTRAP_AFTER.st_size
):
    raise RuntimeError("Concordato implementation bootstrap changed while read.")
_BOOTSTRAP_NAMESPACE = {
    "__file__": _BOOTSTRAP_PATH,
    "__name__": "_concordato_implementation_bootstrap",
}
# The exact stable single-link bootstrap source is verified above.
exec(  # nosec B102
    compile(_BOOTSTRAP_BYTES, _BOOTSTRAP_PATH, "exec"), _BOOTSTRAP_NAMESPACE
)
_BOOTSTRAP_NAMESPACE["activate_implementation_boundary"]()
_SCRIPTS_DIR = _bootstrap_os.path.dirname(_bootstrap_os.path.abspath(__file__))
if _SCRIPTS_DIR not in _bootstrap_sys.path:
    _bootstrap_sys.path.insert(0, _SCRIPTS_DIR)

import argparse
import json
import re
import shutil
import sys
from pathlib import Path
from typing import Any

from docx import Document
from replay_assurance import replay_assurance
from vera_assurance import (  # noqa: E402
    AssuranceContractError,
    load_client_engagement_context_file,
)

__all__ = ["apply_review_edits", "main"]

SUMMARY_DOCX = "concordato_preventivo_review_summary.docx"
REGENERATE_NATIVE_OUTPUT_ACTION = (
    "Regenerate native DOCX/XLSX/PDF outputs before final handoff."
)
FINAL_HANDOFF_ACTION = (
    "Review is recorded; professional conclusion and publication remain withheld."
)
COMPLETE_REVIEW_ACTION = "Complete remaining review decisions before final handoff."


def clean_text(value: object) -> str:
    """Return a stripped string for safe JSON field comparison."""

    return value.strip() if isinstance(value, str) else ""


def _read_json(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError(f"Expected JSON object: {path}")
    return payload


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


def _load_cli_customer_context(
    *,
    client_engagement: Path,
    output_dir: Path,
    persistent_output_dir: Path | None,
    input_paths: list[Path],
) -> dict[str, Any]:
    """Authorize either the canonical output or its MCP transaction copy."""

    context = load_client_engagement_context_file(
        client_engagement,
        expected_workflow_id="concordato-plan-review",
    )
    expected_output = Path(str(context["output_dir"])).resolve()
    persistent_output = (
        persistent_output_dir.expanduser().resolve()
        if persistent_output_dir is not None
        else expected_output
    )
    if persistent_output != expected_output:
        raise AssuranceContractError(
            "persistent Concordato output must be the customer run output root"
        )
    actual_output = output_dir.expanduser().resolve(strict=True)
    if actual_output == expected_output or actual_output.is_relative_to(
        expected_output
    ):
        return load_client_engagement_context_file(
            client_engagement,
            expected_workflow_id="concordato-plan-review",
            input_paths=input_paths,
            output_dir=actual_output,
        )
    try:
        relative = actual_output.relative_to(Path(str(context["run_root"])))
    except ValueError as exc:
        raise AssuranceContractError(
            "Concordato output is outside the customer run"
        ) from exc
    if not (
        len(relative.parts) == 2
        and relative.parts[0].startswith(".generated-review-transaction-")
        and relative.parts[1] == "working"
        and all(
            path.expanduser().resolve().is_relative_to(actual_output)
            for path in input_paths
        )
    ):
        raise AssuranceContractError(
            "Concordato output is outside the customer run and its review transaction"
        )
    return context


def _safe_item_id(value: object) -> str:
    text = clean_text(value) or "item"
    cleaned = "".join(char if char.isalnum() or char in "._-" else "-" for char in text)
    return cleaned.strip("-") or "item"


def _backup_file(output_dir: Path, item_id: str, target_name: str) -> dict[str, Any]:
    source = output_dir / target_name
    if not source.exists():
        return {}
    suffix = source.suffix or ".docx"
    relative = (
        Path("revisions")
        / "originals"
        / f"{source.stem}__{_safe_item_id(item_id)}{suffix}"
    )
    target = output_dir / relative
    target.parent.mkdir(parents=True, exist_ok=True)
    if not target.exists():
        shutil.copy2(source, target)
    return {
        "path": relative.as_posix(),
        "kind": suffix.lstrip(".") or "file",
        "status": "backup_original",
        "source_artifact": target_name,
        "item_id": item_id,
    }


def _upsert_output(outputs: list[dict[str, Any]], record: dict[str, Any]) -> None:
    path = record.get("path")
    for index, output in enumerate(outputs):
        if isinstance(output, dict) and output.get("path") == path:
            outputs[index] = {**output, **record}
            return
    outputs.append(record)


def _memo_effects(effects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [
        effect
        for effect in effects
        if effect.get("action") == "edit"
        and clean_text(effect.get("item_id")) == "codex-review-memo"
        and clean_text(effect.get("edit_value"))
    ]


def _summary_docx_requested(output_dir: Path, final_artifacts: dict[str, Any]) -> bool:
    if (output_dir / SUMMARY_DOCX).exists():
        return True
    outputs = final_artifacts.get("outputs")
    if not isinstance(outputs, list):
        return False
    return any(
        isinstance(output, dict) and clean_text(output.get("path")) == SUMMARY_DOCX
        for output in outputs
    )


def _visible_memo_lines(markdown_text: str) -> list[str]:
    lines: list[str] = []
    for raw_line in markdown_text.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        line = re.sub(r"^#{1,6}\s*", "", line)
        line = re.sub(r"^[-*]\s+", "", line)
        if line:
            lines.append(line)
    return lines


def _required_docx_text(memo_text: str) -> list[str]:
    fragments = ["Memo revisore Codex"]
    fragments.extend(_visible_memo_lines(memo_text))
    return list(dict.fromkeys(fragments))


def _append_memo_to_summary_docx(path: Path, memo_text: str) -> None:
    document = Document(path)
    document.add_heading("Memo revisore Codex", level=1)
    for raw_line in memo_text.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        heading_match = re.match(r"^#{1,6}\s+(.*)$", line)
        if heading_match:
            document.add_heading(heading_match.group(1).strip(), level=2)
            continue
        bullet_match = re.match(r"^[-*]\s+(.*)$", line)
        if bullet_match:
            document.add_paragraph(bullet_match.group(1).strip(), style="List Bullet")
            continue
        document.add_paragraph(line)
    document.save(path)


def _write_review_memo(
    output_dir: Path,
    effect: dict[str, Any],
    memo_text: str,
) -> tuple[dict[str, Any], dict[str, Any] | None]:
    target_name = clean_text(effect.get("target_artifact")) or "codex_run_review.md"
    target_path = output_dir / target_name
    backup_output = _backup_file(
        output_dir,
        clean_text(effect.get("item_id")),
        target_name,
    )
    target_path.parent.mkdir(parents=True, exist_ok=True)
    target_existed = target_path.exists()
    target_path.write_text(memo_text, encoding="utf-8")
    revision_artifact = clean_text(effect.get("revision_artifact"))
    effect["target_artifact"] = target_name
    effect["artifact_update"] = (
        "target_artifact_updated" if target_existed else "target_artifact_created"
    )
    if revision_artifact:
        effect["promoted_from_revision"] = revision_artifact
    if backup_output:
        effect["original_artifact_backup"] = backup_output["path"]
    return (
        {
            "path": target_name,
            "kind": target_path.suffix.lstrip(".") or "md",
            "status": "updated_from_review",
            "item_id": clean_text(effect.get("item_id")),
            "size_bytes": target_path.stat().st_size,
            "required_text": [memo_text],
            "qa_checks": ["nonempty_text", "required_text"],
        },
        backup_output or None,
    )


def _pending_native_paths(effects: list[dict[str, Any]]) -> list[str]:
    paths: list[str] = []
    for effect in effects:
        if not effect.get("requires_native_regeneration"):
            continue
        raw_paths = effect.get("derived_native_regeneration_paths")
        if not isinstance(raw_paths, list):
            raw_paths = [effect.get("target_artifact")]
        for raw_path in raw_paths:
            text = clean_text(raw_path)
            if text:
                paths.append(text)
    return list(dict.fromkeys(paths))


def _application_status(applied: dict[str, Any]) -> str:
    if int(applied.get("blocker_count") or 0) > 0:
        return "blocked"
    if int(applied.get("native_regeneration_count") or 0) > 0:
        return "partial_review_applied"
    if int(applied.get("decision_count") or 0) < int(applied.get("item_count") or 0):
        return "partial_review_applied"
    return "review_applied_assurance_withheld"


def _next_actions(current: list[Any], status: str) -> list[str]:
    next_actions = [
        clean_text(action)
        for action in current
        if clean_text(action) != REGENERATE_NATIVE_OUTPUT_ACTION
    ]
    if status == "review_applied_assurance_withheld":
        next_actions.append(FINAL_HANDOFF_ACTION)
    elif status == "partial_review_applied":
        next_actions.append(COMPLETE_REVIEW_ACTION)
    return list(dict.fromkeys(action for action in next_actions if action))


def apply_review_edits(
    output_dir: Path,
    applied_decisions_path: Path,
    final_artifacts_path: Path,
    *,
    client_context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Refresh Concordato native handoff artifacts after reviewed memo edits."""

    output_dir = output_dir.resolve()
    applied_decisions_path = applied_decisions_path.resolve()
    final_artifacts_path = final_artifacts_path.resolve()
    applied = _read_json(applied_decisions_path)
    final_artifacts = _read_json(final_artifacts_path)
    # The parent already replayed the predecessor closure before opening the
    # bounded transaction. At this point authorized review files exist, so the
    # predecessor whole-tree seal is intentionally stale until finalization.
    assurance_replay = replay_assurance(
        output_dir,
        require_output_closure=False,
        client_context=client_context,
    )
    persisted_review = _read_json(output_dir / "review_payload.json")
    applied_review = applied.get("review_payload")
    if not isinstance(applied_review, dict) or applied_review.get(
        "content_sha256"
    ) != persisted_review.get("content_sha256"):
        raise ValueError("Applied decisions are not bound to review_payload.json")
    effects = [
        effect for effect in applied.get("effects", []) if isinstance(effect, dict)
    ]
    candidate_effects = _memo_effects(effects)
    if not candidate_effects:
        return {
            "ok": True,
            "updated_effect_count": 0,
            "message": "No Concordato memo refresh was required.",
            "applied_decisions": applied,
            "final_artifacts": final_artifacts,
        }
    if not _summary_docx_requested(output_dir, final_artifacts):
        return {
            "ok": True,
            "updated_effect_count": 0,
            "message": "No Concordato summary DOCX is available to refresh.",
            "applied_decisions": applied,
            "final_artifacts": final_artifacts,
        }

    docx_path = output_dir / SUMMARY_DOCX
    if not docx_path.exists():
        raise FileNotFoundError(docx_path)

    memo_text = clean_text(candidate_effects[-1].get("edit_value"))
    memo_output, memo_backup_output = _write_review_memo(
        output_dir,
        candidate_effects[-1],
        memo_text,
    )
    backup_output = _backup_file(
        output_dir,
        clean_text(candidate_effects[-1].get("item_id")),
        SUMMARY_DOCX,
    )
    _append_memo_to_summary_docx(docx_path, memo_text)

    native_regenerated_paths = [SUMMARY_DOCX]
    native_pending = [
        path
        for path in _pending_native_paths(effects)
        if path not in set(native_regenerated_paths)
    ]
    for effect in candidate_effects:
        effect["native_regeneration_status"] = "regenerated"
        effect["native_regenerated_paths"] = native_regenerated_paths

    applied["effects"] = effects
    applied["native_regeneration_count"] = len(native_pending)
    applied["native_regeneration_paths"] = native_pending
    applied["native_regenerated_count"] = len(native_regenerated_paths)
    applied["native_regenerated_paths"] = native_regenerated_paths
    target_update_paths = list(applied.get("target_update_paths") or [])
    if memo_output["path"] not in target_update_paths:
        target_update_paths.append(memo_output["path"])
    applied["target_update_paths"] = target_update_paths
    applied["target_update_count"] = len(target_update_paths)
    original_backup_paths = list(applied.get("original_backup_paths") or [])
    if memo_backup_output and memo_backup_output["path"] not in original_backup_paths:
        original_backup_paths.append(memo_backup_output["path"])
    if backup_output and backup_output["path"] not in original_backup_paths:
        original_backup_paths.append(backup_output["path"])
    applied["original_backup_paths"] = original_backup_paths
    applied["application_status"] = _application_status(applied)

    outputs = [
        output
        for output in final_artifacts.get("outputs", [])
        if isinstance(output, dict)
    ]
    _upsert_output(outputs, memo_output)
    if memo_backup_output:
        _upsert_output(outputs, memo_backup_output)
    _upsert_output(
        outputs,
        {
            "path": SUMMARY_DOCX,
            "kind": "docx",
            "status": "updated_from_review",
            "native_regenerated": True,
            "source_artifact": "codex_run_review.md",
            "size_bytes": docx_path.stat().st_size,
            "required_text": _required_docx_text(memo_text),
            "qa_checks": ["nonempty_text", "required_text"],
        },
    )
    if backup_output:
        _upsert_output(outputs, backup_output)
    final_artifacts["outputs"] = outputs
    final_artifacts["status"] = applied["application_status"]
    final_artifacts["review_status"] = applied["application_status"]
    final_artifacts["final_ready"] = False
    review_application = final_artifacts.setdefault("review_application", {})
    if isinstance(review_application, dict):
        review_application["application_status"] = applied["application_status"]
        review_application["native_regeneration_count"] = applied[
            "native_regeneration_count"
        ]
        review_application["native_regeneration_paths"] = native_pending
        review_application["native_regenerated_count"] = applied[
            "native_regenerated_count"
        ]
        review_application["native_regenerated_paths"] = native_regenerated_paths
        review_application["target_update_count"] = applied["target_update_count"]
        review_application["target_update_paths"] = target_update_paths
        review_application["original_backup_paths"] = original_backup_paths
    final_artifacts["next_actions"] = _next_actions(
        list(final_artifacts.get("next_actions") or []),
        applied["application_status"],
    )

    _write_json(applied_decisions_path, applied)
    _write_json(final_artifacts_path, final_artifacts)
    return {
        "ok": True,
        "updated_effect_count": len(candidate_effects),
        "native_regenerated_paths": native_regenerated_paths,
        "backup_paths": [backup_output["path"]] if backup_output else [],
        "application_status": applied["application_status"],
        "assurance_replay": assurance_replay,
        "applied_decisions": applied,
        "final_artifacts": final_artifacts,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Apply Concordato review edits to downstream artifacts."
    )
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--applied-decisions", type=Path, required=True)
    parser.add_argument("--final-artifacts", type=Path, required=True)
    parser.add_argument("--client-engagement", type=Path, required=True)
    parser.add_argument("--persistent-output-dir", type=Path, help=argparse.SUPPRESS)
    args = parser.parse_args(argv)
    try:
        client_context = _load_cli_customer_context(
            client_engagement=args.client_engagement,
            output_dir=args.output_dir,
            persistent_output_dir=args.persistent_output_dir,
            input_paths=[args.applied_decisions, args.final_artifacts],
        )
    except AssuranceContractError as exc:
        parser.error(str(exc))
    result = apply_review_edits(
        args.output_dir,
        args.applied_decisions,
        args.final_artifacts,
        client_context=client_context,
    )
    sys.stdout.write(json.dumps(result, ensure_ascii=False) + "\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())