skills/auto-preference-learner/scripts/apply_agents_update.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import subprocess
from contextlib import contextmanager
from pathlib import Path
from typing import Any

from _common import (
    STATE_DIR_NAME,
    atomic_write_bytes,
    atomic_write_json,
    codex_home,
    find_git_root,
    isoformat,
    load_json,
    normalize_text,
    sha256_bytes,
    stable_id,
)
from reconcile_agents import (
    MUTATING_ACTIONS,
    ReconcileError,
    apply_decisions,
    decision_fingerprint,
    evidence_ref_record_ids,
    evidence_ref_scopes,
    encode_agents,
    normalized_effective_lines,
    path_key,
    read_agents,
    unified_diff,
)


def append_jsonl(path: Path, items: list[dict[str, Any]]) -> None:
    if not items:
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8", newline="\n") as handle:
        for item in items:
            handle.write(json.dumps(item, ensure_ascii=False) + "\n")
        handle.flush()
        os.fsync(handle.fileno())


@contextmanager
def optimizer_lock(path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = path.open("a+b")
    try:
        if handle.seek(0, os.SEEK_END) == 0:
            handle.write(b"\0")
            handle.flush()
        handle.seek(0)
        if os.name == "nt":
            import msvcrt

            try:
                msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
            except OSError as error:
                raise RuntimeError("another optimizer apply is already running") from error
        else:
            import fcntl

            try:
                fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError as error:
                raise RuntimeError("another optimizer apply is already running") from error
        yield
    finally:
        try:
            handle.seek(0)
            if os.name == "nt":
                import msvcrt

                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                import fcntl

                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        except OSError:
            pass
        handle.close()


def set_confirmed_suggest_commit_preference(state_dir: Path, enabled: bool) -> dict[str, Any]:
    with optimizer_lock(state_dir / ".apply.lock"):
        state_path = state_dir / "state.json"
        state = load_json(state_path, {"version": 1})
        state.setdefault("preferences", {})["confirmed_suggest_git_commit"] = enabled
        state["preferences_updated_at"] = isoformat()
        atomic_write_json(state_path, state)
    return {"confirmed_suggest_git_commit": enabled, "state_path": str(state_path)}


def commit_agents(project_root: Path, target: Path) -> dict[str, Any]:
    git_root = find_git_root(project_root)
    if git_root is None:
        return {"committed": False, "reason": "target is not in a Git repository"}
    try:
        relative = target.resolve().relative_to(git_root.resolve())
    except ValueError:
        return {"committed": False, "reason": "target is outside detected Git root"}
    index_snapshot = subprocess.run(
        ["git", "-C", str(git_root), "ls-files", "--stage", "-z", "--", str(relative)],
        capture_output=True,
        check=False,
    )
    if index_snapshot.returncode != 0:
        raise RuntimeError(f"failed to inspect AGENTS.md index state: {index_snapshot.stderr.decode('utf-8', 'replace').strip()}")
    index_entries: list[tuple[str, str, str]] = []
    for raw_entry in index_snapshot.stdout.split(b"\0"):
        if not raw_entry:
            continue
        metadata, raw_path = raw_entry.split(b"\t", 1)
        mode, object_id, stage = metadata.decode("ascii").split()
        if stage != "0":
            raise RuntimeError("cannot commit an unmerged AGENTS.md index entry")
        index_entries.append((mode, object_id, raw_path.decode("utf-8", "surrogateescape")))

    def restore_index() -> None:
        removed = subprocess.run(
            ["git", "-C", str(git_root), "update-index", "--force-remove", "--", str(relative)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            check=False,
        )
        if removed.returncode != 0:
            raise RuntimeError(f"failed to restore AGENTS.md index state: {removed.stderr.strip()}")
        for mode, object_id, indexed_path in index_entries:
            restored = subprocess.run(
                ["git", "-C", str(git_root), "update-index", "--add", "--cacheinfo", mode, object_id, indexed_path],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                check=False,
            )
            if restored.returncode != 0:
                raise RuntimeError(f"failed to restore AGENTS.md index entry: {restored.stderr.strip()}")

    add = subprocess.run(["git", "-C", str(git_root), "add", "--", str(relative)], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False)
    if add.returncode != 0:
        restore_index()
        raise RuntimeError(f"git add failed: {add.stderr.strip()}")
    status = subprocess.run(["git", "-C", str(git_root), "diff", "--cached", "--quiet", "--", str(relative)], check=False)
    if status.returncode == 0:
        restore_index()
        return {"committed": False, "reason": "no staged AGENTS.md change"}
    commit = subprocess.run(
        ["git", "-C", str(git_root), "commit", "--only", "-m", "chore: update learned Codex preferences", "--", str(relative)],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if commit.returncode != 0:
        restore_index()
        raise RuntimeError(f"git commit failed: {commit.stderr.strip() or commit.stdout.strip()}")
    revision = subprocess.run(["git", "-C", str(git_root), "rev-parse", "HEAD"], capture_output=True, text=True, encoding="utf-8", check=True).stdout.strip()
    return {"committed": True, "commit": revision, "git_root": str(git_root)}


def select_decisions(decisions: list[dict[str, Any]], identifiers: set[str] | None) -> list[dict[str, Any]]:
    if identifiers is None:
        return decisions
    return [item for item in decisions if item.get("decision_id") in identifiers]


def resolve_review_selectors(plan: dict[str, Any], selectors: set[str] | None) -> set[str] | None:
    """Resolve user-facing plan-local numbers while retaining internal ID compatibility."""
    if selectors is None:
        return None
    decisions = [item for item in plan.get("decisions", []) if isinstance(item, dict)]
    internal_ids = {str(item.get("decision_id")) for item in decisions}
    number_to_id = {
        str(item["selection_number"]): str(item.get("decision_id"))
        for item in decisions
        if isinstance(item.get("selection_number"), int) and not isinstance(item.get("selection_number"), bool)
    }
    return {
        selector if selector in internal_ids else number_to_id.get(selector, selector)
        for selector in map(str, selectors)
    }


def rule_provenance_key(decision: dict[str, Any], instruction: str) -> str:
    return stable_id(
        str(decision.get("target")),
        str(decision.get("project_root")),
        normalize_text(instruction),
        prefix="P-",
    )


def decision_scope_key(decision: dict[str, Any]) -> str:
    if decision.get("target") == "global":
        return "global"
    return f"project:{path_key(Path(str(decision.get('project_root') or '.')))}"


def update_state(
    state_dir: Path,
    plan_path: Path,
    plan: dict[str, Any],
    mode: str,
    applied_ids: set[str],
    rejected_ids: set[str],
    failed_ids: set[str],
    checkpoint_record_ids: set[str],
    results: list[dict[str, Any]],
    checkpoint_batch: dict[str, Any] | None = None,
) -> None:
    state_path = state_dir / "state.json"
    state = load_json(
        state_path,
        {
            "version": 1,
            "processed_record_ids": [],
            "pending_decisions": {},
            "rejected_candidates": {},
            "tombstones": {},
            "rule_provenance": {},
            "constraint_revision": 0,
            "scope_revisions": {},
        },
    )
    state["processed_record_ids"] = sorted(set(state.get("processed_record_ids", [])) | checkpoint_record_ids)
    if checkpoint_record_ids and plan.get("review_protocol") == "project-then-global-v1":
        state["pending_global_candidates"] = dict(plan.get("pending_global_candidates", {}))
    if checkpoint_record_ids and isinstance(checkpoint_batch, dict):
        indexed = {
            str(item["record_id"]): item
            for item in state.get("record_index", [])
            if isinstance(item, dict) and item.get("record_id")
        }
        for record in checkpoint_batch.get("records", []):
            if isinstance(record, dict) and str(record.get("record_id")) in checkpoint_record_ids:
                indexed[str(record["record_id"])] = record
        state["record_index"] = sorted(
            indexed.values(), key=lambda item: (item.get("timestamp") or "", item.get("session_id") or "", item.get("turn_ordinal") or 0)
        )
        state["source_cursors"] = dict(checkpoint_batch.get("source_cursors", state.get("source_cursors", {})))
        state["session_scan_cache"] = dict(checkpoint_batch.get("scan_cache", state.get("session_scan_cache", {})))
    pending = state.setdefault("pending_decisions", {})
    rejected_candidates = state.setdefault("rejected_candidates", {})
    tombstones = state.setdefault("tombstones", {})
    provenance = state.setdefault("rule_provenance", {})
    now = isoformat()
    changed_scopes: set[str] = set()
    log_items: list[dict[str, Any]] = []
    application_by_id: dict[str, dict[str, Any]] = {}
    for result in results:
        for identifier in result.get("decision_ids", []):
            application_by_id[identifier] = result

    for decision in plan.get("decisions", []):
        identifier = decision["decision_id"]
        status = "no-op" if decision.get("action") == "no-op" else "needs-evidence" if decision.get("action") == "needs_evidence" else "pending"
        if identifier in rejected_ids:
            status = "rejected"
            rejected_candidates[decision_fingerprint(decision)] = {
                "at": now,
                "decision_id": identifier,
                "target": decision.get("target"),
                "project_root": decision.get("project_root"),
                "instruction": decision.get("instruction"),
                "evidence_refs": decision.get("evidence_refs", []),
            }
            pending.pop(identifier, None)
            changed_scopes.add(decision_scope_key(decision))
        elif identifier in failed_ids:
            status = "error"
        elif identifier in applied_ids:
            action = decision.get("action")
            supersedes = str(decision.get("supersedes_decision_id") or "").strip()
            if supersedes:
                status = "edited-accepted"
                pending.pop(supersedes, None)
            elif action == "no-op":
                status = "no-op"
            elif mode == "auto":
                status = "auto-applied"
            elif action == "remove":
                status = "withdrawn"
            else:
                status = "accepted"
            pending.pop(identifier, None)
            if action in MUTATING_ACTIONS:
                changed_scopes.add(decision_scope_key(decision))
            if action == "remove":
                existing = str(decision.get("existing_instruction") or "")
                provenance_key = rule_provenance_key(decision, existing)
                previous_provenance = provenance.get(provenance_key, {})
                inherited_evidence = (
                    previous_provenance.get("evidence_refs", []) if isinstance(previous_provenance, dict) else []
                )
                tombstone_evidence = sorted(
                    {str(value) for value in [*decision.get("evidence_refs", []), *inherited_evidence]}
                )
                tombstones[decision_fingerprint(decision, existing)] = {
                    "at": now,
                    "decision_id": identifier,
                    "target": decision.get("target"),
                    "project_root": decision.get("project_root"),
                    "instruction": existing,
                    "evidence_refs": tombstone_evidence,
                }
                provenance.pop(provenance_key, None)
            if decision.get("restore_tombstone") and decision.get("instruction"):
                explicit_ids = {str(value) for value in decision.get("restore_tombstone_ids", [])}
                evidence = {str(value) for value in decision.get("evidence_refs", [])}
                for key, entry in list(tombstones.items()):
                    same_target = isinstance(entry, dict) and entry.get("target") == decision.get("target")
                    same_project = decision.get("target") == "global" or (
                        same_target
                        and entry.get("project_root")
                        and decision.get("project_root")
                        and path_key(Path(str(entry["project_root"]))) == path_key(Path(str(decision["project_root"])))
                    )
                    overlapping_evidence = isinstance(entry, dict) and bool(evidence.intersection(str(value) for value in entry.get("evidence_refs", [])))
                    explicit_match = key in explicit_ids or (
                        isinstance(entry, dict) and str(entry.get("decision_id") or "") in explicit_ids
                    )
                    if same_target and same_project and (explicit_match or overlapping_evidence):
                        tombstones.pop(key, None)
                tombstones.pop(decision_fingerprint(decision, str(decision["instruction"])), None)
            if decision.get("override_prior_rejection") and decision.get("instruction"):
                explicit_ids = {str(value) for value in decision.get("override_rejection_ids", [])}
                evidence = {str(value) for value in decision.get("evidence_refs", [])}
                for key, entry in list(rejected_candidates.items()):
                    same_target = isinstance(entry, dict) and entry.get("target") == decision.get("target")
                    same_project = decision.get("target") == "global" or (
                        same_target
                        and entry.get("project_root")
                        and decision.get("project_root")
                        and path_key(Path(str(entry["project_root"]))) == path_key(Path(str(decision["project_root"])))
                    )
                    overlapping_evidence = isinstance(entry, dict) and bool(evidence.intersection(str(value) for value in entry.get("evidence_refs", [])))
                    explicit_match = key in explicit_ids or (
                        isinstance(entry, dict) and str(entry.get("decision_id") or "") in explicit_ids
                    )
                    if same_target and same_project and (explicit_match or overlapping_evidence):
                        rejected_candidates.pop(key, None)
                rejected_candidates.pop(decision_fingerprint(decision, str(decision["instruction"])), None)
            if decision.get("instruction") and action in {"add", "merge", "narrow", "replace"}:
                inherited_evidence = {str(value) for value in decision.get("evidence_refs", [])}
                replaced_instructions: list[str] = []
                if action in {"narrow", "replace"}:
                    replaced_instructions.append(str(decision.get("existing_instruction") or ""))
                elif action == "merge":
                    replaced_instructions.extend(str(value) for value in decision.get("existing_instructions", []))
                for old_instruction in replaced_instructions:
                    old_key = rule_provenance_key(decision, old_instruction)
                    old_entry = provenance.pop(old_key, {})
                    if isinstance(old_entry, dict):
                        inherited_evidence.update(str(value) for value in old_entry.get("evidence_refs", []))
                key = rule_provenance_key(decision, str(decision["instruction"]))
                current_entry = provenance.get(key, {})
                if isinstance(current_entry, dict):
                    inherited_evidence.update(str(value) for value in current_entry.get("evidence_refs", []))
                provenance[key] = {
                    "decision_id": identifier,
                    "evidence_refs": sorted(inherited_evidence),
                    "updated_at": now,
                }
        elif mode in {"suggest", "confirmed-suggest"} and status == "pending":
            pending[identifier] = {"plan_path": str(plan_path), "decision": decision, "created_at": now}
        log_items.append(
            {
                "timestamp": now,
                "mode": mode,
                "status": status,
                "decision": decision,
                "application": application_by_id.get(identifier),
            }
        )

    if changed_scopes:
        state["constraint_revision"] = int(state.get("constraint_revision", 0)) + 1
        scope_revisions = state.setdefault("scope_revisions", {})
        for scope_key in changed_scopes:
            scope_revisions[scope_key] = int(scope_revisions.get(scope_key, 0)) + 1
    state["last_run_at"] = now
    state["last_mode"] = mode
    state["last_run_metadata"] = plan.get("run_metadata", {})
    append_jsonl(state_dir / "decisions.jsonl", log_items)
    reports = state_dir / "reports"
    reports.mkdir(parents=True, exist_ok=True)
    report_name = now.replace(":", "-") + ".json"
    atomic_write_json(reports / report_name, {"timestamp": now, "mode": mode, "plan_path": str(plan_path), "results": results})
    atomic_write_json(state_path, state)


def validation_error(mode: str, message: str) -> dict[str, Any]:
    return {"mode": mode, "mutated_targets": 0, "results": [{"status": "error", "error": message}]}


def lexical_path_key(path: Path) -> str:
    return os.path.normcase(os.path.abspath(os.path.expanduser(str(path))))


def validate_plan_structure(plan: dict[str, Any]) -> str | None:
    top_decisions = plan.get("decisions", [])
    if not isinstance(top_decisions, list):
        return "plan decisions must be a list"
    if plan.get("record_ids") and not top_decisions:
        return "a non-empty batch plan requires at least one explicit decision, including no-op"
    if plan.get("review_protocol") == "project-then-global-v1":
        pending_candidates = plan.get("pending_global_candidates")
        if not isinstance(pending_candidates, dict):
            return "project-then-global plan has invalid pending global candidates"
        if any(
            not isinstance(candidate, dict) or str(candidate.get("candidate_id") or "") != str(identifier)
            for identifier, candidate in pending_candidates.items()
        ):
            return "pending global candidate keys must match candidate_id values"
    edited_acceptance = plan.get("edited_acceptance", False)
    if not isinstance(edited_acceptance, bool):
        return "plan edited_acceptance must be a boolean"
    superseding = [item for item in top_decisions if isinstance(item, dict) and str(item.get("supersedes_decision_id") or "").strip()]
    if edited_acceptance != bool(superseding) or (edited_acceptance and len(superseding) != len(top_decisions)):
        return "plan edited_acceptance is inconsistent with superseding decisions"
    allowed_values = plan.get("allowed_project_roots", [])
    explicit_values = plan.get("explicit_project_roots", [])
    if not isinstance(allowed_values, list) or not isinstance(explicit_values, list):
        return "plan project root allowlists must be lists"
    allowed_keys = {path_key(Path(str(value))) for value in allowed_values}
    explicit_keys = {path_key(Path(str(value))) for value in explicit_values}
    if not explicit_keys.issubset(allowed_keys):
        return "explicit project roots must be part of the target allowlist"
    top_by_id: dict[str, dict[str, Any]] = {}
    selection_numbers: list[int] = []
    for decision in top_decisions:
        if not isinstance(decision, dict):
            return "plan decision must be an object"
        identifier = str(decision.get("decision_id") or "")
        if not identifier or identifier in top_by_id:
            return "plan has a missing or duplicate top-level decision_id"
        top_by_id[identifier] = decision
        decision_target = decision.get("target")
        project_root_value = decision.get("project_root")
        if decision_target == "global":
            if project_root_value is not None:
                return f"global decision {identifier} must not contain project_root"
        elif decision_target == "project":
            if not isinstance(project_root_value, str) or not project_root_value.strip():
                return f"project decision {identifier} has no project_root"
            if path_key(Path(project_root_value)) not in allowed_keys:
                return f"project decision {identifier} root is outside the target allowlist"
        else:
            return f"decision {identifier} has an invalid target"
        selection_number = decision.get("selection_number")
        if decision.get("action") in MUTATING_ACTIONS:
            if not isinstance(selection_number, int) or isinstance(selection_number, bool) or selection_number < 1:
                return "each reviewable decision must have a positive integer selection_number"
            selection_numbers.append(selection_number)
        elif selection_number is not None:
            return "non-reviewable decisions must not have a selection_number"
    if selection_numbers != list(range(1, len(selection_numbers) + 1)):
        return "reviewable decision selection_numbers must be unique and consecutive"

    seen: set[str] = set()
    home = Path(str(plan.get("codex_home") or ""))
    for target in plan.get("targets", []):
        if not isinstance(target, dict) or not isinstance(target.get("decision_ids"), list):
            return "plan target must contain a decision_ids list"
        scope = target.get("scope")
        target_path = Path(str(target.get("path") or ""))
        project_root_value = target.get("project_root")
        if scope == "global":
            if project_root_value is not None or lexical_path_key(target_path) != lexical_path_key(home / "AGENTS.md"):
                return "global target structure is inconsistent"
        elif scope == "project":
            if not isinstance(project_root_value, str) or not project_root_value.strip():
                return "project target has no project_root"
            project_root = Path(project_root_value)
            if lexical_path_key(target_path) != lexical_path_key(project_root / "AGENTS.md"):
                return "project target path is inconsistent with project_root"
        else:
            return "plan target has an unknown scope"

        for value in target["decision_ids"]:
            identifier = str(value or "")
            top = top_by_id.get(identifier)
            if top is None or identifier in seen:
                return "target decision_id is missing from top level or duplicated"
            if scope == "global" and top.get("target") != "global":
                return f"target decision {identifier} has inconsistent global scope"
            if scope == "project" and (
                top.get("target") != "project"
                or lexical_path_key(Path(str(top.get("project_root") or ""))) != lexical_path_key(Path(str(project_root_value)))
            ):
                return f"target decision {identifier} has inconsistent project scope"
            seen.add(identifier)

    expected = {identifier for identifier, decision in top_by_id.items() if decision.get("action") in MUTATING_ACTIONS}
    if seen != expected:
        return "plan targets do not contain each top-level mutating decision exactly once"
    return None


def validate_application(
    plan: dict[str, Any],
    mode: str,
    decision_ids: set[str] | None,
    rejected_ids: set[str],
    accept_all: bool,
    authorization: dict[str, Any] | None,
) -> str | None:
    plan_mode = plan.get("mode")
    expected_plan_mode = "suggest" if mode in {"suggest", "confirmed-suggest"} else "auto"
    if plan_mode != expected_plan_mode:
        return f"{mode} requires a {expected_plan_mode} plan, got {plan_mode!r}"
    if plan.get("errors"):
        return "plan contains reconciliation errors"
    quota = plan.get("run_metadata", {}).get("quota")
    if quota is not None and quota.get("allowed") is not True:
        return "plan quota is blocked or unknown"
    if quota is not None and plan.get("run_metadata", {}).get("reasoning_effort") != "high":
        return "Scheduled plans require reasoning_effort=high"
    structure_error = validate_plan_structure(plan)
    if structure_error:
        return structure_error
    if not isinstance(authorization, dict):
        return "apply requires an independent trusted batch authorization"
    authorization_home = authorization.get("codex_home")
    if not isinstance(authorization_home, str) or not authorization_home.strip():
        return "trusted batch authorization has no codex_home"
    if path_key(Path(str(plan.get("codex_home") or ""))) != path_key(Path(authorization_home)):
        return "plan codex_home differs from trusted batch authorization"
    plan_roots = {path_key(Path(str(value))) for value in plan.get("allowed_project_roots", [])}
    authorized_roots = {path_key(Path(str(value))) for value in authorization.get("allowed_project_roots", [])}
    if plan_roots != authorized_roots:
        return "plan project roots differ from trusted batch authorization"
    plan_explicit = {path_key(Path(str(value))) for value in plan.get("explicit_project_roots", [])}
    authorized_explicit = {path_key(Path(str(value))) for value in authorization.get("explicit_project_roots", [])}
    if plan_explicit != authorized_explicit:
        return "plan explicit project roots differ from trusted batch authorization"
    plan_record_ids = set(map(str, plan.get("record_ids", [])))
    authorized_record_ids = set(map(str, authorization.get("record_ids", [])))
    if plan.get("edited_acceptance") is True:
        if not plan_record_ids.issubset(authorized_record_ids):
            return "edited plan record IDs are outside trusted batch authorization"
    elif plan_record_ids != authorized_record_ids:
        return "plan record IDs differ from trusted batch authorization"
    if any(item.get("action") == "needs_evidence" for item in plan.get("decisions", [])):
        return "plan contains unresolved needs_evidence decisions"
    known_ids = {str(item.get("decision_id")) for item in plan.get("decisions", [])}
    requested = (decision_ids or set()) | rejected_ids
    unknown = requested - known_ids
    if unknown:
        return f"unknown decision ids: {sorted(unknown)}"
    if (decision_ids or set()) & rejected_ids:
        return "the same decision cannot be accepted and rejected"
    if mode == "confirmed-suggest":
        if accept_all and decision_ids:
            return "use either explicit decision ids or --all, not both"
        if not accept_all and not decision_ids:
            return "confirmed-suggest requires at least one --decision-id or explicit --all"
    if plan.get("edited_acceptance") is True and (mode != "confirmed-suggest" or not accept_all):
        return "edited acceptance plan requires confirmed-suggest --all"
    if mode == "auto":
        if decision_ids or accept_all or rejected_ids:
            return "Auto applies the complete freshly reconciled Auto plan and does not accept review selectors"
        for decision in plan.get("decisions", []):
            if decision.get("action") in MUTATING_ACTIONS and (
                decision.get("confidence") != "high" or decision.get("risk_of_overconstraint") != "low"
            ):
                return f"Auto decision {decision.get('decision_id')} is not high-confidence/low-risk"
    trusted_scope_sets = evidence_ref_scopes(authorization)
    trusted_refs = set(trusted_scope_sets)
    ref_scopes = {ref: sorted(scopes) for ref, scopes in trusted_scope_sets.items()}
    record_ids = plan_record_ids
    records_by_ref = evidence_ref_record_ids(authorization)
    covered_records: set[str] = set()
    explicit_project_roots = {path_key(Path(str(value))) for value in plan.get("explicit_project_roots", [])}
    for decision in plan.get("decisions", []):
        evidence_refs = decision.get("evidence_refs")
        if not isinstance(evidence_refs, list) or not evidence_refs:
            return f"decision {decision.get('decision_id')} has no evidence_refs"
        unknown_refs = sorted({str(value) for value in evidence_refs if str(value) not in trusted_refs})
        if unknown_refs:
            return f"decision {decision.get('decision_id')} has untrusted evidence_refs: {unknown_refs}"
        for value in evidence_refs:
            covered_records.update(records_by_ref.get(str(value), set()))
        if decision.get("action") == "no-op" and not str(decision.get("explanation") or "").strip():
            return f"no-op decision {decision.get('decision_id')} has no specific explanation"
        if decision.get("action") not in MUTATING_ACTIONS:
            evidence_projects = {
                scope for value in evidence_refs for scope in trusted_scope_sets.get(str(value), set())
            }
            if len(evidence_projects) > 1:
                return f"decision {decision.get('decision_id')} has evidence spanning multiple projects"
            if decision.get("target") == "project" and decision.get("project_root"):
                target_scope = path_key(Path(str(decision["project_root"])))
                wrong_scope = sorted(
                    {
                        str(value)
                        for value in evidence_refs
                        if not trusted_scope_sets.get(str(value)) or target_scope not in trusted_scope_sets[str(value)]
                    }
                )
                if wrong_scope:
                    return f"decision {decision.get('decision_id')} has evidence_refs from another project: {wrong_scope}"
            continue
        if mode == "auto" and decision.get("target") == "global":
            evidence_projects = {
                scope for value in evidence_refs for scope in trusted_scope_sets.get(str(value), set())
            }
            if len(evidence_projects) < 2:
                return f"Auto global decision {decision.get('decision_id')} requires evidence from two project roots"
        if decision.get("target") == "project" and decision.get("project_root"):
            target_scope = path_key(Path(str(decision["project_root"])))
            wrong_scope = []
            if target_scope not in explicit_project_roots:
                wrong_scope = sorted(
                    {
                        str(value)
                        for value in evidence_refs
                        if not ref_scopes.get(str(value)) or target_scope not in ref_scopes[str(value)]
                    }
                )
            if wrong_scope:
                return f"mutating decision {decision.get('decision_id')} has evidence_refs from another project: {wrong_scope}"
    uncovered_records = sorted(record_ids - covered_records)
    if uncovered_records:
        return f"batch records lack explicit decision coverage: {uncovered_records}"
    return None


def validate_current_state(
    plan: dict[str, Any], state_dir: Path, mode: str, decision_ids: set[str] | None, accept_all: bool
) -> str | None:
    if mode == "suggest":
        return None
    state = load_json(state_dir / "state.json", {})
    if plan.get("edited_acceptance") is True:
        if mode != "confirmed-suggest" or not accept_all:
            return "edited acceptance plan requires confirmed-suggest --all"
        pending = state.get("pending_decisions", {}) if isinstance(state.get("pending_decisions"), dict) else {}
        seen_superseded: set[str] = set()
        for decision in plan.get("decisions", []):
            supersedes = str(decision.get("supersedes_decision_id") or "").strip()
            prior_entry = pending.get(supersedes)
            prior = prior_entry.get("decision") if isinstance(prior_entry, dict) else None
            if not supersedes or supersedes in seen_superseded or not isinstance(prior, dict):
                return f"edited decision {decision.get('decision_id')} no longer supersedes one unique pending decision"
            seen_superseded.add(supersedes)
            if decision.get("target") != prior.get("target"):
                return f"edited decision {decision.get('decision_id')} changed target scope"
            if decision.get("target") == "project" and path_key(Path(str(decision.get("project_root") or "."))) != path_key(
                Path(str(prior.get("project_root") or "."))
            ):
                return f"edited decision {decision.get('decision_id')} changed project scope"
            if not set(map(str, decision.get("evidence_refs", []))).issubset(set(map(str, prior.get("evidence_refs", [])))):
                return f"edited decision {decision.get('decision_id')} changed evidence scope"
    guard = plan.get("state_guard")
    if mode == "auto":
        if not isinstance(guard, dict):
            return "Auto plan has no state freshness guard"
        current_guard = {
            "tombstones": sorted(state.get("tombstones", {})),
            "rejected_candidates": sorted(state.get("rejected_candidates", {})),
            "constraint_revision": int(state.get("constraint_revision", 0)),
        }
        expected_guard = {
            "tombstones": sorted(guard.get("tombstones", [])),
            "rejected_candidates": sorted(guard.get("rejected_candidates", [])),
            "constraint_revision": int(guard.get("constraint_revision", 0)),
        }
        if current_guard != expected_guard:
            return "Auto plan is stale because tombstone or rejection state changed after planning"
        selected = plan.get("decisions", [])
    elif accept_all:
        selected = [
            decision
            for decision in plan.get("decisions", [])
            if plan.get("edited_acceptance") is True or decision.get("action") in MUTATING_ACTIONS
        ]
    else:
        selected = [decision for decision in plan.get("decisions", []) if decision.get("decision_id") in (decision_ids or set())]
    tombstones = state.get("tombstones", {})
    rejected = state.get("rejected_candidates", {})
    for decision in selected:
        if mode == "confirmed-suggest" and decision.get("action") in {"remove", "merge", "narrow", "replace"}:
            if not isinstance(guard, dict):
                return f"decision {decision.get('decision_id')} has no state freshness guard"
            scope_key = decision_scope_key(decision)
            expected_scope_revision = int(guard.get("scope_revisions", {}).get(scope_key, 0))
            current_scope_revision = int(state.get("scope_revisions", {}).get(scope_key, 0))
            if current_scope_revision != expected_scope_revision:
                return f"decision {decision.get('decision_id')} is stale because its rule scope changed after planning"
        if decision.get("action") not in {"add", "merge", "narrow", "replace"}:
            continue
        fingerprint = decision_fingerprint(decision)
        if fingerprint in tombstones and not decision.get("restore_tombstone"):
            return f"decision {decision.get('decision_id')} is blocked by a tombstone created after planning"
        if fingerprint in rejected and not decision.get("override_prior_rejection"):
            return f"decision {decision.get('decision_id')} is blocked by a rejection created after planning"
        decision_evidence = {str(value) for value in decision.get("evidence_refs", [])}
        target_scope = str(decision.get("target") or "")
        project_scope = (
            path_key(Path(str(decision.get("project_root"))))
            if target_scope == "project" and decision.get("project_root")
            else None
        )

        def same_scope(entry: Any) -> bool:
            if not isinstance(entry, dict) or entry.get("target") != target_scope:
                return False
            if target_scope == "global":
                return True
            value = entry.get("project_root")
            return isinstance(value, str) and project_scope == path_key(Path(value))

        tombstone_match = any(
            same_scope(entry)
            and bool(decision_evidence.intersection(str(value) for value in entry.get("evidence_refs", [])))
            for entry in tombstones.values()
        )
        rejection_match = any(
            same_scope(entry)
            and bool(decision_evidence.intersection(str(value) for value in entry.get("evidence_refs", [])))
            for entry in rejected.values()
        )
        if tombstone_match and not decision.get("restore_tombstone"):
            return f"decision {decision.get('decision_id')} reuses feedback evidence withdrawn after planning"
        if rejection_match and not decision.get("override_prior_rejection"):
            return f"decision {decision.get('decision_id')} reuses feedback evidence rejected after planning"
    return None


def _apply_plan_locked(
    plan_path: Path,
    plan: dict[str, Any],
    mode: str,
    decision_ids: set[str] | None,
    rejected_ids: set[str],
    commit: bool,
    state_dir: Path,
    accept_all: bool = False,
    authorization: dict[str, Any] | None = None,
) -> dict[str, Any]:
    decision_ids = resolve_review_selectors(plan, decision_ids)
    rejected_ids = resolve_review_selectors(plan, rejected_ids) or set()
    invalid = validate_application(plan, mode, decision_ids, rejected_ids, accept_all, authorization)
    if invalid is None:
        invalid = validate_current_state(plan, state_dir, mode, decision_ids, accept_all)
    if invalid:
        return validation_error(mode, invalid)

    if mode == "suggest":
        applied_ids: set[str] = set()
        results = [
            {
                "path": item["path"],
                "status": "suggested",
                "diff": item.get("diff", ""),
                "base_sha256": item.get("base_sha256"),
                "new_sha256": item.get("new_sha256"),
                "agents_md_dirty": item.get("agents_md_dirty"),
            }
            for item in plan.get("targets", [])
        ]
        update_state(
            state_dir,
            plan_path,
            plan,
            mode,
            applied_ids,
            rejected_ids,
            set(),
            set(plan.get("record_ids", [])),
            results,
            authorization,
        )
        return {"mode": mode, "mutated_targets": 0, "results": results}

    allowed_roots = {
        path_key(Path(value)): Path(value).expanduser().resolve()
        for value in (authorization or {}).get("allowed_project_roots", [])
    }
    home = Path(str((authorization or {}).get("codex_home"))).expanduser().resolve()
    top_by_id = {str(item["decision_id"]): item for item in plan.get("decisions", [])}
    applied_ids: set[str] = set()
    failed_ids: set[str] = set()
    results: list[dict[str, Any]] = []
    selected_ids = decision_ids
    if mode == "confirmed-suggest" and accept_all:
        selected_ids = {
            item["decision_id"]
            for item in plan.get("decisions", [])
            if plan.get("edited_acceptance") is True or item.get("action") in MUTATING_ACTIONS
        }
    if plan.get("edited_acceptance") is True:
        for decision in plan.get("decisions", []):
            identifier = str(decision["decision_id"])
            if identifier in (selected_ids or set()) and identifier not in rejected_ids and decision.get("action") == "no-op":
                applied_ids.add(identifier)
                results.append(
                    {
                        "status": "no-op",
                        "decision_ids": [identifier],
                        "notes": [str(decision.get("explanation") or "edited instruction is already effective")],
                        "committed": False,
                        "reason": "not required",
                    }
                )
    for target in plan.get("targets", []):
        target_decisions = [top_by_id[str(identifier)] for identifier in target.get("decision_ids", [])]
        target_ids = {item["decision_id"] for item in select_decisions(target_decisions, selected_ids)}
        if mode == "confirmed-suggest" and not target_ids:
            continue
        path = Path(target["path"]).absolute()
        scope = target.get("scope")
        if scope == "global":
            expected = (home / "AGENTS.md").absolute()
            if lexical_path_key(path) != lexical_path_key(expected) or path.is_symlink():
                failed_ids.update(target_ids)
                results.append({"path": str(path), "status": "error", "decision_ids": sorted(target_ids), "error": "global target path mismatch"})
                continue
            project_root = None
        elif scope == "project":
            project_root_value = target.get("project_root")
            project_root = Path(project_root_value).resolve() if project_root_value else None
            if (
                project_root is None
                or not project_root.is_dir()
                or path_key(project_root) not in allowed_roots
                or lexical_path_key(path) != lexical_path_key(project_root / "AGENTS.md")
                or path.is_symlink()
            ):
                failed_ids.update(target_ids)
                results.append({"path": str(path), "status": "error", "decision_ids": sorted(target_ids), "error": "project root no longer exists, is outside trusted roots, or target is not root AGENTS.md"})
                continue
        else:
            failed_ids.update(target_ids)
            results.append({"path": str(path), "status": "error", "decision_ids": sorted(target_ids), "error": "unknown target scope"})
            continue

        decisions = select_decisions(target_decisions, selected_ids)
        decisions = [item for item in decisions if item.get("decision_id") not in rejected_ids]
        if not decisions:
            continue
        current_raw = b""
        current = ""
        updated = ""
        actual_diff = ""
        target_existed = path.exists()
        wrote_target = False
        try:
            current_raw, current = read_agents(path)
            effective_lines = normalized_effective_lines(current)
            if project_root is not None:
                global_path = home / "AGENTS.md"
                if global_path.exists():
                    try:
                        effective_lines.update(normalized_effective_lines(global_path.read_text(encoding="utf-8-sig")))
                    except (OSError, UnicodeDecodeError) as error:
                        raise ReconcileError(f"cannot inspect effective global AGENTS.md: {error}") from error
            runtime_notes: list[str] = []
            for decision in decisions:
                if decision.get("action") != "add":
                    continue
                instruction = str(decision.get("instruction") or "")
                if normalize_text(instruction) in effective_lines:
                    decision["action"] = "no-op"
                    decision["explanation"] = "an equivalent instruction became effective after planning"
                    runtime_notes.append(f"{decision['decision_id']}: latest effective rule made add a no-op")
            updated, notes = apply_decisions(current, decisions)
            notes = [*runtime_notes, *notes]
            actual_diff = unified_diff(path, current, updated)
            if updated != current:
                atomic_write_bytes(path, encode_agents(updated, current_raw))
                wrote_target = True
            should_commit = commit and project_root is not None and wrote_target
            commit_result = commit_agents(project_root, path) if should_commit else {"committed": False, "reason": "not required"}
            ids = {item["decision_id"] for item in decisions}
            applied_ids.update(ids)
            results.append(
                {
                    "path": str(path),
                    "status": "applied" if updated != current else "no-op",
                    "decision_ids": sorted(ids),
                    "notes": notes,
                    "diff": actual_diff,
                    "base_sha256": sha256_bytes(current_raw),
                    "new_sha256": sha256_bytes(encode_agents(updated, current_raw)),
                    **commit_result,
                }
            )
        except (OSError, ReconcileError, RuntimeError, subprocess.SubprocessError) as error:
            rollback_error: str | None = None
            if wrote_target:
                try:
                    if target_existed:
                        atomic_write_bytes(path, current_raw)
                    elif path.exists():
                        path.unlink()
                except OSError as rollback:
                    rollback_error = str(rollback)
            ids = {item["decision_id"] for item in decisions}
            failed_ids.update(ids)
            failure: dict[str, Any] = {"path": str(path), "status": "error", "decision_ids": sorted(ids), "error": str(error)}
            if rollback_error:
                failure["rollback_error"] = rollback_error
            if actual_diff:
                failure.update(
                    {
                        "diff": actual_diff,
                        "base_sha256": sha256_bytes(current_raw),
                        "new_sha256": sha256_bytes(encode_agents(updated, current_raw)),
                    }
                )
            results.append(failure)

    checkpoint_record_ids = set() if failed_ids else set(plan.get("record_ids", []))
    update_state(
        state_dir,
        plan_path,
        plan,
        mode,
        applied_ids,
        rejected_ids,
        failed_ids,
        checkpoint_record_ids,
        results,
        authorization,
    )
    return {"mode": mode, "mutated_targets": sum(item.get("status") == "applied" for item in results), "results": results}


def apply_plan(
    plan_path: Path,
    plan: dict[str, Any],
    mode: str,
    decision_ids: set[str] | None,
    rejected_ids: set[str],
    commit: bool | None,
    state_dir: Path,
    accept_all: bool = False,
    authorization: dict[str, Any] | None = None,
) -> dict[str, Any]:
    try:
        with optimizer_lock(state_dir / ".apply.lock"):
            if commit is None:
                state = load_json(state_dir / "state.json", {})
                preferences = state.get("preferences", {}) if isinstance(state.get("preferences"), dict) else {}
                commit = mode == "confirmed-suggest" and preferences.get("confirmed_suggest_git_commit") is True
            return _apply_plan_locked(
                plan_path,
                plan,
                mode,
                decision_ids,
                rejected_ids,
                commit,
                state_dir,
                accept_all,
                authorization,
            )
    except (OSError, RuntimeError) as error:
        return validation_error(mode, str(error))


def main() -> int:
    parser = argparse.ArgumentParser(description="Record suggestions or safely apply a reconciled AGENTS.md update plan.")
    parser.add_argument("--plan")
    parser.add_argument("--batch", help="Independent trusted batch used to authorize Codex home and project roots.")
    parser.add_argument("--mode", choices=("suggest", "auto", "confirmed-suggest"))
    parser.add_argument(
        "--selection",
        "--decision-id",
        action="append",
        dest="decision_ids",
        help="Accept a visible Suggest number; internal decision IDs remain supported for compatibility.",
    )
    parser.add_argument("--all", action="store_true", dest="accept_all", help="Explicitly accept every mutating decision in a Suggest plan.")
    parser.add_argument(
        "--reject-selection",
        "--reject-id",
        action="append",
        dest="reject_id",
        default=[],
        help="Reject a visible Suggest number; internal decision IDs remain supported for compatibility.",
    )
    parser.add_argument("--state-dir")
    parser.add_argument(
        "--set-confirmed-suggest-commit",
        choices=("yes", "no"),
        help="Persist whether accepted Suggest changes should commit only the target AGENTS.md in Git projects.",
    )
    parser.add_argument("--model")
    parser.add_argument("--reasoning-effort", choices=("low", "medium", "high"))
    parser.add_argument("--output")
    args = parser.parse_args()
    if args.set_confirmed_suggest_commit:
        if args.plan or args.batch:
            parser.error("configure the confirmed-Suggest commit preference separately from applying a plan")
        home = codex_home()
        state_dir = Path(args.state_dir).expanduser().resolve() if args.state_dir else home / STATE_DIR_NAME
        result = set_confirmed_suggest_commit_preference(
            state_dir, args.set_confirmed_suggest_commit == "yes"
        )
        if args.output:
            atomic_write_json(Path(args.output).expanduser().resolve(), result)
        else:
            print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0
    if not args.plan or not args.batch:
        parser.error("--plan and --batch are required when applying or recording a plan")
    plan_path = Path(args.plan).expanduser().resolve()
    plan = load_json(plan_path, {})
    authorization = load_json(Path(args.batch).expanduser().resolve(), {})
    if args.model or args.reasoning_effort:
        metadata = plan.setdefault("run_metadata", {})
        if args.model:
            metadata["model"] = args.model
        if args.reasoning_effort:
            metadata["reasoning_effort"] = args.reasoning_effort
    mode = args.mode or plan.get("mode") or "suggest"
    home = codex_home(authorization.get("codex_home"))
    state_dir = Path(args.state_dir).expanduser().resolve() if args.state_dir else home / STATE_DIR_NAME
    commit = True if mode == "auto" else None if mode == "confirmed-suggest" else False
    result = apply_plan(
        plan_path,
        plan,
        mode,
        set(args.decision_ids) if args.decision_ids else None,
        set(args.reject_id),
        commit,
        state_dir,
        args.accept_all,
        authorization,
    )
    if args.output:
        atomic_write_json(Path(args.output).expanduser().resolve(), result)
    else:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if all(item.get("status") != "error" for item in result["results"]) else 2


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