fix(session,server): lock-safe Session.Messages access + 409-busy guard (#3590)

benchmarks/compaction/main.go

@@ -233,7 +233,7 @@ func run(ctx context.Context) error {
 // compaction input trimmed of its keep-tail, truncated to the context budget,
 // wrapped in the canonical compaction system/user prompts.
 func buildCompactionMessages(sess *session.Session) []chat.Message {
-	messages, _ := sess.CompactionInput()
+	messages, _, _ := sess.CompactionInput()
 	for i := range messages {
 		messages[i].Cost = 0
 		messages[i].CacheControl = false

cmd/root/run.go

@@ -463,6 +463,10 @@ func (f *runExecFlags) runOrExec(ctx context.Context, out *cli.Printer, args []s
 		return f.handleExecMode(ctx, out, rt, sess, args)
 	}
 
+	// startSessionCoordinator wires the App's own RunStream calls into the
+	// same busy guard RunSession/AddMessage/UpdateMessage use, so a
+	// concurrent REST call is correctly rejected with 409 instead of racing
+	// the TUI's live stream (#3590); see app.WithStreamGuard.
 	coordinatorOpt, err := f.startSessionCoordinator(ctx, out, rt, sess)
 	if err != nil {
 		return err
@@ -1104,7 +1108,8 @@ func (f *runExecFlags) createSessionSpawner(agentSource config.Source, sessStore
 		if ctrl != nil {
 			appOpts = append(appOpts, app.WithSnapshotController(ctrl))
 		}
-		if coordinatorOpt := f.recallCoordinatorOpt(spawnCtx, localRt, newSess); coordinatorOpt != nil {
+		coordinatorOpt := f.recallCoordinatorOpt(spawnCtx, localRt, newSess)
+		if coordinatorOpt != nil {
 			appOpts = append(appOpts, coordinatorOpt)
 		}
 

cmd/root/run_listen.go

@@ -21,25 +21,32 @@ import (
 // in-process runtime even when the HTTP control plane is disabled. That lets a
 // background tool wake an idle local TUI by routing through the same injector
 // used by control-plane follow-ups.
+//
+// It also wires app.WithStreamGuard so the App's own direct RunStream calls
+// (Run/Retry/RunWithMessage) hold the same lock RunSession/AddMessage/
+// UpdateMessage use to detect an active stream, closing the gap where a
+// concurrent REST mutation could slip in during an attached/TUI stream (#3590).
 func (f *runExecFlags) recallCoordinatorOpt(ctx context.Context, rt runtime.Runtime, sess *session.Session) app.Opt {
 	sm := server.NewSessionManager(ctx, nil, rt.SessionStore(), 0, &f.runConfig)
-	sm.AttachRuntime(ctx, sess.ID, rt, sess)
+	guard := sm.AttachRuntime(ctx, sess.ID, rt, sess)
 	return func(a *app.App) {
+		app.WithStreamGuard(guard)(a)
 		sm.RegisterFollowUpInjector(sess.ID, a.InjectUserMessage)
 	}
 }
 
 // startSessionCoordinator wires local recall delivery for the in-process
 // runtime and, when --listen is set, exposes that runtime over HTTP so
 // external processes can drive the running TUI (steer, followup, resume, ...).
-// It returns an app.Opt that registers the App as the attached session owner.
+// It returns an app.Opt that registers the App as the attached session owner
+// (see recallCoordinatorOpt for the stream-guard wiring shared by both paths).
 func (f *runExecFlags) startSessionCoordinator(ctx context.Context, out *cli.Printer, rt runtime.Runtime, sess *session.Session) (app.Opt, error) {
 	if f.listenAddr == "" {
 		return f.recallCoordinatorOpt(ctx, rt, sess), nil
 	}
 
 	sm := server.NewSessionManager(ctx, nil, rt.SessionStore(), 0, &f.runConfig)
-	sm.AttachRuntime(ctx, sess.ID, rt, sess)
+	guard := sm.AttachRuntime(ctx, sess.ID, rt, sess)
 
 	ln, err := server.Listen(ctx, f.listenAddr)
 	if err != nil {
@@ -71,6 +78,7 @@ func (f *runExecFlags) startSessionCoordinator(ctx context.Context, out *cli.Pri
 	}()
 
 	return func(a *app.App) {
+		app.WithStreamGuard(guard)(a)
 		sm.RegisterEventSource(sess.ID, func(ctx context.Context, send func(any)) {
 			a.SubscribeWith(ctx, func(msg tea.Msg) {
 				if ev, ok := msg.(runtime.Event); ok {

docs/features/api-server/index.md

@@ -51,6 +51,8 @@ All endpoints are under the `/api` prefix.
 | `PATCH`  | `/api/sessions/:id/title`           | Update session title                                    |
 | `PATCH`  | `/api/sessions/:id/permissions`     | Update session permissions                              |
 | `POST`   | `/api/sessions/:id/fork`            | Fork a session at a user message — creates a new session with messages `[0, message_index)` of the parent (see [Session Forking](#session-forking)) |
+| `POST`   | `/api/sessions/:id/messages`        | Append a message directly to a session's history (bypasses the model). Returns `409 Conflict` while the session has an active run (see [Agent Execution](#agent-execution)). |
+| `PATCH`  | `/api/sessions/:id/messages/:msg_id` | Update an existing message by ID. Returns `409 Conflict` while the session has an active run. |
 | `POST`   | `/api/sessions/:id/resume`          | Resume a paused session (after tool confirmation)       |
 | `POST`   | `/api/sessions/:id/tools/toggle`    | Toggle auto-approve (YOLO) mode                         |
 | `POST`   | `/api/sessions/:id/elicitation`     | Respond to an MCP tool elicitation request              |

pkg/app/app.go

@@ -52,6 +52,7 @@ type App struct {
 	titleGenerating        atomic.Bool                 // True when title generation is in progress
 	titleGen               *sessiontitle.Generator     // Title generator for local runtime (nil for remote)
 	snapshotController     builtins.SnapshotController // Drives /undo, /snapshots, /reset; nil for runtimes that don't capture snapshots
+	streamGuard            sync.Locker                 // Held for the duration of every direct RunStream call; nil when not attached to a SessionManager (see WithStreamGuard)
 
 	startOnce  sync.Once
 	subsMu     sync.Mutex
@@ -121,6 +122,21 @@ func WithSnapshotController(c builtins.SnapshotController) Opt {
 	}
 }
 
+// WithStreamGuard makes the App hold l for the duration of every direct
+// RunStream call (Run, Retry, RunWithMessage). When the App is attached to a
+// SessionManager (the --listen control plane or the local recall
+// coordinator), l is the SAME lock RunSession and AddMessage/UpdateMessage
+// already use (via TryLock) to detect a session's active stream, so a
+// concurrent REST mutation correctly sees the App's own stream as busy
+// instead of racing it (#3590). Without this option (a bare App with no
+// attached SessionManager) there is nothing to race against, so omitting it
+// is safe.
+func WithStreamGuard(l sync.Locker) Opt {
+	return func(a *App) {
+		a.streamGuard = l
+	}
+}
+
 func New(ctx context.Context, rt runtime.Runtime, sess *session.Session, opts ...Opt) *App {
 	app := &App{
 		ctx:              func() context.Context { return context.WithoutCancel(ctx) },
@@ -474,6 +490,9 @@ func (a *App) Run(ctx context.Context, cancel context.CancelFunc, message string
 	}
 
 	go func() {
+		release := a.acquireStreamGuard()
+		defer release()
+
 		if len(attachments) > 0 {
 			multiContent := a.buildUserMultiContent(ctx, message, attachments)
 			a.session.AddMessage(session.UserMessage(message, multiContent...))
@@ -666,6 +685,22 @@ func (a *App) sendEvent(ctx context.Context, event tea.Msg) {
 	}
 }
 
+// acquireStreamGuard locks a.streamGuard (set via WithStreamGuard) and
+// returns the matching release func, or a no-op release when no guard is
+// attached (a bare App with no SessionManager to race against). Callers must
+// hold the lock for the entire direct RunStream call — acquire it before
+// mutating the session and defer the release only after the stream ends —
+// mirroring the order SessionManager.RunSession already uses so a concurrent
+// AddMessage/UpdateMessage/RunSession sees this stream as busy the same way
+// (#3590).
+func (a *App) acquireStreamGuard() func() {
+	if a.streamGuard == nil {
+		return func() {}
+	}
+	a.streamGuard.Lock()
+	return a.streamGuard.Unlock
+}
+
 // processInlineAttachment handles content that is already in memory (e.g. pasted
 // text). The content is appended to textBuilder wrapped in an XML tag for context.
 func (a *App) processInlineAttachment(att messages.Attachment, textBuilder *strings.Builder) {
@@ -690,6 +725,9 @@ func (a *App) Retry(ctx context.Context, cancel context.CancelFunc) {
 	a.cancel = cancel
 
 	go func() {
+		release := a.acquireStreamGuard()
+		defer release()
+
 		streamStarted := false
 		for event := range a.runtime.RunStream(ctx, a.session) {
 			// If context is cancelled, continue draining but don't forward events
@@ -742,6 +780,9 @@ func (a *App) RunWithMessage(ctx context.Context, cancel context.CancelFunc, msg
 	}
 
 	go func() {
+		release := a.acquireStreamGuard()
+		defer release()
+
 		a.session.AddMessage(msg)
 		for event := range a.runtime.RunStream(ctx, a.session) {
 			// If context is cancelled, continue draining but don't forward events

pkg/app/app_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -134,6 +135,80 @@ func (m *retryMockRuntime) RunStream(_ context.Context, sess *session.Session) <
 	return ch
 }
 
+// blockingRunStreamRuntime's RunStream blocks until release is closed (or
+// ctx is cancelled), letting tests observe a streamGuard held for the
+// stream's actual duration instead of racing a fake runtime that returns
+// immediately.
+type blockingRunStreamRuntime struct {
+	mockRuntime
+
+	release chan struct{}
+}
+
+func (r *blockingRunStreamRuntime) RunStream(ctx context.Context, _ *session.Session) <-chan runtime.Event {
+	ch := make(chan runtime.Event)
+	go func() {
+		defer close(ch)
+		select {
+		case <-r.release:
+		case <-ctx.Done():
+		}
+	}()
+	return ch
+}
+
+// TestApp_Run_HoldsStreamGuardForStreamDuration pins the #3590 fix: Run must
+// hold the WithStreamGuard lock for the entire direct RunStream call, not
+// just while scheduling it, so a concurrent SessionManager.AddMessage/
+// UpdateMessage/RunSession sharing the same lock (see AttachRuntime) sees
+// the attached stream as busy for as long as it is genuinely active.
+func TestApp_Run_HoldsStreamGuardForStreamDuration(t *testing.T) {
+	t.Parallel()
+
+	release := make(chan struct{})
+	rt := &blockingRunStreamRuntime{release: release}
+	var guard sync.Mutex
+
+	app := &App{
+		runtime:     rt,
+		session:     session.New(),
+		events:      make(chan tea.Msg, 16),
+		streamGuard: &guard,
+	}
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	app.Run(ctx, cancel, "hello", nil)
+
+	// isLocked probes guard without leaking a spurious lock acquisition:
+	// TryLock succeeding would otherwise leave the test goroutine holding
+	// the mutex, making the later "released" check hang forever.
+	isLocked := func() bool {
+		if guard.TryLock() {
+			guard.Unlock()
+			return false
+		}
+		return true
+	}
+
+	require.Eventually(t, isLocked, time.Second, time.Millisecond, "streamGuard should be held while RunStream is active")
+
+	close(release)
+
+	require.Eventually(t, func() bool { return !isLocked() }, time.Second, time.Millisecond, "streamGuard should be released once RunStream ends")
+}
+
+// TestApp_AcquireStreamGuard_NoopWhenUnset verifies that a bare App with no
+// WithStreamGuard option (the common case: no attached SessionManager) never
+// blocks on a nil lock.
+func TestApp_AcquireStreamGuard_NoopWhenUnset(t *testing.T) {
+	t.Parallel()
+
+	app := &App{}
+	release := app.acquireStreamGuard()
+	require.NotPanics(t, release)
+}
+
 func TestApp_Retry_SuppressesReEmittedUserMessage(t *testing.T) {
 	t.Parallel()
 

pkg/runtime/compactor/compactor.go

@@ -251,8 +251,8 @@ func RunLLM(ctx context.Context, args LLMArgs) (result *Result, err error) {
 // a hook supplies its own summary so the kept-tail policy stays
 // consistent across the two strategies.
 func ComputeFirstKeptEntry(sess *session.Session, contextLimit int64) int {
-	messages, sessIndices := gatherCompactionInput(sess)
-	return firstKeptSessionIndex(sess, sessIndices, compaction.SplitIndexForKeep(messages, keepTokenBudget(contextLimit)))
+	messages, sessIndices, itemCount := gatherCompactionInput(sess)
+	return firstKeptSessionIndex(sessIndices, itemCount, compaction.SplitIndexForKeep(messages, keepTokenBudget(contextLimit)))
 }
 
 // gatherCompactionInput is a thin wrapper around
@@ -272,13 +272,19 @@ func ComputeFirstKeptEntry(sess *session.Session, contextLimit int64) int {
 // past the prior summary, and tracking origin indices in sess.Messages
 // — lives on Session itself so it can run under sess.mu.RLock and stay
 // race-safe against concurrent AddMessage / ApplyCompaction calls.
-func gatherCompactionInput(sess *session.Session) ([]chat.Message, []int) {
-	messages, sessIndices := sess.CompactionInput()
+//
+// The returned itemCount is the SAME snapshot's total item count
+// (session.Session.CompactionInput's third return value); callers must use
+// it — not a fresh sess.ItemCount() call — for any out-of-range sentinel
+// derived from messages/sessIndices, or the boundary can describe a later
+// session length than the snapshot it is supposed to describe (#3590).
+func gatherCompactionInput(sess *session.Session) (messages []chat.Message, sessIndices []int, itemCount int) {
+	messages, sessIndices, itemCount = sess.CompactionInput()
 	for i := range messages {
 		messages[i].Cost = 0
 		messages[i].CacheControl = false
 	}
-	return messages, sessIndices
+	return messages, sessIndices, itemCount
 }
 
 // extractMessages returns the messages to send to the compaction
@@ -297,10 +303,10 @@ func gatherCompactionInput(sess *session.Session) ([]chat.Message, []int) {
 // (contextLimit − summary budget − prompt-overhead), older messages
 // are dropped from the front of the to-compact list to make room.
 func extractMessages(sess *session.Session, _ *agent.Agent, contextLimit int64, additionalPrompt string) ([]chat.Message, int) {
-	messages, sessIndices := gatherCompactionInput(sess)
+	messages, sessIndices, itemCount := gatherCompactionInput(sess)
 
 	splitIdx := compaction.SplitIndexForKeep(messages, keepTokenBudget(contextLimit))
-	firstKeptEntry := firstKeptSessionIndex(sess, sessIndices, splitIdx)
+	firstKeptEntry := firstKeptSessionIndex(sessIndices, itemCount, splitIdx)
 	messages = messages[:splitIdx]
 
 	systemPromptMessage := chat.Message{
@@ -337,13 +343,19 @@ func extractMessages(sess *session.Session, _ *agent.Agent, contextLimit int64,
 // firstKeptSessionIndex translates a split index produced against the
 // chat-message list returned by [gatherCompactionInput] back to an
 // index in sess.Messages, suitable for the new summary's
-// FirstKeptEntry. Out-of-range splits map to len(sess.Messages),
-// matching the "compact everything; keep nothing of the tail"
-// sentinel that session.buildSessionSummaryMessages handles by
-// skipping the conversation loop.
-func firstKeptSessionIndex(sess *session.Session, sessIndices []int, splitIdx int) int {
+// FirstKeptEntry. Out-of-range splits map to itemCount, matching the
+// "compact everything; keep nothing of the tail" sentinel that
+// session.buildSessionSummaryMessages handles by skipping the
+// conversation loop.
+//
+// itemCount MUST come from the same [session.Session.CompactionInput]
+// snapshot as sessIndices (i.e. via gatherCompactionInput), not a fresh
+// sess.ItemCount() call: the live session can already hold an append
+// that landed after the snapshot was taken, which would describe a
+// boundary the snapshot never had (#3590).
+func firstKeptSessionIndex(sessIndices []int, itemCount, splitIdx int) int {
 	if splitIdx >= len(sessIndices) {
-		return len(sess.Messages)
+		return itemCount
 	}
 	return sessIndices[splitIdx]
 }