atlas-iac/services/hermes/scripts/cli_lane_finalization.py

#!/usr/bin/env python3
"""Exact-run database completion and terminal evidence publication."""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

from cli_lane_board import _board_call, _external, _record_board_access_error, _task_value
from cli_lane_capabilities import kanban_capabilities
from cli_lane_config import (
    TerminalIdentity,
    TerminalRecoverySnapshot,
    TerminalSnapshot,
    canonical_run_id,
    utc_now,
)
from cli_lane_evidence import (
    _persist_conflict_evidence,
    _persist_prepared_evidence,
    _retire_terminal_entry,
    _same_terminal_document,
    _write_json_noreplace,
)
from cli_lane_files import _terminal_identity, _terminal_path, state_path
from cli_lane_records import (
    _load_terminal_json,
    _open_small_json_snapshot,
    _open_terminal_snapshot,
    _terminal_record_valid,
)


def _recover_exact_run(
    kanban_db: Any,
    identity: TerminalIdentity | None,
    reason: str,
) -> bool:
    """Make an exact external run retryable after journal recovery fails."""
    if identity is None or canonical_run_id(identity.run_id) is None:
        return False
    if not kanban_capabilities(kanban_db).exact_run_reclaim:
        return False

    def operation(conn: Any) -> bool:
        task = kanban_db.get_task(conn, identity.task_id)
        if task is None:
            return False
        if (
            str(_task_value(task, "status", "")) != "running"
            or _task_value(task, "current_run_id", None) != identity.run_id
            or not _external(task)
        ):
            return False
        return bool(
            kanban_db.reclaim_task(
                conn,
                identity.task_id,
                reason=f"terminal journal recovery failed ({reason}); exact run may retry",
                expected_run_id=identity.run_id,
            )
        )

    try:
        recovered = bool(_board_call(kanban_db, identity.board, operation))
    except Exception as error:
        _record_board_access_error(identity.board, error)
        return False
    if recovered:
        print(
            f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} "
            f"after terminal journal recovery: {reason}",
            file=sys.stderr,
            flush=True,
        )
    return recovered

def _retire_snapshot(
    path: Path,
    snapshot: TerminalSnapshot,
    *,
    authority_name: str | None = None,
) -> str:
    """Retire only the directory entry still naming an opened snapshot inode."""
    return _retire_terminal_entry(
        path,
        snapshot.file_stat,
        board_descriptor=snapshot.directory_descriptor,
        quarantine_descriptor=snapshot.directory_descriptor,
        authority_name=authority_name,
        source_descriptor=snapshot.descriptor,
    )

def _retire_snapshot_after_db(path: Path, snapshot: TerminalSnapshot) -> str:
    """Keep a committed DB result recoverable across retirement fsync errors."""
    try:
        return _retire_snapshot(path, snapshot)
    except OSError as error:
        print(
            f"terminal journal retirement deferred after DB commit: "
            f"{type(error).__name__}: {error}",
            file=sys.stderr,
            flush=True,
        )
        return "deferred"

def _discard_evidence(path: Path) -> str:
    """Identity-safely remove one immutable prepared evidence artifact."""
    snapshot = _open_small_json_snapshot(path)
    if snapshot is None:
        return "missing"
    try:
        return _retire_snapshot(path, snapshot)
    finally:
        snapshot.close()

def _promote_prepared_evidence(
    identity: TerminalIdentity,
    document: dict[str, Any],
    prepared: Path,
    *,
    discard_prepared: bool = True,
) -> tuple[Path, str]:
    """Publish the DB-winning result without overwriting first-writer evidence."""
    committed = _terminal_path(
        state_path(identity.board, identity.task_id),
        identity.run_id,
        "committed",
    )
    committed_document = dict(document)
    committed_document["kanban_state"] = "committed"
    committed_document["committed_at"] = utc_now()
    state = "committed"
    if not _write_json_noreplace(committed, committed_document):
        committed_identity = TerminalIdentity(
            identity.board,
            identity.task_id,
            identity.run_id,
            "committed",
        )
        existing = _load_terminal_json(committed, committed_identity)
        if (
            not _terminal_record_valid(existing, committed_identity)
            or existing.get("kanban_state") != "committed"
            or not _same_terminal_document(existing, committed_document)
        ):
            _persist_conflict_evidence(
                identity,
                document,
                "canonical committed evidence already contains a different result",
            )
            state = "conflict"
    if discard_prepared:
        _discard_evidence(prepared)
    return committed, state

