script-build-and-framework-tmp/script_build_host/tests/test_architecture_failure_journal.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest

from script_build_host.application.failure_policy import classify_script_failure
from script_build_host.application.mission_failure_flow import MissionFailureFlow
from script_build_host.application.mission_failure_resolver import MissionFailureResolver
from script_build_host.application.recovery_planner import (
    MissionRecoveryPlanner,
    RecoveryCommandKind,
)
from script_build_host.domain.errors import ProtocolViolation, ScriptBuildError
from script_build_host.domain.failure_journal import (
    FailureEventType,
    FailureSubject,
    MissionFailureEvent,
    RecoveryAction,
)


def test_outer_protocol_wrapper_does_not_replace_context_root_cause() -> None:
    inner = MissionFailureEvent.observed(
        script_build_id=1,
        root_trace_id="root",
        code="CONTEXT_BUDGET_EXCEEDED",
        summary="validator prompt is too large",
        subject=FailureSubject("task", "attempt", "validation"),
        depth=3,
    )
    outer = MissionFailureEvent(
        "outer",
        1,
        "root",
        FailureEventType.OBSERVED,
        "PROTOCOL_VIOLATION",
        "phase boundary not reached",
        depth=0,
        cause_event_key=inner.event_key,
        occurred_at=datetime.now(UTC),
    )
    resolved = MissionFailureResolver().resolve((inner, outer))
    assert resolved.primary_cause.code == "CONTEXT_BUDGET_EXCEEDED"


def test_failure_resolver_rejects_unknown_cause_link() -> None:
    event = MissionFailureEvent(
        "outer",
        1,
        "root",
        FailureEventType.OBSERVED,
        "PROTOCOL_VIOLATION",
        "wrapper",
        cause_event_key="missing",
    )
    with pytest.raises(ProtocolViolation, match="unknown"):
        MissionFailureResolver().resolve((event,))


def test_unknown_failure_code_fails_closed_without_guessing_prefix() -> None:
    failure = classify_script_failure(
        ScriptBuildError("TASK_MADE_UP", "unknown"),
        source_tool="tool",
    )
    assert failure.code == "INTERNAL_UNCLASSIFIED_FAILURE"
    assert failure.details["observed_code"] == "TASK_MADE_UP"


def test_revalidation_is_not_planned_without_snapshot_and_failed_validation() -> None:
    failure = MissionFailureEvent.observed(
        script_build_id=1,
        root_trace_id="root",
        code="VALIDATION_EVIDENCE_UNAUTHORIZED",
        summary="failed before validation was frozen",
        subject=FailureSubject("task", "attempt", None),
        depth=2,
    )
    task = SimpleNamespace(
        task_id="task",
        current_spec_version=1,
        status=SimpleNamespace(value="needs_replan"),
        validation_ids=[],
    )
    attempt = SimpleNamespace(
        attempt_id="attempt",
        task_id="task",
        spec_version=1,
        snapshot_id=None,
    )
    ledger = SimpleNamespace(
        root_task_id="root-task",
        tasks={
            "root-task": SimpleNamespace(status=SimpleNamespace(value="needs_replan")),
            "task": task,
        },
        attempts={"attempt": attempt},
        validations={},
        operations={},
    )
    command = MissionRecoveryPlanner().plan((failure,), ledger)
    assert command.kind is RecoveryCommandKind.MANUAL_RECONCILIATION


@pytest.mark.asyncio
async def test_new_passed_attempt_recovers_prior_validation_failure() -> None:
    now = datetime.now(UTC)

    class Repository:
        def __init__(self) -> None:
            self.events = ()

        async def append(self, events):
            self.events = events

    repository = Repository()
    attempts = {
        "attempt-old": SimpleNamespace(
            attempt_id="attempt-old",
            created_at=now.isoformat(),
            status=SimpleNamespace(value="submitted"),
            failure=None,
        ),
        "attempt-new": SimpleNamespace(
            attempt_id="attempt-new",
            created_at=(now + timedelta(seconds=2)).isoformat(),
            status=SimpleNamespace(value="submitted"),
            failure=None,
        ),
    }
    validations = {
        "validation-old": SimpleNamespace(
            validation_id="validation-old",
            attempt_id="attempt-old",
            created_at=(now + timedelta(seconds=1)).isoformat(),
            status=SimpleNamespace(value="error"),
            verdict=None,
            failure=SimpleNamespace(code="CONTEXT_BUDGET_EXCEEDED", message="too large"),
        ),
        "validation-new": SimpleNamespace(
            validation_id="validation-new",
            attempt_id="attempt-new",
            created_at=(now + timedelta(seconds=3)).isoformat(),
            status=SimpleNamespace(value="completed"),
            verdict=SimpleNamespace(value="passed"),
            failure=None,
        ),
    }
    task = SimpleNamespace(
        task_id="task",
        attempt_ids=list(attempts),
        validation_ids=list(validations),
        status=SimpleNamespace(value="completed"),
        updated_at=(now + timedelta(seconds=4)).isoformat(),
    )
    ledger = SimpleNamespace(
        tasks={"task": task},
        attempts=attempts,
        validations=validations,
    )
    events = await MissionFailureFlow(repository).sync_ledger(
        script_build_id=1,
        root_trace_id="root",
        ledger=ledger,
    )
    recovery = next(item for item in events if item.event_type is FailureEventType.RECOVERY)
    assert recovery.code == RecoveryAction.ATTEMPT_SUCCEEDED
    assert recovery.successor_attempt_id == "attempt-new"