Build the T08 domain core with the engine behind a seam
approval-engine APPROVAL-WP-0002-T01 is still progress and its namespace has no pods, so the live end-to-end proof cannot run. Built everything that does not depend on it, with the engine behind a Protocol plus a fake carrying its real refusal semantics, so its arrival is a wiring change rather than a build. - memo.py: the Decision Memo, versions, binding document. Principal, Scope, Awareness and Hat are dataclasses rather than dicts because the canonicalizer requires a shape and a missing key should fail at construction rather than deep inside hashing — which is exactly how it failed twice while building this. Field names follow the governed canonicalizer (item_id, severity, locator): the published vectors are the contract, so the object was aligned to them rather than the reverse. - presentation.py: the sole writer of view_hash. One writer, one canonicalizer, one place to audit. Acknowledgment is an explicit method call and nothing infers it from scroll, dwell or focus. - disposition.py: verbs and guards G_NOAGENT, G_STEP, G_PRES, G_ACK, G_REASONS, G_SEALED. accept is ABSENT from weak steps rather than present-and-disabled, because a greyed-out accept still teaches the wrong model. Only accept reaches the engine; a memo return is not represented there at all. - provenance.py: claim routes per A-16. assert_human_control_dischargeable refuses a registration-supplied human, so PR-11's limitation fires at the point of use instead of sitting in a document. - evidence.py: local outbox, commitment-only records carrying the GH-DEC-2026-014 §4 existence assertion, per-class reconciliation counts, and a custody-locator guard that rejects credentialed URLs (PR-12). - approval_client.py: 409 duplicate_approver is success, 409 conflict terminal, 503 fail-closed, approval:consume refused before a token is requested. 87 tests pass, including every negative case in the Use Case Catalog and that a fail-closed outcome is recorded as a stance application with no verb field — never as a decline, because the human did not make one. T08 stays progress: the live proof is the remainder. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01V3W1dQG7GFFM9d94jFx7iR Assistant: claude-code Assistant-Model: opus Assistant-Process: 1565372@bnt-lap001 Assistant-Session: 16bb2f25-b34c-49ef-8e94-5fec3567a568
This commit is contained in:
parent
4379576abd
commit
9e1f77e32b
18 changed files with 1503 additions and 10 deletions
BIN
tests/__pycache__/test_skeleton.cpython-312-pytest-7.4.4.pyc
Normal file
BIN
tests/__pycache__/test_skeleton.cpython-312-pytest-7.4.4.pyc
Normal file
Binary file not shown.
456
tests/test_skeleton.py
Normal file
456
tests/test_skeleton.py
Normal file
|
|
@ -0,0 +1,456 @@
|
|||
"""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",
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue