fix(state): reject duplicate metadata CAS keys (W33)

agent/run_checkpoint_custody.py

@@ -0,0 +1,160 @@
+"""Private turn-bound handles; SessionDB remains the sole durable custody owner.
+
+No database is opened here and no renewal timer is created. Uncertain mutations
+retain a handle requiring reconciliation; neither teardown nor a later turn
+retries them. Tokens and lease holders never appear in returned summaries.
+"""
+from dataclasses import dataclass, field
+import os
+import threading
+
+from hermes_state_runs import RunCustody, RunCustodyError
+from scripts import run_checkpoint_claim as claim_client
+from scripts.run_checkpoint_claim import ClaimOutcomeUnknown, ClaimRefusal
+
+
+@dataclass
+class _Handle:
+    holder: str = field(repr=False)
+    value: RunCustody | None = field(default=None, repr=False)
+    status: str = "pending"
+    error: str = "CLAIM_OUTCOME_UNKNOWN"
+
+
+class TurnRunCustody:
+    """Per-agent capability handles, not a second persistent lease/fence."""
+
+    def __init__(self, db):
+        self.db = db
+        self._lock = threading.RLock()
+        self._active_holder = None
+        self._handles: dict[str, _Handle] = {}
+
+    def begin_turn(self, holder):
+        with self._lock:
+            if type(holder) is not str or not holder:
+                raise ClaimRefusal("INVALID_LIVE_BINDING")
+            if self._active_holder is not None:
+                raise ClaimRefusal("TURN_ALREADY_ACTIVE")
+            self._active_holder = holder
+
+    def _active(self, holder):
+        if not holder or self._active_holder != holder:
+            raise ClaimRefusal("TURN_NOT_ACTIVE")
+
+    def _handle(self, holder, run_id, generation):
+        handle = self._handles.get(run_id)
+        if handle is None:
+            raise ClaimRefusal("CUSTODY_HANDLE_REQUIRED")
+        if handle.status != "owned":
+            raise ClaimOutcomeUnknown(handle.error)
+        if handle.holder != holder:
+            raise ClaimRefusal("TURN_NOT_ACTIVE")
+        if handle.value is None or handle.value.generation != generation:
+            raise ClaimRefusal("FENCE_MISMATCH")
+        return handle
+
+    def claim(self, holder, *, session_id, run_id, **kwargs):
+        with self._lock:
+            self._active(holder)
+            if run_id in self._handles:
+                if self._handles[run_id].status != "owned":
+                    raise ClaimOutcomeUnknown(self._handles[run_id].error)
+                if self._handles[run_id].value.generation != kwargs.get("expected_generation"):
+                    raise ClaimRefusal("FENCE_MISMATCH")
+                raise ClaimRefusal("OWNER_ACTIVE")
+            if len(self._handles) >= 16:
+                raise ClaimRefusal("CUSTODY_HANDLE_LIMIT")
+            handle = _Handle(holder)
+            self._handles[run_id] = handle
+
+            def captured(value):
+                handle.value = value
+
+            try:
+                result = claim_client.claim_from_files(self.db, run_id=run_id,
+                    expected_session_id=session_id, expected_lease_holder=holder,
+                    controller_pid=os.getpid(), _on_claim=captured, **kwargs)
+            except ClaimRefusal:
+                del self._handles[run_id]
+                raise
+            except ClaimOutcomeUnknown as exc:
+                handle.status, handle.error = "unknown", str(exc)
+                raise
+            except BaseException:  # Cancellation can follow a committed native mutation.
+                handle.status, handle.error = "unknown", "CLAIM_OUTCOME_UNKNOWN"
+                raise ClaimOutcomeUnknown(handle.error) from None
+            if not isinstance(handle.value, RunCustody):
+                handle.status, handle.error = "unknown", "CLAIM_HANDLE_UNKNOWN"
+                raise ClaimOutcomeUnknown(handle.error)
+            handle.status = "owned"
+            return result
+
+    @staticmethod
+    def _summary(value, status):
+        return {"status": status, "run_id": value.run_id, "generation": value.generation,
+                "custody_changed": True, "resume_authorized": False,
+                "downstream_effects_executed": False}
+
+    def _mutate(self, handle, operation, **kwargs):
+        value = handle.value
+        # Mark uncertainty before crossing the native effect boundary, including
+        # cancellation between commit and Python return/readback.
+        handle.status, handle.error = "pending", "CUSTODY_OUTCOME_UNKNOWN"
+        try:
+            updated = operation(value.run_id, owner_token=value.owner_token,
+                                expected_generation=value.generation, **kwargs)
+        except RunCustodyError as exc:
+            handle.status = "owned"
+            raise ClaimRefusal(exc.code) from None
+        except BaseException:  # Cancellation can follow a committed native mutation.
+            handle.status, handle.error = "unknown", "CUSTODY_OUTCOME_UNKNOWN"
+            raise ClaimOutcomeUnknown(handle.error) from None
+        # Retain the native return before readback; an ACK/readback failure must
+        # not discard the token or be turned into an automatic second mutation.
+        handle.value = updated
+        try:
+            if self.db.read_run_custody(updated.run_id) != updated:
+                raise ClaimOutcomeUnknown("CUSTODY_READBACK_MISMATCH")
+        except BaseException:  # Cancellation can follow a committed native mutation.
+            handle.status, handle.error = "unknown", "CUSTODY_READBACK_UNKNOWN"
+            raise ClaimOutcomeUnknown(handle.error) from None
+        handle.status = "owned" if updated.disposition == "active" else "released"
+        return updated
+
+    def refresh(self, holder, *, run_id, expected_generation, ttl_seconds):
+        with self._lock:
+            self._active(holder)
+            handle = self._handle(holder, run_id, expected_generation)
+            value = self._mutate(handle, self.db.refresh_run_custody, ttl_seconds=ttl_seconds)
+            return self._summary(value, "refresh_observed")
+
+    def release(self, holder, *, run_id, expected_generation):
+        with self._lock:
+            handle = self._handle(holder, run_id, expected_generation)
+            value = self._mutate(handle, self.db.release_run_custody)
+            del self._handles[run_id]
+            return self._summary(value, "release_observed")
+
+    def finish_turn(self, holder):
+        with self._lock:
+            if not holder or self._active_holder != holder:
+                return []
+            # Close admission before any cleanup; late claims cannot outlive
+            # this turn even while the native session lease still exists.
+            self._active_holder = None
+            errors = []
+            for run_id, handle in list(self._handles.items()):
+                if handle.holder != holder:
+                    continue
+                if handle.status == "released":
+                    del self._handles[run_id]
+                    continue
+                if handle.status == "owned":
+                    try:
+                        self.release(holder, run_id=run_id, expected_generation=handle.value.generation)
+                    except (ClaimRefusal, ClaimOutcomeUnknown) as exc:
+                        handle.status, handle.error = "unknown", str(exc)
+                if handle.status != "owned" and run_id in self._handles:
+                    errors.append({"run_id": run_id, "status": "unknown", "code": handle.error})
+            return errors

