fix(case): validate original intent and bind its running receipt

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa

src/warden/case_run.py

@@ -180,16 +180,76 @@ class OriginalIntent:
     """Durable record of the original mutating dispatch, written before dispatch.
 
     Binds the original Invocation (attempt, step, owner, target, actor, principal,
-    lane), all pins, the admitted adapter source and the tenant/realm provenance
-    the caller was admitted with. It is evidence of what Warden dispatched, not
-    owner proof of creation: owners still authenticate created-only provenance.
+    lane), all pins and the admitted adapter source. Current admit leaves tenant
+    and realm empty; their reserved fields are not admitted provenance. This is
+    local dispatch correspondence, not owner proof of creation. Owners still
+    authenticate original created-only provenance independently.
     """
 
     binding: str
     invocation: Invocation
     source: str
 
+    def validate(self) -> None:
+        """Fixed bounded metadata only; tenant/realm are not authority claims.
+
+        Current admit leaves tenant/realm empty. Nonempty future coordinates need
+        separate owner provenance; serializability alone must never supply it.
+        """
+        if (
+            type(self.invocation) is not Invocation
+            or type(self.invocation.pins) is not Pins
+        ):
+            raise Refused("exact original invocation required")
+        i = self.invocation
+        fields = (i.step, i.owner, i.target, i.principal, i.lane, i.attempt_id)
+        if (
+            not all(
+                type(value) is str and IDENTIFIER.fullmatch(value) for value in fields
+            )
+            or type(i.actor) is not str
+            or i.actor not in ("human", "agent")
+            or type(i.operation) is not str
+            or i.operation not in ("run", "rollback")
+            or not all(
+                type(value) is str and (not value or IDENTIFIER.fullmatch(value))
+                for value in (i.tenant, i.realm)
+            )
+            or type(self.source) is not str
+            or not DIGEST.fullmatch(self.source)
+            or type(self.binding) is not str
+            or self.binding != f"{i.step}:{i.target}"
+        ):
+            raise Refused("invalid original intent metadata")
+        if set(Invocation.__dataclass_fields__) != {
+            "step",
+            "owner",
+            "target",
+            "actor",
+            "principal",
+            "lane",
+            "pins",
+            "attempt_id",
+            "operation",
+            "tenant",
+            "realm",
+        }:
+            raise Refused("original invocation field contract changed")
+        if not all(
+            type(v) is str and DIGEST.fullmatch(v)
+            for v in (
+                i.pins.case,
+                i.pins.case_type,
+                i.pins.adapter,
+                i.pins.catalog,
+                i.pins.acceptance,
+                i.pins.run_sheet,
+            )
+        ):
+            raise Refused("invalid original intent pins")
+
     def payload(self) -> str:
+        self.validate()
         return json.dumps(
             {"schema": INTENT_SCHEMA, "binding": self.binding, "source": self.source}
             | asdict(self.invocation),
@@ -199,13 +259,44 @@ class OriginalIntent:
 
     @classmethod
     def decode(cls, raw: str) -> OriginalIntent:
-        obj = json.loads(raw)
-        fields = {f for f in Invocation.__dataclass_fields__}
-        if set(obj) != fields | {"schema", "binding", "source"} or obj["schema"] != INTENT_SCHEMA:
-            raise Refused("unknown original intent schema")
-        inv = {k: obj[k] for k in fields}
-        inv["pins"] = Pins(**inv["pins"])
-        return cls(obj["binding"], Invocation(**inv), obj["source"])
+        def unique(pairs):
+            result = {}
+            for key, value in pairs:
+                if key in result:
+                    raise Refused("duplicate original intent field")
+                result[key] = value
+            return result
+
+        try:
+            if type(raw) is not str or len(raw.encode()) > MAX_SOURCE_BYTES:
+                raise Refused("original intent input exceeds bound")
+            obj = json.loads(raw, object_pairs_hook=unique)
+            fields = {
+                "step",
+                "owner",
+                "target",
+                "actor",
+                "principal",
+                "lane",
+                "pins",
+                "attempt_id",
+                "operation",
+                "tenant",
+                "realm",
+            }
+            if (
+                type(obj) is not dict
+                or set(obj) != fields | {"schema", "binding", "source"}
+                or obj["schema"] != INTENT_SCHEMA
+            ):
+                raise Refused("unknown original intent schema")
+            inv = {k: obj[k] for k in fields}
+            inv["pins"] = Pins(**inv["pins"])
+            result = cls(obj["binding"], Invocation(**inv), obj["source"])
+            result.validate()
+            return result
+        except (TypeError, ValueError, KeyError):
+            raise Refused("invalid original intent record") from None
 
 
 @dataclass(frozen=True, slots=True)
@@ -256,28 +347,76 @@ class Journal:
             "CREATE TABLE IF NOT EXISTS intents (seq INTEGER PRIMARY KEY, attempt_id TEXT NOT NULL UNIQUE, binding TEXT NOT NULL, payload TEXT NOT NULL)"
         )
 
