"""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", }