apps/desktop/package.json

@@ -157,7 +157,7 @@
     "bippy": "0.5.43",
     "concurrently": "10.0.4",
     "cross-env": "10.1.0",
-    "electron": "40.10.2",
+    "electron": "41.10.3",
     "electron-builder": "^26.8.1",
     "esbuild": "^0.28.1",
     "eslint": "^9.39.4",
@@ -176,7 +176,7 @@
     "wait-on": "^9.0.5"
   },
   "build": {
-    "electronVersion": "40.10.2",
+    "electronVersion": "41.10.3",
     "appId": "com.recursiveintell.ares",
     "productName": "Ares",
     "executableName": "Ares",

hermes_state.py

@@ -94,6 +94,7 @@
     _PREVIEW_SCAFFOLDED_SQL,
 )
 from hermes_state_portability import SessionPortabilityMixin
+from hermes_state_runs import SessionRunCustodyMixin
 from hermes_state_schema import SessionSchemaMixin
 from hermes_state_search import SessionSearchMixin
 
@@ -4279,7 +4280,7 @@ def classify_session_status(
     return SESSION_STATUS_COMPLETE
 
 
-class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin):
+class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin, SessionRunCustodyMixin):
     """
     SQLite-backed session storage with FTS5 search.
 
@@ -14201,8 +14202,15 @@ def _do(conn):
     def compare_and_set_meta_many(
         self, items: List[Tuple[str, Optional[str], str]]
     ) -> bool:
-        """Atomically publish several meta rows if every preimage matches."""
+        """Atomically publish several meta rows if every preimage matches.
+
+        Each normalized key must occur only once. Otherwise two writes can
+        validate the same preimage and silently overwrite one another within
+        an apparently successful batch. Reject before entering the transaction.
+        """
         normalized = [(str(key), expected, str(value)) for key, expected, value in items]
+        if len({key for key, _expected, _value in normalized}) != len(normalized):
+            raise ValueError("duplicate keys in metadata compare-and-set batch")
         if not normalized:
             return True