From 45cf4de7e6f7e1bd23ce1fba6631323bed713a3b Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Fri, 31 Jul 2026 22:44:58 +0530 Subject: [PATCH] Refuse a second session on another session's checkpoint thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionStore.create() guarded id reuse and let thread_id reuse straight through, and sharing a thread was an approval-gate bypass entirely inside the session API: the second session got a fresh record with no holds, resumed the first one's checkpointed boundary, and — because interrupt_before does not re-fire for a boundary it has already stopped at — ran the held gated node with no approval ever given, while the first session's record still said the hold was open. The guard is a SELECT inside the existing BEGIN IMMEDIATE transaction, so the refusal (ThreadInUseError, naming both the thread and the session holding it) is atomic with the insert: no session row and no transition row left behind, including under two processes racing on one store. No schema change — a UNIQUE index taken at open would brick an existing store already carrying duplicates. The default path, thread id equals session id, is unchanged: ids are unique, and a reused id still lands as SessionExistsError. Co-Authored-By: Claude Fable 5 --- grapharc/session/__init__.py | 2 ++ grapharc/session/errors.py | 18 +++++++++++++++++ grapharc/session/store.py | 31 ++++++++++++++++++++++++---- tests/test_session.py | 39 ++++++++++++++++++++++++++++++++++++ 4 files changed, 86 insertions(+), 4 deletions(-) diff --git a/grapharc/session/__init__.py b/grapharc/session/__init__.py index e3eb96a..4849fde 100644 --- a/grapharc/session/__init__.py +++ b/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", diff --git a/grapharc/session/errors.py b/grapharc/session/errors.py index 41db511..cdaadf9 100644 --- a/grapharc/session/errors.py +++ b/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", ] diff --git a/grapharc/session/store.py b/grapharc/session/store.py index 7b6659c..ca0df40 100644 --- a/grapharc/session/store.py +++ b/grapharc/session/store.py @@ -43,6 +43,7 @@ SessionBusy, SessionExistsError, SessionTerminated, + ThreadInUseError, UnknownSessionError, ) from grapharc.session.events import EventKind, SessionEvent @@ -316,9 +317,30 @@ 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, " @@ -326,7 +348,7 @@ def create( ( session_id, graph, - thread_id or session_id, + thread, SessionStatus.CREATED.value, now, now, @@ -334,8 +356,9 @@ def create( ), ) 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 diff --git a/tests/test_session.py b/tests/test_session.py index 3a9943d..a4a9744 100644 --- a/tests/test_session.py +++ b/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.