informed-decision/tests/test_skeleton.py

"""Walking skeleton — the negative cases are as load-bearing as the happy path.

Each negative case here is one from `docs/specs/UseCaseCatalog.md` and protects
a named invariant. If one of these starts passing by doing the forbidden thing,
the surface has become something this repository says it must not be.
"""

from __future__ import annotations

import pytest

from informed_decision.approval_client import (
    ApprovalEngineError,
    FakeApprovalEngine,
    assert_scopes_permissible,
    is_success,
)
from informed_decision.disposition import (
    Actor,
    ActorKind,
    DispositionRefused,
    Verb,
    legal_verbs,
    record,
)
from informed_decision.evidence import (
    CustodyLocatorRejected,
    EventClass,
    Outbox,
    assert_custody_locator_safe,
    commit_disposition,
    commit_presentation,
    commit_stance_application,
    heartbeat,
)
from informed_decision.memo import (
    Awareness,
    BindingLevel,
    BindingSlice,
    Hat,
    Highlight,
    Memo,
    PacketItem,
    Principal,
    Scope,
    StepKind,
)
from informed_decision.presentation import render
from informed_decision.provenance import (
    Claim,
    HumanControlNotDischargeable,
    Route,
    assert_human_control_dischargeable,
)

CUSTODY = "informed-decision:presentations"
HUMAN = Actor("bernd", ActorKind.PERSON)
AGENT = Actor("drafter-bot", ActorKind.AGENT)


def make_memo(**over) -> Memo:
    kw = dict(
        id="memo-1",
        version=1,
        question="Approve rotation of the production database credential for T-1183?",
        requested_act="approve",
        binding_level=BindingLevel.ORGANIZATIONAL,
        brief="The credential is 400 days old.",
        binding=BindingSlice(
            principal=Principal(id="p-1", kind="person", display_name="Bernd Worsch"),
            target=Scope(kind="tenant", id="tenant:acme", label="ACME", environment="prod"),
        ),
        step_kind=StepKind.APPROVE,
        packet=(PacketItem("doc-1", "Change request", "sha256:" + "a" * 64),),
        highlights=(
            Highlight("h-1", "doc-1", "Target is production", required_ack=True, severity="critical"),
        ),
        approval_id="appr-1",
    )
    kw.update(over)
    return Memo(**kw)


# -------------------------------------------------------------------------
# Happy path
# -------------------------------------------------------------------------


def test_one_approval_end_to_end_in_process():
    memo = make_memo()
    engine = FakeApprovalEngine({"appr-1": {"status": "requested", "required_count": 1}})
    outbox = Outbox()

    pres = render(
        memo,
        principal_sub="bernd",
        tenant=Claim("tenant:platform", Route.REGISTRATION),
        principal_type=Claim("human", Route.REGISTRATION),
    )
    outbox.append(commit_presentation(pres, custody=CUSTODY))

    pres = pres.with_ack("h-1")
    disp = record(memo, pres, Verb.ACCEPT, HUMAN)
    outbox.append(commit_disposition(disp, custody=CUSTODY))

    assert disp.reaches_approval_engine
    result = engine.add_entry(memo.approval_id, "bernd")

    assert result.correlation == ("appr-1", "bernd", "2026-09-10T15:00:00Z")
    assert result.status == "approved"
    assert len(outbox.pending) == 2


def test_the_presentation_is_reachable_from_the_correlation_triple():
    """DoD-3 — satisfied by the triple, not by a hash on the entry."""
    memo = make_memo()
    pres = render(memo, principal_sub="bernd").with_ack("h-1")
    engine = FakeApprovalEngine({"appr-1": {"status": "requested", "required_count": 1}})
    result = engine.add_entry("appr-1", "bernd")
    assert pres.approval_id == result.approval_id
    assert pres.principal_sub == result.subject
    assert pres.view_hash


# -------------------------------------------------------------------------
# NC-01 — accept on a Kenntnisnahme step
# -------------------------------------------------------------------------


