ops-warden/tests/test_case_run.py

"""Injected owner harness; these receipts are not native owner attestations."""

from dataclasses import asdict, replace
import json
import os
import subprocess
import sys
import sqlite3
import pytest
import yaml
from warden.case_receipt_state import Outcome, Refused
from warden.case_run import Adapter, Descriptor, Executor, Journal, OwnerResult, digest


def lane_entry(lane="issuer-lane", owner="owner", **over):
    """Fixture catalog entry (TEST ONLY; never a real routing pointer)."""
    entry = {
        "id": lane,
        "title": "fixture lane",
        "need_keywords": [lane],
        "owner_repo": owner,
        "subsystem": "fixture",
        "warden_executes": False,
        "wiki_ref": "wiki/fixture.md",
        "canon_ref": "fixture/canon.md",
        "reviewed": "2026-10-17",
        "status": "active",
        "risk": "high",
        "workload_ref": {"applicability": "not-applicable", "reason": "fixture"},
        "delegation": {
            "mode": "native",
            "intended_owner": owner,
            "reviewed": "2026-10-17",
        },
    }
    entry.update(over)
    return entry


def fixture_catalog(*entries):
    return yaml.safe_dump({"entries": list(entries) or [lane_entry()]}).encode()


CATALOG = fixture_catalog()

MINT_STEP = {
    "id": "mint-secret",
    "owner": "owner",
    "lane": "issuer-lane",
    "warden_executes": True,
    "attended": False,
}


def make_rig(tmp_path):
    desc = Descriptor(
        "warden.case.execution.v1",
        "mint-secret",
        "owner",
        "owner/mint@v1",
        "sha256:" + "1" * 64,
        "human",
        "issuer-lane",
        "client_id",
        "mint",
        ("case-policy", "case-role"),
        True,
        ("parent-session", "broker-grant"),
    )
    descriptor = json.dumps(asdict(desc)).encode()
    typ = yaml.safe_dump(
        {
            "spec": {
                "id": "test",
                "version": "1",
                "status": "acknowledged",
                "parameters": [
                    {
                        "name": "client_id",
                        "sensitivity": "operational",
                        "case_settable": True,
                    }
                ],
                "execution_descriptors": {"mint-secret": digest(descriptor)},
                "follow_through": [dict(MINT_STEP)],
            }
        }
    ).encode()
    case = yaml.safe_dump(
        {
            "spec": {
                "id": "case-test",
                "type": "test@1",
                "status": "materialized",
                "acceptance": "acceptance",
                "parameters": {"client_id": "fixture-client"},
            }
        }
    ).encode()
    calls = []

    def invoke(inv):
        calls.append(inv)
        return bound(
            inv,
            OwnerResult(
                Outcome.SUCCESS,
                "owner-receipt",
                ("case-policy", "case-role"),
                True,
                session_revocation_evidence="session-revoked",
                grant_revocation_evidence="grant-revoked",
            ),
        )

    adapter = Adapter(
        "owner/mint@v1",
        desc.adapter_source,
        invoke,
        lambda inv, result: result.evidence == "owner-receipt",
        lambda inv, objects: bound(
            inv,
            OwnerResult(
                Outcome.ROLLED_BACK,
                "owner-receipt",
                (),
                True,
                session_revocation_evidence="session-revoked",
                grant_revocation_evidence="grant-revoked",
            ),
        ),
        lambda inv: bound(
            inv,
            OwnerResult(
                Outcome.NO_CHANGE,
                "owner-receipt",
                (),
                True,
                session_revocation_evidence="session-revoked",
                grant_revocation_evidence="grant-revoked",
            ),
        ),
    )
    verified = {
        ("case-acceptance", digest(case), "acceptance"),
        ("ops-mason-materialization", digest(case), "materialization"),
        ("owner-descriptor-acceptance:owner", digest(descriptor), "descriptor"),
        ("type-machine-seam-acceptance", digest(typ), "descriptor"),
        ("catalog-routing-acceptance", digest(CATALOG), "descriptor"),
        ("caller-session:human:operator:issuer-lane", digest(case), "session"),
        ("session-revocation:issuer-lane", digest(case), "session-revoked"),
        ("grant-revocation:issuer-lane", digest(case), "grant-revoked"),
    }

    def verifier(purpose, source, evidence):
        if purpose.startswith(("session-revocation:", "grant-revocation:")):
            scope, lane, operation, attempt = purpose.split(":")
            return (
                source == digest(case)
                and lane == "issuer-lane"
                and operation in ("run", "rollback", "reconcile")
                and evidence
                == ("session-" if scope == "session-revocation" else "grant-") + attempt
            )
        return (purpose, source, evidence) in verified

    executor = Executor(
        Journal(tmp_path / "private"), {"owner/mint@v1": adapter}, verifier
    )
    kwargs = dict(
        principal="operator",
        actor="human",
        lane="issuer-lane",
        acceptance_evidence="acceptance",
        materialization_evidence="materialization",
        descriptor_evidence="descriptor",
        session_evidence="session",
    )
    admitted = executor.admit(case, typ, descriptor, CATALOG, **kwargs)
    return (executor, admitted, calls, (case, typ, descriptor, CATALOG), kwargs)


