@@ -112,6 +112,15 @@ type SessionAgentCall struct {
112112 // (in-process / local callers like AppWorkspace), behavior is
113113 // unchanged and no accept tracking applies.
114114 Accepted *AcceptedRun
115+ // acceptSeq carries the accept sequence of the handle that produced
116+ // this call after it has been enqueued and its Accepted handle
117+ // stripped. The queue-drain paths compare it against a session's
118+ // cancel mark so a follow-up queued before a cancel is dropped while
119+ // one queued after the cancel survives. 0 means untracked (an
120+ // in-process enqueue with no accept reservation), which the drain
121+ // paths treat as covered by any present mark, preserving the
122+ // pre-sequence behavior.
123+ acceptSeq uint64
115124}
116125
117126type SessionAgent interface {
@@ -167,20 +176,31 @@ type sessionAgent struct {
167176 // BeginAccepted increments it; only AcceptedRun.Close decrements
168177 // it.
169178 acceptedRuns *csync.Map[string, int]
170− // pendingCancels records sessions whose dispatched-but-not-yet-
171− // running call should observe a cancellation request. It is only
172− // set by Cancel when acceptedRuns > 0, so an idle Escape never
173− // poisons the next prompt.
174− pendingCancels *csync.Map[string, struct{}]
179+ // cancelMark records, per session, a high-water accept sequence: an
180+ // accepted handle is canceled by it iff the handle's sequence is at
181+ // or below the mark. Cancel raises the mark to the latest sequence
182+ // assigned at cancel time, so a single Cancel covers every prompt
183+ // accepted-but-not-yet-active then, while a prompt accepted later
184+ // (higher sequence) is never poisoned. Absent or 0 means no pending
185+ // cancel. It is only raised by Cancel when acceptedRuns > 0, so an
186+ // idle Escape never records a mark.
187+ cancelMark *csync.Map[string, uint64]
175188 // dispatchMuCreate guards lazy creation of per-session entries in
176189 // dispatchMu so two goroutines can't race to lock different mutex
177190 // instances for the same session.
178191 dispatchMuCreate sync.Mutex
179− // acceptedMu serializes increments/decrements of acceptedRuns. It
192+ // acceptedMu serializes increments/decrements of acceptedRuns and
193+ // the assignment of accept sequence numbers from acceptSeqGen. It
180194 // is separate from dispatchMu so AcceptedRun.Close (which may run
181195 // while Run holds dispatchMu for the same session) does not
182196 // deadlock by re-entering the dispatch lock.
183197 acceptedMu sync.Mutex
198+ // acceptSeqGen is the monotonic source of accept sequence numbers.
199+ // Each BeginAccepted increments it under acceptedMu and stamps the
200+ // returned handle, so sequences strictly increase in accept order
201+ // across the agent. Cancel uses its current value as the per-session
202+ // high-water mark.
203+ acceptSeqGen uint64
184204}
185205
186206type SessionAgentOptions struct {
@@ -218,7 +238,7 @@ func NewSessionAgent(
218238 activeRequests: csync.NewMap[string, context.CancelFunc](),
219239 dispatchMu: csync.NewMap[string, *sync.Mutex](),
220240 acceptedRuns: csync.NewMap[string, int](),
221− pendingCancels: csync.NewMap[string, struct{}](),
241+ cancelMark: csync.NewMap[string, uint64](),
222242 }
223243}
224244
@@ -231,7 +251,12 @@ func NewSessionAgent(
231251type AcceptedRun struct {
232252 agent *sessionAgent
233253 sessionID string
234− done atomic.Bool
254+ // seq is the monotonic accept sequence stamped by BeginAccepted. A
255+ // cancel covers this handle iff seq is at or below the session's
256+ // cancel mark, so a handle accepted after a cancel (higher seq) is
257+ // never poisoned by it.
258+ seq uint64
259+ done atomic.Bool
235260}
236261
237262// Close decrements the accept counter for this reservation. It is safe
@@ -263,19 +288,30 @@ func (a *sessionAgent) BeginAccepted(sessionID string) *AcceptedRun {
263288 defer a.acceptedMu.Unlock()
264289 count, _ := a.acceptedRuns.Get(sessionID)
265290 a.acceptedRuns.Set(sessionID, count+1)
266− return &AcceptedRun{agent: a, sessionID: sessionID}
291+ a.acceptSeqGen++
292+ return &AcceptedRun{agent: a, sessionID: sessionID, seq: a.acceptSeqGen}
267293}
268294
269295// endAccepted decrements the accept counter for sessionID. It is only
270296// called via AcceptedRun.Close. It uses a dedicated lock (not the
271297// per-session dispatch mutex) so it can run while Run holds dispatchMu
272298// for the same session without deadlocking.
299+//
300+// When the count reaches zero the session's cancel mark is dropped: no
301+// accepted handle remains for it to cover, and any handle accepted later
302+// gets a strictly higher sequence that the mark would not match anyway.
303+// Handles canceled on entry never reach RunComplete, so this is the only
304+// place that clears the mark for an all-canceled batch. Sibling handles
305+// covered by the same mark are serialized on the per-session dispatch
306+// mutex and read the mark before they Close, so this never clears it out
307+// from under a covered handle still waiting to enter Run.
273308func (a *sessionAgent) endAccepted(sessionID string) {
274309 a.acceptedMu.Lock()
275310 defer a.acceptedMu.Unlock()
276311 count, ok := a.acceptedRuns.Get(sessionID)
277312 if !ok || count <= 1 {
278313 a.acceptedRuns.Del(sessionID)
314+ a.cancelMark.Del(sessionID)
279315 return
280316 }
281317 a.acceptedRuns.Set(sessionID, count-1)
@@ -311,20 +347,44 @@ func (a *sessionAgent) enqueueCall(call SessionAgentCall) {
311347 existing = []SessionAgentCall{}
312348 }
313349 queued := call
350+ if call.Accepted != nil {
351+ // Preserve the accept sequence after the handle is stripped so
352+ // the queue-drain paths can tell a follow-up queued before a
353+ // cancel (covered by the mark) from one queued after it.
354+ queued.acceptSeq = call.Accepted.seq
355+ }
314356 queued.OnComplete = nil
315357 queued.Accepted = nil
316358 existing = append(existing, queued)
317359 a.messageQueue.Set(call.SessionID, existing)
318360}
319361
320−// clearPendingCancel removes any pending-cancel record for sessionID. It
321−// takes the per-session dispatch lock so it is ordered against Cancel and
322−// the dispatch handoff.
362+// clearPendingCancel removes any pending-cancel mark for sessionID. It
363+// takes the per-session dispatch lock so it is ordered against Cancel
364+// and the dispatch handoff.
323365func (a *sessionAgent) clearPendingCancel(sessionID string) {
324366 mu := a.sessionMu(sessionID)
325367 mu.Lock()
326368 defer mu.Unlock()
327− a.pendingCancels.Del(sessionID)
369+ a.cancelMark.Del(sessionID)
370+}
371+
372+// canceledBySeq reports whether an accepted handle or queued call with
373+// the given accept sequence is covered by a pending cancel for the
374+// session. Callers must hold the session's dispatch mutex. A tracked
375+// sequence (seq > 0) is covered only when it is at or below the cancel
376+// high-water mark, so a prompt accepted after the cancel (higher seq) is
377+// never poisoned. An untracked sequence (seq == 0, an in-process enqueue
378+// with no accept reservation) is covered whenever any mark is present,
379+// preserving the pre-sequence behavior. The mark is not consumed: it
380+// stays so every sibling handle it covers observes the same cancel, and
381+// a later handle (higher seq) ignores it regardless.
382+func (a *sessionAgent) canceledBySeq(sessionID string, seq uint64) bool {
383+ mark, ok := a.cancelMark.Get(sessionID)
384+ if !ok || mark == 0 {
385+ return false
386+ }
387+ return seq == 0 || seq <= mark
328388}
329389
330390// persistCanceledTurn writes the user/assistant records for a turn that
@@ -356,6 +416,26 @@ func (a *sessionAgent) persistCanceledTurn(ctx context.Context, call SessionAgen
356416 return a.messages.Update(writeCtx, assistant)
357417}
358418
419+// publishRunComplete emits the authoritative terminal event for a turn.
420+// It honors the per-call OnComplete hook when set (so the coordinator can
421+// coalesce retries) and otherwise falls back to the RunComplete broker.
422+// ctx is used only for the bounded-blocking must-deliver publish; the
423+// terminal payload is supplied by the caller. This is the single emit path
424+// shared by the streaming defer and the cancel-on-entry early return so a
425+// caller waiting on RunComplete (e.g. `crush run` with a RunID) always
426+// observes exactly one terminal event regardless of which Run branch ends
427+// the turn.
428+func (a *sessionAgent) publishRunComplete(ctx context.Context, call SessionAgentCall, complete notify.RunComplete) {
429+ if call.OnComplete != nil {
430+ call.OnComplete(complete)
431+ return
432+ }
433+ if a.runComplete == nil {
434+ return
435+ }
436+ a.runComplete.PublishMustDeliver(ctx, pubsub.UpdatedEvent, complete)
437+}
438+
359439// ValidateCall performs the cheap structural validation that
360440// sessionAgent.Run requires before a call can be dispatched: a call must
361441// carry either a non-empty prompt or a text attachment, and it must name a
@@ -394,22 +474,39 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (result *
394474 // Serialize the accepted -> (cancel-on-entry | queued |
395475 // active) transition against a concurrent Cancel. Cancel takes
396476 // the same per-session lock, so every cancel observes at least
397− // one of: pendingCancels, an activeRequests entry, or a
477+ // one of: a cancel mark, an activeRequests entry, or a
398478 // messageQueue entry it then clears.
399479 mu := a.sessionMu(call.SessionID)
400480 mu.Lock()
401481
402− if _, pending := a.pendingCancels.Get(call.SessionID); pending {
482+ if a.canceledBySeq(call.SessionID, call.Accepted.seq) {
403483 // Cancel-on-entry: a cancel arrived while this run was
404− // dispatched but not yet active. Consume the pending
405− // cancel, release the accept reservation, drop the lock,
406− // and persist a canceled turn without entering Stream.
407− a.pendingCancels.Del(call.SessionID)
484+ // dispatched but not yet active, and this handle's accept
485+ // sequence is at or below the session's cancel mark. The
486+ // mark is left in place so sibling handles it also covers
487+ // observe the same cancel; release the accept reservation,
488+ // drop the lock, and persist a canceled turn without
489+ // entering Stream.
490+ //
491+ // This path returns before the streaming defer that
492+ // publishes RunComplete is installed, so emit the terminal
493+ // event explicitly. Without it, a caller waiting on
494+ // RunComplete for this RunID (e.g. `crush run`, which
495+ // ignores message events and blocks on RunComplete) would
496+ // hang on an immediately-canceled accepted run.
408497 call.Accepted.Close()
409498 mu.Unlock()
499+ complete := notify.RunComplete{
500+ SessionID: call.SessionID,
501+ RunID: call.RunID,
502+ Cancelled: true,
503+ }
410504 if err := a.persistCanceledTurn(ctx, call, false); err != nil {
505+ complete.Error = err.Error()
506+ a.publishRunComplete(ctx, call, complete)
411507 return nil, err
412508 }
509+ a.publishRunComplete(ctx, call, complete)
413510 return nil, nil
414511 }
415512
@@ -579,14 +676,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (result *
579676 // the authoritative terminal event so a momentarily-full
580677 // subscriber channel can't silently drop it and hang
581678 // non-interactive clients waiting on RunComplete.
582− if call.OnComplete != nil {
583− call.OnComplete(complete)
584− return
585− }
586− if a.runComplete == nil {
587− return
588− }
589− a.runComplete.PublishMustDeliver(ctx, pubsub.UpdatedEvent, complete)
679+ a.publishRunComplete(ctx, call, complete)
590680 }()
591681
592682 history, files := a.preparePrompt(msgs, largeModel.CatwalkCfg.SupportsImages, call.Attachments...)
@@ -621,24 +711,26 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (result *
621711 // Use latest tools (updated by SetTools when MCP tools change).
622712 prepared.Tools = a.tools.Copy()
623713
624− // Drain queued follow-up prompts, but skip them if a cancel
625− // was recorded for the session while they sat in the queue:
626− // a cancel that arrived after the queue insertion must not
627− // let the queued prompt run as part of this step.
714+ // Drain queued follow-up prompts, but skip any covered by a
715+ // cancel recorded while they sat in the queue: a cancel that
716+ // arrived after a prompt was queued must not let it run as
717+ // part of this step. Coverage is per-call by accept sequence
718+ // so a follow-up queued after the cancel (higher seq) is
719+ // still folded in.
628720 dispatchLock := a.sessionMu(call.SessionID)
629721 dispatchLock.Lock()
630− _, canceled := a.pendingCancels.Get(call.SessionID)
631722 queuedCalls, _ := a.messageQueue.Get(call.SessionID)
632723 a.messageQueue.Del(call.SessionID)
633724 dispatchLock.Unlock()
634− if !canceled {
635− for _, queued := range queuedCalls {
636− userMessage, createErr := a.createUserMessage(callContext, queued)
637− if createErr != nil {
638− return callContext, prepared, createErr
639− }
640− prepared.Messages = append(prepared.Messages, userMessage.ToAIMessage()...)
725+ for _, queued := range queuedCalls {
726+ if a.canceledBySeq(call.SessionID, queued.acceptSeq) {
727+ continue
641728 }
729+ userMessage, createErr := a.createUserMessage(callContext, queued)
730+ if createErr != nil {
731+ return callContext, prepared, createErr
732+ }
733+ prepared.Messages = append(prepared.Messages, userMessage.ToAIMessage()...)
642734 }
643735
644736 prepared.Messages = a.workaroundProviderMediaLimitations(prepared.Messages, largeModel)
@@ -1008,20 +1100,37 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (result *
10081100 // closing the dequeue -> re-register window.
10091101 mu := a.sessionMu(call.SessionID)
10101102 mu.Lock()
1011− if _, pending := a.pendingCancels.Get(call.SessionID); pending {
1103+ queuedMessages, _ := a.messageQueue.Get(call.SessionID)
1104+ if mark, ok := a.cancelMark.Get(call.SessionID); ok && mark > 0 && len(queuedMessages) > 0 {
10121105 // A cancel was recorded for this session (e.g. it arrived while
1013− // this run was active and a follow-up had been accepted). Drop
1014− // the queue instead of running it and consume the marker.
1015− a.pendingCancels.Del(call.SessionID)
1106+ // this run was active and follow-ups had been queued). Drop the
1107+ // queued prompts it covers (accept sequence at or below the
1108+ // mark, or untracked); keep any queued after the cancel (higher
1109+ // sequence) so they still run.
1110+ var kept []SessionAgentCall
1111+ for _, q := range queuedMessages {
1112+ if q.acceptSeq == 0 || q.acceptSeq <= mark {
1113+ continue
1114+ }
1115+ kept = append(kept, q)
1116+ }
1117+ queuedMessages = kept
1118+ a.messageQueue.Set(call.SessionID, kept)
1119+ }
1120+ if len(queuedMessages) == 0 {
1121+ // No queued work. Clear the cancel mark only when no accepted
1122+ // run remains in flight that it might still cover; otherwise a
1123+ // sibling prompt (sequence at or below the mark) waiting to
1124+ // enter Run would lose its cancellation. When accepted runs are
1125+ // gone, this also clears a stale mark so it can't catch a
1126+ // future run.
10161127 a.messageQueue.Del(call.SessionID)
1017− mu.Unlock()
1018− return result, err
1019− }
1020− queuedMessages, ok := a.messageQueue.Get(call.SessionID)
1021− if !ok || len(queuedMessages) == 0 {
1022− // No queued work. Clear any stale pending-cancel entry as a
1023− // safety net so it can't catch a future run (no-op when unset).
1024− a.pendingCancels.Del(call.SessionID)
1128+ a.acceptedMu.Lock()
1129+ inFlight, _ := a.acceptedRuns.Get(call.SessionID)
1130+ a.acceptedMu.Unlock()
1131+ if inFlight == 0 {
1132+ a.cancelMark.Del(call.SessionID)
1133+ }
10251134 mu.Unlock()
10261135 return result, err
10271136 }
@@ -1619,17 +1728,27 @@ func (a *sessionAgent) Cancel(sessionID string) {
16191728 }
16201729
16211730 // Record a pending cancel only when a dispatched-but-not-yet-active
1622− // run exists. This catches a run still in the goroutine scheduler or
1731+ // run exists. This catches runs still in the goroutine scheduler or
16231732 // about to enter Run's busy-queue branch, while leaving an idle
16241733 // session untouched. Active and accepted are not mutually exclusive:
16251734 // when a run is active and a follow-up has been accepted, both the
16261735 // cancel above and this pending record fire.
1736+ //
1737+ // Raise the session's cancel mark to the latest accept sequence
1738+ // assigned so far. Every prompt currently accepted-but-not-yet-
1739+ // active has a sequence at or below that value, so one cancel covers
1740+ // all of them; a prompt accepted after this cancel gets a strictly
1741+ // higher sequence and is never poisoned. Using max keeps repeated
1742+ // cancels idempotent while the same prompts are in flight and lets a
1743+ // later cancel extend coverage to prompts accepted since.
16271744 a.acceptedMu.Lock()
16281745 count, ok := a.acceptedRuns.Get(sessionID)
1746+ mark := a.acceptSeqGen
16291747 a.acceptedMu.Unlock()
16301748 if ok && count > 0 {
1631− slog.Debug("Recording pending cancel for accepted run", "session_id", sessionID)
1632− a.pendingCancels.Set(sessionID, struct{}{})
1749+ slog.Debug("Recording cancel mark for accepted runs", "session_id", sessionID, "count", count, "mark", mark)
1750+ existing, _ := a.cancelMark.Get(sessionID)
1751+ a.cancelMark.Set(sessionID, max(existing, mark))
16331752 }
16341753
16351754 if a.QueuedPrompts(sessionID) > 0 {