git.iko.soy/crush
on main help sign in
fix(server): ignore background errors from unrelated prompts

Carry the submitted prompt identity through error events so command-line runs fail only for their own prompt. Older session-scoped errors remain supported for compatibility. Co-Authored-By: Charm Crush <crush@charm.land>

commita6a8459c07cc2088cd1ce012171b0cc9c05a1339 authorChristian Rocha <christian@rocha.is> dateMon 1 Jun 2026, 18:11 (5 months ago) parent34995e9
8 files changed+156−4
compare with main
unifiedsplit
internal/agent/notify/notify.go+6 internal/backend/agent.go+1 internal/cmd/run.go+20−4 internal/cmd/run_stream_test.go+88 internal/proto/agent.go+7 internal/server/events.go+1 internal/server/events_test.go+32 internal/workspace/client_workspace.go+1
internal/agent/notify/notify.go+6
@@ -23,6 +23,12 @@ type Notification struct {
2323 SessionTitle string
2424 Type Type
2525 ProviderID string
26+ // RunID, when non-empty, is the caller-supplied correlator
27+ // (proto.AgentMessage.RunID) for the run that produced this
28+ // notification. It lets observers attribute a TypeAgentError to a
29+ // specific request rather than to any in-flight run on the
30+ // session. Empty when no caller set one.
31+ RunID string
2632 // Message carries the error text for TypeAgentError. Other
2733 // notification types ignore it.
2834 Message string
internal/backend/agent.go+1
@@ -85,6 +85,7 @@ func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.
8585
8686 ws.AgentNotifications().Publish(pubsub.CreatedEvent, notify.Notification{
8787 SessionID: msg.SessionID,
88+ RunID: msg.RunID,
8889 Type: notify.TypeAgentError,
8990 Message: err.Error(),
9091 })
internal/cmd/run.go+20−4
@@ -409,11 +409,27 @@ func (s *runStream) handle(ev any, stopSpinner func()) (done bool, err error) {
409409 return true, nil
410410
411411 case pubsub.Event[proto.AgentEvent]:
412− if e.Payload.Error != nil {
413− stop()
414− return true, fmt.Errorf("agent error: %w", e.Payload.Error)
412+ if e.Payload.Error == nil {
413+ return false, nil
415414 }
416− return false, nil
415+ // Attribute the error to our run before treating it as
416+ // fatal. Async errors from an unrelated workspace run share
417+ // this channel, so a foreign failure must not abort us:
418+ // - if the event carries a RunID, it is the authoritative
419+ // correlator: it must match our run exactly, otherwise it
420+ // belongs to a different request and we ignore it.
421+ // - if the event carries no RunID (older server), fall back
422+ // to SessionID: it must be present and match our session,
423+ // otherwise we ignore it.
424+ if e.Payload.RunID != "" {
425+ if e.Payload.RunID != s.runID {
426+ return false, nil
427+ }
428+ } else if e.Payload.SessionID == "" || e.Payload.SessionID != s.sessionID {
429+ return false, nil
430+ }
431+ stop()
432+ return true, fmt.Errorf("agent error: %w", e.Payload.Error)
417433 }
418434 return false, nil
419435}
internal/cmd/run_stream_test.go+88
@@ -2,6 +2,7 @@ package cmd
22
33import (
44 "bytes"
5+ "errors"
56 "testing"
67 "time"
78
@@ -307,6 +308,93 @@ func TestRunStream_RunIDSuppressesLiveMessagesAndPrintsRunComplete(t *testing.T)
307308 require.Equal(t, "streamed prefix final", buf.String())
308309}
309310
311+// TestRunStream_AgentErrorRunIDFiltersForeign verifies that an async
312+// agent error carrying a non-empty RunID is fatal only when it matches
313+// our run. A foreign RunID is ignored regardless of the event's
314+// SessionID, because RunID is the authoritative correlator and async
315+// errors share the agent event channel: without strict RunID matching
316+// an unrelated workspace failure would abort our run.
317+func TestRunStream_AgentErrorRunIDFiltersForeign(t *testing.T) {
318+ t.Parallel()
319+
320+ // Foreign RunID with a matching session is still foreign.
321+ s := &runStream{sessionID: "S", runID: "run-mine", out: &bytes.Buffer{}, read: map[string]int{}}
322+ done, err := s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
323+ Type: proto.AgentEventTypeError,
324+ SessionID: "S",
325+ RunID: "run-other",
326+ Error: errors.New("foreign boom"),
327+ }}, nil)
328+ require.NoError(t, err, "foreign RunID error must not abort our run")
329+ require.False(t, done)
330+
331+ // Foreign RunID with a different session is ignored.
332+ done, err = s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
333+ Type: proto.AgentEventTypeError,
334+ SessionID: "other",
335+ RunID: "run-other",
336+ Error: errors.New("foreign boom"),
337+ }}, nil)
338+ require.NoError(t, err, "foreign RunID error must not abort our run")
339+ require.False(t, done)
340+
341+ // Foreign RunID with a missing session is ignored.
342+ done, err = s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
343+ Type: proto.AgentEventTypeError,
344+ RunID: "run-other",
345+ Error: errors.New("foreign boom"),
346+ }}, nil)
347+ require.NoError(t, err, "foreign RunID error must not abort our run")
348+ require.False(t, done)
349+
350+ // Matching RunID is fatal.
351+ done, err = s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
352+ Type: proto.AgentEventTypeError,
353+ SessionID: "S",
354+ RunID: "run-mine",
355+ Error: errors.New("my boom"),
356+ }}, nil)
357+ require.Error(t, err, "matching RunID error must be fatal")
358+ require.True(t, done)
359+}
360+
361+// TestRunStream_AgentErrorNoRunIDFiltersBySession verifies the
362+// compatibility fallback: when the event carries no RunID, attribution
363+// falls back to SessionID. An error for another session or with an
364+// empty session is ignored, while an error for our own session is fatal
365+// so a real failure is never dropped.
366+func TestRunStream_AgentErrorNoRunIDFiltersBySession(t *testing.T) {
367+ t.Parallel()
368+
369+ s := &runStream{sessionID: "S", out: &bytes.Buffer{}, read: map[string]int{}}
370+
371+ // Empty RunID for another session is ignored.
372+ done, err := s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
373+ Type: proto.AgentEventTypeError,
374+ SessionID: "other",
375+ Error: errors.New("foreign boom"),
376+ }}, nil)
377+ require.NoError(t, err, "error for another session must not abort our run")
378+ require.False(t, done)
379+
380+ // Empty RunID with an empty session is ignored.
381+ done, err = s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
382+ Type: proto.AgentEventTypeError,
383+ Error: errors.New("foreign boom"),
384+ }}, nil)
385+ require.NoError(t, err, "error with no session must not abort our run")
386+ require.False(t, done)
387+
388+ // Empty RunID with a matching session is fatal.
389+ done, err = s.handle(pubsub.Event[proto.AgentEvent]{Payload: proto.AgentEvent{
390+ Type: proto.AgentEventTypeError,
391+ SessionID: "S",
392+ Error: errors.New("my boom"),
393+ }}, nil)
394+ require.Error(t, err, "error for our own session must be fatal")
395+ require.True(t, done)
396+}
397+
310398// TestRunStream_NoRunIDFallsBackToSessionID preserves the older
311399// behaviour for callers (and tests) that don't supply a RunID:
312400// SessionID-only matching still terminates the stream on the
internal/proto/agent.go+7
@@ -31,6 +31,13 @@ type AgentEvent struct {
3131 Message Message `json:"message"`
3232 Error error `json:"error,omitempty"`
3333
34+ // RunID echoes the caller-supplied AgentMessage.RunID for the run
35+ // that produced this event. It lets observers (notably
36+ // `crush run`) attribute an error event to a specific request
37+ // instead of to any in-flight run on the session. Empty when no
38+ // caller set one.
39+ RunID string `json:"run_id,omitempty"`
40+
3441 // When summarizing.
3542 SessionID string `json:"session_id,omitempty"`
3643 SessionTitle string `json:"session_title,omitempty"`
internal/server/events.go+1
@@ -89,6 +89,7 @@ func wrapEvent(ev any) *pubsub.Payload {
8989 payload := proto.AgentEvent{
9090 SessionID: e.Payload.SessionID,
9191 SessionTitle: e.Payload.SessionTitle,
92+ RunID: e.Payload.RunID,
9293 Type: proto.AgentEventType(e.Payload.Type),
9394 }
9495 if e.Payload.Type == notify.TypeAgentError {
internal/server/events_test.go+32
@@ -123,6 +123,38 @@ func TestRunCompleteToProto_RoundTrip(t *testing.T) {
123123 require.False(t, decoded.Payload.Cancelled)
124124}
125125
126+// TestAgentErrorToProto_PreservesRunID verifies that an async agent
127+// error notification carries its originating RunID (and SessionID)
128+// through the SSE envelope. Without these correlators, `crush run`
129+// cannot tell whether an error event belongs to its own run and
130+// would abort on any unrelated workspace failure.
131+func TestAgentErrorToProto_PreservesRunID(t *testing.T) {
132+ t.Parallel()
133+
134+ src := pubsub.Event[notify.Notification]{
135+ Type: pubsub.CreatedEvent,
136+ Payload: notify.Notification{
137+ SessionID: "S",
138+ RunID: "run-99",
139+ Type: notify.TypeAgentError,
140+ Message: "boom",
141+ },
142+ }
143+
144+ env := wrapEvent(src)
145+ require.NotNil(t, env)
146+ require.Equal(t, pubsub.PayloadTypeAgentEvent, env.Type)
147+
148+ var decoded pubsub.Event[proto.AgentEvent]
149+ require.NoError(t, json.Unmarshal(env.Payload, &decoded))
150+ require.Equal(t, proto.AgentEventTypeError, decoded.Payload.Type)
151+ require.Equal(t, "S", decoded.Payload.SessionID)
152+ require.Equal(t, "run-99", decoded.Payload.RunID,
153+ "RunID must survive so observers can attribute the error to its run")
154+ require.NotNil(t, decoded.Payload.Error)
155+ require.Equal(t, "boom", decoded.Payload.Error.Error())
156+}
157+
126158// TestRunCompleteToProto_Error verifies that error- and cancel-shaped
127159// RunComplete events round-trip cleanly so clients can distinguish
128160// "agent failed" (returns non-zero from `crush run`) from "agent
internal/workspace/client_workspace.go+1
@@ -706,6 +706,7 @@ func (w *ClientWorkspace) translateEvent(ev any) tea.Msg {
706706 n := notify.Notification{
707707 SessionID: e.Payload.SessionID,
708708 SessionTitle: e.Payload.SessionTitle,
709+ RunID: e.Payload.RunID,
709710 Type: notify.Type(e.Payload.Type),
710711 }
711712 if e.Payload.Error != nil {