@pytest.fixture
def rig(tmp_path):
    return make_rig(tmp_path)


def test_commit_running_is_visible_from_separate_connection_before_dispatch(rig):
    executor, args, calls, _, _ = rig
    old = executor.adapters[args[0].adapter]

    def inspect(inv):
        other = sqlite3.connect(executor.journal.directory / "journal.sqlite")
        payload = json.loads(
            other.execute(
                "SELECT payload FROM receipts ORDER BY seq DESC LIMIT 1"
            ).fetchone()[0]
        )
        assert payload["receipts"][-1]["outcome"] == "running"
        return old.invoke(inv)

    executor.adapters[old.reference] = replace(old, invoke=inspect)
    assert executor.run(*args).last.outcome is Outcome.SUCCESS
    with pytest.raises(Refused):
        executor.run(*args)
    assert len(calls) == 1


def test_crash_and_fresh_process_resume_never_duplicates_effect(rig):
    executor, args, calls, sources, kwargs = rig
    old = executor.adapters[args[0].adapter]

    def crash(inv):
        calls.append(inv)
        raise KeyboardInterrupt

    executor.adapters[old.reference] = replace(old, invoke=crash)
    with pytest.raises(KeyboardInterrupt):
        executor.run(*args)
    fresh = Executor(
        Journal(executor.journal.directory),
        {old.reference: old},
        executor.evidence_verifier,
    )
    admitted = fresh.admit(*sources, **kwargs)
    with pytest.raises(Refused):
        fresh.run(*admitted, retry=True)
    state = fresh.journal.load("mint-secret:fixture-client", admitted[2])
    assert state.last.outcome is Outcome.UNCERTAIN
    assert len(calls) == 1


def test_lost_result_preserved_and_cas_reconciliation_does_not_mint(rig):
    executor, args, calls, _, _ = rig
    old = executor.adapters[args[0].adapter]

    def lost(inv):
        calls.append(inv)
        raise TimeoutError("SECRET-HIDDEN")

    executor.adapters[old.reference] = replace(old, invoke=lost)
    assert executor.run(*args).last.outcome is Outcome.UNCERTAIN
    assert executor.reconcile(*args).last.outcome is Outcome.NO_CHANGE
    with pytest.raises(Refused):
        executor.run(*args, retry=True)
    assert len(calls) == 1
    assert (
        b"SECRET-HIDDEN"
        not in (executor.journal.directory / "journal.sqlite").read_bytes()
    )


def test_failed_before_change_requires_explicit_retry_and_safe_owner(rig):
    executor, args, calls, _, _ = rig
    old = executor.adapters[args[0].adapter]

    def no_change(inv):
        calls.append(inv)
        return bound(
            inv,
            OwnerResult(
                Outcome.FAILED_BEFORE,
                "owner-receipt",
                (),
                True,
                session_revocation_evidence="session-revoked",
                grant_revocation_evidence="grant-revoked",
            ),
        )

    executor.adapters[old.reference] = replace(old, invoke=no_change)
    assert executor.run(*args).last.outcome is Outcome.FAILED_BEFORE
    with pytest.raises(Refused):
        executor.run(*args)
    assert executor.run(*args, retry=True).last.outcome is Outcome.FAILED_BEFORE
    assert len(calls) == 2


