Refuse a second session on another session's checkpoint thread

grapharc/session/__init__.py

@@ -23,6 +23,7 @@
     SessionError,
     SessionExistsError,
     SessionTerminated,
+    ThreadInUseError,
     UnknownGraphError,
     UnknownSessionError,
 )
@@ -68,6 +69,7 @@
     "SessionStore",
     "SessionTerminated",
     "StatusChange",
+    "ThreadInUseError",
     "TurnResult",
     "UnknownGraphError",
     "UnknownSessionError",

grapharc/session/errors.py

@@ -21,6 +21,23 @@ class SessionExistsError(SessionError):
     """A session with that id is already recorded in this store."""
 
 
+class ThreadInUseError(SessionError):
+    """The checkpoint thread already belongs to another session in this store.
+
+    A checkpoint thread belongs to exactly one session. A second session on the
+    same thread would resume the first one's checkpointed boundary with a fresh
+    record carrying no holds — and a gated node the first session is holding
+    would run with no approval ever given.
+    """
+
+    def __init__(self, thread_id: str, session_id: str) -> None:
+        super().__init__(
+            f"thread {thread_id!r} already belongs to session {session_id!r}"
+        )
+        self.thread_id = thread_id
+        self.session_id = session_id
+
+
 class UnknownGraphError(SessionError):
     """The session names a graph this process has not registered.
 
@@ -92,6 +109,7 @@ class SessionContractError(SessionError):
     "SessionError",
     "SessionExistsError",
     "SessionTerminated",
+    "ThreadInUseError",
     "UnknownGraphError",
     "UnknownSessionError",
 ]

grapharc/session/store.py

@@ -43,6 +43,7 @@
     SessionBusy,
     SessionExistsError,
     SessionTerminated,
+    ThreadInUseError,
     UnknownSessionError,
 )
 from grapharc.session.events import EventKind, SessionEvent
@@ -316,26 +317,48 @@ def create(
         thread_id: str | None = None,
         metadata: dict[str, Any] | None = None,
     ) -> SessionRecord:
-        """Record a new session. The graph's checkpoint thread defaults to the id."""
+        """Record a new session. The graph's checkpoint thread defaults to the id.
+
+        A checkpoint thread belongs to exactly one session: a `thread_id`
+        another session already holds is refused with `ThreadInUseError`,
+        checked and inserted under one write lock so two creates racing on one
+        thread cannot both land. Ids are unique, so the default path — thread
+        id equals session id — can never collide on a fresh id.
+        """
         now = _now()
+        thread = thread_id or session_id
         with self._transaction() as conn:
+            # Under BEGIN IMMEDIATE, so check-then-insert is one atomic claim.
+            # A second session on an existing thread would resume the first
+            # one's checkpointed boundary with a record carrying no holds, and
+            # a gated node the first session is holding would run unapproved —
+            # `interrupt_before` does not re-fire for a boundary it has
+            # already stopped at.
+            holder = conn.execute(
+                "SELECT id FROM sessions WHERE thread_id = ?", (thread,)
+            ).fetchone()
+            # A holder with this very id is an id reuse, not a thread theft:
+            # the insert below refuses it as `SessionExistsError`.
+            if holder is not None and holder["id"] != session_id:
+                raise ThreadInUseError(thread, holder["id"])
             try:
                 conn.execute(
                     "INSERT INTO sessions (id, graph, thread_id, status, created_at, "
                     "updated_at, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
                     (
                         session_id,
                         graph,
-                        thread_id or session_id,
+                        thread,
                         SessionStatus.CREATED.value,
                         now,
                         now,
                         json.dumps(metadata or {}),
                     ),
                 )
             except sqlite3.IntegrityError as exc:
-                # Reusing an id would silently attach a new session to another
-                # session's checkpoint thread.
+                # Same guarantee, id column: a checkpoint thread belongs to
+                # exactly one session, and reusing an id would silently attach
+                # a new session to another session's checkpoint thread.
                 raise SessionExistsError(
                     f"session {session_id!r} already exists in {self.path}"
                 ) from exc

tests/test_session.py

@@ -48,6 +48,7 @@
     SessionStatus,
     SessionStore,
     SessionTerminated,
+    ThreadInUseError,
     UnknownGraphError,
     UnknownSessionError,
     decision_in,
@@ -252,6 +253,44 @@ def test_reusing_a_session_id_is_refused(manager):
     assert len(manager.list()) == 1
 
 
+def test_reusing_another_sessions_thread_is_refused(manager):
+    """A checkpoint thread belongs to exactly one session, explicit ids included."""
+    manager.create(GRAPH_NAME, session_id="first", thread_id="shared")
+
+    with pytest.raises(ThreadInUseError) as caught:
+        manager.create(GRAPH_NAME, session_id="second", thread_id="shared")
+    # The error names both sides, so a caller can find the session it collided with.
+    assert caught.value.thread_id == "shared"
+    assert caught.value.session_id == "first"
+
+    # The refusal is atomic: no session row, no transition row left behind.
+    assert [r.id for r in manager.list()] == ["first"]
+    assert manager.store.get("second") is None
+    assert manager.store.history("second") == []
+
+
+def test_a_second_session_on_one_thread_cannot_run_a_held_gated_node(manager):
+    """Regression for the approval-gate bypass a shared thread used to open.
+
+    A second session on the first one's thread got a fresh record with no
+    holds, resumed the checkpointed boundary, and — `interrupt_before` does
+    not re-fire for a boundary it has already stopped at — ran the gated node
+    with no approval ever given, while the first session's record still said
+    the hold was open. The refusal at `create()` is what closes it.
+    """
+    a = manager.create(GRAPH_NAME, session_id="a", thread_id="shared-thread")
+    a.run()
+    assert a.status is SessionStatus.AWAITING_APPROVAL
+    assert [h.node for h in a.record.pending_approvals] == [APPROVAL_NODE]
+
+    with pytest.raises(ThreadInUseError, match="shared-thread"):
+        manager.create(GRAPH_NAME, session_id="b", thread_id="shared-thread")
+
+    # The gated node never ran, and the first session's hold is still the truth.
+    assert APPROVAL_NODE not in a.state()["log"]
+    assert a.status is SessionStatus.AWAITING_APPROVAL
+
+
 def test_a_session_terminated_mid_turn_is_reported_not_resurrected(manager, registry):
     """Terminal has to survive a turn that is still finishing.