@pytest.mark.parametrize(
    "kind", [StepKind.INFORM, StepKind.COMMENT, StepKind.REVIEW, StepKind.ACKNOWLEDGE]
)
def test_accept_is_absent_from_weak_steps_not_merely_disabled(kind):
    assert Verb.ACCEPT not in legal_verbs(kind)
    assert Verb.ACKNOWLEDGE in legal_verbs(kind)


def test_accept_on_a_weak_step_is_refused_at_the_api(): 
    memo = make_memo(step_kind=StepKind.REVIEW, highlights=())
    pres = render(memo, principal_sub="bernd")
    with pytest.raises(DispositionRefused) as e:
        record(memo, pres, Verb.ACCEPT, HUMAN)
    assert e.value.guard == "G_STEP"
    assert "Kenntnisnahme is not approval" in str(e.value)


# -------------------------------------------------------------------------
# NC-02 — bind with unacknowledged required highlights
# -------------------------------------------------------------------------


def test_bind_without_required_ack_fails_closed_and_creates_nothing():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd")
    with pytest.raises(DispositionRefused) as e:
        record(memo, pres, Verb.ACCEPT, HUMAN)
    assert e.value.guard == "G_ACK"
    assert "h-1" in str(e.value)


def test_acknowledgment_is_explicit_never_inferred():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd")
    assert pres.acked_highlight_ids == frozenset()
    assert pres.with_ack("h-1").acked_highlight_ids == {"h-1"}


# -------------------------------------------------------------------------
# NC-03 — agents never bind. There is no upstream backstop.
# -------------------------------------------------------------------------


@pytest.mark.parametrize("verb", [Verb.ACCEPT, Verb.DECLINE, Verb.ACKNOWLEDGE])
def test_agent_cannot_perform_a_binding_verb(verb):
    memo = make_memo()
    pres = render(memo, principal_sub="bot").with_ack("h-1")
    with pytest.raises(DispositionRefused) as e:
        record(memo, pres, verb, AGENT)
    assert e.value.guard == "G_NOAGENT"


def test_agent_may_still_comment():
    memo = make_memo()
    pres = render(memo, principal_sub="bot")
    assert record(memo, pres, Verb.COMMENT, AGENT).verb is Verb.COMMENT


# -------------------------------------------------------------------------
# NC-05 / NC-06 — the binding/awareness split, on live objects
# -------------------------------------------------------------------------


def test_awareness_never_enters_view_hash():
    memo = make_memo()
    plain = render(memo, principal_sub="bernd")
    oriented = render(
        memo,
        principal_sub="bernd",
        awareness=Awareness(
            proposed_hat=Hat("hat:fc", "Finance Controller"),
            proposed_hat_source="last_used",
            situation_note="changed after the presentation was taken",
        ),
    )
    assert plain.view_hash == oriented.view_hash
    assert plain.awareness_hash != oriented.awareness_hash


def test_changing_the_act_scope_changes_view_hash():
    memo = make_memo()
    before = render(memo, principal_sub="bernd").view_hash
    moved = memo.next_version(
        binding=BindingSlice(
            principal=memo.binding.principal,
            target=Scope(kind="tenant", id="tenant:beta", label="Beta GmbH"),
        )
    )
    assert render(moved, principal_sub="bernd").view_hash != before


def test_the_tenant_claim_is_not_the_act_scope():
    """PR-08 — two facts, one field upstream; never conflated here."""
    memo = make_memo()
    a = render(memo, principal_sub="bernd", tenant=Claim("tenant:platform", Route.REGISTRATION))
    b = render(memo, principal_sub="bernd", tenant=Claim("tenant:coulomb", Route.DIRECTORY))
    assert a.view_hash == b.view_hash, "the tenant claim must not reach view_hash"
    assert memo.binding.target.id == "tenant:acme"


# -------------------------------------------------------------------------
# NC-07 — no silent version upgrade
# -------------------------------------------------------------------------


def test_bind_against_a_stale_presentation_is_refused():
    memo = make_memo()
    stale = render(memo, principal_sub="bernd").with_ack("h-1")
    advanced = memo.next_version(brief="revised")
    with pytest.raises(DispositionRefused) as e:
        record(advanced, stale, Verb.ACCEPT, HUMAN)
    assert e.value.guard == "G_PRES"


# -------------------------------------------------------------------------
# NC-08 — return is structured, and is not decline
# -------------------------------------------------------------------------


def test_return_without_coded_reasons_is_refused():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd")
    with pytest.raises(DispositionRefused) as e:
        record(memo, pres, Verb.RETURN, HUMAN, note="please clarify")
    assert e.value.guard == "G_REASONS"


def test_return_is_distinguishable_from_decline_and_never_reaches_the_engine():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd").with_ack("h-1")
    returned = record(memo, pres, Verb.RETURN, HUMAN, reasons=("insufficient-context",))
    declined = record(memo, pres, Verb.DECLINE, HUMAN)
    assert returned.verb is not declined.verb
    assert not returned.reaches_approval_engine
    assert not declined.reaches_approval_engine


def test_only_accept_reaches_the_engine():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd").with_ack("h-1")
    assert record(memo, pres, Verb.ACCEPT, HUMAN).reaches_approval_engine
    for verb in (Verb.COMMENT, Verb.DISCUSS, Verb.FORWARD, Verb.ESCALATE):
        assert not record(memo, pres, verb, HUMAN).reaches_approval_engine


# -------------------------------------------------------------------------
# Engine semantics
# -------------------------------------------------------------------------


def test_duplicate_approver_is_success_not_failure():
    assert is_success(ApprovalEngineError(409, "duplicate_approver"))
    assert not is_success(ApprovalEngineError(409, "conflict"))
    assert not is_success(ApprovalEngineError(503, "store_unavailable"))


def test_consume_scope_is_refused_before_a_token_is_requested():
    with pytest.raises(ValueError, match="approval:consume"):
        assert_scopes_permissible(("openid", "approval:read", "approval:consume"))


def test_store_unavailable_fails_closed():
    engine = FakeApprovalEngine({"appr-1": {"status": "requested"}})
    engine.available = False
    with pytest.raises(ApprovalEngineError) as e:
        engine.add_entry("appr-1", "bernd")
    assert e.value.status == 503


def test_terminal_approval_conflicts():
    engine = FakeApprovalEngine({"appr-1": {"status": "revoked"}})
    with pytest.raises(ApprovalEngineError) as e:
        engine.add_entry("appr-1", "bernd")
    assert (e.value.status, e.value.reason) == (409, "conflict")


# -------------------------------------------------------------------------
# GH-DEC-2026-016 §5 / PR-11 — humanity provenance
# -------------------------------------------------------------------------


def test_human_control_not_dischargeable_on_a_registration_supplied_claim():
    with pytest.raises(HumanControlNotDischargeable, match="registration-supplied"):
        assert_human_control_dischargeable(Claim("human", Route.REGISTRATION))


@pytest.mark.parametrize("route", [Route.DIRECTORY, Route.AUTHENTICATION])
def test_human_control_dischargeable_when_asserted_about_the_person(route):
    assert_human_control_dischargeable(Claim("human", route))


def test_service_principal_is_refused_outright():
    with pytest.raises(HumanControlNotDischargeable, match="not 'human'"):
        assert_human_control_dischargeable(Claim("service", Route.DIRECTORY))


def test_claims_are_stored_with_their_route_never_bare():
    memo = make_memo()
    pres = render(
        memo,
        principal_sub="bernd",
        tenant=Claim("tenant:platform", Route.REGISTRATION),
        principal_type=Claim("human", Route.REGISTRATION),
    )
    data = commit_presentation(pres, custody=CUSTODY).data
    assert data["tenant"] == {"value": "tenant:platform", "route": "registration-supplied"}
    assert data["principal_type"]["route"] == "registration-supplied"