@pytest.mark.parametrize(
    "result",
    [
        OwnerResult(
            Outcome.SUCCESS,
            "owner-receipt",
            ("shared-policy",),
            True,
            session_revocation_evidence="session-revoked",
            grant_revocation_evidence="grant-revoked",
        ),
        OwnerResult(
            Outcome.SUCCESS,
            "owner-receipt",
            (),
            False,
            session_revocation_evidence="session-revoked",
            grant_revocation_evidence="grant-revoked",
        ),
        OwnerResult(
            Outcome.SUCCESS,
            "unknown-receipt",
            (),
            True,
            session_revocation_evidence="session-revoked",
            grant_revocation_evidence="grant-revoked",
        ),
        {"outcome": "verified-success", "stdout": "SECRET-HIDDEN"},
    ],
)
def test_unknown_secret_payload_created_scope_or_missing_revocation_never_succeeds(
    rig, result, capsys
):
    executor, args, _, _, _ = rig
    old = executor.adapters[args[0].adapter]

    def malicious(inv):
        print("SECRET-HIDDEN" * 1000)
        return result

    executor.adapters[old.reference] = replace(old, invoke=malicious)
    assert executor.run(*args).last.outcome is Outcome.UNCERTAIN
    assert capsys.readouterr().out == ""
    assert (
        b"SECRET-HIDDEN"
        not in (executor.journal.directory / "journal.sqlite").read_bytes()
    )


def test_rollback_reverses_only_verified_case_created_objects(rig):
    executor, args, _, _, _ = rig
    old = executor.adapters[args[0].adapter]
    undos = []

    def undo(inv, objects):
        undos.append(objects)
        return old.rollback(inv, objects)

    executor.adapters[old.reference] = replace(old, rollback=undo)
    executor.run(*args)
    assert executor.rollback(*args).last.outcome is Outcome.ROLLED_BACK
    assert undos == [("case-role", "case-policy")]
    assert "shared-policy" not in undos[0]


def test_uncertain_rollback_cannot_replay(rig):
    executor, args, _, _, _ = rig
    old = executor.adapters[args[0].adapter]
    executor.run(*args)
    executor.adapters[old.reference] = replace(
        old,
        rollback=lambda inv, objs: bound(
            inv,
            OwnerResult(
                Outcome.ROLLED_BACK,
                "owner-receipt",
                (),
                False,
                session_revocation_evidence="session-revoked",
                grant_revocation_evidence="grant-revoked",
            ),
        ),
    )
    assert executor.rollback(*args).last.outcome is Outcome.ROLLBACK_UNCERTAIN
    with pytest.raises(Refused):
        executor.rollback(*args)


@pytest.mark.parametrize(
    "change", ["materialization", "type-prose", "actor", "lane", "source", "target"]
)
def test_exact_admission_requires_owner_evidence_and_closed_source_pins(rig, change):
    executor, _, _, sources, kwargs = rig
    case, typ, descriptor, catalog = sources
    kwargs = dict(kwargs)
    if change == "materialization":
        kwargs["materialization_evidence"] = "bare-json"
    if change == "type-prose":
        obj = yaml.safe_load(typ)
        obj["spec"].pop("execution_descriptors")
        typ = yaml.safe_dump(obj).encode()
    if change == "actor":
        kwargs["actor"] = "agent"
    if change == "lane":
        kwargs["lane"] = "other-lane"
    if change == "source":
        obj = json.loads(descriptor)
        obj["adapter_source"] = "sha256:" + "2" * 64
        descriptor = json.dumps(obj).encode()
    if change == "target":
        obj = yaml.safe_load(case)
        obj["spec"]["parameters"]["client_id"] = "other-target"
        case = yaml.safe_dump(obj).encode()
    with pytest.raises(Refused):
        executor.admit(case, typ, descriptor, catalog, **kwargs)


def test_agent_env_cannot_dispatch_after_human_admission(rig, monkeypatch):
    executor, args, calls, _, _ = rig
    monkeypatch.setenv("WARDEN_AGENT_ID", "test-agent")
    with pytest.raises(Refused):
        executor.run(*args)
    assert calls == []


def test_forged_invocation_cannot_bypass_admission(rig):
    executor, args, calls, _, _ = rig
    with pytest.raises(Refused):
        executor.run(args[0], replace(args[1], target="other-target"), args[2])
    assert calls == []


