From e490fb6d30926c906ba1375cbd2c2a5e80aa23a7 Mon Sep 17 00:00:00 2001 From: ysyneu Date: Tue, 10 Nov 2026 18:49:46 -0800 Subject: [PATCH] feat(safari): steer follow-ups into a running turn and watch the session A message sent while the session's turn is still running no longer waits for that turn to finish. `safari send` reads turn_in_flight from session/get and sets steering only then, so the running turn takes the message in at its next step and answers it; an idle session still runs the message as a turn of its own. The answer to a steered message comes from the running turn, under that turn's invocation, so `safari watch` now follows the session instead of one invocation: - --root-invocation-id marks where output starts (the request's first frame) and which request must settle before the watch can end; replies from every turn after that point are printed. - A turn that does not answer through the reply tool (one started from the web console) has its final answer, named by the settle's final_event_id, printed as a result reply. - A steered settle is written before the running turn answers, so it never ends the watch; the next non-steered settle with no standing tasks, followed by an idle session/get, does. - Re-arming with --root-invocation-id and --since replays the session from its start and skips everything up to the cursor silently, so the request's own settle is still seen. A done candidate inside the skipped part only ends the watch when nothing new follows it. Requires go-flashduty v0.16.1 for SessionMessageRequest.Steering. --- go.mod | 2 +- go.sum | 4 +- internal/cli/safari_send.go | 17 ++- internal/cli/safari_send_test.go | 24 ++++ internal/cli/safari_watch.go | 220 ++++++++++++++++++++++++------ internal/cli/safari_watch_test.go | 198 +++++++++++++++++++++++++-- 6 files changed, 406 insertions(+), 59 deletions(-) diff --git a/go.mod b/go.mod index cbd4317..f1ac899 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/flashcatcloud/flashduty-cli go 1.26.0 require ( - github.com/flashcatcloud/go-flashduty v0.16.0 + github.com/flashcatcloud/go-flashduty v0.16.1 github.com/mattn/go-runewidth v0.0.30 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 diff --git a/go.sum b/go.sum index 70aad9a..25947f3 100644 --- a/go.sum +++ b/go.sum @@ -1,8 +1,8 @@ github.com/clipperhouse/uax29/v2 v2.2.0 h1:ChwIKnQN3kcZteTXMgb1wztSgaU+ZemkgWdohwgs8tY= github.com/clipperhouse/uax29/v2 v2.2.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= -github.com/flashcatcloud/go-flashduty v0.16.0 h1:YHrd9NaJZReWv1u1iUSLh89HNCeFP+ZaL1tBvLFHrA8= -github.com/flashcatcloud/go-flashduty v0.16.0/go.mod h1:YpHiTYXR5NXBI/rGRZfUy537XMkhdCkwA8NW1QoRHwk= +github.com/flashcatcloud/go-flashduty v0.16.1 h1:CzEa53FYKaHI3lpbqnXcdCj5km5SZhH14xDg1OAZvt0= +github.com/flashcatcloud/go-flashduty v0.16.1/go.mod h1:YpHiTYXR5NXBI/rGRZfUy537XMkhdCkwA8NW1QoRHwk= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/mattn/go-runewidth v0.0.30 h1:+KUuiDA4fF0R1p5FeueHefjDm+GIM+kWfFnDjybOPgk= diff --git a/internal/cli/safari_send.go b/internal/cli/safari_send.go index d412590..da50707 100644 --- a/internal/cli/safari_send.go +++ b/internal/cli/safari_send.go @@ -97,10 +97,14 @@ The message is sent on the CLI channel (reply-only delivery): the agent answers through its reply tool and the answers are read with 'safari watch'. If the server does not accept that channel, the command fails instead of falling back. +Sent to a session whose turn is still running, the message is taken into that +turn at its next step instead of waiting for the turn to finish, and the +running turn answers it. + Follow the answer with: flashduty safari watch --session-id --root-invocation-id -API: POST /safari/session/message, POST /safari/attachment/upload`, +API: POST /safari/session/message, POST /safari/session/get, POST /safari/attachment/upload`, Example: ` flashduty safari send "why is checkout latency up?" flashduty safari send --session-id sess_01 --file trace.json "does this match?" cat question.md | flashduty safari send --team-id 42 --env-kind byoc`, @@ -178,6 +182,17 @@ API: POST /safari/session/message, POST /safari/attachment/upload`, req.EntryKind = safariEntryKind } + if sessionID != "" { + // A message sent while a turn runs joins that turn at its + // next step instead of waiting for a turn of its own; an + // idle session runs it as an ordinary turn. + info, _, err := ctx.Client.Sessions.ReadInfo(cctx, &flashduty.SessionGetRequest{SessionID: sessionID, Limit: 1}) + if err != nil { + return err + } + req.Steering = info.Session.TurnInFlight + } + ack, _, err := ctx.Client.Sessions.WriteMessage(cctx, req) if err != nil { return err diff --git a/internal/cli/safari_send_test.go b/internal/cli/safari_send_test.go index 8d1f5a4..9d67f22 100644 --- a/internal/cli/safari_send_test.go +++ b/internal/cli/safari_send_test.go @@ -24,6 +24,7 @@ type sendServer struct { uploadBody string ackMode string createCalled map[string]any + turnInFlight bool } func newSendServer(t *testing.T, ackMode string) *sendServer { @@ -49,6 +50,10 @@ func newSendServer(t *testing.T, ackMode string) *sendServer { b, _ := io.ReadAll(f) s.uploadName, s.uploadBody = hdr.Filename, string(b) reply(map[string]any{"object_key": "attachments/a1/" + hdr.Filename, "mime": "text/plain", "filename": hdr.Filename, "size": len(b), "attachment_id": "a1"}) + case "/safari/session/get": + reply(map[string]any{"events": []any{}, "pending_messages": []any{}, "session": map[string]any{ + "session_id": "sess-7", "is_running": s.turnInFlight, "turn_in_flight": s.turnInFlight, + }}) case "/safari/session/message": _ = json.NewDecoder(r.Body).Decode(&s.message) sid, _ := s.message["session_id"].(string) @@ -115,6 +120,25 @@ func TestSafariSendExistingSessionOmitsNewSessionFields(t *testing.T) { } } +func TestSafariSendSteersOnlyIntoARunningTurn(t *testing.T) { + for _, running := range []bool{true, false} { + t.Run(map[bool]string{true: "running", false: "idle"}[running], func(t *testing.T) { + saveAndResetGlobals(t) + s := newSendServer(t, "reply_only") + s.turnInFlight = running + if _, err := execCommand("safari", "send", "--session-id", "sess-7", "and the db?"); err != nil { + t.Fatalf("send: %v", err) + } + if strings.Join(s.paths, ",") != "/safari/session/get,/safari/session/message" { + t.Errorf("calls = %v", s.paths) + } + if got, _ := s.message["steering"].(bool); got != running { + t.Errorf("steering = %v, want %v (body %v)", s.message["steering"], running, s.message) + } + }) + } +} + func TestSafariSendFilesCreateSessionThenUploadThenMessage(t *testing.T) { saveAndResetGlobals(t) s := newSendServer(t, "reply_only") diff --git a/internal/cli/safari_watch.go b/internal/cli/safari_watch.go index ede6b5c..2190f6a 100644 --- a/internal/cli/safari_watch.go +++ b/internal/cli/safari_watch.go @@ -22,6 +22,8 @@ const ( replyToolName = "reply" replyPurposeResult = "result" replyPurposeProgress = "progress" + // settledReasonSteered settles a message the running turn took in. + settledReasonSteered = "steered" ) // watchEvent is the one user-facing line `safari watch` prints. Only the fields @@ -45,32 +47,96 @@ type settledPayload struct { RootInvocationID string `json:"root_invocation_id"` Reason string `json:"reason"` StandingTasks []json.RawMessage `json:"standing_tasks"` + // FinalEventID names the turn's answer. A turn that is not reply-only + // (one started from the web console, say) answers in that event's text + // instead of through the reply tool; FinalKind set means it has nothing + // to deliver. + FinalEventID string `json:"final_event_id"` + FinalKind string `json:"final_kind"` + DeliveryMode string `json:"delivery_mode"` } // frameOutcome is what one frame means for the watch loop. type frameOutcome struct { Events []watchEvent - // Settled is true for a session_turn_settled frame of the watched root with - // no standing tasks: the root has no more automatic work, so the request - // is a candidate for "fully done" (confirmed against is_running). + // Settled is true for a settle with no standing tasks once the followed + // request has settled: the request is then a candidate for "fully done" + // (confirmed against is_running). Settled bool // Fatal is true for a frame after which the session will produce nothing // more (environment provisioning failed before the turn started). Fatal bool } -// watchState turns raw stream frames into events. It owns the root filter, the -// dedupe set and the resume cursor, none of which depend on the transport. +// watchState turns raw stream frames into events. It owns the request +// tracking, the dedupe set and the resume cursor, none of which depend on the +// transport. +// +// The watch follows the session, not one invocation: a message sent while +// another turn runs may be taken into that turn, and its answer then comes +// from that turn's invocation. rootID (the request being followed) only marks +// where output starts — its first frame — and which settle makes the request +// a done candidate. type watchState struct { rootID string - seen map[string]struct{} + // started is set once output begins: at the followed request's first + // frame, or immediately when no request is followed. + started bool + // rootSettled is set once the followed request has settled (answered in a + // turn of its own, or taken into the running turn). + rootSettled bool + seen map[string]struct{} + // answers holds, per root, the visible text of its latest persisted + // assistant frame, so a turn that answers without the reply tool can still + // be printed from its settle's final_event_id. A finished turn clears it. + answers map[string]heldAnswer + + // When an earlier watch is re-armed with --root-invocation-id and --since, + // the stream is replayed from the session start so the request's own + // frames are seen again. Persisted frames up to quietAt (and, at exactly + // quietAt, up to and including quietID) were already printed: they update + // the state but print nothing. + quietAt int64 + quietID string + quietPassed bool + // quietDone is set when the last persisted frame was a quiet done + // candidate. It cannot end the stream — later frames may still follow in + // the replay — so it is acted on only if the stream ends with nothing new. + quietDone bool cursorAt int64 cursorID string } +type heldAnswer struct { + eventID, invocation, text string +} + func newWatchState(rootID string, since int64) *watchState { - return &watchState{rootID: rootID, seen: map[string]struct{}{}, cursorAt: since} + return &watchState{ + rootID: rootID, started: rootID == "", rootSettled: rootID == "", + seen: map[string]struct{}{}, answers: map[string]heldAnswer{}, cursorAt: since, + } +} + +// quiet reports whether a persisted frame was already printed by an earlier +// watch. Live-only frames (no event id) are never quiet. +func (w *watchState) quiet(f *flashduty.SessionEventFrame) bool { + switch { + case f.EventID == "" || w.quietAt == 0 || f.CreatedAt > w.quietAt: + return false + case f.CreatedAt < w.quietAt: + return true + case w.quietID == "": + // --since alone resumes at that millisecond inclusive. + return false + case w.quietPassed: + return false + } + if f.EventID == w.quietID { + w.quietPassed = true + } + return true } // frameRoot is the root invocation a frame belongs to. A frame that carries no @@ -85,7 +151,14 @@ func frameRoot(f *flashduty.SessionEventFrame) string { // handle processes one frame. Persisted frames always advance the resume // cursor, even when filtered out, so a reconnect never replays them. func (w *watchState) handle(f *flashduty.SessionEventFrame) frameOutcome { + quiet := w.quiet(f) out := w.classify(f) + if quiet { + w.quietDone = out.Settled + out.Events, out.Settled = nil, false + } else if f.EventID != "" { + w.quietDone = false + } for i := range out.Events { out.Events[i].CreatedAt = f.CreatedAt } @@ -104,23 +177,42 @@ func (w *watchState) classify(f *flashduty.SessionEventFrame) frameOutcome { } w.seen[f.EventID] = struct{}{} } - if w.rootID != "" && frameRoot(f) != w.rootID { - return frameOutcome{} + if !w.started { + if frameRoot(f) != w.rootID { + return frameOutcome{} + } + w.started = true } switch f.Type { case frameTypeSettled: var p settledPayload _ = json.Unmarshal(f.Payload, &p) - if w.rootID != "" && p.RootInvocationID != "" && p.RootInvocationID != w.rootID { - return frameOutcome{} - } root := p.RootInvocationID if root == "" { root = frameRoot(f) } - ev := watchEvent{Type: "settled", RootInvocationID: root, Reason: p.Reason, EventID: f.EventID, StandingTasks: len(p.StandingTasks)} - return frameOutcome{Events: []watchEvent{ev}, Settled: len(p.StandingTasks) == 0} + var events []watchEvent + if a := w.answers[root]; a.eventID != "" && a.eventID == p.FinalEventID && p.DeliveryMode != safariReplyOnlyMode && p.FinalKind == "" { + // Carries the settle's own event id, so the line's + // created_at/event_id pair is one real stream position to resume + // after. + events = append(events, watchEvent{Type: "reply", Purpose: replyPurposeResult, Message: a.text, InvocationID: a.invocation, EventID: f.EventID}) + } + if root == w.rootID { + w.rootSettled = true + } + events = append(events, watchEvent{Type: "settled", RootInvocationID: root, Reason: p.Reason, EventID: f.EventID, StandingTasks: len(p.StandingTasks)}) + // A steered settle is written while the absorbing turn still runs and + // is about to answer, so it is never the session's last frame. + steered := p.Reason == settledReasonSteered + if !steered { + // The session runs one turn at a time: a finished turn leaves no + // other answer pending. + clear(w.answers) + } + done := w.rootSettled && len(p.StandingTasks) == 0 && !steered + return frameOutcome{Events: events, Settled: done} case frameTypeEnvFailed: var p struct { ErrorMessage string `json:"error_message"` @@ -140,15 +232,35 @@ func (w *watchState) classify(f *flashduty.SessionEventFrame) frameOutcome { if f.IsError { return frameOutcome{Events: []watchEvent{{Type: "error", Message: errorFrameMessage(f), Code: f.ErrorCode, EventID: f.EventID}}} } + if text := visibleText(f); text != "" && f.EventID != "" { + w.answers[frameRoot(f)] = heldAnswer{eventID: f.EventID, invocation: f.InvocationID, text: text} + } } return frameOutcome{} } +// visibleText is a frame's answer text: its text parts without the model's +// thinking. +func visibleText(f *flashduty.SessionEventFrame) string { + var c adkContent + if len(f.Payload) == 0 || json.Unmarshal(f.Payload, &c) != nil { + return "" + } + var texts []string + for _, p := range c.Parts { + if t := strings.TrimSpace(p.Text); t != "" && !p.Thought { + texts = append(texts, t) + } + } + return strings.Join(texts, "\n") +} + // adkContent is the slice of an ADK content payload the reply tool result -// lives in. +// and the answer text live in. type adkContent struct { Parts []struct { Text string `json:"text"` + Thought bool `json:"thought"` FunctionResponse *struct { Name string `json:"name"` Response struct { @@ -276,19 +388,20 @@ var errFatalFrame = errors.New("session produced a terminal error") // run blocks until the request is fully done (nil), an unrecoverable error, or // ctx cancellation. // -// Done rule: a session_turn_settled frame for the watched root that lists no -// standing tasks means the root has no more automatic work; the watcher then -// asks POST /safari/session/get and exits only when is_running is false (no -// in-flight turn, so no queued follow-up turn or notification round is about -// to start). A settle that lists standing tasks keeps the watch open: those -// background tasks post later rounds into the same root. +// Done rule: once the followed request has settled — in a turn of its own, or +// as "steered" when the running turn took it in — the next settle that is not +// steered and lists no standing tasks (its own, or the absorbing turn's) makes +// the watch ask POST /safari/session/get, and it exits only when is_running is +// false (no in-flight turn, so no queued follow-up turn or notification round +// is about to start). A settle that lists standing tasks keeps the watch open: +// those background tasks post later rounds into the session. // -// Why the first such settle is final for the root: the server persists a -// settle only when the whole session has no open tasks and no queued -// notifications (logic/session/actor/settled.go, maybePersistSessionTurnSettled), -// and stamps standing_tasks with the running bash_background/monitor tasks -// (model/task/task.go, ListSessionRunningProcessTasks). With both empty nothing -// can start another round under that root; a user follow-up opens a new root. +// Why such a settle is final: the server persists it only when the whole +// session has no open tasks and no queued notifications, and stamps +// standing_tasks with the running background shell and monitor tasks. With +// both empty nothing can start another round; a new message opens a new root. +// A steered settle is different — it is written while the absorbing turn still +// runs, before that turn answers — so it never ends the watch. func (r *watchRunner) run(ctx context.Context) error { failures := 0 settledSeen := false @@ -312,6 +425,11 @@ func (r *watchRunner) run(ctx context.Context) error { started := time.Now() progressed, settled, err := r.streamOnce(ctx) + if err == nil && r.state.quietDone { + // A re-armed watch replayed up to a done candidate the earlier + // watch already printed, and nothing new followed it. + settled = true + } if err == nil && time.Since(started) >= r.healthyAfter { progressed = true } @@ -448,22 +566,31 @@ Events: {"type":"settled","root_invocation_id":"...","reason":"...","event_id":"...","created_at":1700000000000} {"type":"error","message":"...","created_at":1700000000000} -Only messages the agent explicitly sends with its reply tool are printed; tool -calls, thinking and ordinary assistant text are not. Every event carries created_at (Unix ms of its frame). Events are de-duplicated -by event id and the stream is resumed from the last seen event after a drop. +Printed: messages the agent sends with its reply tool, and the final answer of +a turn that does not use the reply tool (one started from the web console). +Tool calls, thinking and other assistant text are not. Every event carries +created_at (Unix ms of its frame). Events are de-duplicated by event id and the +stream is resumed from the last seen event after a drop. -To re-arm a watch later, pass the last line's created_at as --since and its -event_id as --since-event-id: replay then starts strictly after that event. ---since alone resumes at that millisecond inclusive, so the frame at exactly -that time (and any sibling in the same millisecond) is delivered again. +The watch follows the whole session. A message sent while another turn is +running is taken into that turn ('safari send' asks for this), so its answer +comes from that turn, and replies to other people's messages in the same +session are printed too. + +With --root-invocation-id (printed by 'safari send') output starts at that +request and the watch ends once it is done. Without it, --since (Unix +milliseconds) bounds the replay; the default is "now". -With --root-invocation-id (printed by 'safari send') only that request's frames -count. Without it, --since (Unix milliseconds) bounds the replay; the default -is "now". +To re-arm a watch later, pass the last line's created_at as --since and its +event_id as --since-event-id: nothing up to and including that event is +printed again. --since alone resumes at that millisecond inclusive, so the +frame at exactly that time (and any sibling in the same millisecond) is +delivered again. -The command exits 0 once the request is fully done: the root has settled with -no background tasks left and the session reports no running turn. It exits -non-zero on an unrecoverable error. +The command exits 0 once the request is fully done: it has settled (answered +in a turn of its own, or taken into the running turn), a turn has then settled +with no background tasks left, and the session reports no running turn. It +exits non-zero on an unrecoverable error. API: POST /safari/session/subscribe, POST /safari/session/get`, Example: ` flashduty safari watch --session-id sess_01 --root-invocation-id inv_01`, @@ -477,7 +604,14 @@ API: POST /safari/session/subscribe, POST /safari/session/get`, since = time.Now().UnixMilli() } r := newWatchRunner(ctx.Client, sessionID, rootID, since, ctx.Writer, ctx.Cmd.ErrOrStderr()) - if sinceEventID != "" { + switch { + case rootID != "" && since != 0: + // The request's own frames may predate the cursor; replay + // from the session start, silently up to the cursor, so + // its settle is still seen. + r.state.cursorAt = 0 + r.state.quietAt, r.state.quietID = since, sinceEventID + case sinceEventID != "": r.state.cursorID = sinceEventID r.state.seen[sinceEventID] = struct{}{} } @@ -492,8 +626,8 @@ API: POST /safari/session/subscribe, POST /safari/session/get`, }, } cmd.Flags().StringVar(&sessionID, "session-id", "", "Session to follow (required)") - cmd.Flags().StringVar(&rootID, "root-invocation-id", "", "Only follow this request (the root_invocation_id printed by 'safari send')") - cmd.Flags().Int64Var(&since, "since", 0, "Replay from this Unix-millisecond timestamp (default: now, or the start of the session with --root-invocation-id)") + cmd.Flags().StringVar(&rootID, "root-invocation-id", "", "Start at this request and end once it is done (the root_invocation_id printed by 'safari send')") + cmd.Flags().Int64Var(&since, "since", 0, "Print only what comes after this Unix-millisecond timestamp (default: now; with --root-invocation-id the session is replayed from its start and everything up to this point is skipped silently)") cmd.Flags().StringVar(&sinceEventID, "since-event-id", "", "With --since: resume strictly after the event with this id (the event_id of the last line seen)") _ = cmd.MarkFlagRequired("session-id") return cmd diff --git a/internal/cli/safari_watch_test.go b/internal/cli/safari_watch_test.go index 3ca38c3..570e792 100644 --- a/internal/cli/safari_watch_test.go +++ b/internal/cli/safari_watch_test.go @@ -121,25 +121,171 @@ func TestWatchStateDedupeAndCursor(t *testing.T) { } } -func TestWatchStateRootFilter(t *testing.T) { +func steeredFrame(eventID, root string, createdAt int64) *flashduty.SessionEventFrame { + payload := fmt.Sprintf(`{"root_invocation_id":%q,"reason":"steered"}`, root) + return &flashduty.SessionEventFrame{ + EventID: eventID, CreatedAt: createdAt, Type: "session_turn_settled", + RootInvocationID: root, Payload: json.RawMessage(payload), + } +} + +func TestWatchStateStartsAtRequestThenFollowsSession(t *testing.T) { w := newWatchState("rootA", 0) - if out := w.handle(replyFrame("e1", "invB", "rootB", "result", "other", 1)); len(out.Events) != 0 { - t.Errorf("other root emitted: %+v", out.Events) + // Before the request's first frame: earlier history is not printed. + if out := w.handle(replyFrame("e1", "invB", "rootB", "result", "earlier", 1)); len(out.Events) != 0 { + t.Errorf("history before the request emitted: %+v", out.Events) } - // Filtered frames still advance the resume cursor. + // Skipped frames still advance the resume cursor. if w.cursorID != "e1" { t.Errorf("cursor id = %q", w.cursorID) } - if out := w.handle(replyFrame("e2", "invA", "rootA", "result", "mine", 2)); len(out.Events) != 1 { - t.Errorf("own root dropped: %+v", out.Events) - } // A frame without a root id belongs to the root whose own invocation it is. - bare := replyFrame("e3", "rootA", "", "result", "bare", 3) - if out := w.handle(bare); len(out.Events) != 1 { - t.Errorf("bare invocation frame dropped: %+v", out.Events) + if out := w.handle(replyFrame("e2", "rootA", "", "progress", "mine", 2)); len(out.Events) != 1 { + t.Errorf("own request dropped: %+v", out.Events) + } + // From then on the whole session is followed. + if out := w.handle(replyFrame("e3", "invB", "rootB", "result", "theirs", 3)); len(out.Events) != 1 { + t.Errorf("other root after the request dropped: %+v", out.Events) + } + // Another root settling is printed but is not done while the request is open. + if out := w.handle(settleFrame("e4", "rootB", 4, 0)); len(out.Events) != 1 || out.Settled { + t.Errorf("other root's settle before the request settled: %+v", out) + } + if out := w.handle(settleFrame("e5", "rootA", 5, 0)); !out.Settled { + t.Errorf("own settle must be a done candidate: %+v", out) + } +} + +func TestWatchStateSteeredRequestIsAnsweredByTheRunningTurn(t *testing.T) { + w := newWatchState("rootA", 0) + // The running turn (rootH) took the message in: its own root settles as + // steered first, then the running turn answers and settles. + out := w.handle(steeredFrame("e1", "rootA", 1)) + if len(out.Events) != 1 || out.Events[0].Reason != "steered" { + t.Fatalf("steered settle = %+v", out) + } + if out.Settled { + t.Error("a steered settle is written before the absorbing turn answers; it must not be a done candidate") + } + if out = w.handle(replyFrame("e2", "invH", "rootH", "result", "answer to both", 2)); len(out.Events) != 1 { + t.Errorf("absorbing turn's reply dropped: %+v", out.Events) + } + if out = w.handle(settleFrame("e3", "rootH", 3, 0)); !out.Settled { + t.Errorf("absorbing turn's settle must be a done candidate: %+v", out) + } +} + +func nativeFrame(eventID, invocation, root, payload string, createdAt int64) *flashduty.SessionEventFrame { + return &flashduty.SessionEventFrame{ + EventID: eventID, CreatedAt: createdAt, Type: "assistant", + InvocationID: invocation, RootInvocationID: root, Payload: json.RawMessage(payload), + } +} + +func TestWatchStatePrintsFinalAnswerOfTurnWithoutReplyTool(t *testing.T) { + settled := func(eventID, root, extra string) *flashduty.SessionEventFrame { + return &flashduty.SessionEventFrame{EventID: eventID, CreatedAt: 9, Type: "session_turn_settled", RootInvocationID: root, + Payload: json.RawMessage(`{"root_invocation_id":"` + root + `","reason":"completed"` + extra + `}`)} + } + t.Run("web turn", func(t *testing.T) { + w := newWatchState("", 0) + if out := w.handle(nativeFrame("a1", "iW", "rW", `{"parts":[{"text":"draft"}]}`, 1)); len(out.Events) != 0 { + t.Errorf("working text printed: %+v", out.Events) + } + w.handle(nativeFrame("a2", "iW", "rW", `{"parts":[{"text":"thinking","thought":true},{"text":"Disk is full."}]}`, 2)) + out := w.handle(settled("s", "rW", `,"final_event_id":"a2"`)) + if len(out.Events) != 2 || out.Events[0].Type != "reply" || out.Events[0].Message != "Disk is full." || out.Events[0].Purpose != "result" || out.Events[0].EventID != "s" || out.Events[0].InvocationID != "iW" { + t.Fatalf("events = %+v", out.Events) + } + if len(w.answers) != 0 { + t.Errorf("held answers not released at settle: %v", w.answers) + } + }) + t.Run("reply-only turn", func(t *testing.T) { + w := newWatchState("", 0) + w.handle(nativeFrame("a1", "iC", "rC", `{"parts":[{"text":"wrap-up text"}]}`, 1)) + out := w.handle(settled("s", "rC", `,"final_event_id":"a1","delivery_mode":"reply_only"`)) + if len(out.Events) != 1 || out.Events[0].Type != "settled" { + t.Errorf("reply-only final text printed: %+v", out.Events) + } + }) + t.Run("nothing to deliver", func(t *testing.T) { + w := newWatchState("", 0) + w.handle(nativeFrame("a1", "iW", "rW", `{"parts":[{"text":"partial"}]}`, 1)) + out := w.handle(settled("s", "rW", `,"final_event_id":"a1","final_kind":"no_deliverable"`)) + if len(out.Events) != 1 { + t.Errorf("undeliverable final printed: %+v", out.Events) + } + }) +} + +func TestWatchStateResumeQuietlyReplaysWithoutReprinting(t *testing.T) { + w := newWatchState("rootA", 0) + w.quietAt, w.quietID = 5, "e5" + frames := []*flashduty.SessionEventFrame{ + steeredFrame("e2", "rootA", 2), + replyFrame("e4", "invH", "rootH", "result", "seen before", 5), + replyFrame("e5", "invH", "rootH", "result", "seen before too", 5), + replyFrame("e6", "invH", "rootH", "result", "new", 5), + settleFrame("e7", "rootH", 7, 0), + } + var printed []string + var done bool + for _, f := range frames { + out := w.handle(f) + for _, ev := range out.Events { + printed = append(printed, ev.Type+":"+ev.Message) + } + done = out.Settled + } + if strings.Join(printed, "|") != "reply:new|settled:" { + t.Errorf("printed = %q", printed) + } + if !done { + t.Error("the request settled before the cursor; a later settle must still be a done candidate") + } +} + +func TestWatchStateQuietSettleDoesNotEndTheReplay(t *testing.T) { + // The earlier watch saw the request settle at 5 but the session kept + // running, so it stayed open and printed up to e10 before it was stopped. + w := newWatchState("rootA", 0) + w.quietAt, w.quietID = 10, "e10" + if out := w.handle(settleFrame("e5", "rootA", 5, 0)); out.Settled || len(out.Events) != 0 { + t.Fatalf("quiet settle = %+v, want silent and not done", out) + } + if !w.quietDone { + t.Error("a quiet done candidate must be remembered") + } + w.handle(replyFrame("e10", "iB", "rootB", "progress", "printed before", 10)) + if out := w.handle(replyFrame("e12", "iB", "rootB", "result", "new", 12)); len(out.Events) != 1 { + t.Errorf("frame after the cursor = %+v", out) } - if out := w.handle(settleFrame("e4", "rootB", 4, 0)); len(out.Events) != 0 || out.Settled { - t.Errorf("other root's settle leaked: %+v", out) + if w.quietDone { + t.Error("a new frame must clear the quiet done candidate") + } + if out := w.handle(settleFrame("e13", "rootB", 13, 0)); !out.Settled { + t.Errorf("later settle = %+v, want done candidate", out) + } + // Live-only frames are never quiet. + w.quietAt, w.quietPassed = 99, false + if out := w.handle(&flashduty.SessionEventFrame{Type: "error", Payload: json.RawMessage(`{"message":"boom"}`)}); len(out.Events) != 1 { + t.Errorf("live error frame silenced: %+v", out) + } +} + +func TestWatchRunEndsAReplayThatIsAllAlreadyPrinted(t *testing.T) { + s := &watchServer{ + streams: []string{sse(replyFrame("e1", "rootA", "", "result", "done", 3), settleFrame("e2", "rootA", 5, 0))}, + running: []bool{false}, + } + r, out, _ := newTestWatchRunner(t, s, "rootA") + r.state.quietAt, r.state.quietID = 5, "e2" + if err := r.run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + if out.Len() != 0 { + t.Errorf("reprinted:\n%s", out.String()) } } @@ -325,6 +471,34 @@ func TestWatchRunStandingTasksKeepWatchOpen(t *testing.T) { } } +func TestWatchRunWaitsForTheTurnThatTookTheMessageIn(t *testing.T) { + // The message was taken into a running turn: its own root settles as + // steered while that turn still runs, so the watch must keep listening + // until the running turn answers and settles. + // The session already reads idle: a watch that stopped reading at the + // steered settle to confirm idleness would exit before the answer. + s := &watchServer{ + streams: []string{sse( + replyFrame("e0", "iH", "rH", "progress", "before the message", 1), + steeredFrame("e1", "r1", 2), + replyFrame("e2", "iH", "rH", "result", "answer", 3), + settleFrame("e3", "rH", 4, 0), + )}, + running: []bool{false}, + } + r, out, _ := newTestWatchRunner(t, s, "r1") + if err := r.run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + text := out.String() + if strings.Contains(text, "before the message") { + t.Errorf("output before the request:\n%s", text) + } + if !strings.Contains(text, `"message":"answer"`) || strings.Count(text, `"type":"settled"`) != 2 { + t.Errorf("stdout:\n%s", text) + } +} + func TestWatchRunResumesFromCursorAfterDrop(t *testing.T) { // The stream drops after one reply without [DONE]; the reconnect must carry // the cursor, and the replayed frame must not print twice.