# -------------------------------------------------------------------------
# GH-DEC-2026-014 §4 — the existence assertion, and PR-12
# -------------------------------------------------------------------------


def test_every_commitment_carries_the_existence_assertion():
    memo = make_memo()
    pres = render(memo, principal_sub="bernd").with_ack("h-1")
    disp = record(memo, pres, Verb.ACCEPT, HUMAN)
    for c in (
        commit_presentation(pres, custody=CUSTODY),
        commit_disposition(disp, custody=CUSTODY),
        commit_stance_application(
            memo_id=memo.id, memo_version=1, binding_level="organizational",
            binding_level_state="present", stance="fail_closed",
            dependency="access-engine", custody=CUSTODY,
        ),
    ):
        assert c.data["content_exists"] is True
        assert c.data["custody"] == CUSTODY


def test_commitment_carries_no_content():
    """Commitment-only: never the brief, never the packet."""
    memo = make_memo()
    data = commit_presentation(render(memo, principal_sub="bernd"), custody=CUSTODY).data
    flat = str(data)
    assert memo.brief not in flat
    assert "Change request" not in flat
    assert data["view_hash"]


@pytest.mark.parametrize(
    "locator",
    [
        "https://user:pw@store.example/obj",
        "https://store.example/obj?token=abc123",
        "https://store.example/o?access_token=x",
        "",
    ],
)
def test_secret_shaped_custody_locators_are_rejected(locator):
    with pytest.raises(CustodyLocatorRejected):
        assert_custody_locator_safe(locator)


def test_stable_identifier_custody_locator_is_accepted():
    assert_custody_locator_safe("informed-decision:memo/memo-1/presentation/pres-9")


# -------------------------------------------------------------------------
# Stance application is not a decline
# -------------------------------------------------------------------------


def test_fail_closed_is_recorded_as_a_stance_not_as_a_decline():
    c = commit_stance_application(
        memo_id="memo-1", memo_version=1, binding_level=None,
        binding_level_state="absent", stance="fail_closed",
        dependency="access-engine", custody=CUSTODY,
    )
    assert c.event_class is EventClass.STANCE_APPLICATION
    assert "verb" not in c.data
    assert c.data["stance_applied"] == "fail_closed"


def test_stance_application_pins_decision_attributable_false():
    """GH-DEC-2026-010 inherited: we cannot show access-engine said it."""
    c = commit_stance_application(
        memo_id="m", memo_version=1, binding_level="organizational",
        binding_level_state="present", stance="fail_closed",
        dependency="access-engine", custody=CUSTODY,
    )
    assert c.data["decision_attributable"] is False


# -------------------------------------------------------------------------
# Outbox
# -------------------------------------------------------------------------


def test_drain_failure_leaves_the_record_pending_never_lost():
    outbox = Outbox()
    memo = make_memo()
    outbox.append(commit_presentation(render(memo, principal_sub="b"), custody=CUSTODY))

    def failing_sink(_):
        raise RuntimeError("audit-core unreachable")

    with pytest.raises(RuntimeError):
        outbox.drain(failing_sink)
    assert len(outbox.pending) == 1
    assert outbox.drained == ()


def test_drain_delivers_and_counts_reconcile_per_class():
    outbox = Outbox()
    memo = make_memo()
    pres = render(memo, principal_sub="b").with_ack("h-1")
    outbox.append(commit_presentation(pres, custody=CUSTODY))
    outbox.append(commit_disposition(record(memo, pres, Verb.ACCEPT, HUMAN), custody=CUSTODY))
    sent: list[dict] = []
    assert outbox.drain(sent.append) == 2
    assert outbox.counts_by_class() == {
        EventClass.PRESENTATION.value: 1,
        EventClass.DISPOSITION.value: 1,
    }
    assert all("type" in e and "data" in e for e in sent)


