tegwick
d61ba327fa
type() excluded bool even though bool subclasses int, so schema fields requires_new_bind, required_ack, and sealed failed as invalid_type. Unknown leaves still fail closed. No submitter, token, or write-back change. Assistant: grok Assistant-Session: 01a10286-9185-7bb2-b411-89cf6ab4c0c6
399 lines
14 KiB
Python
399 lines
14 KiB
Python
"""Dev fixture for hub memo forwarding. No live intake, token, or submitter."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from api.services.informed_decision_contract import (
|
|
PROVENANCE_NOTE,
|
|
ContractError,
|
|
build_envelope,
|
|
correlate_resolution,
|
|
resolution_applies,
|
|
stable_idempotency_key,
|
|
validate_envelope,
|
|
)
|
|
from api.services.memo_forwarding import DevIntakeFixture
|
|
from api.services.write_idempotency import route_rule_for
|
|
|
|
DECISION = "62aa95e4-3ccc-4867-8806-5e3db005b483"
|
|
MEMO = "infd-20261003-p01"
|
|
OTHER = "infd-20261003-p02"
|
|
APPROVAL = "ba5ce2d8-8b6d-40be-af89-2e8c147029a3"
|
|
BINDING = "sha256:" + "ab" * 32
|
|
TOKEN = "dev-fixture-not-a-credential"
|
|
|
|
|
|
class Clock:
|
|
def __init__(self) -> None:
|
|
self.now = 1_000.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
|
|
class Transport:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def __call__(self, method, url, body, headers):
|
|
self.calls.append((method, url, body, dict(headers)))
|
|
item = self.responses.pop(0)
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
|
|
|
|
def envelope(**overrides):
|
|
memo = {"id": MEMO, "question": "Accept the forwarded draft?"}
|
|
packet = [{"item_id": "brief", "media_type": "text/plain", "content": "Bounded brief."}]
|
|
intent = {
|
|
"action": "accept",
|
|
"principal": "pending-human-session",
|
|
"purpose": "Review the draft.",
|
|
"target": {"id": MEMO, "type": "decision-memo", "system": "informed-decision"},
|
|
}
|
|
fields = dict(
|
|
decision_id=DECISION,
|
|
memo=memo,
|
|
packet=packet,
|
|
memo_class="work-record-accept",
|
|
review_group="net-kingdom-admins",
|
|
approval_intent=intent,
|
|
drafted_by="claude@the-custodian",
|
|
resolves="on-answer",
|
|
)
|
|
fields.update(overrides)
|
|
return build_envelope(**fields)
|
|
|
|
|
|
def receipt(memo_id=MEMO):
|
|
return {
|
|
"memo_id": memo_id,
|
|
"approval_id": APPROVAL,
|
|
"approval_binding_digest": BINDING,
|
|
"echo": "should-not-be-kept",
|
|
}
|
|
|
|
|
|
def fixture(tmp_path, transport, clock=None, token=lambda: TOKEN):
|
|
return DevIntakeFixture(
|
|
tmp_path / "forward.sqlite",
|
|
transport=transport,
|
|
intake_base_url="http://intake.test",
|
|
hub_base_url="http://hub.test",
|
|
token_reader=token,
|
|
clock=clock or Clock(),
|
|
)
|
|
|
|
|
|
def stored_bytes(path) -> bytes:
|
|
data = Path(path).read_bytes()
|
|
for suffix in ("-wal", "-shm"):
|
|
side = Path(str(path) + suffix)
|
|
if side.exists():
|
|
data += side.read_bytes()
|
|
return data
|
|
|
|
|
|
def test_idempotency_key_is_stable_and_drafted_by_is_not_identity():
|
|
first = envelope()
|
|
second = envelope()
|
|
assert first["idempotency_key"] == second["idempotency_key"]
|
|
assert first["idempotency_key"] == stable_idempotency_key(DECISION, MEMO)
|
|
assert "drafted_by" not in first["idempotency_key"]
|
|
assert "submitter" not in first
|
|
changed = dict(first)
|
|
changed["idempotency_key"] = "unstable-key-value"
|
|
with pytest.raises(ContractError) as error:
|
|
validate_envelope(changed)
|
|
assert error.value.code == "idempotency_key_not_stable"
|
|
|
|
|
|
def test_envelope_is_bounded_and_does_not_echo_secrets():
|
|
huge = envelope()
|
|
huge["packet"][0]["content"] = "x" * (256 * 1024 + 1)
|
|
with pytest.raises(ContractError) as too_big:
|
|
validate_envelope(huge)
|
|
assert too_big.value.code == "packet_too_large"
|
|
leaked = envelope()
|
|
leaked["packet"][0]["content"] = "-----BEGIN PRIVATE KEY-----"
|
|
with pytest.raises(ContractError) as secret:
|
|
validate_envelope(leaked)
|
|
assert secret.value.code == "secret_marker"
|
|
assert "PRIVATE KEY" not in str(secret.value)
|
|
extra = envelope()
|
|
extra["highlights"] = ["not-a-notice"]
|
|
with pytest.raises(ContractError) as unknown:
|
|
validate_envelope(extra)
|
|
assert unknown.value.code == "unknown_field"
|
|
|
|
|
|
def test_governed_boolean_leaves_round_trip():
|
|
checked = envelope(memo={
|
|
"id": MEMO,
|
|
"question": "Accept the forwarded draft?",
|
|
"sealed": False,
|
|
"binding": {"target": {"requires_new_bind": True}},
|
|
"highlights": [{"required_ack": True}],
|
|
})
|
|
memo = checked["memo"]
|
|
assert memo["sealed"] is False
|
|
assert memo["binding"]["target"]["requires_new_bind"] is True
|
|
assert memo["highlights"][0]["required_ack"] is True
|
|
rejected = envelope()
|
|
rejected["memo"]["marker"] = object()
|
|
with pytest.raises(ContractError) as error:
|
|
validate_envelope(rejected)
|
|
assert error.value.code == "invalid_type"
|
|
|
|
|
|
def test_pose_replay_conflict_and_asserted_provenance(tmp_path):
|
|
clock = Clock()
|
|
transport = Transport([(201, receipt())])
|
|
fix = fixture(tmp_path, transport, clock)
|
|
posed = fix.submit(envelope())
|
|
assert posed.status == "delivered"
|
|
assert posed.drafted_by == "claude@the-custodian"
|
|
assert posed.drafted_by_provenance == "asserted"
|
|
assert posed.provenance_note == PROVENANCE_NOTE
|
|
assert "asserted provenance" in posed.provenance_note
|
|
assert set(posed.receipt) == {"memo_id", "approval_id", "approval_binding_digest"}
|
|
sent = transport.calls[0]
|
|
assert sent[0:2] == ("POST", "http://intake.test/v1/memos")
|
|
assert sent[2]["drafted_by"] == "claude@the-custodian"
|
|
assert "drafted_by_provenance" not in sent[2]
|
|
assert "submitter" not in sent[2]
|
|
assert sent[3]["Authorization"] == f"Bearer {TOKEN}"
|
|
assert TOKEN.encode() not in stored_bytes(fix.path)
|
|
|
|
restarted = DevIntakeFixture(
|
|
fix.path,
|
|
transport=Transport([(201, receipt())]),
|
|
intake_base_url="http://intake.test",
|
|
hub_base_url="http://hub.test",
|
|
token_reader=lambda: TOKEN,
|
|
clock=clock,
|
|
)
|
|
replay = restarted.submit(envelope())
|
|
assert replay.status == "replay"
|
|
assert replay.receipt == posed.receipt
|
|
assert restarted.transport.calls == []
|
|
|
|
conflict = envelope()
|
|
conflict["packet"][0]["content"] = "Different brief."
|
|
blocked = restarted.submit(conflict)
|
|
assert blocked.status == "blocked"
|
|
assert blocked.detail == "idempotency_conflict"
|
|
kept = restarted.pose(posed.idempotency_key)
|
|
assert kept["state"] == "delivered"
|
|
assert "Different brief." not in kept["envelope_json"]
|
|
assert restarted.transport.calls == []
|
|
|
|
|
|
def test_delivery_failure_keeps_local_evidence_and_retries(tmp_path):
|
|
clock = Clock()
|
|
transport = Transport([
|
|
ConnectionRefusedError("down"),
|
|
(503, {"error": "intake_unavailable", "body": "LEAKED-BODY"}),
|
|
(201, receipt()),
|
|
])
|
|
fix = fixture(tmp_path, transport, clock)
|
|
failed = fix.submit(envelope())
|
|
assert failed.status == "pending"
|
|
assert failed.detail == "ConnectionRefusedError"
|
|
row = fix.pose(failed.idempotency_key)
|
|
assert row["state"] == "pending"
|
|
assert "Bounded brief." in row["envelope_json"]
|
|
assert "LEAKED-BODY" not in row["envelope_json"]
|
|
assert fix.run_due() is None
|
|
|
|
clock.now += 2
|
|
again = fix.run_due()
|
|
assert again.status == "pending"
|
|
assert again.detail == "http_503"
|
|
assert b"LEAKED-BODY" not in stored_bytes(fix.path)
|
|
assert fix.run_due() is None
|
|
|
|
clock.now += 4
|
|
delivered = fix.run_due()
|
|
assert delivered.status == "delivered"
|
|
assert delivered.receipt["memo_id"] == MEMO
|
|
assert fix.run_due() is None
|
|
assert len(transport.calls) == 3
|
|
|
|
|
|
def test_intake_conflict_blocks_without_retry(tmp_path):
|
|
transport = Transport([(409, {"error": "idempotency_conflict", "packet": "LEAK"})])
|
|
fix = fixture(tmp_path, transport)
|
|
blocked = fix.submit(envelope())
|
|
assert blocked.status == "blocked"
|
|
assert blocked.detail == "idempotency_conflict"
|
|
assert b"LEAK" not in stored_bytes(fix.path)
|
|
assert fix.run_due() is None
|
|
assert len(transport.calls) == 1
|
|
|
|
|
|
def test_absent_token_is_not_invented(tmp_path):
|
|
transport = Transport([(201, receipt())])
|
|
fix = fixture(tmp_path, transport, token=None)
|
|
pending = fix.submit(envelope())
|
|
assert pending.status == "pending"
|
|
assert pending.detail == "projected_token_absent"
|
|
assert transport.calls == []
|
|
assert fix.pose(pending.idempotency_key)["state"] == "pending"
|
|
|
|
def boom():
|
|
raise RuntimeError("sk-or-this-must-not-be-stored-0123456789")
|
|
|
|
fix.token_reader = boom
|
|
fix.clock = Clock()
|
|
fix.clock.now = 1_000 + 2
|
|
retried = fix.run_due()
|
|
assert retried.status == "pending"
|
|
assert retried.detail == "RuntimeError"
|
|
assert transport.calls == []
|
|
assert b"sk-or-this-must-not-be-stored" not in stored_bytes(fix.path)
|
|
|
|
|
|
def test_live_hosts_and_runtime_binding_are_refused(tmp_path):
|
|
with pytest.raises(ValueError):
|
|
DevIntakeFixture(
|
|
tmp_path / "no.sqlite",
|
|
transport=Transport([]),
|
|
intake_base_url="http://127.0.0.1:8000",
|
|
hub_base_url="http://hub.test",
|
|
)
|
|
root = Path(__file__).resolve().parents[1] / "api"
|
|
main = (root / "main.py").read_text()
|
|
assert "memo_notifications" in main
|
|
assert "memo_forwarding" not in main
|
|
assert "DevIntakeFixture" not in main
|
|
owned = [
|
|
root / "main.py",
|
|
root / "services" / "memo_forwarding.py",
|
|
root / "services" / "informed_decision_contract.py",
|
|
root / "services" / "memo_notification.py",
|
|
root / "routers" / "memo_notifications.py",
|
|
root / "models" / "memo_notification.py",
|
|
]
|
|
for path in owned:
|
|
text = path.read_text()
|
|
assert "INFD_HUB_WRITEBACK" not in text
|
|
assert "IntakeDeliveryWorker" not in text
|
|
assert "/var/run/secrets" not in text
|
|
assert "token_file" not in text
|
|
assert route_rule_for("POST", "/informed-decision/notifications") is None
|
|
|
|
|
|
@pytest.mark.parametrize("verb,serves", [
|
|
("return", {"hub_decision_id": DECISION, "resolves": "on-answer"}),
|
|
("discuss", {"hub_decision_id": DECISION, "resolves": "on-answer"}),
|
|
("accept", {"hub_decision_id": DECISION, "resolves": "never"}),
|
|
("decline", None),
|
|
])
|
|
def test_non_terminal_answers_do_not_resolve(tmp_path, verb, serves):
|
|
assert resolution_applies(verb, serves) is False
|
|
transport = Transport([(200, {})])
|
|
fix = fixture(tmp_path, transport)
|
|
status = fix.observe_resolution(
|
|
decision_id=DECISION, memo_id=MEMO, verb=verb, serves=serves, payload={"rationale": "no"},
|
|
)
|
|
assert status == "not_applicable"
|
|
assert transport.calls == []
|
|
|
|
|
|
def test_resolution_failure_retries_and_other_memo_blocks(tmp_path):
|
|
clock = Clock()
|
|
payload = {
|
|
"decided_by": "Bernd Worsch via informed-decision",
|
|
"rationale": f"accept in informed-decision memo {MEMO} v1; ",
|
|
"write_log": False,
|
|
}
|
|
transport = Transport([
|
|
ConnectionRefusedError("down"),
|
|
(200, {"status": "resolved"}),
|
|
(409, {"error": "already resolved"}),
|
|
(200, {"rationale": "resolved by hand"}),
|
|
])
|
|
fix = fixture(tmp_path, transport, clock)
|
|
failed = fix.observe_resolution(
|
|
decision_id=DECISION, memo_id=MEMO, verb="accept",
|
|
serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
|
|
)
|
|
assert failed == "failed"
|
|
assert fix.resolutions()[0]["payload_json"]
|
|
assert fix.run_due_resolution() is None
|
|
clock.now += 2
|
|
assert fix.run_due_resolution() == "resolved"
|
|
assert [row["status"] for row in fix.resolutions()] == ["failed", "resolved"]
|
|
assert fix.run_due_resolution() is None
|
|
|
|
conflict = fix.observe_resolution(
|
|
decision_id=DECISION, memo_id=OTHER, verb="decline",
|
|
serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
|
|
)
|
|
assert conflict == "conflict"
|
|
calls = len(transport.calls)
|
|
assert transport.calls[-2][0] == "POST"
|
|
assert transport.calls[-1][0] == "GET"
|
|
assert fix.observe_resolution(
|
|
decision_id=DECISION, memo_id=OTHER, verb="decline",
|
|
serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
|
|
) == "conflict"
|
|
assert len(transport.calls) == calls
|
|
|
|
|
|
def test_unreadable_409_stays_retryable(tmp_path):
|
|
clock = Clock()
|
|
transport = Transport([
|
|
(409, {}),
|
|
(503, {}),
|
|
(409, {}),
|
|
(200, {"rationale": f"decline in informed-decision memo {MEMO} v1; "}),
|
|
])
|
|
fix = fixture(tmp_path, transport, clock)
|
|
payload = {"rationale": "kept locally", "decided_by": "reviewer", "write_log": False}
|
|
assert fix.observe_resolution(
|
|
decision_id=DECISION, memo_id=MEMO, verb="decline",
|
|
serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
|
|
) == "failed"
|
|
assert fix.resolutions()[0]["error_type"] == "resolution_unreadable"
|
|
clock.now += 2
|
|
assert fix.run_due_resolution() == "already_resolved"
|
|
|
|
|
|
async def test_hub_409_still_returns_the_existing_resolution(client):
|
|
domain = await client.post("/domains/", json={"slug": "testdomain", "name": "Test Domain"})
|
|
assert domain.status_code == 201, domain.text
|
|
topic = await client.post("/topics/", json={
|
|
"slug": "testtopic", "title": "Test Topic", "domain": "testdomain",
|
|
})
|
|
assert topic.status_code == 201, topic.text
|
|
created = await client.post("/decisions/", json={
|
|
"title": "Should the draft stand?", "topic_id": topic.json()["id"],
|
|
})
|
|
assert created.status_code == 201, created.text
|
|
decision_id = created.json()["id"]
|
|
rationale = f"accept in informed-decision memo {MEMO} v1; approval {APPROVAL}"
|
|
resolved = await client.post(f"/decisions/{decision_id}/resolve", json={
|
|
"rationale": rationale,
|
|
"decided_by": "Bernd Worsch via informed-decision",
|
|
"write_log": False,
|
|
})
|
|
assert resolved.status_code == 200, resolved.text
|
|
conflict = await client.post(f"/decisions/{decision_id}/resolve", json={
|
|
"rationale": "resolved by hand",
|
|
"decided_by": "someone else",
|
|
"write_log": False,
|
|
})
|
|
assert conflict.status_code == 409
|
|
current = await client.get(f"/decisions/{decision_id}")
|
|
assert current.status_code == 200
|
|
body = current.json()
|
|
assert body["rationale"] == rationale
|
|
assert correlate_resolution(MEMO, post_status=409, existing=body) == "already_resolved"
|
|
assert correlate_resolution(OTHER, post_status=409, existing=body) == "conflict"
|