Prepare purpose image release with bounded custody protocol and owner reviews

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_purpose_owner_mutation_negatives.py

@@ -0,0 +1,66 @@
+"""Owner review: malformed/advisory results cannot mutate lifecycle or replay keys.
+
+Synthetic source checks only; no native custody/provider/approval acceptance.
+"""
+import pytest
+from informed_decision.lifecycle_service import LifecycleError
+from test_lifecycle import workspace
+from test_lifecycle_service import caller, service
+from test_policy_client import PolicyTransport
+
+
+VARIANTS = ["readback", "enforcement", "unknown", None, False, 0, [], {},
+            "NULL_BINDING", "LIST_BINDING"]
+
+
+def domain_state(store):
+    with store._connection() as db:
+        tables = [row[0] for row in db.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
+            if row[0] != "policy_observations"]
+        return {table: tuple(tuple(row) for row in db.execute(
+            'SELECT * FROM "' + table + '" ORDER BY rowid')) for table in tables}
+
+
+@pytest.mark.parametrize("operation", ["register", "replay", "publish", "annotate", "revise"])
+@pytest.mark.parametrize("purpose", VARIANTS)
+def test_advisory_or_malformed_refusal_preserves_domain_and_retry_key(workspace, operation, purpose):
+    transport = PolicyTransport()
+    app, memo = service(workspace, transport)
+    actor = caller()
+    prior = app.register(actor, memo, key="initial") if operation != "register" else None
+    actions = {
+        "register": lambda: app.register(actor, memo, key="same-attempt"),
+        "replay": lambda: app.register(actor, memo, key="initial"),
+        "publish": lambda: app.publish(actor, memo.id, 1,
+            audience={"groups": ["reviewers"], "subjects": []}, key="same-attempt"),
+        "annotate": lambda: app.annotate(actor, memo.id, 1, kind="comment",
+            body={"text": "independent owner check"}, key="same-attempt"),
+        "revise": lambda: app.revise(actor, memo.next_version(brief="independent owner check"),
+            expected_version=1, summary="review", key="same-attempt"),
+    }
+    before = domain_state(app.repository.store)
+    observations = len(app.repository.store.policy_observations())
+    def contaminate(data):
+        if purpose == "NULL_BINDING":
+            data["binding"] = None
+        elif purpose == "LIST_BINDING":
+            data["binding"] = []
+        else:
+            data["binding"]["purpose"] = purpose
+    transport.change = contaminate
+    with pytest.raises(LifecycleError) as error:
+        actions[operation]()
+    assert error.value.status == 503 and error.value.code == "invalid_decision"
+    assert domain_state(app.repository.store) == before
+    rows = app.repository.store.policy_observations()
+    assert len(rows) == observations + 1
+    assert rows[-1]["outcome"] == "invalid_decision"
+    # Refusal retains its observation but consumes no domain idempotency key.
+    transport.change = lambda data: None
+    result = actions[operation]()
+    if operation == "replay":
+        assert result == prior
+        assert domain_state(app.repository.store) == before
+    else:
+        assert domain_state(app.repository.store) != before