audit-core/tests/test_run_receipt_findings.py

"""AUDIT-WP-0018-T03. A started run with no receipt is a stream finding.

The bounded executor emits ``execution.run.started`` once it has consumed its
approval, and ``execution.attribution.receipt`` when the run ends. A crash in
between leaves only the start in custody. The scope overlay declares the pair
(``closing_classes``) and its bound; ``GET /v1/stream-findings`` reports every
start whose receipt has not arrived by then. Audit-core does not infer what
the run did.
"""

import copy
import hashlib
import json
import os
import uuid
from datetime import datetime, timedelta, timezone

import pytest

from audit_core import stream_findings
from audit_core.interface import AuditEvent
from audit_core.senders import SenderIdentity, SenderRegistry
from audit_core.sqlite_backend import SQLiteAuditBackend
from audit_core.stream_findings import UNCLOSED_OPENING, evaluate_closing

from test_bounded_executor_sender import (  # noqa: E402  (tests/ is on sys.path)
    RUN_ID,
    app,  # noqa: F401  (fixture)
    post,
    receipt_event,
    started_event,
)
from test_stream_findings import invoke  # noqa: E402

SOURCE = "net-kingdom.bounded-executor"
STARTED = "execution.run.started"
RECEIPT = "execution.attribution.receipt"
PG_URL = os.environ.get("AUDIT_CORE_TEST_DATABASE_URL")


def _postgres_available() -> bool:
    if not PG_URL:
        return False
    try:
        import psycopg

        with psycopg.connect(PG_URL, connect_timeout=3):
            return True
    except Exception:
        return False


HAVE_PG = _postgres_available()


@pytest.fixture(params=["sqlite", "postgres"])
def backend(request, tmp_path):
    if request.param == "sqlite":
        yield SQLiteAuditBackend(str(tmp_path / "runs.db"))
        return
    if not HAVE_PG:
        pytest.skip("no PostgreSQL reachable (set AUDIT_CORE_TEST_DATABASE_URL)")
    from audit_core.postgres_backend import PostgresAuditBackend

    schema = f"runs_{uuid.uuid4().hex[:12]}"
    instance = PostgresAuditBackend(PG_URL, schema=schema)
    try:
        yield instance
    finally:
        with instance.pool.connection() as conn:
            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
        instance.close()


def store(backend, event_id, action, correlation_id, *, source=SOURCE, data=None):
    details = {"data": data or {"k": "v"}}
    if correlation_id is not None:
        details["correlation_id"] = correlation_id
    event = AuditEvent(
        event_id=event_id, source=source, action=action, resource="admission:a",
        outcome="recorded", tenant="platform", scope="tenant", details=details,
        observed_at="2026-11-03T10:00:00+00:00",
    )
    digest = hashlib.sha256(json.dumps(event.as_record(), sort_keys=True).encode()).hexdigest()
    backend.accept(event, digest)


EPOCH = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()


# --- backend query, both backends --------------------------------------------


def test_a_start_without_a_receipt_is_unclosed(backend):
    store(backend, "run:r1:started", STARTED, "r1")
    rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)
    assert [(r["event_id"], r["correlation_id"]) for r in rows] == [("run:r1:started", "r1")]
    assert rows[0]["accepted_at"]
    assert rows[0]["window_not_after"] is None


def test_a_receipt_on_the_same_correlation_closes_the_start(backend):
    store(backend, "run:r1:started", STARTED, "r1")
    store(backend, "run:r1:receipt", RECEIPT, "r1")
    store(backend, "run:r2:started", STARTED, "r2")
    rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)
    assert [r["correlation_id"] for r in rows] == ["r2"]


def test_a_receipt_from_another_source_does_not_close(backend):
    store(backend, "run:r1:started", STARTED, "r1")
    store(backend, "other:r1:receipt", RECEIPT, "r1", source="someone-else")
    rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)
    assert [r["correlation_id"] for r in rows] == ["r1"]


