APPROVAL-WP-0007-T02 done: actor_consumes; T05 principal_approves split out

Founder decision 2026-11-01, relayed by the net-kingdom coordinator: "Agree to
split APPROVAL-WP-0007-T02: build actor_consumes now (opt-in at issue; the
bounded executor will set it); principal_approves keeps waiting for the KeyCape
subject mapping."

- Optional actor_consumes (boolean) at issue, schema v8. Consume refuses with
  403 unless the consuming subject equals binding.actor, also on idempotent
  retries; internal calls without a consumer fail closed. Inherited on
  supersession (existing-successor mismatch 409), in audit details, served in
  GET/claim only when declared so undeclared claims keep exact bytes. Legacy
  rows migrate without the flag.
- GH-DEC-2026-026 Ruling 6: under human_control, add_entry refuses (403) the
  binding actor as approver. human_control already covered a service actor;
  this covers a human actor.
- Workplan: T02 done (actor_consumes); new T05 principal_approves (wait).
- Not released or deployed: needs a v8 image and founder-run rollout.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5502@bnt-lap001
Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196

tests/test_actor_consumes.py

@@ -0,0 +1,190 @@
+"""APPROVAL-WP-0007-T02: opt-in consumer = actor, declared at issue.
+
+Also GH-DEC-2026-026 Ruling 6: under human_control the actor never approves
+its own approval.
+"""
+import dataclasses
+import json
+from pathlib import Path
+
+import jsonschema
+import pytest
+
+from approval_engine.api import App, call
+from approval_engine.auth import Identity, StaticTokenAuthenticator
+from approval_engine.errors import Conflict, Forbidden, Unprocessable
+from approval_engine.store import Engine, LATEST_SCHEMA_VERSION
+from tests.conftest import FROZEN, binding, validity
+
+SCHEMA = json.loads(
+    (Path(__file__).parents[1] / "schemas" / "approval_claim.schema.json").read_text()
+)
+ACTOR = "agt-secrets-engine"  # binding() default actor
+
+
+def _identity(subject, principal_type="service"):
+    return Identity(
+        subject=subject,
+        issuer="https://keycape.example",
+        audiences=("approval-engine",),
+        principal_type=principal_type,
+        tenant="tenant:platform",
+        roles=frozenset(),
+        scopes=frozenset({"approval:create", "approval:read", "approval:approve",
+                          "approval:consume", "approval:supersede"}),
+        assurance={"level": "aal2", "amr": ["pwd", "otp"], "source": "key-cape"},
+        evidence_ref=f"test-identity:{subject}",
+    )
+
+
+@pytest.fixture
+def multi_app(engine):
+    return App(engine, StaticTokenAuthenticator({
+        "actor": _identity(ACTOR),
+        "other": _identity("svc-other"),
+        "human": _identity("user:admin", "human"),
+    }))
+
+
+def _approved(engine, **kwargs):
+    obj = engine.create(binding(), validity(), **kwargs)
+    return engine.add_entry(obj.id, "user:admin", principal_type="human")
+
+
+def test_undeclared_keeps_any_consumer_and_exact_shape(engine):
+    obj = _approved(engine)
+    assert obj.actor_consumes is False
+    assert "actor_consumes" not in obj.as_dict()["binding"]
+    claim = engine.claim(obj.id)
+    assert "actor_consumes" not in claim["binding"]
+    jsonschema.validate(claim, SCHEMA)
+    result = engine.consume(obj.id, obj.binding_digest, consumer=_identity("svc-other"))
+    assert result["status"] == "consumed"
+
+
+def test_declared_false_is_the_same_as_undeclared(engine):
+    obj = _approved(engine, actor_consumes=False)
+    assert "actor_consumes" not in engine.claim(obj.id)["binding"]
+    engine.consume(obj.id, obj.binding_digest, consumer=_identity("svc-other"))
+
+
+@pytest.mark.parametrize("value", ["true", 1, None, [], {}])
+def test_declaration_must_be_a_boolean(multi_app, engine, value):
+    status, _ = call(multi_app, "POST", "/v1/approvals", {
+        "binding": binding(), "validity": validity(), "actor_consumes": value,
+    }, authorization="Bearer actor")
+    assert status == 422
+    with pytest.raises(Unprocessable, match="actor_consumes"):
+        engine.create(binding(), validity(), actor_consumes=value)
+
+
+def test_wrong_consumer_is_refused_and_actor_consumes_over_http(multi_app, engine):
+    status, created = call(multi_app, "POST", "/v1/approvals", {
+        "binding": binding(), "validity": validity(),
+        "human_control": True, "actor_consumes": True,
+    }, authorization="Bearer actor")
+    assert status == 201, created
+    assert created["binding"]["actor_consumes"] is True
+    status, _ = call(multi_app, "POST", f"/v1/approvals/{created['id']}/entries", {},
+                     authorization="Bearer human")
+    assert status == 200
+    claim = engine.claim(created["id"])
+    assert claim["binding"]["actor_consumes"] is True and claim["valid_now"]
+    jsonschema.validate(claim, SCHEMA)
+
+    digest = created["binding"]["digest"]
+    status, refused = call(multi_app, "POST", f"/v1/approvals/{created['id']}/consume",
+                           {"request_digest": digest}, authorization="Bearer other")
+    assert status == 403, refused
+    assert engine.get(created["id"]).status == "approved"
+
+    status, spent = call(multi_app, "POST", f"/v1/approvals/{created['id']}/consume",
+                         {"request_digest": digest}, authorization="Bearer actor")
+    assert status == 200 and spent["status"] == "consumed" and spent["idempotent"] is False
+
+    # Another service gets no receipt from an idempotent retry either.
+    status, _ = call(multi_app, "POST", f"/v1/approvals/{created['id']}/consume",
+                     {"request_digest": digest}, authorization="Bearer other")
+    assert status == 403
+    status, again = call(multi_app, "POST", f"/v1/approvals/{created['id']}/consume",
+                         {"request_digest": digest}, authorization="Bearer actor")
+    assert status == 200 and again["idempotent"] is True
+
+
+def test_internal_call_without_consumer_fails_closed(engine):
+    obj = _approved(engine, actor_consumes=True)
+    with pytest.raises(Forbidden, match="actor_consumes"):
+        engine.consume(obj.id, obj.binding_digest)
+    assert engine.get(obj.id).status == "approved"
+    assert engine.consume(obj.id, obj.binding_digest, consumer=_identity(ACTOR))["status"] == "consumed"
+
+
+def test_supersession_inherits_and_refuses_existing_mismatch(engine):
+    parent = engine.create(binding(), validity(), actor_consumes=True)
+    plain = engine.create(binding(), validity())
+    with pytest.raises(Conflict, match="actor_consumes"):
+        engine.supersede(parent.id, plain.id)
+    assert engine.get(parent.id).status == "requested"
+    child = engine.get(engine.supersede(parent.id)["successor_id"])
+    assert child.actor_consumes is True
+    assert engine.claim(child.id)["binding"]["actor_consumes"] is True
+    engine.add_entry(child.id, "user:admin", principal_type="human")
+    with pytest.raises(Forbidden):
+        engine.consume(child.id, child.binding_digest, consumer=_identity("svc-other"))
+
+
+def test_declaration_is_on_audit_events_and_outside_the_act_digest(engine):
+    plain = engine.create(binding(), validity())
+    declared = engine.create(binding(), validity(), actor_consumes=True)
+    assert plain.binding_digest == declared.binding_digest
+    engine.add_entry(declared.id, "user:admin", principal_type="human")
+    event = json.loads(engine._conn().execute(
+        "SELECT payload_json FROM outbox WHERE approval_id=?", (declared.id,)).fetchone()[0])
+    assert event["details"]["actor_consumes"] is True
+    plain_event_rows = engine._conn().execute(
+        "SELECT payload_json FROM outbox WHERE approval_id=?", (plain.id,)).fetchall()
+    assert all("actor_consumes" not in json.loads(r[0])["details"] for r in plain_event_rows)
+
+
+def test_legacy_v7_rows_migrate_without_the_flag(tmp_path):
+    path = tmp_path / "v7.sqlite"
+    eng = Engine(path, clock=lambda: FROZEN)
+    obj = eng.create(binding(), validity())
+    eng.add_entry(obj.id, "user:admin", principal_type="human")
+    eng._conn().execute("ALTER TABLE approvals DROP COLUMN actor_consumes")
+    eng._conn().execute("PRAGMA user_version=7")
+    eng._conn().commit()
+    eng.close()
+    upgraded = Engine(path, clock=lambda: FROZEN)
+    survivor = upgraded.get(obj.id)
+    assert survivor.actor_consumes is False
+    assert "actor_consumes" not in upgraded.claim(obj.id)["binding"]
+    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 8
+    upgraded.consume(obj.id, obj.binding_digest, consumer=_identity("svc-other"))
+    upgraded.close()
+
+
+def test_human_control_refuses_the_actor_as_approver(engine):
+    """GH-DEC-2026-026 Ruling 6. A human actor may create a human_control
+    request but may not approve it; another human may."""
+    obj = engine.create(binding(actor="user:operator"), validity(), human_control=True)
+    with pytest.raises(Forbidden, match="actor"):
+        engine.add_entry(obj.id, "user:operator", principal_type="human")
+    assert engine.get(obj.id).entries == []
+    assert engine.add_entry(obj.id, "user:admin", principal_type="human").status == "approved"
+
+
+def test_actor_may_still_approve_without_human_control(engine):
+    obj = engine.create(binding(actor="user:operator"), validity())
+    assert engine.add_entry(obj.id, "user:operator", principal_type="human").status == "approved"
+
+
+def test_human_actor_self_approval_refused_over_http(engine):
+    app = App(engine, StaticTokenAuthenticator({"op": _identity("user:operator", "human")}))
+    status, created = call(app, "POST", "/v1/approvals", {
+        "binding": binding(actor="user:operator"), "validity": validity(), "human_control": True,
+    }, authorization="Bearer op")
+    assert status == 201
+    status, refused = call(app, "POST", f"/v1/approvals/{created['id']}/entries", {},
+                           authorization="Bearer op")
+    assert status == 403, refused

