git.iko.soy/crush
on main help sign in
fix(server): complete background prompts that fail early
d411ab5Christian Rocha
internal/agent/agenttest/coordinator.go+1
compare with main
unifiedsplit
@@ -57,6 +57,7 @@ func NewCoordinator(
57 selected := config.SelectedModel{Provider: providerID, Model: modelID}
57 selected := config.SelectedModel{Provider: providerID, Model: modelID}
58 cfg.Config().Models[config.SelectedModelTypeLarge] = selected
58 cfg.Config().Models[config.SelectedModelTypeLarge] = selected
59 cfg.Config().Models[config.SelectedModelTypeSmall] = selected
59 cfg.Config().Models[config.SelectedModelTypeSmall] = selected
60+ cfg.SetupAgents()
60
61
61 // Keep buildTools light: no sub-agent or agentic-fetch construction.
62 // Keep buildTools light: no sub-agent or agentic-fetch construction.
62 coderCfg := cfg.Config().Agents[config.AgentCoder]
63 coderCfg := cfg.Config().Agents[config.AgentCoder]
internal/agent/coordinator.go+5
@@ -294,6 +294,11 @@ func (c *coordinator) run(ctx context.Context, accept *AcceptedRun, sessionID st
294
294
295 if hasLatest && c.runComplete != nil {
295 if hasLatest && c.runComplete != nil {
296 c.runComplete.PublishMustDeliver(ctx, pubsub.UpdatedEvent, latest)
296 c.runComplete.PublishMustDeliver(ctx, pubsub.UpdatedEvent, latest)
297+ // Signal to the dispatcher (backend.runAgent) that the
298+ // authoritative terminal RunComplete for this run was already
299+ // emitted, so it does not publish a duplicate fallback for the
300+ // error it is about to receive.
301+ MarkRunCompletePublished(ctx)
297 }
302 }
298 return result, originalErr
303 return result, originalErr
299}
304}
internal/agent/run_marker.gonew file+52
@@ -0,0 +1,52 @@
1+package agent
2+
3+import (
4+ "context"
5+ "sync/atomic"
6+)
7+
8+// runCompleteMarkerKey is the unexported context key carrying a
9+// [runCompleteMarker] from the dispatch boundary (backend.runAgent)
10+// down into the coordinator. It lets the dispatcher learn whether the
11+// coordinator already published the authoritative terminal
12+// notify.RunComplete for the run, so a fallback terminal event is only
13+// emitted when one is actually missing (e.g. an error returned before
14+// sessionAgent.Run ever executed). It avoids a breaking change to the
15+// Coordinator interface.
16+type runCompleteMarkerKey struct{}
17+
18+// runCompleteMarker records whether a terminal RunComplete has been
19+// published for a run. It is shared by pointer through the context so
20+// a publish deep in the call stack is observable by the dispatcher
21+// after the call returns.
22+type runCompleteMarker struct {
23+ published atomic.Bool
24+}
25+
26+// WithRunCompleteMarker returns ctx carrying a fresh marker the
27+// coordinator can flag via [MarkRunCompletePublished] once it emits the
28+// run's terminal RunComplete. Callers read the result with
29+// [RunCompletePublished]. Attaching the marker is optional: code paths
30+// without one simply skip the dedup signal.
31+func WithRunCompleteMarker(ctx context.Context) context.Context {
32+ return context.WithValue(ctx, runCompleteMarkerKey{}, &runCompleteMarker{})
33+}
34+
35+// MarkRunCompletePublished records that the authoritative terminal
36+// RunComplete has been published for the run carried by ctx. It is a
37+// no-op when no marker is present (e.g. the in-process/local Run path,
38+// which is not dispatched through backend.runAgent).
39+func MarkRunCompletePublished(ctx context.Context) {
40+ if m, ok := ctx.Value(runCompleteMarkerKey{}).(*runCompleteMarker); ok {
41+ m.published.Store(true)
42+ }
43+}
44+
45+// RunCompletePublished reports whether [MarkRunCompletePublished] was
46+// called on ctx's marker. It returns false when no marker is present.
47+func RunCompletePublished(ctx context.Context) bool {
48+ if m, ok := ctx.Value(runCompleteMarkerKey{}).(*runCompleteMarker); ok {
49+ return m.published.Load()
50+ }
51+ return false
52+}
internal/app/app.go+8
@@ -185,6 +185,14 @@ func (app *App) AgentNotifications() *pubsub.Broker[notify.Notification] {
185 return app.agentNotifications
185 return app.agentNotifications
186}
186}
187
187
188+// RunCompletions returns the broker for the authoritative per-run
189+// terminal RunComplete events. The dispatcher (backend.runAgent) uses
190+// it to emit a reliable terminal event when a run fails before the
191+// coordinator could publish one of its own.
192+func (app *App) RunCompletions() *pubsub.Broker[notify.RunComplete] {
193+ return app.runCompletions
194+}
195+
188// resolveSession resolves which session to use for a non-interactive run
196// resolveSession resolves which session to use for a non-interactive run
189// If continueSessionID is set, it looks up that session by ID
197// If continueSessionID is set, it looks up that session by ID
190// If useLast is set, it returns the most recently updated top-level session
198// If useLast is set, it returns the most recently updated top-level session
internal/backend/agent.go+33−5
@@ -61,14 +61,27 @@ func (b *Backend) SendMessage(workspaceID string, msg proto.AgentMessage) error
61// runAgent executes an accepted agent run for the workspace. It owns the
61// runAgent executes an accepted agent run for the workspace. It owns the
62// accept reservation (releasing it on return) and the runWG ticket added
62// accept reservation (releasing it on return) and the runWG ticket added
63// by SendMessage. The run is bound to the workspace context so its
63// by SendMessage. The run is bound to the workspace context so its
64−// lifetime is independent of any client's HTTP request. On a non-cancel
64+// lifetime is independent of any client's HTTP request.
65+//
65−// error it surfaces the failure to observers via a notify.TypeAgentError
66+// On a non-cancel error it surfaces the failure to observers via a
67+// notify.TypeAgentError notification (lossy, best-effort). That alone is
68+// not a reliable terminal signal: the agent-event fan-in uses lossy
69+// subscribers, so a `crush run` caller blocking on its RunID could hang
70+// if the event is dropped. To guarantee termination, when msg.RunID is
71+// non-empty and the coordinator did not already publish the run's
72+// authoritative terminal RunComplete (e.g. the error was returned before
73+// sessionAgent.Run executed, such as a readyWg or UpdateModels failure),
74+// runAgent emits an errored RunComplete on the must-deliver
75+// runCompletions broker so the waiter observes a deterministic terminal
66−// notification; context.Canceled is expected (the FinishReasonCanceled
76+// event. context.Canceled is expected (sessionAgent.Run already
67−// marker is already published by sessionAgent.Run) and swallowed.
77+// publishes the cancelled terminal marker) and produces no error
78+// terminal event.
68//
79//
69// When msg.RunID is non-empty it is attached to the context via
80// When msg.RunID is non-empty it is attached to the context via
70// agent.WithRunID so the coordinator can stamp the terminal
81// agent.WithRunID so the coordinator can stamp the terminal
71−// notify.RunComplete event with that correlator.
82+// notify.RunComplete event with that correlator. A run-complete marker
83+// is also attached so the coordinator can report whether it published
84+// the terminal event, letting runAgent avoid a duplicate fallback.
72func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.AcceptedRun) {
85func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.AcceptedRun) {
73 defer ws.runWG.Done()
86 defer ws.runWG.Done()
74 defer accept.Close()
87 defer accept.Close()
@@ -77,6 +90,7 @@ func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.
77 if msg.RunID != "" {
90 if msg.RunID != "" {
78 ctx = agent.WithRunID(ctx, msg.RunID)
91 ctx = agent.WithRunID(ctx, msg.RunID)
79 }
92 }
93+ ctx = agent.WithRunCompleteMarker(ctx)
80
94
81 _, err := ws.AgentCoordinator.RunAccepted(ctx, accept, msg.SessionID, msg.Prompt, proto.AttachmentsToMessage(msg.Attachments)...)
95 _, err := ws.AgentCoordinator.RunAccepted(ctx, accept, msg.SessionID, msg.Prompt, proto.AttachmentsToMessage(msg.Attachments)...)
82 if err == nil || errors.Is(err, context.Canceled) {
96 if err == nil || errors.Is(err, context.Canceled) {
@@ -89,6 +103,20 @@ func (b *Backend) runAgent(ws *Workspace, msg proto.AgentMessage, accept *agent.
89 Type: notify.TypeAgentError,
103 Type: notify.TypeAgentError,
90 Message: err.Error(),
104 Message: err.Error(),
91 })
105 })
106+
107+ // Reliable terminal fallback. Only needed when a RunID waiter
108+ // exists and the coordinator has not already emitted the run's
109+ // terminal RunComplete; otherwise this would be a duplicate.
110+ if msg.RunID == "" || agent.RunCompletePublished(ctx) {
111+ return
112+ }
113+ if rc := ws.RunCompletions(); rc != nil {
114+ rc.PublishMustDeliver(ctx, pubsub.UpdatedEvent, notify.RunComplete{
115+ SessionID: msg.SessionID,
116+ RunID: msg.RunID,
117+ Error: err.Error(),
118+ })
119+ }
92}
120}
93
121
94// GetAgentInfo returns the agent's model and busy status.
122// GetAgentInfo returns the agent's model and busy status.
internal/backend/agent_runcomplete_test.gonew file+162
@@ -0,0 +1,162 @@
1+package backend
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+ "time"
8+
9+ "charm.land/fantasy"
10+ "github.com/charmbracelet/crush/internal/agent"
11+ "github.com/charmbracelet/crush/internal/app"
12+ "github.com/charmbracelet/crush/internal/message"
13+ "github.com/charmbracelet/crush/internal/proto"
14+ "github.com/google/uuid"
15+ "github.com/stretchr/testify/require"
16+)
17+
18+// errorCoordinator is a minimal agent.Coordinator whose RunAccepted
19+// returns a configurable error. When markPublished is true it stamps
20+// the run-complete marker on the context before returning, simulating a
21+// real coordinator that already published the run's authoritative
22+// terminal RunComplete (so runAgent must not emit a duplicate fallback).
23+type errorCoordinator struct {
24+ err error
25+ markPublished bool
26+}
27+
28+func (c *errorCoordinator) Run(ctx context.Context, sessionID, prompt string, attachments ...message.Attachment) (*fantasy.AgentResult, error) {
29+ return nil, c.err
30+}
31+
32+func (c *errorCoordinator) RunAccepted(ctx context.Context, accept *agent.AcceptedRun, sessionID, prompt string, attachments ...message.Attachment) (*fantasy.AgentResult, error) {
33+ if c.markPublished {
34+ agent.MarkRunCompletePublished(ctx)
35+ }
36+ return nil, c.err
37+}
38+
39+func (c *errorCoordinator) BeginAccepted(sessionID string) *agent.AcceptedRun { return nil }
40+func (c *errorCoordinator) Cancel(string) {}
41+func (c *errorCoordinator) CancelAll() {}
42+func (c *errorCoordinator) IsBusy() bool { return false }
43+func (c *errorCoordinator) IsSessionBusy(string) bool { return false }
44+func (c *errorCoordinator) QueuedPrompts(string) int { return 0 }
45+func (c *errorCoordinator) QueuedPromptsList(string) []string { return nil }
46+func (c *errorCoordinator) ClearQueue(string) {}
47+func (c *errorCoordinator) Summarize(context.Context, string) error { return nil }
48+func (c *errorCoordinator) Model() agent.Model { return agent.Model{} }
49+func (c *errorCoordinator) UpdateModels(context.Context) error { return nil }
50+
51+// insertRunCompleteWorkspace installs a workspace backed by a real
52+// app.App (so the runCompletions broker exists) with the given
53+// coordinator and a workspace run context derived from base.
54+func insertRunCompleteWorkspace(t *testing.T, b *Backend, base context.Context, coord agent.Coordinator) *Workspace {
55+ t.Helper()
56+ a := app.NewForTest(base)
57+ a.AgentCoordinator = coord
58+ t.Cleanup(a.ShutdownForTest)
59+ ws := &Workspace{
60+ ID: uuid.New().String(),
61+ Path: t.TempDir(),
62+ resolvedPath: t.TempDir(),
63+ clients: make(map[string]*clientState),
64+ shutdownFn: func() {},
65+ }
66+ ws.App = a
67+ ws.ctx, ws.cancel = context.WithCancel(base)
68+ b.mu.Lock()
69+ b.workspaces.Set(ws.ID, ws)
70+ b.pathIndex[ws.resolvedPath] = ws.ID
71+ b.mu.Unlock()
72+ return ws
73+}
74+
75+// TestRunAgent_PreRunErrorPublishesTerminalRunComplete proves that an
76+// error returned from RunAccepted before the coordinator could publish
77+// its own terminal event (e.g. a readyWg or UpdateModels failure,
78+// modeled here by a stub coordinator) still yields a reliable terminal
79+// RunComplete for the run's RunID. Without it, a `crush run` caller
80+// blocking on that RunID would hang because the lossy TypeAgentError
81+// event is not a guaranteed terminal signal.
82+func TestRunAgent_PreRunErrorPublishesTerminalRunComplete(t *testing.T) {
83+ t.Parallel()
84+ b, _ := newTestBackend(t)
85+ runErr := errors.New("update models failed")
86+ ws := insertRunCompleteWorkspace(t, b, context.Background(), &errorCoordinator{err: runErr})
87+
88+ subCtx, cancel := context.WithCancel(context.Background())
89+ defer cancel()
90+ ch := ws.RunCompletions().Subscribe(subCtx)
91+
92+ err := b.SendMessage(ws.ID, proto.AgentMessage{SessionID: "S1", RunID: "run-1", Prompt: "hi"})
93+ require.NoError(t, err)
94+
95+ select {
96+ case ev := <-ch:
97+ require.Equal(t, "run-1", ev.Payload.RunID,
98+ "the terminal RunComplete must carry the dispatched RunID")
99+ require.Equal(t, "S1", ev.Payload.SessionID)
100+ require.Equal(t, runErr.Error(), ev.Payload.Error,
101+ "the fallback terminal event must be marked errored")
102+ require.False(t, ev.Payload.Cancelled)
103+ case <-time.After(2 * time.Second):
104+ t.Fatal("no terminal RunComplete published for a pre-run error; a run waiter would hang")
105+ }
106+}
107+
108+// TestRunAgent_NoFallbackWhenCoordinatorPublished ensures the fallback
109+// is suppressed when the coordinator already emitted the run's
110+// authoritative terminal RunComplete, so callers never observe a
111+// duplicate terminal event for the same RunID.
112+func TestRunAgent_NoFallbackWhenCoordinatorPublished(t *testing.T) {
113+ t.Parallel()
114+ b, _ := newTestBackend(t)
115+ runErr := errors.New("stream failed after publishing terminal event")
116+ ws := insertRunCompleteWorkspace(t, b, context.Background(),
117+ &errorCoordinator{err: runErr, markPublished: true})
118+
119+ subCtx, cancel := context.WithCancel(context.Background())
120+ defer cancel()
121+ ch := ws.RunCompletions().Subscribe(subCtx)
122+
123+ err := b.SendMessage(ws.ID, proto.AgentMessage{SessionID: "S1", RunID: "run-1", Prompt: "hi"})
124+ require.NoError(t, err)
125+
126+ // Wait for the dispatched run goroutine to return so any publish
127+ // has already happened.
128+ ws.runWG.Wait()
129+
130+ select {
131+ case ev := <-ch:
132+ t.Fatalf("runAgent published a duplicate terminal RunComplete: %+v", ev.Payload)
133+ case <-time.After(200 * time.Millisecond):
134+ }
135+}
136+
137+// TestRunAgent_CancellationPublishesNoErrorTerminal verifies that a
138+// context.Canceled result from RunAccepted produces no errored terminal
139+// RunComplete from runAgent: cancellation is sessionAgent.Run's
140+// responsibility (it publishes the cancelled marker) and the dispatcher
141+// must not synthesize an error terminal for it.
142+func TestRunAgent_CancellationPublishesNoErrorTerminal(t *testing.T) {
143+ t.Parallel()
144+ b, _ := newTestBackend(t)
145+ ws := insertRunCompleteWorkspace(t, b, context.Background(),
146+ &errorCoordinator{err: context.Canceled})
147+
148+ subCtx, cancel := context.WithCancel(context.Background())
149+ defer cancel()
150+ ch := ws.RunCompletions().Subscribe(subCtx)
151+
152+ err := b.SendMessage(ws.ID, proto.AgentMessage{SessionID: "S1", RunID: "run-1", Prompt: "hi"})
153+ require.NoError(t, err)
154+
155+ ws.runWG.Wait()
156+
157+ select {
158+ case ev := <-ch:
159+ t.Fatalf("cancellation must not publish a terminal RunComplete: %+v", ev.Payload)
160+ case <-time.After(200 * time.Millisecond):
161+ }
162+}