def test_a_start_without_a_correlation_id_can_never_close(backend):
    store(backend, "run:x:started", STARTED, None)
    store(backend, "run:x:receipt", RECEIPT, None)
    rows = backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)
    assert [r["event_id"] for r in rows] == ["run:x:started"]


def test_the_per_run_window_is_surfaced_and_a_malformed_one_is_not(backend):
    store(backend, "run:r1:started", STARTED, "r1",
          data={"window": {"not_after": "2026-11-03T12:00:00Z"}})
    store(backend, "run:r2:started", STARTED, "r2", data={"window": {"not_after": 5}})
    rows = {r["correlation_id"]: r for r in
            backend.unclosed_openings(SOURCE, STARTED, RECEIPT, EPOCH)}
    assert rows["r1"]["window_not_after"] == "2026-11-03T12:00:00Z"
    assert rows["r2"]["window_not_after"] is None


def test_only_starts_accepted_since_the_lookback_are_examined(backend):
    store(backend, "run:r1:started", STARTED, "r1")
    future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
    assert backend.unclosed_openings(SOURCE, STARTED, RECEIPT, future) == []


# --- evaluator ---------------------------------------------------------------

EXECUTOR = SenderIdentity(
    name="nk-bounded-executor", tokens=("t",), sources=frozenset({SOURCE}),
    tenants=frozenset({"platform"}), evidence_kind="load-bearing",
    closing_classes=((STARTED, RECEIPT, 7200),),
)
T0 = datetime(2026, 11, 3, 10, 0, tzinfo=timezone.utc)


def rows(**extra):
    row = {"event_id": "run:r1:started", "correlation_id": "r1",
           "accepted_at": T0.isoformat(), "window_not_after": None}
    row.update(extra)
    return lambda source, opening, closing, since: [row]


def test_no_finding_inside_the_bound_and_its_grace():
    # 2 h bound plus the 1.5x grace allowance (1 h): due at 13:00.
    assert evaluate_closing([EXECUTOR], rows(), now=T0 + timedelta(hours=2, minutes=59)) == []


def test_a_finding_past_the_bound_and_grace():
    findings = evaluate_closing([EXECUTOR], rows(), now=T0 + timedelta(hours=3, minutes=1))
    assert len(findings) == 1
    body = findings[0].as_dict()
    assert body["kind"] == UNCLOSED_OPENING
    assert body["class"] == STARTED and body["closed_by"] == RECEIPT
    assert body["correlation_id"] == "r1"
    assert body["bound"] == "declared"
    assert body["overdue_seconds"] == 60
    assert "does not infer what the work did" in body["means"]


def test_a_per_run_window_tightens_the_deadline():
    early = rows(window_not_after=(T0 + timedelta(minutes=30)).isoformat())
    findings = evaluate_closing([EXECUTOR], early, now=T0 + timedelta(hours=1, minutes=31))
    assert [f.bound for f in findings] == ["window.not_after"]


def test_a_per_run_window_never_extends_the_declared_bound():
    late = rows(window_not_after=(T0 + timedelta(days=1)).isoformat())
    findings = evaluate_closing([EXECUTOR], late, now=T0 + timedelta(hours=3, minutes=1))
    assert [f.bound for f in findings] == ["declared"]


def test_the_lookback_is_the_custody_window():
    seen = {}

    def spy(source, opening, closing, since):
        seen["since"] = since
        return []

    evaluate_closing([EXECUTOR], spy, now=T0)
    assert seen["since"] == (T0 - timedelta(days=30)).isoformat()


def test_a_wildcard_source_is_not_held_to_a_close():
    wild = SenderIdentity(
        name="w", tokens=("w",), sources=frozenset({"*"}),
        closing_classes=((STARTED, RECEIPT, 60),),
    )
    assert evaluate_closing([wild], rows(), now=T0 + timedelta(days=2)) == []


# --- declaration and overlay -------------------------------------------------