def test_journal_rejects_unsafe_directory_and_symlink(tmp_path):
    unsafe = tmp_path / "unsafe"
    unsafe.mkdir(mode=493)
    with pytest.raises(Refused):
        Journal(unsafe)
    private = tmp_path / "private"
    private.mkdir(mode=448)
    (private / "journal.sqlite").symlink_to(tmp_path / "outside")
    with pytest.raises(OSError):
        Journal(private)


def test_process_lock_blocks_parallel_dispatch(rig):
    executor, args, calls, _, _ = rig
    with executor.journal.locked():
        with pytest.raises(Refused):
            executor.run(*args)
    assert calls == []


def test_changed_catalog_pins_cannot_inherit_old_success(rig):
    executor, args, calls, sources, kwargs = rig
    executor.run(*args)
    with pytest.raises(Refused):
        executor.admit(*sources[:3], fixture_catalog(lane_entry(title="changed")), **kwargs)
    assert len(calls) == 1


def test_private_journal_does_not_retain_owner_payload_or_accept_unknown_schema(rig):
    executor, args, _, _, _ = rig
    executor.run(*args)
    row = executor.journal.db.execute(
        "SELECT seq,payload FROM receipts ORDER BY seq DESC LIMIT 1"
    ).fetchone()
    payload = json.loads(row[1])
    assert payload["journal_schema"] == "warden.case.step-journal.v1"
    assert payload["evidence_digest"] == digest(b"owner-receipt")
    assert payload["session_revoked"] is True
    assert "owner-receipt" not in row[1]
    payload["stdout"] = "UNTRUSTED-OUTPUT"
    executor.journal.db.execute(
        "UPDATE receipts SET payload=? WHERE seq=?", (json.dumps(payload), row[0])
    )
    with pytest.raises(Refused):
        executor.run(*args)


def test_expired_caller_session_cannot_use_cached_admission(rig):
    executor, args, calls, _, _ = rig
    old = executor.evidence_verifier
    executor.evidence_verifier = lambda purpose, source, evidence: (
        False
        if purpose.startswith("caller-session:")
        else old(purpose, source, evidence)
    )
    with pytest.raises(Refused):
        executor.run(*args)
    assert calls == []


@pytest.mark.parametrize("index", [0, 1, 2, 3])
def test_source_sizes_are_bounded_before_parsing(rig, index):
    executor, _, _, sources, kwargs = rig
    oversized = list(sources)
    oversized[index] = b"x" * 262145
    with pytest.raises(Refused, match="bounded"):
        executor.admit(*oversized, **kwargs)


def test_owner_fd_and_inherited_subprocess_output_is_contained(rig, capfd):
    executor, args, _, _, _ = rig
    old = executor.adapters[args[0].adapter]

    def leaking(inv):
        os.write(1, b"SYNTHETIC-SECRET")
        os.write(2, b"SYNTHETIC-SECRET")
        subprocess.run([sys.executable, "-c", 'print("SYNTHETIC-SECRET")'], check=True)
        return old.invoke(inv)

    executor.adapters[old.reference] = replace(old, invoke=leaking)
    assert executor.run(*args).last.outcome is Outcome.SUCCESS
    observed = capfd.readouterr()
    assert "SYNTHETIC-SECRET" not in observed.out + observed.err


def test_evidence_provider_output_and_error_never_escape(rig, capfd):
    executor, _, calls, sources, kwargs = rig

    def broken(*args):
        print("SYNTHETIC-SECRET")
        os.write(2, b"SYNTHETIC-SECRET")
        raise RuntimeError("SYNTHETIC-SECRET")

    executor.evidence_verifier = broken
    with pytest.raises(Refused) as caught:
        executor.admit(*sources, **kwargs)
    assert "SYNTHETIC-SECRET" not in str(caught.value)
    observed = capfd.readouterr()
    assert "SYNTHETIC-SECRET" not in observed.out + observed.err
    assert calls == []


@pytest.mark.parametrize("missing", ["session", "grant"])
def test_broker_grant_and_parent_session_have_separate_required_evidence(rig, missing):
    executor, args, _, _, _ = rig
    old = executor.adapters[args[0].adapter]
    result = old.invoke(args[1])
    if missing == "session":
        result = replace(result, session_revocation_evidence=None)
    else:
        result = replace(result, grant_revocation_evidence=None)
    executor.adapters[old.reference] = replace(old, invoke=lambda inv: result)
    assert executor.run(*args).last.outcome is Outcome.UNCERTAIN


def test_mint_descriptor_cannot_omit_broker_grant_scope(rig):
    descriptor = json.loads(rig[3][2])
    descriptor["revocation_scopes"] = ["parent-session"]
    with pytest.raises(Refused):
        Descriptor.decode(json.dumps(descriptor).encode())


def test_unsafe_owner_step_cannot_retry_even_after_verified_failed_before(
    rig, tmp_path
):
    executor, args, _, sources, kwargs = rig
    descriptor = replace(args[0], retry_safe=False)
    descriptor_source = json.dumps(asdict(descriptor)).encode()
    type_doc = yaml.safe_load(sources[1])
    type_doc["spec"]["execution_descriptors"][descriptor.step] = digest(
        descriptor_source
    )
    type_source = yaml.safe_dump(type_doc).encode()
    old_verifier = executor.evidence_verifier

    def verifier(purpose, source, evidence):
        if purpose == "owner-descriptor-acceptance:owner":
            return source == digest(descriptor_source) and evidence == "descriptor"
        if purpose == "type-machine-seam-acceptance":
            return source == digest(type_source) and evidence == "descriptor"
        return old_verifier(purpose, source, evidence)

    adapter = executor.adapters[descriptor.adapter]

    def failed(inv):
        return bound(
            inv,
            OwnerResult(
                Outcome.FAILED_BEFORE,
                "owner-receipt",
                (),
                True,
                "session-revoked",
                "grant-revoked",
            ),
        )

    fresh = Executor(
        Journal(tmp_path / "unsafe-retry"),
        {adapter.reference: replace(adapter, invoke=failed)},
        verifier,
    )
    admitted = fresh.admit(
        sources[0], type_source, descriptor_source, sources[3], **kwargs
    )
    assert fresh.run(*admitted).last.outcome is Outcome.FAILED_BEFORE
    with pytest.raises(Refused):
        fresh.run(*admitted, retry=True)


def bound(inv, result):
    return replace(
        result,
        attempt_id=inv.attempt_id,
        operation=inv.operation,
        session_revocation_evidence=("session-" + inv.attempt_id)
        if result.session_revocation_evidence
        else None,
        grant_revocation_evidence=("grant-" + inv.attempt_id)
        if result.grant_revocation_evidence
        else None,
    )


def test_old_grant_and_session_proofs_cannot_verify_a_new_retry(rig):
    executor, args, _, _, _ = rig
    adapter = executor.adapters[args[0].adapter]
    retained = []

    def attempts(inv):
        if not retained:
            result = bound(
                inv,
                OwnerResult(
                    Outcome.FAILED_BEFORE,
                    "owner-receipt",
                    (),
                    True,
                    "session-revoked",
                    "grant-revoked",
                ),
            )
            retained.append(result)
            return result
        # Even a caller rewriting labels cannot authenticate prior revocation
        # evidence for the newly persisted attempt identity.
        return replace(retained[0], attempt_id=inv.attempt_id, operation=inv.operation)

    executor.adapters[adapter.reference] = replace(adapter, invoke=attempts)
    assert executor.run(*args).last.outcome is Outcome.FAILED_BEFORE
    assert executor.run(*args, retry=True).last.outcome is Outcome.UNCERTAIN


def test_owner_interrupt_payload_is_sanitized_with_running_intent_retained(rig, capfd):
    executor, args, _, _, _ = rig
    adapter = executor.adapters[args[0].adapter]

    def interrupted(inv):
        os.write(2, b"SYNTHETIC-SECRET")
        raise KeyboardInterrupt("SYNTHETIC-SECRET")

    executor.adapters[adapter.reference] = replace(adapter, invoke=interrupted)
    with pytest.raises(KeyboardInterrupt) as caught:
        executor.run(*args)
    assert "SYNTHETIC-SECRET" not in str(caught.value)
    assert "SYNTHETIC-SECRET" not in "".join(capfd.readouterr())
    assert (
        executor.journal.load("mint-secret:fixture-client", args[2]).last.outcome
        is Outcome.RUNNING
    )


