Keep original case provenance while revalidating a fresh caller session

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a14bfc-9643-7872-aa12-98326dcf25e7

tests/test_case_current_session_reference.py

@@ -0,0 +1,105 @@
+"""Current caller authentication never replaces historical dispatch provenance."""
+from dataclasses import replace
+
+import pytest
+
+from tests.test_case_run import make_rig
+from warden.case_receipt_state import Outcome, Refused
+from warden.case_run import Executor, Journal
+
+
+def fresh_verifier(executor, *, allow=True):
+    original = executor.evidence_verifier
+    def verify(purpose, source, evidence):
+        if purpose.startswith("caller-session:") and evidence == "fresh-session":
+            return allow
+        return original(purpose, source, evidence)
+    executor.evidence_verifier = verify
+    return verify
+
+
+def rows(executor):
+    return tuple(executor.journal.db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+                 for table in ("receipts", "intents"))
+
+
+def readmit(executor, sources, kwargs, reference="session"):
+    return executor.admit(*sources, **{**kwargs, "session_evidence": "fresh-session",
+                                      "original_session_reference": reference})
+
+
+def test_normal_new_session_cannot_rewrite_original_journal_pins(tmp_path):
+    executor, admitted, _, sources, kwargs = make_rig(tmp_path)
+    executor.run(*admitted)
+    fresh_verifier(executor)
+    before = rows(executor)
+    changed = executor.admit(*sources, **{**kwargs,"session_evidence":"fresh-session"})
+    assert changed[1].pins != admitted[1].pins
+    with pytest.raises(Refused,match="journal pins changed"):
+        executor.rollback(*changed)
+    assert rows(executor) == before
+
+
+def test_fresh_process_inverse_retains_original_pins_without_replay(tmp_path):
+    executor, admitted, calls, sources, kwargs = make_rig(tmp_path)
+    assert executor.run(*admitted).last.outcome is Outcome.SUCCESS
+    before = rows(executor)
+    fresh = Executor(Journal(executor.journal.directory), executor.adapters,
+                     fresh_verifier(executor))
+    renewed = readmit(fresh, sources, kwargs)
+    assert renewed[1].pins == admitted[1].pins
+    assert rows(fresh) == before
+    with pytest.raises(Refused):
+        fresh.run(*renewed)
+    assert len(calls) == 1
+    assert fresh.rollback(*renewed).last.outcome is Outcome.ROLLED_BACK
+
+
+@pytest.mark.parametrize("kind", ["missing-intent", "orphan-intent", "wrong-principal", "wrong-reference", "same-as-current", "invalid-reference", "inactive-current"])
+def test_original_reference_is_never_authentication_or_new_intent(tmp_path, kind):
+    executor, admitted, calls, sources, kwargs = make_rig(tmp_path)
+    if kind != "missing-intent":
+        executor.run(*admitted)
+    if kind == "orphan-intent":
+        executor.journal.db.execute("DELETE FROM intents")
+    if kind == "wrong-principal":
+        kwargs = {**kwargs,"principal":"another-operator"}
+    fresh_verifier(executor, allow=kind != "inactive-current")
+    before = rows(executor)
+    reference = {"wrong-reference": "unrelated-session", "same-as-current": "fresh-session",
+                 "invalid-reference": "not valid"}.get(kind, "session")
+    with pytest.raises(Refused):
+        readmit(executor, sources, kwargs, reference)
+    assert rows(executor) == before
+
+
+def test_unknown_effect_stays_unknown_after_current_readmission(tmp_path):
+    executor, admitted, calls, sources, kwargs = make_rig(tmp_path)
+    old = executor.adapters[admitted[0].adapter]
+    def interrupted(inv):
+        calls.append(inv)
+        raise KeyboardInterrupt
+    executor.adapters[old.reference] = replace(old, invoke=interrupted)
+    with pytest.raises(KeyboardInterrupt):
+        executor.run(*admitted)
+    fresh_verifier(executor)
+    before = rows(executor)
+    renewed = readmit(executor, sources, kwargs)
+    assert rows(executor) == before
+    with pytest.raises(Refused):
+        executor.run(*renewed, retry=True)
+    state = executor.journal.load("mint-secret:fixture-client", renewed[2])
+    assert state.last.outcome is Outcome.UNCERTAIN
+    assert len(calls) == 1
+
+
+def test_current_session_revalidated_after_readmission_before_journal_write(tmp_path):
+    executor, admitted, _, sources, kwargs = make_rig(tmp_path)
+    executor.run(*admitted)
+    fresh_verifier(executor)
+    renewed = readmit(executor, sources, kwargs)
+    executor.evidence_verifier = lambda *_: False
+    before = rows(executor)
+    with pytest.raises(Refused, match="caller session no longer verified"):
+        executor.rollback(*renewed)
+    assert rows(executor) == before