def _finalize_document_db(
    kanban_db: Any,
    identity: TerminalIdentity,
    document: dict[str, Any],
) -> str:
    """Apply one validated exact-run result to Kanban under DB run guards."""
    if canonical_run_id(identity.run_id) is None:
        return "stale"

    def operation(conn: Any) -> str:
        task = kanban_db.get_task(conn, identity.task_id)
        if task is None:
            return "stale"
        status = str(_task_value(task, "status", ""))
        if status == "done":
            return (
                "committed"
                if (
                    _task_value(task, "completed_run_id", None) == identity.run_id
                    and str(_task_value(task, "result", "") or "")
                    == document["result"]
                )
                else "stale"
            )
        current_run_id = _task_value(task, "current_run_id", None)
        completion_guard: dict[str, int]
        if status == "running" and current_run_id == identity.run_id:
            if not kanban_capabilities(kanban_db).exact_run_completion:
                return "deferred"
            completion_guard = {"expected_run_id": identity.run_id}
        elif status in {"ready", "blocked", "triage"} and current_run_id is None:
            # The journal may have survived an older post-persistence error
            # path that ended its run. The patched DB verifies atomically that
            # this is still the latest ended run before allowing completion.
            if not kanban_capabilities(kanban_db).ended_run_replay:
                return "deferred"
            completion_guard = {"replay_ended_run_id": identity.run_id}
        else:
            return "stale"
        completed = kanban_db.complete_task(
            conn,
            identity.task_id,
            result=document["result"],
            summary=document["summary"],
            metadata=document["metadata"],
            **completion_guard,
        )
        if completed:
            return "committed"
        # A duplicate finalizer can lose the guarded UPDATE to an identical
        # first writer. Re-read rather than reporting a false pending state.
        latest = kanban_db.get_task(conn, identity.task_id)
        latest_status = (
            str(_task_value(latest, "status", "")) if latest is not None else ""
        )
        if latest_status == "done":
            return (
                "committed"
                if (
                    _task_value(latest, "completed_run_id", None) == identity.run_id
                    and str(_task_value(latest, "result", "") or "")
                    == document["result"]
                )
                else "stale"
            )
        # A guarded completion can be retried only while this exact live run
        # still owns the task. An ended, replaced, or nonexistent run cannot
        # become authoritative later and must converge to conflict evidence.
        if (
            latest_status == "running"
            and _task_value(latest, "current_run_id", None) == identity.run_id
        ):
            return "pending"
        return "stale"

    return str(_board_call(kanban_db, identity.board, operation))

def _resolve_pending_after_winner(
    path: Path,
    identity: TerminalIdentity,
    winning_document: dict[str, Any],
) -> bool:
    """Retire duplicates and preserve differing replacements as conflicts."""
    for _attempt in range(8):
        snapshot = _open_terminal_snapshot(path, identity)
        if snapshot is None:
            return False
        document = snapshot.document
        if not _terminal_record_valid(document, identity):
            snapshot.close()
            return False
        differing = not _same_terminal_document(document, winning_document)
        try:
            if differing:
                _persist_conflict_evidence(
                    identity,
                    document,
                    "different valid result lost the exact-run first-writer race",
                )
            retirement = _retire_snapshot_after_db(path, snapshot)
        finally:
            snapshot.close()
        if retirement in {"retired", "missing"}:
            if differing:
                print(
                    f"preserved terminal result conflict for {identity.board}/"
                    f"{identity.task_id} run {identity.run_id}",
                    file=sys.stderr,
                    flush=True,
                )
            return differing
        if retirement not in {"replacement", "replacement-staged", "collision"}:
            return differing
    return False

def _finalize_terminal_record(
    kanban_db: Any,
    path: Path,
    _record: dict[str, Any] | None = None,
    *,
    snapshot: TerminalRecoverySnapshot | None = None,
) -> str:
    """Commit one inode-pinned exact-run journal, or leave it safely pending."""
    identity = _terminal_identity(path)
    if identity is None or identity.state != "pending":
        if snapshot is not None:
            snapshot.close()
        return "invalid"
    pinned: TerminalSnapshot | TerminalRecoverySnapshot | None = snapshot
    if pinned is None:
        pinned = _open_terminal_snapshot(path, identity)
    if pinned is None or pinned.document is None:
        if pinned is not None:
            pinned.close()
        return "invalid"
    document = pinned.document
    try:
        if not _terminal_record_valid(document):
            return "invalid"
        if not _terminal_record_valid(document, identity):
            return "foreign"
        prepared = _persist_prepared_evidence(identity, document)
        outcome = _finalize_document_db(kanban_db, identity, document)
        if outcome == "committed":
            _committed, evidence_state = _promote_prepared_evidence(
                identity,
                document,
                prepared,
            )
            retirement = _retire_snapshot_after_db(path, pinned)
        elif outcome == "stale":
            _persist_conflict_evidence(
                identity,
                document,
                "valid terminal result no longer matches the authoritative task run",
            )
            _discard_evidence(prepared)
            retirement = _retire_snapshot_after_db(path, pinned)
            outcome = "conflict"
            evidence_state = "conflict"
        else:
            return outcome
    finally:
        pinned.close()
    if outcome == "committed" and retirement not in {"retired", "missing"}:
        _resolve_pending_after_winner(path, identity, document)
    if evidence_state == "conflict":
        print(
            f"terminal first-writer conflict recorded for {identity.board}/"
            f"{identity.task_id} run {identity.run_id}",
            file=sys.stderr,
            flush=True,
        )
    return outcome