a6a8459 · crush · git.iko.soy

git.iko.soy/crush

on main help sign in

fix(server): ignore background errors from unrelated prompts

a6a8459Christian Rocha5 months ago

internal/agent/notify/notify.go+6

compare with main

unifiedsplit

@@ -23,6 +23,12 @@ type Notification struct {

23 SessionTitle string

24 Type Type

25 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

26 // Message carries the error text for TypeAgentError. Other

32 // Message carries the error text for TypeAgentError. Other

27 // notification types ignore it.

33 // notification types ignore it.

28 Message string

34 Message string

internal/backend/agent.go+1

@@ -85,6 +85,7 @@ func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.

86 ws.AgentNotifications().Publish(pubsub.CreatedEvent, notify.Notification{

87 SessionID: msg.SessionID,

88+ RunID: msg.RunID,

88 Type: notify.TypeAgentError,

89 Type: notify.TypeAgentError,

89 Message: err.Error(),

90 Message: err.Error(),

90 })

91 })

internal/cmd/run.go+20−4

@@ -409,11 +409,27 @@ func (s *runStream) handle(ev any, stopSpinner func()) (done bool, err error) {

409 return true, nil

410

411 case pubsub.Event[proto.AgentEvent]:

412− if e.Payload.Error != nil {

412+ if e.Payload.Error == nil {

413− stop()

413+ return false, nil

414− return true, fmt.Errorf("agent error: %w", e.Payload.Error)

415 }

414 }

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 {

416− return false, nil

429+ return false, nil

430+ }

431+ stop()

432+ return true, fmt.Errorf("agent error: %w", e.Payload.Error)

417 }

433 }

418 return false, nil

434 return false, nil

419}

435}

internal/cmd/run_stream_test.go+88

@@ -2,6 +2,7 @@ package cmd

3import (

4 "bytes"

5+ "errors"

5 "testing"

6 "testing"

6 "time"

7 "time"

@@ -307,6 +308,93 @@ func TestRunStream_RunIDSuppressesLiveMessagesAndPrintsRunComplete(t *testing.T)

307 require.Equal(t, "streamed prefix final", buf.String())

308 require.Equal(t, "streamed prefix final", buf.String())

308}

309}

309

310

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+

310// TestRunStream_NoRunIDFallsBackToSessionID preserves the older

398// TestRunStream_NoRunIDFallsBackToSessionID preserves the older

311// behaviour for callers (and tests) that don't supply a RunID:

399// behaviour for callers (and tests) that don't supply a RunID:

312// SessionID-only matching still terminates the stream on the

400// SessionID-only matching still terminates the stream on the

internal/proto/agent.go+7

@@ -31,6 +31,13 @@ type AgentEvent struct {

31 Message Message `json:"message"`

32 Error error `json:"error,omitempty"`

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+

34 // When summarizing.

41 // When summarizing.

35 SessionID string `json:"session_id,omitempty"`

42 SessionID string `json:"session_id,omitempty"`

36 SessionTitle string `json:"session_title,omitempty"`

43 SessionTitle string `json:"session_title,omitempty"`

internal/server/events.go+1

@@ -89,6 +89,7 @@ func wrapEvent(ev any) *pubsub.Payload {

89 payload := proto.AgentEvent{

90 SessionID: e.Payload.SessionID,

91 SessionTitle: e.Payload.SessionTitle,

92+ RunID: e.Payload.RunID,

92 Type: proto.AgentEventType(e.Payload.Type),

93 Type: proto.AgentEventType(e.Payload.Type),

93 }

94 }

94 if e.Payload.Type == notify.TypeAgentError {

95 if e.Payload.Type == notify.TypeAgentError {

internal/server/events_test.go+32

@@ -123,6 +123,38 @@ func TestRunCompleteToProto_RoundTrip(t *testing.T) {

123 require.False(t, decoded.Payload.Cancelled)

124}

125

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+

126// TestRunCompleteToProto_Error verifies that error- and cancel-shaped

158// TestRunCompleteToProto_Error verifies that error- and cancel-shaped

127// RunComplete events round-trip cleanly so clients can distinguish

159// RunComplete events round-trip cleanly so clients can distinguish

128// "agent failed" (returns non-zero from `crush run`) from "agent

160// "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 {

706 n := notify.Notification{

707 SessionID: e.Payload.SessionID,

708 SessionTitle: e.Payload.SessionTitle,

709+ RunID: e.Payload.RunID,

709 Type: notify.Type(e.Payload.Type),

710 Type: notify.Type(e.Payload.Type),

710 }

711 }

711 if e.Payload.Error != nil {

712 if e.Payload.Error != nil {