def test_attempt_identity_changes_with_exact_catalog_revision_and_new_journal(
    rig, tmp_path
):
    executor, args, calls, sources, kwargs = rig
    executor.run(*args)
    original_attempt = calls[-1].attempt_id
    new_catalog = fixture_catalog(lane_entry(title="independently reviewed v2"))
    old_verifier = executor.evidence_verifier

    def verifier(purpose, source, evidence):
        if purpose == "catalog-routing-acceptance":
            return source == digest(new_catalog) and evidence == "descriptor"
        return old_verifier(purpose, source, evidence)

    fresh = Executor(Journal(tmp_path / "new-revision"), executor.adapters, verifier)
    revised = fresh.admit(*sources[:3], new_catalog, **kwargs)
    assert fresh.run(*revised).last.outcome is Outcome.SUCCESS
    assert calls[-1].attempt_id != original_attempt
    assert calls[-1].pins.catalog != calls[0].pins.catalog


def test_secret_sensitivity_parameter_cannot_reach_admission_or_journal(rig):
    executor, _, _, sources, kwargs = rig
    type_doc = yaml.safe_load(sources[1])
    type_doc["spec"]["parameters"][0]["sensitivity"] = "secret"
    source = yaml.safe_dump(type_doc).encode()
    old = executor.evidence_verifier
    executor.evidence_verifier = lambda purpose, digest_value, evidence: (
        True
        if purpose == "type-machine-seam-acceptance"
        else old(purpose, digest_value, evidence)
    )
    with pytest.raises(Refused, match="secret-sensitivity"):
        executor.admit(sources[0], source, sources[2], sources[3], **kwargs)


def _lose_result(executor, args, calls):
    old = executor.adapters[args[0].adapter]

    def lost(inv):
        calls.append(inv)
        raise TimeoutError("lost")

    seen = []

    def by_original(ctx):
        seen.append(ctx)
        return old.reconcile(ctx.attempt)

    executor.adapters[old.reference] = replace(
        old, invoke=lost, reconcile=None, reconcile_original=by_original
    )
    assert executor.run(*args).last.outcome is Outcome.UNCERTAIN
    return seen


def test_original_intent_survives_restart_separate_from_reconcile_attempt(rig):
    executor, args, calls, _, _ = rig
    seen = _lose_result(executor, args, calls)
    executor.journal = Journal(executor.journal.directory)  # restart: fresh handle
    assert executor.reconcile(*args).last.outcome is Outcome.NO_CHANGE
    (ctx,) = seen
    original = ctx.original.invocation
    assert original.attempt_id == calls[0].attempt_id
    assert original.operation == "run" and ctx.attempt.operation == "reconcile"
    assert ctx.attempt.attempt_id != original.attempt_id
    assert original.pins == args[1].pins
    assert (original.principal, original.tenant, original.realm) == (
        args[1].principal,
        args[1].tenant,
        args[1].realm,
    )
    assert ctx.original.source == executor.adapters[args[0].adapter].source


def test_missing_original_intent_refuses_and_stays_uncertain(rig):
    executor, args, calls, _, _ = rig
    seen = _lose_result(executor, args, calls)
    executor.journal.db.execute("DELETE FROM intents")
    with pytest.raises(Refused, match="original intent record missing"):
        executor.reconcile(*args)
    assert seen == []
    binding = f"{args[1].step}:{args[1].target}"
    assert executor.journal.load(binding, args[2]).last.outcome is Outcome.UNCERTAIN


@pytest.mark.parametrize(
    "change", [{"tenant": "other-tenant"}, {"realm": "other-realm"}, {"lane": "x"}]
)
def test_original_intent_mismatch_refuses(rig, change):
    executor, args, calls, _, _ = rig
    seen = _lose_result(executor, args, calls)
    with pytest.raises(Refused):
        executor.reconcile(args[0], replace(args[1], **change), args[2])
    assert seen == []


def test_original_intent_source_mismatch_refuses(rig):
    executor, args, calls, _, _ = rig
    _lose_result(executor, args, calls)
    binding = f"{args[1].step}:{args[1].target}"
    with pytest.raises(Refused, match="source mismatch"):
        executor.journal.original_intent(binding, calls[0], "other-source")