diff --git a/cli/internal/daemon/client.go b/cli/internal/daemon/client.go index ea3c003..ae98d5a 100644 --- a/cli/internal/daemon/client.go +++ b/cli/internal/daemon/client.go @@ -195,16 +195,10 @@ func (c *ChatClient) parseResponseError(res *http.Response) error { return errors.New(fallback) } -func (c *ChatClient) Send(ctx context.Context, content string, image *ImageAttachment) (*SendResult, error) { +func (c *ChatClient) submitPayload(ctx context.Context, payload map[string]any) (*SendResult, error) { reqCtx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() - payload := map[string]any{ - "content": content, - } - if image != nil { - payload["image"] = image - } if c.clientID != "" { payload["clientId"] = c.clientID } @@ -255,6 +249,22 @@ func (c *ChatClient) Send(ctx context.Context, content string, image *ImageAttac }, nil } +func (c *ChatClient) Send(ctx context.Context, content string, image *ImageAttachment) (*SendResult, error) { + payload := map[string]any{ + "content": content, + } + if image != nil { + payload["image"] = image + } + return c.submitPayload(ctx, payload) +} + +func (c *ChatClient) Continue(ctx context.Context) (*SendResult, error) { + return c.submitPayload(ctx, map[string]any{ + "type": "continue", + }) +} + func (c *ChatClient) ReplaceWorkspace(ctx context.Context, workspace string) (*WorkspaceUpdate, error) { if c.agentID == "" { return nil, errors.New("cannot replace workspace without agent ID") diff --git a/cli/internal/daemon/client_test.go b/cli/internal/daemon/client_test.go index 91d968d..a8e34f4 100644 --- a/cli/internal/daemon/client_test.go +++ b/cli/internal/daemon/client_test.go @@ -823,3 +823,40 @@ func TestSnapshotEventsAreMarkedReplayed(t *testing.T) { t.Fatalf("only the snapshot after a reset is history: %v", replayed) } } + +func TestChatClientContinue(t *testing.T) { + var receivedPath string + var receivedBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + receivedPath = r.URL.Path + _ = json.NewDecoder(r.Body).Decode(&receivedBody) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "queued": false}) + })) + defer server.Close() + + client := NewChatClient(ChatClientOptions{ + BaseURL: server.URL, + AgentID: "session-1", + ClientID: "test-client", + }) + + res, err := client.Continue(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !res.OK { + t.Fatal("expected OK true") + } + if receivedPath != "/sessions/session-1/events" { + t.Fatalf("expected /sessions/session-1/events, got %s", receivedPath) + } + if receivedBody["type"] != "continue" { + t.Fatalf("expected type continue, got %v", receivedBody) + } + if receivedBody["clientId"] != "test-client" { + t.Fatalf("expected clientId test-client, got %v", receivedBody) + } +} diff --git a/cli/internal/tui/chat.go b/cli/internal/tui/chat.go index b2baef0..fa10ac6 100644 --- a/cli/internal/tui/chat.go +++ b/cli/internal/tui/chat.go @@ -107,6 +107,7 @@ type ChatTurnSentMsg struct { Generation int64 Prompt string Image *daemon.ImageAttachment + Continue bool OK bool Queued bool Err error @@ -1284,11 +1285,13 @@ func (m ChatModel) update(msg tea.Msg) (ChatModel, tea.Cmd) { if errors.As(msg.Err, &wsErr) { m.Status.Running = false m.Status.Idle = true - if len(m.pendingUsers) > 0 { - m.pendingUsers = m.pendingUsers[1:] - } - if m.TextArea.Value() == "" { - m.TextArea.SetValue(msg.Prompt) + if !msg.Continue { + if len(m.pendingUsers) > 0 { + m.pendingUsers = m.pendingUsers[1:] + } + if m.TextArea.Value() == "" { + m.TextArea.SetValue(msg.Prompt) + } } if m.AttachedImage == nil && msg.Image != nil { m.AttachedImage = msg.Image @@ -1301,10 +1304,14 @@ func (m ChatModel) update(msg tea.Msg) (ChatModel, tea.Cmd) { ti.Cursor.Style = DefaultStyles.Cursor ti.Cursor.SetMode(cursor.CursorStatic) m.RecoveryInput = ti + recoveryPrompt := msg.Prompt + if msg.Continue { + recoveryPrompt = "" + } m.WorkspaceRecovery = &WorkspaceRecoveryState{ Missing: wsErr.Workspace, Replacement: wsErr.Workspace, - Prompt: msg.Prompt, + Prompt: recoveryPrompt, Image: msg.Image, } m.refreshViewportContent() @@ -1317,15 +1324,17 @@ func (m ChatModel) update(msg tea.Msg) (ChatModel, tea.Cmd) { } else { m.appendSettledEntry(HistoryEntry{Kind: EntryError, Text: msg.Err.Error()}) } - if m.TextArea.Value() == "" { - m.TextArea.SetValue(msg.Prompt) + if !msg.Continue { + if m.TextArea.Value() == "" { + m.TextArea.SetValue(msg.Prompt) + } + if len(m.pendingUsers) > 0 { + m.pendingUsers = m.pendingUsers[1:] + } } if m.AttachedImage == nil && msg.Image != nil { m.AttachedImage = msg.Image } - if len(m.pendingUsers) > 0 { - m.pendingUsers = m.pendingUsers[1:] - } m.refreshViewportContent() return m, nil } @@ -1519,6 +1528,23 @@ func (m *ChatModel) sendTurnCmd(content string, image *daemon.ImageAttachment) t } } +func (m *ChatModel) sendContinueCmd() tea.Cmd { + client := m.Client + sessID := m.SessionID + gen := m.Generation + + return func() tea.Msg { + if client == nil { + return ChatTurnSentMsg{SessionID: sessID, Generation: gen, Prompt: ".", Continue: true, Err: fmt.Errorf("no client available")} + } + res, err := client.Continue(context.Background()) + if err != nil { + return ChatTurnSentMsg{SessionID: sessID, Generation: gen, Prompt: ".", Continue: true, Err: err} + } + return ChatTurnSentMsg{SessionID: sessID, Generation: gen, Prompt: ".", Continue: true, OK: res.OK, Queued: res.Queued} + } +} + func (m *ChatModel) interruptCmd() tea.Cmd { client := m.Client sessID := m.SessionID @@ -1563,6 +1589,24 @@ func (m *ChatModel) submitInput(input string, cmds *[]tea.Cmd) { return } + if input == "." { + if len(m.pendingUsers) >= MaxPendingUsers { + m.AddError("too many pending turns; wait for current turn to complete") + m.refreshViewportContent() + return + } + + m.ClearNotices() + m.isSending = true + m.reseedMood() + m.sentHere = true + m.Follow = true + m.refreshViewportContent() + + *cmds = append(*cmds, m.sendContinueCmd(), m.startAnimation()) + return + } + if len(m.pendingUsers) >= MaxPendingUsers { m.AddError("too many pending turns; wait for current turn to complete") m.refreshViewportContent() @@ -1792,10 +1836,12 @@ func (m *ChatModel) handleStreamEvent(evt daemon.StreamEvent) { Timestamp: time.Now().UnixMilli(), } m.appendSettledEntry(entry) - if m.turn != nil { - m.turn.tools++ - m.turn.touch(evt.Timestamp) + if m.turn == nil { + ts := time.Now().UnixMilli() + m.turn = &openTurn{start: ts, last: ts} } + m.turn.tools++ + m.turn.touch(evt.Timestamp) case daemon.EventMessage: m.ToolProgressText = "" @@ -1822,9 +1868,14 @@ func (m *ChatModel) handleStreamEvent(evt daemon.StreamEvent) { Timestamp: ts, } m.appendSettledEntry(entry) - if m.turn != nil { - m.turn.touch(evt.Timestamp) + if m.turn == nil { + start := ts + if start == 0 { + start = time.Now().UnixMilli() + } + m.turn = &openTurn{start: start, last: start} } + m.turn.touch(evt.Timestamp) case daemon.EventNote: m.settleActiveStream() @@ -1919,9 +1970,11 @@ func (t *openTurn) begun() bool { } func (m *ChatModel) turnIsLive() { - if m.turn != nil { - m.turn.live = true + if m.turn == nil { + ts := time.Now().UnixMilli() + m.turn = &openTurn{start: ts, last: ts} } + m.turn.live = true } // stretch is one run of a phase: a thinking or replying spell, or one tool diff --git a/cli/internal/tui/pending_test.go b/cli/internal/tui/pending_test.go index b4f8dff..6aa6a23 100644 --- a/cli/internal/tui/pending_test.go +++ b/cli/internal/tui/pending_test.go @@ -1,6 +1,7 @@ package tui import ( + "fmt" "strings" "testing" "time" @@ -45,3 +46,55 @@ func TestPendingMessageIsAGreyedOutCopyInTheTranscript(t *testing.T) { t.Fatalf("settling moved the transcript:\n%s", settled) } } + +func TestDotContinueDoesNotAppearInUI(t *testing.T) { + lipgloss.SetColorProfile(termenv.ANSI256) + t.Cleanup(func() { lipgloss.SetColorProfile(termenv.Ascii) }) + m := NewChatModel(&daemon.Session{ID: "s"}, nil) + m.Client = daemon.NewChatClient(daemon.ChatClientOptions{BaseURL: "http://127.0.0.1:1", AgentID: "s"}) + m.SetSize(80, 30) + m.appendSettledEntry(HistoryEntry{Kind: EntryAssistant, Speaker: "albedo", Text: "ready", Timestamp: time.Now().UnixMilli() - 1000}) + + var cmds []tea.Cmd + m.submitInput(".", &cmds) + + if len(m.pendingUsers) != 0 { + t.Fatalf("expected 0 pending users for dot continue, got %d", len(m.pendingUsers)) + } + if len(m.pendingRows()) != 0 { + t.Fatalf("expected 0 pending rows for dot continue, got %d", len(m.pendingRows())) + } + afterView := ansi.Strip(m.Viewport.View()) + if strings.Contains(afterView, "│ .") || strings.Contains(afterView, "│ continue") || strings.Contains(afterView, "│ you") { + t.Fatalf("transcript should not contain dot or continue prompt:\n%s", afterView) + } + if !m.isSending { + t.Fatal("expected isSending to be true") + } + if len(cmds) == 0 { + t.Fatal("expected cmds to be returned for continue submission") + } +} + +func TestDotContinueErrorDoesNotCorruptPendingUsers(t *testing.T) { + m := NewChatModel(&daemon.Session{ID: "s"}, nil) + m.pendingUsers = []PendingUserTurn{{Text: "real user prompt", At: time.Now().UnixMilli()}} + m.TextArea.SetValue("") + + errMsg := ChatTurnSentMsg{ + SessionID: "s", + Generation: m.Generation, + Prompt: ".", + Continue: true, + Err: fmt.Errorf("network failure"), + } + + um, _ := m.Update(errMsg) + + if len(um.pendingUsers) != 1 || um.pendingUsers[0].Text != "real user prompt" { + t.Fatalf("pending user turn was corrupted or popped: %+v", um.pendingUsers) + } + if um.TextArea.Value() == "." { + t.Fatalf("text area was overwritten with dot: %q", um.TextArea.Value()) + } +} diff --git a/src/albedo/daemon/conversation.gleam b/src/albedo/daemon/conversation.gleam index cfaf7f9..a97e827 100644 --- a/src/albedo/daemon/conversation.gleam +++ b/src/albedo/daemon/conversation.gleam @@ -1,6 +1,7 @@ import albedo/daemon/events import albedo/daemon/images import albedo/daemon/note +import albedo/daemon/notice import albedo/daemon/store import albedo/daemon/transcript import albedo/daemon/usage @@ -511,9 +512,9 @@ pub fn latest_user(inputs: List(types.Input)) -> Option(String) { list.fold(inputs, None, fn(latest, input) { case input { types.User(text) | types.UserImage(text, _) -> - case note.parse(text), is_webhook(text) { - None, False -> Some(text) - _, _ -> latest + case notice.is_notice(text), note.parse(text), is_webhook(text) { + False, None, False -> Some(text) + _, _, _ -> latest } _ -> latest } diff --git a/src/albedo/daemon/events.gleam b/src/albedo/daemon/events.gleam index c6ac2bb..bdbc107 100644 --- a/src/albedo/daemon/events.gleam +++ b/src/albedo/daemon/events.gleam @@ -1,4 +1,5 @@ import albedo/daemon/note +import albedo/daemon/notice import albedo/daemon/store import albedo/daemon/transcript import albedo/daemon/usage @@ -248,9 +249,13 @@ fn render( ) -> List(String) { case entry.input { types.User(value) -> - case note.parse(value) { - Some(#(origin, body)) -> [user(body, origin, None, entry.timestamp)] - None -> [user(value, "chat", None, entry.timestamp)] + case notice.is_notice(value) { + True -> [] + False -> + case note.parse(value) { + Some(#(origin, body)) -> [user(body, origin, None, entry.timestamp)] + None -> [user(value, "chat", None, entry.timestamp)] + } } types.UserImage(value, image) -> [ user_image(value, "user", None, entry.timestamp, image), diff --git a/src/albedo/daemon/history.gleam b/src/albedo/daemon/history.gleam index 12763bf..54429b9 100644 --- a/src/albedo/daemon/history.gleam +++ b/src/albedo/daemon/history.gleam @@ -2,6 +2,7 @@ import albedo/daemon/conversation import albedo/daemon/events +import albedo/daemon/notice import albedo/daemon/store import albedo/harness/extensions/lcm/graph as lcm_graph import albedo/openai_api/types @@ -167,9 +168,16 @@ fn recent_item(row: Row) -> Result(Item, Nil) { } } case row.input { - types.User(text) -> item(User, text, recent_excerpt) + types.User(text) -> + case notice.is_notice(text) { + True -> Error(Nil) + False -> item(User, text, recent_excerpt) + } types.UserImage(text, image) -> - item(User, text <> image_label(image), recent_excerpt) + case notice.is_notice(text) { + True -> Error(Nil) + False -> item(User, text <> image_label(image), recent_excerpt) + } types.Assistant(text) -> item(Assistant, text, recent_excerpt) types.ToolOutput(_, _, _) -> Error(Nil) types.Replay(replay) -> diff --git a/src/albedo/daemon/notice.gleam b/src/albedo/daemon/notice.gleam new file mode 100644 index 0000000..7289830 --- /dev/null +++ b/src/albedo/daemon/notice.gleam @@ -0,0 +1,15 @@ +//// System notices injected into the transcript that direct the model without +//// appearing as user bubbles in the client view. + +import gleam/string + +pub const open = "" +pub const close = "" + +pub fn wrap(text: String) -> String { + open <> "\n" <> text <> "\n" <> close +} + +pub fn is_notice(text: String) -> Bool { + string.starts_with(text, open) +} diff --git a/src/albedo/daemon/server.gleam b/src/albedo/daemon/server.gleam index 6857f84..0ec8eba 100644 --- a/src/albedo/daemon/server.gleam +++ b/src/albedo/daemon/server.gleam @@ -1079,28 +1079,51 @@ fn daemon_route( ) Post, "events" -> { let decoder = { - use text <- decode.field("content", decode.string) - use client_id <- decode.optional_field( - "clientId", - "", + use kind <- decode.optional_field( + "type", + "user", decode.string, ) - use submitted_image <- decode.optional_field( - "image", - None, - decode.optional(submitted_image_decoder()), - ) - decode.success(#(text, client_id, submitted_image)) + case kind { + "continue" | "resume" -> { + use client_id <- decode.optional_field( + "clientId", + "", + decode.string, + ) + decode.success(#(True, "", client_id, None)) + } + _ -> { + use text <- decode.field("content", decode.string) + use client_id <- decode.optional_field( + "clientId", + "", + decode.string, + ) + use submitted_image <- decode.optional_field( + "image", + None, + decode.optional(submitted_image_decoder()), + ) + decode.success(#(False, text, client_id, submitted_image)) + } + } } case body(req, decoder) |> result.map_error(session.Rejected) |> result.try(fn(submission) { - use image <- result.try( - validate_submitted_image(submission.2) - |> result.map_error(session.Rejected), - ) - session.submit(worker, submission.0, submission.1, image) + let #(is_continue, text, client_id, raw_image) = submission + case is_continue { + True -> session.submit_continue(worker, client_id) + False -> { + use image <- result.try( + validate_submitted_image(raw_image) + |> result.map_error(session.Rejected), + ) + session.submit(worker, text, client_id, image) + } + } }) { Ok(queued) -> diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index 8f039eb..3c137c2 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -37,6 +37,13 @@ const restart_note = Submission( None, ) +pub const continue_prompt = " +continue your unfinished task, by resuming the most recent intent. +if interrupted mid-step, just pick it back up from where it stopped. +never pause to summarize progress, re-confirm the plan, or ask whether to proceed. +just continue. +" + pub type Session = Subject(Message) @@ -255,6 +262,16 @@ pub fn submit( )) } +pub fn submit_continue( + session: Session, + client_id: String, +) -> Result(Bool, SubmissionError) { + actor.call(session, 10_000, Submit( + Submission("", continue_prompt, client_id, turn.Continue, None), + _, + )) +} + /// This session's materialized command catalog and the context that runs them. /// Answered through the kernel's session, so the list matches its bindings. pub fn commands( @@ -519,10 +536,14 @@ fn handle(state: State, message: Message) { } |> start_run(kernel, client, model_history) process.send(reply, Ok(False)) - actor.continue(session_state.emit( - state, - session_submission.event(submission, timestamp), - )) + actor.continue(case submission.source { + turn.Continue -> state + _ -> + session_state.emit( + state, + session_submission.event(submission, timestamp), + ) + }) } } } diff --git a/src/albedo/daemon/session_submission.gleam b/src/albedo/daemon/session_submission.gleam index 0486110..87a7369 100644 --- a/src/albedo/daemon/session_submission.gleam +++ b/src/albedo/daemon/session_submission.gleam @@ -41,6 +41,9 @@ pub fn emit( timestamp: Int, ) -> session_state.State(message) { list.fold(submissions, state, fn(state, submission) { - session_state.emit(state, event(submission, timestamp)) + case submission.source { + turn.Continue -> state + _ -> session_state.emit(state, event(submission, timestamp)) + } }) } diff --git a/src/albedo/daemon/turn.gleam b/src/albedo/daemon/turn.gleam index 721a05f..1d9073b 100644 --- a/src/albedo/daemon/turn.gleam +++ b/src/albedo/daemon/turn.gleam @@ -22,6 +22,7 @@ pub type Source { /// Something an extension tells the agent, such as a user's work ledger /// change. `origin` labels it in the transcript. Note(origin: String) + Continue } pub type Submission { @@ -79,6 +80,7 @@ pub fn source_name(source: Source) -> String { JobWake -> "bash" Webhook(_) -> "webhook" Note(origin) -> origin + Continue -> "continue" } } @@ -90,7 +92,7 @@ pub fn admit( case activity, bounded(submission), submission.source { _, False, _ -> Reject(Oversized) _, True, Note(_) -> room(queued) - Running(_), True, Chat -> room(queued) + Running(_), True, Chat | Running(_), True, Continue -> room(queued) Running(_), True, JobWake | Running(_), True, Webhook(_) -> Reject(Busy) Resting, True, _ | Interrupted, True, _ -> Start } @@ -118,7 +120,9 @@ fn bounded(submission: Submission) -> Bool { /// Whether a finished run should start another for this queue: notes alone /// wait for the user's next message. pub fn starts_turn(queued: List(Submission)) -> Bool { - list.any(queued, fn(submission) { submission.source == Chat }) + list.any(queued, fn(submission) { + submission.source == Chat || submission.source == Continue + }) } pub fn running(activity: Activity) -> Option(Run) { diff --git a/test/daemon/events_test.gleam b/test/daemon/events_test.gleam index 93a0290..4ee08e1 100644 --- a/test/daemon/events_test.gleam +++ b/test/daemon/events_test.gleam @@ -1,5 +1,6 @@ import albedo/daemon/events as view import albedo/daemon/note +import albedo/daemon/session import albedo/daemon/transcript import albedo/harness/runtime import albedo/openai_api/types @@ -83,6 +84,18 @@ pub fn snapshot_surfaces_saved_reasoning_apart_from_the_answer_test() { runtime.stop(host) } +pub fn snapshot_omits_continue_notice_test() { + let assert Ok(host) = runtime.start(":memory:") + let events = + view.snapshot( + runtime.ledger(host), + [transcript.Entry(types.User(session.continue_prompt), None, None)], + None, + ) + types_of(events) |> should.equal([]) + runtime.stop(host) +} + /// A reloaded transcript labels a note by its origin and shows its body, and /// labels a typed message the way the live stream does. pub fn snapshot_labels_notes_by_origin_test() { diff --git a/test/daemon/turn_test.gleam b/test/daemon/turn_test.gleam index 3436d35..1881535 100644 --- a/test/daemon/turn_test.gleam +++ b/test/daemon/turn_test.gleam @@ -142,3 +142,15 @@ pub fn a_full_note_queue_does_not_block_an_idle_chat_test() { turn.admit(turn.Resting, chat("hello"), turn.queue_limit) |> should.equal(turn.Start) } + +fn cont(text: String) -> turn.Submission { + turn.Submission("", text, "client", turn.Continue, None) +} + +pub fn continue_submission_starts_turn_and_queues_when_busy_test() { + turn.admit(turn.Resting, cont("continue"), 0) |> should.equal(turn.Start) + turn.admit(turn.Interrupted, cont("continue"), 0) |> should.equal(turn.Start) + turn.admit(running("a"), cont("continue"), 0) |> should.equal(turn.Queue) + turn.starts_turn([cont("continue")]) |> should.be_true + turn.starts_turn([note("note"), cont("continue")]) |> should.be_true +}