tests/test_human_control.py

@@ -144,7 +144,7 @@ def test_legacy_v4_rows_are_not_reclassified_from_human_entries(tmp_path):
     assert survivor.entries[0]["principal_type"] == "human"
     assert survivor.human_control is False
     assert upgraded.claim(obj.id)["binding"]["human_control"] is False
-    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 7
+    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 8
     upgraded.close()
 
 

tests/test_required_assurance.py

@@ -159,5 +159,5 @@ def test_legacy_v6_rows_migrate_to_null_floor(tmp_path):
     survivor = upgraded.get(obj.id)
     assert survivor.status == "approved" and survivor.required_assurance is None
     assert upgraded.claim(obj.id)["valid_now"] is True
-    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 7
+    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 8
     upgraded.close()

tests/test_tenant_provenance.py

@@ -79,7 +79,7 @@ def test_v5_migration_preserves_unknown_provenance_and_existing_evidence(tmp_pat
     with pytest.raises(StoreUnavailable, match="migrate"):
         Engine(path, auto_migrate=False)
     engine = Engine(path, clock=lambda: FROZEN)
-    assert engine.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 7
+    assert engine.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 8
     assert engine.get(obj.id).entries[0]["tenant_source"] is None
     assert engine.get(obj.id).binding_digest == digest
     assert engine.get(obj.id).human_control is True