fix(session,server): lock-safe Session.Messages access + 409-busy guard (#3590)#3606
Draft
aheritier wants to merge 8 commits into
Draft
fix(session,server): lock-safe Session.Messages access + 409-busy guard (#3590)#3606aheritier wants to merge 8 commits into
aheritier wants to merge 8 commits into
Conversation
pkg/runtime/loop.go, runtime.go, and compactor.go read sess.Messages (or len(sess.Messages)) directly in several hot paths, racing a concurrent HTTP AddMessage or compaction on the same live session and risking a wrong UserMessageEvent.SessionPosition or an index-out-of-range panic: - appendSteerAndEmit and the follow-up injection path in loop.go now use AddMessage's returned index instead of a separate len(sess.Messages)-1 read. - The initial-turn UserMessage emission in loop.go and firstKeptSessionIndex in compactor.go now use sess.ItemCount() instead of len(sess.Messages). - EmitStartupInfo's session-restore LastMessage reconstruction in runtime.go now iterates a MessagesSnapshot() instead of sess.Messages. Refs #3590
InMemorySessionStore.GetSession returns the live, shared *Session pointer, not a copy. userMessageOrdinalToItemIndex iterated s.Messages directly to translate a user-message ordinal into an item index, racing a concurrent AddMessage on that same session (e.g. the HTTP AddMessage handler racing a TUI fork action) and risking an index computed against a torn read. It now walks s.MessagesSnapshot() instead. Refs #3590
session.Session.mu now makes AddMessage/branch/fork race-free against a live RunStream, but a message injected or edited mid-stream (mid-tool-call in particular) can still desynchronize the in-flight turn from what the model/tools expect. As defense in depth on top of the locking, reject the mutation outright while the session has an active stream. The check reuses the same activeRuntimes.streaming lock RunSession already uses to serialize concurrent RunSession calls (TryLock/Unlock), so it can never race a stream that starts between the check and the mutation: both require sm.mux, which AddMessage/UpdateMessage hold for their entire body. The existing ErrSessionBusy sentinel (already mapped to 409 for runAgent) is reused and now also returned by AddMessage/UpdateMessage; server.go maps it to 409 Conflict for both endpoints. Documents the two message endpoints (previously missing from the API reference) and their new 409 behavior. Refs #3590
…rship AddMessage/UpdateMessage/RunSession detect an active stream via activeRuntimes.streaming, but that lock was only ever acquired by RunSession itself. Runtimes registered via AttachRuntime (the TUI's --listen control plane and local recall coordinator) stream directly through pkg/app.App.Run/Retry/RunWithMessage, which called runtime.RunStream without ever touching that lock, so a REST AddMessage/UpdateMessage (or a second RunSession) during a genuinely active attached/TUI stream wrongly succeeded instead of 409ing. AttachRuntime now returns the same lock (sync.Locker) RunSession and AddMessage/UpdateMessage already TryLock. A new app.WithStreamGuard option lets the App hold it for the duration of every direct RunStream call, acquired before mutating the session and released only once the stream ends, mirroring RunSession's own ordering. cmd/root wires this option into both the --listen and local-recall paths. Adds SessionManager and App-level regression tests: driving a real attached stream through the returned lock (not through RunSession) now gets AddMessage/UpdateMessage to correctly return ErrSessionBusy, and a blocking-runtime App test pins that the lock is held for the stream's actual duration, not just while scheduling it. Refs #3590
…own snapshot count gatherCompactionInput took a locked snapshot via Session.CompactionInput, but firstKeptSessionIndex's out-of-range sentinel (the 'compact everything, keep nothing' boundary) called sess.ItemCount() again instead of reusing that snapshot's own length. ItemCount is lock-safe on its own, but it can observe an AddMessage/ApplyCompaction that landed after CompactionInput's snapshot was taken, so the sentinel could describe a longer session than the one messages/sessIndices actually cover, breaking the invariant that messages appended after the snapshot stay in the kept tail. CompactionInput now returns the snapshot's own item count alongside messages and sessIndices; gatherCompactionInput and firstKeptSessionIndex thread that count through instead of re-querying the live session, so the sentinel always describes the SAME snapshot the split index was computed against. Adds a deterministic test (no goroutines/-race needed: this was a logic error, not a data race) that gathers the compaction input, lets a message race in afterwards, and asserts the out-of-range sentinel still matches the original snapshot count rather than the now-longer live session. Also replaces the now-obsolete TestFirstKeptSessionIndexConcurrent (firstKeptSessionIndex no longer touches the session) with TestGatherCompactionInputConcurrent, which keeps -race coverage on gatherCompactionInput's own snapshot read. Refs #3590
…tion AddMessage/UpdateMessage TryLock activeRuntimes.streaming to reject a mutation while a stream is active, but released the lock immediately after the check, before performing the append/update. Holding sm.mux for the rest of the method does not close the gap for attached runtimes (see AttachRuntime): App.acquireStreamGuard only ever acquires streaming, never sm.mux, so an attached stream could start after the busy check passed but before/during the REST mutation. Both methods now hold streaming via defer until the mutation (store write, plus the in-memory session update for AddMessage) has fully completed. TryLock stays non-blocking, so a busy REST call still returns immediately instead of waiting on an in-progress attached stream, and the attached guard's plain Lock() cannot deadlock against it. Adds a deterministic blocking-store regression test (mirroring the reviewer's TestReview_AttachedGuardCannotStartBetweenBusyCheckAndMutation probe) for both AddMessage and UpdateMessage: a fake store blocks mid-write so the test can assert the attached-stream guard stays blocked until the mutation returns. Confirmed both tests fail without the defer fix and pass with it. Refs #3590
TestApp_Run_HoldsStreamGuardForStreamDuration declares 'var guard sync.Mutex' but the file never imported sync, so the package failed to compile (pkg/app/app_test.go:169:12: undefined: sync). task dev and the -race suite could not even build. Refs #3590
aheritier
added
area/api
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes #3590 —
Session.muis documented to protectMessages, but severalhot paths read/iterated
sess.Messageswithout the lock, racing with HTTPAddMessage/compaction that mutate a live session. Risks: data races,wrong
SessionPositionin emitted events, and index-out-of-range panics ifcompaction truncates concurrently.
What changed
Lock-safe accessors (
pkg/session)Session.AddMessagenow returns the appended item's index, computed underthe same write lock as the append (no TOCTOU) — callers no longer do a racy
len(sess.Messages)-1.Session.ItemCount()— lockedlen(Messages).Session.MessagesSnapshot()— exported, deep-copying snapshot (mirrors theexisting
snapshotItemsprecedent: copy under lock, release, then act).Call sites fixed
pkg/runtime/loop.go(steer/follow-up/initial UserMessage emission)pkg/runtime/runtime.go(session-restore LastMessage reconstruction)pkg/runtime/compactor/compactor.go(firstKeptSessionIndex boundary)pkg/server/session_manager.goForkSession — walks a locked snapshotinstead of the live shared
*Session.Messagespkg/session/branch.go— branch/fork paths take the RLock likeClone409-busy guard
AddMessage/UpdateMessagenow reuse the existingactiveRuntimes.streamingmutex +
ErrSessionBusysentinel thatRunSessionalready uses, held acrossthe full mutation so a stream cannot start between the busy check and the
write. Coverage was extended to attached/TUI direct-
RunStreamownership(new
app.WithStreamGuard), which previously bypassed the guard. Handlers mapErrSessionBusy→409 Conflict, mirroringrunAgent.Compaction snapshot consistency
Session.CompactionInputnow returns the snapshot's own item count, threadedthrough to the out-of-range sentinel so it always describes the compaction
snapshot rather than a later live length.
Testing
New
-raceregression tests across pkg/session, pkg/runtime,pkg/runtime/compactor, pkg/server, pkg/app — covering concurrent
AddMessage/branch/fork vs mutation, EmitStartupInfo vs AddMessage, the
attached-stream 409 guard (blocking-store, both AddMessage and UpdateMessage),
and the compaction snapshot-count invariant. All verified to fail against the
pre-fix code and pass now.
task dev(build + full tests + lint + 19 custom cops): greengo test -race -count=2 ./pkg/session/... ./pkg/runtime/... ./pkg/server/... ./pkg/app/...: greenNotes
TryLock, so a busy RESTcall returns 409 immediately rather than waiting on the App-held stream lock;
no path re-acquires the streaming mutex while held.
syncimport topkg/app/app_test.go);merge-tree check confirms a clean auto-merge.