fix(chat): fence managed session resume by goal lifetime

loopx/attached_session.py

@@ -5,14 +5,19 @@
 import hashlib
 import math
 import time
-from collections.abc import Iterator, Mapping
-from contextlib import contextmanager
+from collections.abc import Callable, Iterator, Mapping
+from contextlib import AbstractContextManager, contextmanager, nullcontext
 from pathlib import Path
 from typing import Any
 
 from .chat import normalize_agent_response
-from .chat_store import CHAT_SESSION_MODE_ATTACHED, ChatSessionStore
+from .chat_store import (
+    CHAT_SESSION_MODE_ATTACHED,
+    CHAT_SESSION_MODE_MANAGED,
+    ChatSessionStore,
+)
 from .control_plane.effect_runtime import effect_runtime_result
+from .control_plane.collaboration import conversation_scope
 from .control_plane.goals.source_session_registry_state import (
     current_goal_ref,
     guard_path,
@@ -31,6 +36,7 @@
 ATTACHED_SESSION_UPSTREAM_MODE = "host_broker"
 MAX_ATTACHED_SESSION_CLAIM_WAIT_SECONDS = 1_800.0
 ATTACHED_SESSION_CLAIM_POLL_INTERVAL_SECONDS = 0.1
+ManagedSessionGoalCommit = Callable[[], AbstractContextManager[object]]
 
 
 def _binding_lock_path(
@@ -109,6 +115,7 @@ def _lifecycle_decision(
     session: Mapping[str, Any] | None = None,
     turn: Mapping[str, Any] | None = None,
     candidates: list[dict[str, Any]] | None = None,
+    subject: str = "attached",
 ) -> dict[str, Any]:
     facts: dict[str, Any] = {
         "operation": operation,
@@ -131,7 +138,7 @@ def _lifecycle_decision(
     if not isinstance(decision, dict):
         raise RuntimeError("Chat session lifecycle decision must be an object")
     if decision.get("kind") == "reject":
-        raise ValueError(f"attached Chat session rejected: {decision.get('code')}")
+        raise ValueError(f"{subject} Chat session rejected: {decision.get('code')}")
     return decision
 
 
@@ -528,6 +535,67 @@ def _attached_session_lifetime(
         yield session, registry, current_ref
 
 
+def managed_session_goal_commit(
+    *,
+    store: ChatSessionStore,
+    registry_path: Path | None,
+    session_id: str,
+) -> ManagedSessionGoalCommit:
+    """Return a repeatable short commit guard for managed owner-Goal resume."""
+    session = store.load_session(session_id)
+    if (
+        session is None
+        or session.get("status") == "closed"
+        or session.get("session_mode") != CHAT_SESSION_MODE_MANAGED
+        or conversation_scope(session)["kind"] != "owner_goal"
+    ):
+        return nullcontext
+    goal_id = str(session.get("goal_id") or "")
+    source_registry_path = _source_registry_path_or_legacy(
+        store=store,
+        registry_path=registry_path,
+        goal_id=goal_id,
+        agent_id=str(session.get("agent_id") or ""),
+        channel_id=str(session.get("channel_id") or f"goal.{goal_id}"),
+    )
+    if source_registry_path is None:
+        return nullcontext
+    registry = load_project_registry(source_registry_path)
+    if registry.get("profile_id") != SOURCE_SESSION_PROFILE_ID:
+        return nullcontext
+    original_identity = (goal_id, session.get("goal_instance_id"))
+
+    @contextmanager
+    def commit() -> Iterator[object]:
+        with exclusive_cross_runtime_file_lock(
+            guard_path(source_registry_path, goal_id),
+            operation="source_session_goal_lifetime",
+        ):
+            registry = load_project_registry(source_registry_path)
+            session = store.load_session(session_id)
+            if session is None or session.get("status") == "closed":
+                raise KeyError("chat session was not found")
+            if session.get("session_mode") != CHAT_SESSION_MODE_MANAGED:
+                raise ValueError("the selected Session is not a managed runtime session")
+            if (
+                registry.get("profile_id") != SOURCE_SESSION_PROFILE_ID
+                or (session.get("goal_id"), session.get("goal_instance_id"))
+                != original_identity
+            ):
+                raise ValueError("managed Chat session rejected: stale_goal_instance")
+            current_ref, _goal = current_goal_ref(registry, goal_id=goal_id)
+            _lifecycle_decision(
+                operation="admit",
+                registry=registry,
+                current_ref=current_ref,
+                session=session,
+                subject="managed",
+            )
+            yield session
+
+    return commit
+
+
 def select_current_attached_session(
     *,
     store: ChatSessionStore,

loopx/chat_loopx_mode.py

@@ -649,10 +649,27 @@ def _woken(intent, session_id, turn_id, *, created, now):
             created=created, woken_at=now,
         )
 
-    def recover(self, session_id, adapter):
+    @staticmethod
+    def observe_recovery(
+        adapter: Any,
+        goal_commit: Callable[[], Any],
+    ) -> dict[str, Any]:
+        # Reject a known-stale identity before pausing the native transport.
+        # Provider reads remain outside the short commit guard.
+        with goal_commit():
+            pass
         driver = CodexGoalDriver(adapter.session)
         driver.pause()
-        self.store.update_session(session_id, native_goal=driver.compact(driver.read()))
+        return driver.compact(driver.read())
+
+    def commit_recovery(
+        self,
+        session_id: str,
+        native_goal: dict[str, Any] | None,
+    ) -> None:
+        if native_goal is None:
+            return
+        self.store.update_session(session_id, native_goal=native_goal)
         for row in self.store.loopx_ingress(session_id):
             if row["status"] == "delivering":
                 self.store.update_ingress_receipt(session_id, row["client_ingress_id"], status="uncertain")