def test_heartbeat_is_an_ordinary_event_with_the_same_envelope():
    hb = heartbeat(EventClass.DISPOSITION)
    env = hb.as_envelope()
    assert env["type"] == "audit-core.heartbeat"
    assert env["data"] == {
        "class": "informed-decision.disposition",
        "assertion": "nothing-to-report",
    }


# -------------------------------------------------------------------------
# GH-DEC-2026-015 — nesting, activated 2026-09-10
#
# approval-engine stated the presentation exclusion as normative and tested
# (docs/approval-claim.md "Presentation exclusion", and
# tests/test_claim_contract.py::test_presentation_changes_cannot_change_the_approved_act,
# which pins the input set from BOTH sides — widening and narrowing).
# -------------------------------------------------------------------------

DIGEST = "sha256:" + "b" * 64
OTHER_DIGEST = "sha256:" + "c" * 64


def test_carried_digest_enters_view_hash():
    plain = make_memo()
    nested = make_memo(approval_binding_digest=DIGEST)
    assert render(nested, principal_sub="b").view_hash != render(plain, principal_sub="b").view_hash


def test_a_different_act_digest_changes_view_hash():
    a = render(make_memo(approval_binding_digest=DIGEST), principal_sub="b")
    b = render(make_memo(approval_binding_digest=OTHER_DIGEST), principal_sub="b")
    assert a.view_hash != b.view_hash


def test_act_scope_is_no_longer_independently_canonicalized_when_nested():
    """The act has exactly one canonicalization, computed by its owner.

    Two memos differing only in act-scope must now hash alike, because the
    scope is act material already covered by the carried digest. Canonicalizing
    it again here is the partial recomputation GH-DEC-2026-015 removed.
    """
    acme = make_memo(approval_binding_digest=DIGEST)
    beta = make_memo(
        approval_binding_digest=DIGEST,
        binding=BindingSlice(
            principal=acme.binding.principal,
            target=Scope(kind="tenant", id="tenant:beta", label="Beta GmbH"),
        ),
    )
    assert render(acme, principal_sub="b").view_hash == render(beta, principal_sub="b").view_hash


def test_act_scope_still_binds_when_there_is_no_carried_digest():
    """L0/L2 are unchanged: with no approval there is no digest to defer to."""
    acme = make_memo(approval_id=None)
    beta = make_memo(
        approval_id=None,
        binding=BindingSlice(
            principal=acme.binding.principal,
            target=Scope(kind="tenant", id="tenant:beta", label="Beta GmbH"),
        ),
    )
    assert render(acme, principal_sub="b").view_hash != render(beta, principal_sub="b").view_hash


def test_the_approver_still_binds_when_nested():
    """`this person was shown this presentation of this act` — the person half.

    approval-engine's `principal` is the party on whose behalf; ours is the
    approver being bound. Dropping ours would remove who was shown this.
    """
    a = make_memo(approval_binding_digest=DIGEST)
    b = make_memo(
        approval_binding_digest=DIGEST,
        binding=BindingSlice(
            principal=Principal(id="p-2", kind="person", display_name="Someone Else"),
            target=a.binding.target,
        ),
    )
    assert render(a, principal_sub="x").view_hash != render(b, principal_sub="x").view_hash


def test_presentation_material_still_moves_view_hash_when_nested():
    base = make_memo(approval_binding_digest=DIGEST)
    relocalized = make_memo(approval_binding_digest=DIGEST, locale="de")
    assert render(base, principal_sub="b").view_hash != render(relocalized, principal_sub="b").view_hash


@pytest.mark.parametrize("bad", ["deadbeef", "sha256:zz", "sha1:" + "a" * 40, "sha256:" + "A" * 64])
def test_a_malformed_carried_digest_is_refused(bad):
    with pytest.raises(ValueError, match="carried, never computed"):
        make_memo(approval_binding_digest=bad)


def test_a_carried_digest_without_its_approval_is_refused():
    with pytest.raises(ValueError, match="approval it belongs to"):
        make_memo(approval_binding_digest=DIGEST, approval_id=None)