def _overlay(tmp_path, closing):
    scope = tmp_path / "scope.json"
    scope.write_text(json.dumps([{"name": "x", "closing_classes": closing}]))
    secret = json.dumps([{
        "name": "x", "tokens": ["x"], "sources": [SOURCE],
        "closing_classes": {STARTED: {"closed_by": RECEIPT, "within_seconds": 7200}},
    }])
    return SenderRegistry.from_env({
        "AUDIT_CORE_SENDERS": secret, "AUDIT_CORE_SENDERS_SCOPE_PATH": str(scope),
    })


def test_the_overlay_may_shorten_the_bound(tmp_path):
    reg = _overlay(tmp_path, {STARTED: {"closed_by": RECEIPT, "within_seconds": 3600}})
    assert reg.authenticate("Bearer x").closing_classes == ((STARTED, RECEIPT, 3600),)


@pytest.mark.parametrize("closing", [
    {STARTED: {"closed_by": RECEIPT, "within_seconds": 7201}},
    {STARTED: {"closed_by": "something.else", "within_seconds": 7200}},
    {"other.start": {"closed_by": RECEIPT, "within_seconds": 60}},
])
def test_the_overlay_may_not_lengthen_rebind_or_remove(tmp_path, closing):
    with pytest.raises(ValueError):
        _overlay(tmp_path, closing)


@pytest.mark.parametrize("closing", [
    {STARTED: {"closed_by": RECEIPT}},
    {STARTED: {"closed_by": RECEIPT, "within_seconds": 0}},
    {STARTED: {"closed_by": STARTED, "within_seconds": 60}},
    [STARTED, RECEIPT, 60],
])
def test_a_malformed_declaration_is_refused(closing):
    secret = json.dumps([{"name": "x", "tokens": ["x"], "sources": [SOURCE],
                          "closing_classes": closing}])
    with pytest.raises(ValueError):
        SenderRegistry.from_env({"AUDIT_CORE_SENDERS": secret})


# --- end to end on the executor's own envelopes and the Git overlay ----------


@pytest.fixture
def three_hours_later(monkeypatch):
    real = stream_findings.evaluate_closing
    later = datetime.now(timezone.utc) + timedelta(hours=3, minutes=1)
    monkeypatch.setattr(stream_findings, "evaluate_closing",
                        lambda ids, unclosed: real(ids, unclosed, now=later))


def _closing_findings(app):  # noqa: F811
    status, body = invoke(app, "/v1/stream-findings", token="operator")
    assert status.startswith("200")
    return [f for f in body["stream_findings"] if f["kind"] == UNCLOSED_OPENING]


def test_a_fresh_start_is_not_yet_a_finding(app):  # noqa: F811
    assert post(app, started_event())[0] == 202
    assert _closing_findings(app) == []


def test_a_start_with_no_receipt_past_the_bound_is_reported(app, three_hours_later):  # noqa: F811
    assert post(app, started_event())[0] == 202
    findings = _closing_findings(app)
    assert [(f["sender"], f["correlation_id"]) for f in findings] == [
        ("nk-bounded-executor", RUN_ID)
    ]
    assert findings[0]["event_id"] == f"run:{RUN_ID}:started"
    # Identifiers and timing only: nothing from the run's data is disclosed.
    assert "data" not in findings[0] and "details" not in findings[0]


def test_the_receipt_clears_the_finding(app, three_hours_later):  # noqa: F811
    assert post(app, started_event())[0] == 202
    assert post(app, receipt_event())[0] == 202
    assert _closing_findings(app) == []


def test_another_run_receipt_does_not_clear_this_one(app, three_hours_later):  # noqa: F811
    assert post(app, started_event())[0] == 202
    other = copy.deepcopy(receipt_event())
    other["id"] = "run:other:receipt"
    other["correlation_id"] = "other"
    other["data"]["run_id"] = "other"
    assert post(app, other)[0] == 202
    assert [f["correlation_id"] for f in _closing_findings(app)] == [RUN_ID]