diff --git a/pkg/app/app.go b/pkg/app/app.go index d23e6f082..b2d3bbdd7 100644 --- a/pkg/app/app.go +++ b/pkg/app/app.go @@ -421,6 +421,7 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill // so the supervisor marks the session idle. go func() { events := make(chan runtime.Event, defaultRuntimeEventBuffer) + var failed atomic.Bool go func() { defer close(events) result, err := a.runtime.RunSkillFork(ctx, a.session, skillstool.RunSkillArgs{ @@ -430,16 +431,41 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill switch { case errors.Is(err, runtime.ErrUnsupported): slog.WarnContext(ctx, "Runtime does not support fork-mode skills; skill not executed", "skill", skillName) + failed.Store(true) a.sendEvent(ctx, runtime.Error(fmt.Sprintf("Skill %q cannot run: this runtime does not support fork-mode skills.", skillName))) case err != nil: slog.ErrorContext(ctx, "Failed to run fork-mode skill", "skill", skillName, "error", err) + failed.Store(true) a.sendEvent(ctx, runtime.Error(fmt.Sprintf("Skill %q failed: %v", skillName, err))) case result != nil && result.IsError: + failed.Store(true) a.sendEvent(ctx, runtime.Error(result.Output)) } }() + // sawStop tracks ANY StreamStoppedEvent forwarded here — unlike + // forwardRunStreamEvents' root-only rule (#4136), every event on this + // channel belongs to the fork's own sub-session, so a root-session + // check would always be false here and synthesize a spurious + // duplicate stop on every fork run, successful or not. + var ( + sawStop bool + lastSessionID string + agentName string + ) for event := range events { + if scoped, ok := event.(runtime.SessionScoped); ok { + if id := scoped.GetSessionID(); id != "" { + lastSessionID = id + } + } + if name := event.GetAgentName(); name != "" { + agentName = name + } + if _, ok := event.(*runtime.StreamStoppedEvent); ok { + sawStop = true + } + if ctx.Err() != nil { if _, ok := event.(*runtime.StreamStoppedEvent); ok { // ctx is cancelled; detach cancellation but keep its trace @@ -450,6 +476,10 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill } a.sendEvent(ctx, event) } + + if !sawStop { + a.synthesizeStreamStopped(ctx, cmp.Or(lastSessionID, a.session.ID), agentName, failed.Load()) + } }() } @@ -758,7 +788,36 @@ func mustSkipMirroredElicitation(rt runtime.Runtime) bool { // RunWithMessage pass nil. func (a *App) forwardRunStreamEvents(ctx context.Context, ch <-chan runtime.Event, filter func(event runtime.Event) (forward bool)) { skipMirroredElicitation := mustSkipMirroredElicitation(a.runtime) + + // sawRootStop/sawRootError/agentName drive the #4136 fallback below: the + // runtime documents channel close (not receipt of StreamStoppedEvent) as + // the terminal signal, and may drop the event under back-pressure + // (LocalRuntime.finalizeEventChannel). Only a root-session stop counts — + // a sub-session (delegation/fork) stop leaves the root stream running. + var ( + sawRootStop bool + sawRootError bool + agentName string + ) + for event := range ch { + isRoot := isRootSessionEvent(event, a.session.ID) + if isRoot { + if name := event.GetAgentName(); name != "" { + agentName = name + } + switch event.(type) { + case *runtime.StreamStoppedEvent: + // Tracked here — before the cancellation/filter/dedupe checks + // below — so a filter that happens to veto a genuine root stop + // can never leave this bookkeeping thinking none arrived and + // synthesize a spurious duplicate. + sawRootStop = true + case *runtime.ErrorEvent: + sawRootError = true + } + } + // If context is cancelled, continue draining but don't forward events // — except StreamStoppedEvent, which must always propagate so the // supervisor can mark the session as no longer running. @@ -791,6 +850,51 @@ func (a *App) forwardRunStreamEvents(ctx context.Context, ch <-chan runtime.Even a.sendEvent(ctx, event) } + + if !sawRootStop { + a.synthesizeStreamStopped(ctx, a.session.ID, agentName, sawRootError) + } +} + +// isRootSessionEvent reports whether event belongs to the given root +// session rather than a sub-session (delegation or fork-skill child). +// Events that don't implement [runtime.SessionScoped], or that carry an +// empty SessionID, are treated as root-scoped by convention — matching how +// [runtime.SessionScoped] consumers elsewhere (e.g. the supervisor's +// isTopLevelStream) already interpret an absent session id. +func isRootSessionEvent(event runtime.Event, sessionID string) bool { + scoped, ok := event.(runtime.SessionScoped) + if !ok { + return true + } + id := scoped.GetSessionID() + return id == "" || id == sessionID +} + +// synthesizeStreamStopped sends a StreamStoppedEvent for sessionID when the +// runtime's event channel closed without forwarding one (#4136). Consumers +// (the chat page, the supervisor) clear their "Working…" state only on +// receipt of this event and treat channel close as the terminal signal only +// in the runtime's documentation, not in their own code — so a dropped +// event (see LocalRuntime.finalizeEventChannel's non-blocking emit) leaves +// them stuck indefinitely without this fallback. +func (a *App) synthesizeStreamStopped(ctx context.Context, sessionID, agentName string, sawError bool) { + reason := runtime.TurnEndReasonNormal + switch { + case ctx.Err() != nil: + reason = runtime.TurnEndReasonCanceled + case sawError: + reason = runtime.TurnEndReasonError + } + if agentName == "" { + agentName = a.runtime.CurrentAgentName(ctx) + } + slog.WarnContext(ctx, "runtime stream closed without a StreamStoppedEvent; synthesizing one", + "session_id", sessionID, "reason", reason) + // ctx may already be cancelled; detach cancellation but keep its trace + // context so the synthesized stop still reaches subscribers, mirroring + // the ctx-cancelled StreamStoppedEvent forwarding above. + a.sendEvent(context.WithoutCancel(ctx), runtime.StreamStopped(sessionID, agentName, reason)) } // acquireStreamGuard locks a.streamGuard (set via WithStreamGuard) and diff --git a/pkg/app/app_test.go b/pkg/app/app_test.go index d95e9318f..0721721b0 100644 --- a/pkg/app/app_test.go +++ b/pkg/app/app_test.go @@ -1240,3 +1240,215 @@ func TestElicitationDeliveryAcrossEntryPoints(t *testing.T) { }) } } + +// scriptedStreamMockRuntime replays a fixed sequence of events on RunStream +// and then closes the channel, without appending a StreamStoppedEvent +// unless the script itself contains one. This drives the #4136 regression +// tests below through every shape the drain loop must fall back for: no +// stop at all, a real root stop, a sub-session-only stop, a mid-stream +// cancellation, and a root error — uniformly across every entry point in +// elicitationEntryPoints. +type scriptedStreamMockRuntime struct { + mockRuntime + + script []runtime.Event + // cancelAfterScript, when set, is called after every scripted event has + // been pushed onto the channel and before it closes, simulating ctx + // being cancelled mid-stream (e.g. Esc) with the channel closing right + // after — without it ever carrying a StreamStoppedEvent of its own. + cancelAfterScript context.CancelFunc +} + +func (m *scriptedStreamMockRuntime) RunStream(_ context.Context, _ *session.Session) <-chan runtime.Event { + ch := make(chan runtime.Event, len(m.script)+1) + go func() { + defer close(ch) + for _, ev := range m.script { + ch <- ev + } + if m.cancelAfterScript != nil { + m.cancelAfterScript() + } + }() + return ch +} + +// collectUntilQuietWindow is how long collectUntilQuiet waits for a lull +// before deciding the producer is done. +const collectUntilQuietWindow = 500 * time.Millisecond + +// collectUntilQuiet drains events until no further message arrives within +// collectUntilQuietWindow, returning everything collected in order. The +// #4136 regression tests below have no single sentinel event to stop on (a +// synthesized StreamStoppedEvent looks just like a real one, and a +// sub-session stop can arrive before it), so a fixed quiet window is the +// simplest way to know forwardRunStreamEvents' goroutine has finished. +func collectUntilQuiet(t *testing.T, events <-chan tea.Msg) []tea.Msg { + t.Helper() + + var collected []tea.Msg + for { + select { + case msg := <-events: + collected = append(collected, msg) + case <-time.After(collectUntilQuietWindow): + return collected + } + } +} + +// streamStoppedEvents returns every *runtime.StreamStoppedEvent in msgs, in +// order. +func streamStoppedEvents(msgs []tea.Msg) []*runtime.StreamStoppedEvent { + var stops []*runtime.StreamStoppedEvent + for _, msg := range msgs { + if stop, ok := msg.(*runtime.StreamStoppedEvent); ok { + stops = append(stops, stop) + } + } + return stops +} + +// TestForwardRunStreamEvents_SynthesizesRootStreamStopped is the regression +// coverage for #4136 (TUI stuck on "Working…" after the stream ends): the +// runtime documents the channel close, not receipt of StreamStoppedEvent, +// as the terminal signal, and drops the event under back-pressure +// (LocalRuntime.finalizeEventChannel's non-blocking emit) — a fast local +// model streaming faster than the persistence observer can keep up +// reliably triggers this. Run/Retry/RunWithMessage all funnel through +// forwardRunStreamEvents, so every case is driven across all three via +// elicitationEntryPoints, mirroring TestElicitationDeliveryAcrossEntryPoints. +func TestForwardRunStreamEvents_SynthesizesRootStreamStopped(t *testing.T) { + t.Parallel() + + for _, ep := range elicitationEntryPoints { + t.Run(ep.name, func(t *testing.T) { + t.Parallel() + + t.Run("no stop event: synthesizes exactly one, reason normal, as the last event", func(t *testing.T) { + t.Parallel() + + sess := session.New() + rt := &scriptedStreamMockRuntime{script: []runtime.Event{ + runtime.StreamStarted(sess.ID, "mock"), + runtime.AgentChoice("mock", sess.ID, "partial content"), + }} + events := make(chan tea.Msg, 16) + app := &App{runtime: rt, session: sess, events: events} + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + ep.invoke(app, ctx, cancel) + + collected := collectUntilQuiet(t, events) + require.NotEmpty(t, collected, "the script's own events must still be forwarded") + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized") + assert.Equal(t, sess.ID, stops[0].SessionID) + assert.Equal(t, "normal", stops[0].Reason) + assert.Equal(t, "mock", stops[0].AgentName, "must fall back to the last observed root-session agent name") + assert.Same(t, tea.Msg(stops[0]), collected[len(collected)-1], + "the synthesized stop must be the last event forwarded") + }) + + t.Run("real root stop: no duplicate is synthesized", func(t *testing.T) { + t.Parallel() + + sess := session.New() + rt := &scriptedStreamMockRuntime{script: []runtime.Event{ + runtime.StreamStarted(sess.ID, "mock"), + runtime.AgentChoice("mock", sess.ID, "partial content"), + runtime.StreamStopped(sess.ID, "mock", "normal"), + }} + events := make(chan tea.Msg, 16) + app := &App{runtime: rt, session: sess, events: events} + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + ep.invoke(app, ctx, cancel) + + collected := collectUntilQuiet(t, events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "the real StreamStoppedEvent must not be duplicated") + assert.Equal(t, sess.ID, stops[0].SessionID) + assert.Equal(t, "normal", stops[0].Reason) + }) + + t.Run("sub-session stop only: root stop is still synthesized", func(t *testing.T) { + t.Parallel() + + sess := session.New() + const subSessionID = "sub-session" + rt := &scriptedStreamMockRuntime{script: []runtime.Event{ + runtime.StreamStarted(sess.ID, "mock"), + runtime.StreamStopped(subSessionID, "worker", "normal"), + }} + events := make(chan tea.Msg, 16) + app := &App{runtime: rt, session: sess, events: events} + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + ep.invoke(app, ctx, cancel) + + collected := collectUntilQuiet(t, events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 2, "the sub-session stop must not satisfy the root fallback") + assert.Equal(t, subSessionID, stops[0].SessionID, "the sub-session's own stop is forwarded first") + assert.Equal(t, sess.ID, stops[1].SessionID, "a root stop must still be synthesized") + assert.Equal(t, "normal", stops[1].Reason) + assert.Equal(t, "mock", stops[1].AgentName, + "must use the root StreamStarted's agent name, not the sub-session's") + assert.Same(t, tea.Msg(stops[1]), collected[len(collected)-1], + "the synthesized root stop must be the last event forwarded") + }) + + t.Run("ctx cancelled mid-stream: synthesized with reason canceled", func(t *testing.T) { + t.Parallel() + + sess := session.New() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + rt := &scriptedStreamMockRuntime{ + script: []runtime.Event{ + runtime.StreamStarted(sess.ID, "mock"), + runtime.AgentChoice("mock", sess.ID, "partial content"), + }, + cancelAfterScript: cancel, + } + events := make(chan tea.Msg, 16) + app := &App{runtime: rt, session: sess, events: events} + + ep.invoke(app, ctx, cancel) + + collected := collectUntilQuiet(t, events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized") + assert.Equal(t, sess.ID, stops[0].SessionID) + assert.Equal(t, "canceled", stops[0].Reason) + }) + + t.Run("root error event: synthesized with reason error", func(t *testing.T) { + t.Parallel() + + sess := session.New() + rt := &scriptedStreamMockRuntime{script: []runtime.Event{ + runtime.StreamStarted(sess.ID, "mock"), + runtime.ErrorForSession(sess.ID, "boom"), + }} + events := make(chan tea.Msg, 16) + app := &App{runtime: rt, session: sess, events: events} + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + ep.invoke(app, ctx, cancel) + + collected := collectUntilQuiet(t, events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized") + assert.Equal(t, sess.ID, stops[0].SessionID) + assert.Equal(t, "error", stops[0].Reason) + assert.Equal(t, "mock", stops[0].AgentName) + }) + }) + } +} diff --git a/pkg/app/skills_fork_test.go b/pkg/app/skills_fork_test.go index 5ce21937c..afd86fa11 100644 --- a/pkg/app/skills_fork_test.go +++ b/pkg/app/skills_fork_test.go @@ -33,6 +33,10 @@ type skillFakeRuntime struct { calls []skillstool.RunSkillArgs emitted []runtime.Event stopCall atomic.Bool + // skipStop, when true, omits the unconditional StreamStoppedEvent this + // mock otherwise emits — used by the #4136 regression tests below to + // exercise RunSkillFork's own channel-close fallback. + skipStop bool } func (f *skillFakeRuntime) CurrentAgentSkillsToolset() *skillstool.ToolSet { @@ -43,13 +47,17 @@ func (f *skillFakeRuntime) RunSkillFork(_ context.Context, sess *session.Session f.mu.Lock() f.calls = append(f.calls, args) emitted := f.emitted + skipStop := f.skipStop f.mu.Unlock() for _, ev := range emitted { sink.Emit(ev) } - // Always emit StreamStoppedEvent so the App's drain loop terminates. - sink.Emit(runtime.StreamStopped(sess.ID, "", "")) + if !skipStop { + // Emit StreamStoppedEvent so the App's drain loop terminates without + // relying on its own #4136 fallback (see skipStop for the opposite). + sink.Emit(runtime.StreamStopped(sess.ID, "", "")) + } f.stopCall.Store(true) return tools.ResultSuccess("done"), nil } @@ -247,3 +255,69 @@ func TestApp_SlashSkill_NonFork_E2E(t *testing.T) { assert.Empty(t, rt.recordedCalls(), "Runtime.RunSkillFork must not be called for non-fork skills") assert.False(t, rt.stopCall.Load()) } + +// TestApp_RunSkillFork_SynthesizesStreamStoppedWhenMissing is part of the +// #4136 regression coverage: RunSkillFork's drain loop mirrors +// forwardRunStreamEvents and has the identical gap. If the runtime's +// events channel closes without a StreamStoppedEvent (dropped under +// back-pressure, see LocalRuntime.finalizeEventChannel's non-blocking +// emit), the App must still synthesize one so the TUI's "Working…" +// indicator clears. +func TestApp_RunSkillFork_SynthesizesStreamStoppedWhenMissing(t *testing.T) { + t.Parallel() + + skill := writeSkill(t, "commit", true /* fork */, "# Commit\nPlease commit.\n") + st := skillstool.New([]skills.Skill{skill}, filepath.Dir(skill.FilePath)) + + rt := &skillFakeRuntime{ + mockRuntime: &mockRuntime{}, + skillset: st, + skipStop: true, + } + + sess := session.New() + a := New(t.Context(), rt, sess) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + a.RunSkillFork(ctx, cancel, "commit", "please commit", nil) + + require.Eventually(t, func() bool { return rt.stopCall.Load() }, time.Second, 10*time.Millisecond, + "RunSkillFork goroutine should finish even without an explicit stop event") + + collected := collectUntilQuiet(t, a.events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized") + assert.Equal(t, "normal", stops[0].Reason) + assert.Equal(t, sess.ID, stops[0].SessionID, + "with no sub-session id ever observed, the fallback must use the parent session's own id") + assert.Equal(t, "mock", stops[0].AgentName, "must fall back to Runtime.CurrentAgentName since no event carried one") +} + +// TestApp_RunSkillFork_DoesNotDuplicateRealStreamStopped pins the flip +// side: a fork run that does emit its own StreamStoppedEvent (the common +// case) must not also get a synthesized duplicate. +func TestApp_RunSkillFork_DoesNotDuplicateRealStreamStopped(t *testing.T) { + t.Parallel() + + skill := writeSkill(t, "commit", true /* fork */, "# Commit\nPlease commit.\n") + st := skillstool.New([]skills.Skill{skill}, filepath.Dir(skill.FilePath)) + + rt := &skillFakeRuntime{ + mockRuntime: &mockRuntime{}, + skillset: st, + } + + a := New(t.Context(), rt, session.New()) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + a.RunSkillFork(ctx, cancel, "commit", "please commit", nil) + + require.Eventually(t, func() bool { return rt.stopCall.Load() }, time.Second, 10*time.Millisecond, + "RunSkillFork goroutine should finish") + + collected := collectUntilQuiet(t, a.events) + stops := streamStoppedEvents(collected) + require.Len(t, stops, 1, "the real StreamStoppedEvent must not be duplicated") +} diff --git a/pkg/runtime/hooks.go b/pkg/runtime/hooks.go index b0ef5a212..6e53d8876 100644 --- a/pkg/runtime/hooks.go +++ b/pkg/runtime/hooks.go @@ -152,13 +152,35 @@ func (r *LocalRuntime) executeTurnStartHooks(ctx context.Context, sess *session. } // Reason values reported in [hooks.Input.Reason] when [hooks.EventTurnEnd] -// fires. The runtime guarantees that turn_end runs once per turn that -// fired turn_start, no matter how the turn exited; the reason classifies -// which exit path the runtime took. +// fires. The same values (including the exported trio below, plus +// hook_blocked/loop_detected/budget_exceeded) are also reported verbatim in +// [runtime.StreamStoppedEvent.Reason] — loop.go passes its turnEndReason* +// classification straight through to StreamStopped. The runtime guarantees +// that turn_end runs once per turn that fired turn_start, no matter how the +// turn exited; the reason classifies which exit path the runtime took. const ( - // turnEndReasonNormal — the model finished the turn cleanly and the + // TurnEndReasonNormal — the model finished the turn cleanly and the // run loop is about to break out (no further iterations). - turnEndReasonNormal = "normal" + TurnEndReasonNormal = "normal" + // TurnEndReasonError — the model call failed and the runtime is + // shutting down the run (handleStreamError returned non-retry). + TurnEndReasonError = "error" + // TurnEndReasonCanceled — the turn ended because the stream context + // was cancelled (e.g. user Ctrl+C). Includes deferred firing on + // any return path while ctx is done. + TurnEndReasonCanceled = "canceled" +) + +// The remaining turnEndReason values below also end up in +// StreamStoppedEvent.Reason via loop.go's streamReason/ls.exitReason +// plumbing, same as the exported trio above. They stay unexported because +// nothing outside pkg/runtime constructs one of these values directly — +// consumers (e.g. the TUI's isSuccessfulStop) only ever compare against the +// string, never need to produce it. +const ( + // turnEndReasonNormal aliases [TurnEndReasonNormal] to avoid churning + // the call sites below. + turnEndReasonNormal = TurnEndReasonNormal // turnEndReasonContinue — the turn finished cleanly and the loop is // about to start a new iteration (e.g. after tool calls, or after a // stop with a queued follow-up). @@ -166,13 +188,10 @@ const ( // turnEndReasonSteered — the turn finished and was followed by // drained steered messages, prompting a new iteration. turnEndReasonSteered = "steered" - // turnEndReasonError — the model call failed and the runtime is - // shutting down the run (handleStreamError returned non-retry). - turnEndReasonError = "error" - // turnEndReasonCanceled — the turn ended because the stream context - // was cancelled (e.g. user Ctrl+C). Includes deferred firing on - // any return path while ctx is done. - turnEndReasonCanceled = "canceled" + // turnEndReasonError aliases [TurnEndReasonError]. + turnEndReasonError = TurnEndReasonError + // turnEndReasonCanceled aliases [TurnEndReasonCanceled]. + turnEndReasonCanceled = TurnEndReasonCanceled // turnEndReasonHookBlocked — a hook (before_llm_call or // post_tool_use) signalled run termination via a deny verdict. turnEndReasonHookBlocked = "hook_blocked"