-    def original_intent(self, binding: str, current: Invocation, source: str) -> OriginalIntent:
+    def original_intent(
+        self, binding: str, current: Invocation, source: str
+    ) -> OriginalIntent:
         """Latest original run intent for binding, refused unless it matches exactly.
 
         Missing or mismatched records refuse; the step stays uncertain.
         """
         row = self.db.execute(
-            "SELECT payload FROM intents WHERE binding=? ORDER BY seq DESC LIMIT 1",
+            "SELECT attempt_id,payload FROM intents WHERE binding=? ORDER BY seq DESC LIMIT 1",
             (binding,),
         ).fetchone()
         if row is None:
             raise Refused("original intent record missing; step stays uncertain")
-        intent = OriginalIntent.decode(row[0])
+        intent = OriginalIntent.decode(row[1])
         o = intent.invocation
+        if row[0] != o.attempt_id:
+            raise Refused("original intent attempt identity differs")
         if intent.binding != binding or o.operation != "run":
             raise Refused("original intent binding mismatch")
         if o.pins != current.pins:
             raise Refused("original intent pins mismatch")
         if intent.source != source:
             raise Refused("original intent source mismatch")
-        for name in ("step", "owner", "target", "principal", "lane", "tenant", "realm"):
+        for name in (
+            "step",
+            "owner",
+            "target",
+            "actor",
+            "principal",
+            "lane",
+            "tenant",
+            "realm",
+        ):
             if getattr(o, name) != getattr(current, name):
                 raise Refused(f"original intent {name} mismatch")
+        # Match the persisted original running receipt. An orphan context or a
+        # payload from another attempt cannot become the current lookup key.
+        receipts = self.db.execute(
+            "SELECT payload FROM receipts WHERE binding=? ORDER BY seq DESC LIMIT ?",
+            (binding, MAX_HISTORY + 1),
+        ).fetchall()
+        if len(receipts) > MAX_HISTORY:
+            raise Refused("original intent receipt history exceeds bound")
+        matched = False
+        try:
+            for (raw,) in receipts:
+                if type(raw) is not str or len(raw.encode()) > MAX_SOURCE_BYTES:
+                    raise Refused("original receipt input exceeds bound")
+                payload = json.loads(raw)
+                if (
+                    payload.get("attempt_id") != o.attempt_id
+                    or payload.get("operation") != "run"
+                ):
+                    continue
+                last = payload.get("receipts", [])[-1]
+                if last.get("outcome") != "running":
+                    continue
+                if (
+                    payload.get("journal_schema") != "warden.case.step-journal.v1"
+                    or payload.get("step_id") != o.step
+                    or payload.get("owner") != o.owner
+                    or payload.get("pins") != asdict(o.pins)
+                ):
+                    raise Refused("original running receipt binding differs")
+                matched = True
+                break
+        except (TypeError, ValueError, KeyError, IndexError, AttributeError):
+            raise Refused("invalid original running receipt") from None
+        if not matched:
+            raise Refused("original running receipt missing; no backfill")
         return intent
 
     @contextmanager
@@ -355,6 +494,23 @@ class Journal:
         invocation: Invocation | None = None,
         intent: OriginalIntent | None = None,
     ):
+        if intent is not None:
+            if type(intent) is not OriginalIntent:
+                raise Refused("exact original intent required")
+            intent.validate()
+            expected = (
+                Outcome.RUNNING
+                if intent.invocation.operation == "run"
+                else Outcome.ROLLBACK_RUNNING
+            )
+            if (
+                intent.binding != binding
+                or intent.invocation != invocation
+                or intent.invocation.pins != state.pins
+                or not state.last
+                or state.last.outcome is not expected
+            ):
+                raise Refused("original intent and running receipt differ")
         self.db.execute("BEGIN IMMEDIATE")
         try:
             if intent is not None: