feat(tui): keep unresolved turns in sight
Retain uncertain turns and creation handles across navigation. Expired receipts leave a visible note and stop the polling loop.
45 unmodified lines46474849505152535455565758596039 unmodified lines10010110210310410510610710810911011111211311411546 unmodified lines162163164165166167168169170171172173174101 unmodified linesExpand all276277278279280281282283284285286287288289290291264292293266267294295296297298299300301302303304181 unmodified linesExpand all486487488489490491492493494495496497498499500501502503504178 unmodified linesExpand all68368468568668768868969069169269369484 unmodified lines77978078178278378478578678778878946 unmodified lines836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887249 unmodified linesExpand all1137113811391140114111421143114411451146114711481149212 unmodified linesExpand all45 unmodified lines Err error ID string}
type sessionCreationRecoveryMsg struct { Handle *daemon.OperationHandle Gen int}
type sessionCreatedMsg struct { Handle *daemon.OperationHandle Err error Session daemon.Session Gen int}39 unmodified lines Provider string Gen int}
type pendingSessionIntents struct { users []PendingUserTurn continuations map[string]pendingOperation}
type AppModel struct { pendingBySession map[string]pendingSessionIntents pendingCreations map[string]pendingOperation ActiveSession *daemon.Session openBrowser func(url string) Conn *daemon.Connection Profiles config.Profiles46 unmodified lines}
func (m *AppModel) newChatModel(session *daemon.Session) ChatModel { chat := NewChatModel(session, daemon.NewChatClient(m.Conn, session.ID)) if pending, ok := m.pendingBySession[session.ID]; ok { chat.pendingUsers = slices.Clone(pending.users) chat.pendingContinuations = maps.Clone(pending.continuations) delete(m.pendingBySession, session.ID) } chat.Flags.Thinking = m.SessionPicker.prefs.Thinking chat.Flags.Tools = m.SessionPicker.prefs.Tools chat.Notices = slices.Clone(m.Notices) chat.graphemes = m.Graphemes101 unmodified linesExpand all}
func (m *AppModel) createSessionCmd(gen int, workspace string) tea.Cmd { conn := m.Conn handle, preparationErr := daemon.NewCreation(map[string]string{"workspace": workspace}) if preparationErr == nil { if m.pendingCreations == nil { m.pendingCreations = make(map[string]pendingOperation) } m.pendingCreations[handle.ID] = pendingOperation{Handle: handle} } return func() tea.Msg { if preparationErr != nil { return sessionCreatedMsg{Err: preparationErr, Gen: gen} } if conn == nil { return sessionCreatedMsg{Err: errors.New("daemon connection unavailable"), Gen: gen} return sessionCreatedMsg{Err: errors.New("daemon connection unavailable"), Gen: gen, Handle: handle} } s, err := daemon.CreateSession(context.Background(), conn, map[string]string{"workspace": workspace}) return sessionCreatedMsg{Session: s, Err: err, Gen: gen} session, err := daemon.CreateSessionOperation(context.Background(), conn, handle) return sessionCreatedMsg{Session: session, Err: err, Gen: gen, Handle: handle} }}func creationRecoveryCmd(handle *daemon.OperationHandle, gen int) tea.Cmd { return tea.Tick(15*time.Second, func(time.Time) tea.Msg { return sessionCreationRecoveryMsg{Handle: handle, Gen: gen} })}
func (m *AppModel) loadProfilesCmd(provider string, gen int) tea.Cmd { conn := m.Conn return func() tea.Msg {181 unmodified linesExpand all m.SessionPicker.SetSessions(m.Sessions, m.ActiveSession)}
func (m *AppModel) setChatSession(s daemon.Session, prepend bool) tea.Cmd { m.SessionGen++ if m.pendingBySession == nil { m.pendingBySession = make(map[string]pendingSessionIntents) } if m.Chat.SessionID != "" { if len(m.Chat.pendingUsers) == 0 && len(m.Chat.pendingContinuations) == 0 { delete(m.pendingBySession, m.Chat.SessionID) } else { m.pendingBySession[m.Chat.SessionID] = pendingSessionIntents{users: slices.Clone(m.Chat.pendingUsers), continuations: maps.Clone(m.Chat.pendingContinuations)} } } m.Chat.Close() session := s m.ActiveSession = &session if prepend {178 unmodified linesExpand all sid, forward = sm.SessionID, true if m.ActiveSession != nil && sid == m.ActiveSession.ID && sm.Event.Type == daemon.EventUser && !sm.Event.Replayed { m.ClearNotices() } case ChatOperationPollMsg: sid, forward = sm.SessionID, true case ChatOperationResolvedMsg: sid, forward = sm.SessionID, true case ChatTurnSentMsg: sid, forward = sm.SessionID, true if m.ActiveSession != nil && sid == m.ActiveSession.ID && sm.Err == nil { m.ClearNotices()84 unmodified lines } return m, nil
case sessionDeletedMsg: if msg.Err == nil { delete(m.pendingBySession, msg.ID) } if msg.Err != nil { m.AddError(operationError(msg.Err, "Could not delete the session: ", "Session may have been deleted; refresh the session list before trying again.")) return m, nil }46 unmodified lines m.SessionPicker.Prune(msg.Sessions) m.updateSessionPickerItems() return m, m.SessionPicker.PreviewCmd()
case sessionCreationRecoveryMsg: if pending, exists := m.pendingCreations[msg.Handle.ID]; !exists || pending.Expired { return m, nil } conn, handle, gen := m.Conn, msg.Handle, msg.Gen return m, func() tea.Msg { session, err := daemon.ResolveCreation(context.Background(), conn, handle) return sessionCreatedMsg{Session: session, Err: err, Handle: handle, Gen: gen} }
case sessionCreatedMsg: if msg.Gen != m.SessionGen && (msg.Handle == nil || m.pendingCreations[msg.Handle.ID].Handle == nil) { return m, nil } if msg.Handle != nil && daemon.IsOperationExpired(msg.Err) { pending := m.pendingCreations[msg.Handle.ID] pending.Expired = true m.pendingCreations[msg.Handle.ID] = pending m.AddError("Session operation " + msg.Handle.ID + " expired; its outcome is unresolved.") return m, nil } if msg.Err != nil && msg.Handle != nil { if _, uncertain := errors.AsType[*daemon.UncertainOutcomeError](msg.Err); uncertain { m.AddError("Session admission is uncertain. Checking operation " + msg.Handle.ID + ".") return m, creationRecoveryCmd(msg.Handle, msg.Gen) } if api, ok := errors.AsType[*daemon.APIError](msg.Err); ok && (api.Code == "operation_unknown" || api.StatusCode >= 500) { return m, creationRecoveryCmd(msg.Handle, msg.Gen) } } if msg.Handle != nil { delete(m.pendingCreations, msg.Handle.ID) } if msg.Gen != m.SessionGen { if msg.Err != nil { m.AddError(msg.Err.Error()) return m, nil } if !slices.ContainsFunc(m.Sessions, func(session daemon.Session) bool { return session.ID == msg.Session.ID }) { m.Sessions = append(m.Sessions, msg.Session) } m.updateSessionPickerItems() m.ClearNotices() m.AddNotice("Recovered the created session " + msg.Session.ID) return m, nil } if msg.Err != nil && m.State == AppStateFolderPicker { m.FolderPicker.Refused(errors.New(operationError(msg.Err, "", "Session may have been created; check the session list before creating another.")))249 unmodified linesExpand all m.ProfileGen++ return m, m.loadProfilesCmd(msg.Name, m.ProfileGen)
case ChatExecuteCommandMsg: if slices.ContainsFunc(m.CommandCatalog, func(command daemon.SessionCommand) bool { return command.Name == msg.Name && command.Skill }) { var commands []tea.Cmd m.Chat.handleSubmittedCommand(strings.TrimSpace(msg.Name+" "+msg.Args), &commands) return m, tea.Batch(commands...) } switch msg.Name { case "/login": return m, m.openLogin(msg.Args) case "/model":212 unmodified linesExpand all49 unmodified lines50515253545554555657585960616263646566141 unmodified linesExpand all49 unmodified lines // skips the second rendering that marks wraps for a copy until it settles. Live bool}
// pending is how far a message of yours has got: sent and waiting for the// daemon, or accepted and queued behind the running turn.// pending records whether a user message is awaiting admission, queued for// delivery, or unresolved after its operation receipt expires.type pending int
const ( settled pending = iota sending queued unresolved)
func (e *HistoryEntry) ComputeSize() int64 { size := int64(len(e.Speaker) + len(e.Text) + len(e.ClientID) + len(e.ToolName) + len(e.ToolResult) + 64)141 unmodified linesExpand all4 unmodified lines567891011121394 unmodified lines10810911011111111211311211311511411711811911511611711811912012112212312412512612712812913013113213313413513611 unmodified lines148149150151152153154155156157158159160161162163164165166167142143144145168169170171172173174175149 unmodified linesExpand all325326327328329303304305306307330331332333334335336337338481 unmodified linesExpand all8208218228237978248258268278289 unmodified lines83883984084184284384484584684784884981585085185285385485585685785885986086186286386486538 unmodified lines90490590690786690890991091191297 unmodified lines10101011101210139721014101510161017101810191020102110221023102410251026102710281029342 unmodified linesExpand all137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271331142814291430143114321333143314341435143614371438143914401441144214431340134114441445144614471448144918 unmodified lines1468146914701471136814721473147414751476147714781479148014811482148362 unmodified lines154615471548154915501551155215531554155515561557155815591560144315611562156314461447144814491450145114521453145414551564156515661457145815671568156915701571157228 unmodified lines16011602160316041495160516061607160816091 unmodified line16111612161316141505150615071615161616171618161916201621162216231624162516261627162816291630163116321633150915101511151215131514163416351636163716385 unmodified lines16441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167077 unmodified lines1748174917501751161317521753175417553 unmodified lines175917601761176216251626176317641765176617671768202 unmodified linesExpand all197119721973197418371975197619771978197935 unmodified lines201520162017201818812019202020212022202370 unmodified lines2094209520962097196020982099210021012102215 unmodified linesExpand all2318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023414 unmodified lines "cmp" "context" "errors" "fmt" "maps" "math" "math/rand/v2" "path/filepath" "regexp"94 unmodified lines SessionID string Generation int64}
type ChatTurnSentMsg struct { Err error Image *daemon.ImageAttachmenttype ChatOperationResolvedMsg struct { SessionID string Prompt string Generation int64 Continue bool OK bool Queued bool Handle *daemon.OperationHandle Receipt daemon.OperationReceipt Err error}
type ChatTurnSentMsg struct { Handle *daemon.OperationHandle OperationID string Err error Image *daemon.ImageAttachment SessionID string Prompt string Generation int64 Continue bool OK bool Queued bool}
type ChatInterruptMsg struct { Err error SessionID string Generation int6411 unmodified lines StreamKindText ActiveStreamKind = "text" StreamKindThinking ActiveStreamKind = "thinking")
type pendingOperation struct { Handle *daemon.OperationHandle Expired bool}
type ChatOperationPollMsg struct { SessionID string Generation int64 Handle *daemon.OperationHandle}
type PendingUserTurn struct { Expired bool Handle *daemon.OperationHandle OperationID string BlockingReason string Image *daemon.ImageAttachment Text string At int64 Queued bool Image *daemon.ImageAttachment Text string At int64 Queued bool}
const ( MaxLiveStreamBytes = 64 * 1024149 unmodified linesExpand all burstRows []string Notices Notices settledLines []string
pendingUsers []PendingUserTurn pendingUsers []PendingUserTurn Glances []PageGlance frameLines []string CommandMenu CommandMenuModel Viewport viewport.Model pendingContinuations map[string]pendingOperation Glances []PageGlance frameLines []string CommandMenu CommandMenuModel Viewport viewport.Model
transcript transcriptState TextArea textarea.Model burstRowsKey burstRowsKey481 unmodified linesExpand all// pendingRows are your messages the daemon has not echoed yet, greyed out at// the end of the transcript where they will settle. Each follows whatever is// live above it, so the rails and names join up as they will once settled.func (m ChatModel) pendingRows() []string { if len(m.pendingUsers) == 0 { if len(m.pendingUsers) == 0 && len(m.pendingContinuations) == 0 { return nil } before := slices.Clone(m.History.Entries()) if m.transcript.activeKind != StreamKindNone && m.transcript.activeText != "" {9 unmodified lines state := sending if p.Queued { state = queued } text := p.Text if p.Expired { state = unresolved text += "\nOperation " + p.OperationID + " expired; its outcome is unknown." } if p.BlockingReason != "" && !p.Expired { text += "\nWaiting: " + p.BlockingReason } entry := HistoryEntry{Kind: EntryUser, Speaker: "You", Text: p.Text, Timestamp: p.At, Pending: state} entry := HistoryEntry{Kind: EntryUser, Speaker: "You", Text: text, Timestamp: p.At, Pending: state} block, _ := m.Renderer.Block(before, entry, m.Flags) rows = append(rows, block...) before = append(before, entry) } for _, id := range slices.Sorted(maps.Keys(m.pendingContinuations)) { if m.pendingContinuations[id].Expired { entry := HistoryEntry{Kind: EntryError, Text: "Continue operation " + id + " expired; its outcome is unresolved."} block, _ := m.Renderer.Block(before, entry, m.Flags) rows = append(rows, block...) } } return rows}
// jumpToYou scrolls to the start of your previous or next message. Past38 unmodified lines }}
func (m ChatModel) animating() bool { return m.isSending || len(m.pendingUsers) > 0 || m.Stopping || m.Progress != nil || m.Status.Running && !m.Status.Idle return m.isSending || m.pendingSendCount() > 0 || m.Stopping || m.Progress != nil || m.Status.Running && !m.Status.Idle}
func (m *ChatModel) startAnimation() tea.Cmd { if m.animationActive || !m.animating() {97 unmodified lines return tea.Tick(750*time.Millisecond, func(time.Time) tea.Msg { return ChatStatusPollMsg{SessionID: id, Generation: generation} })}
func (m ChatModel) Init() tea.Cmd { return tea.Batch(m.startStreamSubscription(), m.statusCmd()) commands := []tea.Cmd{m.startStreamSubscription(), m.statusCmd()} for _, pending := range m.pendingUsers { if pending.Handle != nil && !pending.Expired { commands = append(commands, m.queryOperationCmd(pending.Handle)) } } for _, pending := range m.pendingContinuations { if !pending.Expired { commands = append(commands, m.queryOperationCmd(pending.Handle)) } } return tea.Batch(commands...)}
// Update handles msg, then fetches older history when it left you at the top.func (m ChatModel) Update(msg tea.Msg) (ChatModel, tea.Cmd) {342 unmodified linesExpand all
case ChatStreamClosedMsg: return m, nil
case ChatOperationPollMsg: if msg.SessionID != m.SessionID || msg.Generation != m.Generation || !m.operationRecoverable(msg.Handle.ID) { return m, nil } return m, m.queryOperationCmd(msg.Handle)
case ChatOperationResolvedMsg: if msg.SessionID != m.SessionID || msg.Generation != m.Generation { return m, nil } index := slices.IndexFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.OperationID == msg.Handle.ID }) if !m.operationRecoverable(msg.Handle.ID) { return m, nil } if daemon.IsOperationExpired(msg.Err) { if index >= 0 { m.pendingUsers[index].Expired = true } else { pending := m.pendingContinuations[msg.Handle.ID] pending.Expired = true m.pendingContinuations[msg.Handle.ID] = pending } m.refreshViewportContent() return m, nil } if msg.Err == nil && (msg.Receipt.Status == "rejected" || msg.Receipt.DeliveryStatus == "cancelled" || msg.Receipt.DeliveryStatus == "committed") { delete(m.pendingContinuations, msg.Handle.ID) if rejection := msg.Receipt.Rejection(); rejection != nil { m.ClearNotices() m.AddError(rejection.Error()) } if index >= 0 { m.pendingUsers = slices.Delete(m.pendingUsers, index, index+1) } m.refreshViewportContent() return m, nil } if index >= 0 && msg.Err == nil { m.pendingUsers[index].Queued = true m.pendingUsers[index].BlockingReason = msg.Receipt.BlockingReason m.refreshViewportContent() } return m, m.resolveOperationCmd(msg.Handle)
case ChatTurnSentMsg: if msg.SessionID != m.SessionID || msg.Generation != m.Generation { return m, nil } m.isSending = false if msg.Err != nil { m.interruptDeferred = false if _, uncertain := errors.AsType[*daemon.UncertainOutcomeError](msg.Err); uncertain { message := "Message may have been accepted; check the conversation before resending." if msg.Handle != nil && m.operationRecoverable(msg.Handle.ID) && daemon.IsOperationExpired(msg.Err) { return m.Update(ChatOperationResolvedMsg{SessionID: m.SessionID, Generation: m.Generation, Handle: msg.Handle, Err: msg.Err}) } message := "Message admission is uncertain. Checking its operation receipt." if msg.Continue { message = "Continue may have been accepted; check the conversation before trying again." message = "Continue admission is uncertain. Checking its operation receipt." } m.AddError(message) m.refreshViewportContent() if msg.Handle != nil && m.operationRecoverable(msg.Handle.ID) { return m, m.resolveOperationCmd(msg.Handle) } return m, nil } delete(m.pendingContinuations, msg.OperationID) if !msg.Continue { if len(m.pendingUsers) > 0 { m.pendingUsers = m.pendingUsers[1:] if i := slices.IndexFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.OperationID == msg.OperationID }); i >= 0 { m.pendingUsers = slices.Delete(m.pendingUsers, i, i+1) } if m.TextArea.Value() == "" { m.TextArea.SetValue(msg.Prompt) }18 unmodified lines }
if msg.Queued { if i := slices.IndexFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.Text == msg.Prompt && !p.Queued return p.OperationID == msg.OperationID && !p.Queued }); i >= 0 { m.pendingUsers[i].Queued = true } } if msg.Handle != nil && (msg.Continue || slices.ContainsFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.OperationID == msg.OperationID })) { cmds = append(cmds, m.resolveOperationCmd(msg.Handle)) }
// If user pressed Esc while this send was in flight, dispatch interrupt now that send is accepted if m.interruptDeferred { m.interruptDeferred = false62 unmodified lines m.submitInput(prompt, &cmds) return tea.Batch(cmds...)}
func (m ChatModel) queryOperationCmd(handle *daemon.OperationHandle) tea.Cmd { client, id, gen, ctx := m.client, m.SessionID, m.Generation, m.streamCtx return func() tea.Msg { receipt, err := client.ResolveOperation(ctx, handle) return ChatOperationResolvedMsg{SessionID: id, Generation: gen, Handle: handle, Receipt: receipt, Err: err} }}func (m ChatModel) resolveOperationCmd(handle *daemon.OperationHandle) tea.Cmd { id, gen := m.SessionID, m.Generation return tea.Tick(15*time.Second, func(time.Time) tea.Msg { return ChatOperationPollMsg{SessionID: id, Generation: gen, Handle: handle} })}func (m *ChatModel) sendCmd(prompt string, image *daemon.ImageAttachment, isCont bool) tea.Cmd {func (m *ChatModel) sendCmd(handle *daemon.OperationHandle, prompt string, image *daemon.ImageAttachment, isCont bool) tea.Cmd { client, id, gen := m.client, m.SessionID, m.Generation return func() tea.Msg { msg := ChatTurnSentMsg{SessionID: id, Generation: gen, Prompt: prompt, Image: image, Continue: isCont} var ( res *daemon.SendResult err error ) if isCont { res, err = client.Continue(context.Background()) } else { res, err = client.Send(context.Background(), prompt, image) } msg := ChatTurnSentMsg{SessionID: id, Generation: gen, Prompt: prompt, Image: image, Continue: isCont, Handle: handle, OperationID: handle.ID} result, err := client.SubmitOperation(context.Background(), handle) msg.Err = err if err == nil && res != nil { msg.OK, msg.Queued = res.OK, res.Queued if err == nil && result != nil { msg.OK, msg.Queued = result.OK, result.Queued } return msg }}28 unmodified lines m.handleSubmittedCommand(input, cmds) return }
if len(m.pendingUsers) >= MaxPendingUsers { if m.pendingSendCount() >= MaxPendingUsers { m.AddError("Too many messages are waiting. Wait for the current reply to finish before sending another message.") m.refreshViewportContent() return }1 unmodified line m.ClearNotices() m.TurnFailed = false m.Stopped = false
cmd := m.sendCmd(".", nil, true) if input != "." { img := m.AttachedImage continuation := input == "." image := m.AttachedImage if continuation { image = nil } handle, err := m.client.PrepareTurn(input, image, continuation) if err != nil { m.AddError(err.Error()) return } cmd := m.sendCmd(handle, input, image, continuation) if continuation { if m.pendingContinuations == nil { m.pendingContinuations = make(map[string]pendingOperation) } m.pendingContinuations[handle.ID] = pendingOperation{Handle: handle} } if !continuation { m.AttachedImage = nil m.pendingUsers = append(m.pendingUsers, PendingUserTurn{ Text: input, Image: img, At: time.Now().UnixMilli(), }) cmd = m.sendCmd(input, img, false) m.pendingUsers = append(m.pendingUsers, PendingUserTurn{Text: input, Image: image, At: time.Now().UnixMilli(), Handle: handle, OperationID: handle.ID}) } m.isSending, m.sentHere, m.Follow = true, true, true m.reseedMood() m.refreshViewportContent()5 unmodified lines m.syncLayout() trimmed := strings.TrimSpace(input) emit := func(msg tea.Msg) { *cmds = append(*cmds, func() tea.Msg { return msg }) }
name, args, _ := strings.Cut(trimmed, " ") if slices.ContainsFunc(m.CommandMenu.Catalog, func(command daemon.SessionCommand) bool { return command.Name == name && command.Skill }) { if m.pendingSendCount() >= MaxPendingUsers { m.AddError("Too many messages are waiting.") return } handle, err := m.client.PrepareSkill(name, args) if err != nil { m.AddError(err.Error()) return } m.pendingUsers = append(m.pendingUsers, PendingUserTurn{Text: trimmed, At: time.Now().UnixMilli(), Handle: handle, OperationID: handle.ID}) m.isSending, m.sentHere, m.Follow = true, true, true m.TurnFailed, m.Stopped = false, false *cmds = append(*cmds, m.sendCmd(handle, trimmed, nil, false), m.startAnimation()) m.refreshViewportContent() return }
switch trimmed { case "/agents": emit(ChatOpenAgentsMsg{}) case "/a", "/sessions":77 unmodified lines m.Usage = nil m.ClearNotices() m.TurnFailed, m.Stopping, m.Stopped = false, false, false m.Follow, m.scrollOffset = true, 0 m.pendingUsers = nil m.olderBefore, m.olderMore, m.loadingOlder = evt.Before, evt.More, false case daemon.EventCommitted: m.History.Stamp(evt.Seq) case daemon.EventRetry:3 unmodified lines m.clearAction() if !evt.Replayed { m.ClearNotices() } if evt.ClientID == m.client.ClientID() { if i := slices.IndexFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.Text == evt.Text }); i >= 0 { if evt.OperationID != "" { if i := slices.IndexFunc(m.pendingUsers, func(p PendingUserTurn) bool { return p.OperationID == evt.OperationID }); i >= 0 { m.pendingUsers = slices.Delete(m.pendingUsers, i, i+1) } } case daemon.EventText, daemon.EventThinking:202 unmodified linesExpand all return "stopped" } return "" } if m.isSending || len(m.pendingUsers) > 0 && !m.Status.Running { if m.isSending || m.pendingSendCount() > 0 && !m.Status.Running { return "preparing" } if m.Progress != nil { if m.Progress.Phase == "generating" {35 unmodified linesfunc (m ChatModel) phaseMood() mood { switch { case m.Stopping: return moodStopping case m.isSending || len(m.pendingUsers) > 0 && !m.Status.Running: case m.isSending || m.pendingSendCount() > 0 && !m.Status.Running: return moodPreparing case m.Progress != nil: return moodWorking case m.transcript.activeKind == StreamKindText:70 unmodified lines rows = append(rows, "") }
view := m.Viewport.View() if m.History.Len() == 0 && m.transcript.activeText == "" && m.ToolProgressText == "" { if m.History.Len() == 0 && len(m.pendingUsers) == 0 && m.transcript.activeText == "" && m.ToolProgressText == "" { textWidth := max(1, min(m.Renderer.BodyWidth, m.Viewport.Width())-railWidth) var emptyRows []string for _, line := range wrapOrChunkLine("What would you like to work on?", textWidth) { emptyRows = append(emptyRows, m.Renderer.rail(laneNone)+m.Styles.Faint.Render(line))215 unmodified linesExpand all default: return strings.TrimSuffix(fmt.Sprintf("%.1f", float64(n)/1_000_000), ".0") + "m" }}
func (m ChatModel) pendingSendCount() int { count := 0 for _, pending := range m.pendingUsers { if !pending.Expired { count++ } } return count}
func (m ChatModel) operationRecoverable(id string) bool { for _, pending := range m.pendingUsers { if pending.OperationID == id { return !pending.Expired } } pending, exists := m.pendingContinuations[id] return exists && !pending.Expired}97 unmodified lines9899100101102103104105106107108658 unmodified linesExpand all97 unmodified lines } if entry.Pending == queued { meta = append(meta, "queued") } if entry.Pending == unresolved { meta = append(meta, "unresolved") } plate := markChrome + style.Render(strings.ToLower(who)) if len(meta) > 0 { plate += r.Styles.Faint.Render(" · " + strings.Join(meta, " · ")) }658 unmodified linesExpand all20 unmodified lines21222324252627282930313231333433343536373839404 unmodified lines4546474847495051525319 unmodified lines73747576777879778081828384858687888589909192937 unmodified lines10110210310410110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314410510610710866 unmodified lines17517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723820 unmodified lines image := &daemon.ImageAttachment{} m.AttachedImage = image var commands []tea.Cmd m.submitInput("check this", &commands) handle := m.pendingUsers[0].Handle m.pendingUsers[0].Queued = queued m.TextArea.Reset() m.interruptDeferred = true
m, cmd := m.Update(ChatTurnSentMsg{ SessionID: "s", Generation: m.Generation, Prompt: "check this", Image: image, Handle: handle, OperationID: handle.ID, Err: fmt.Errorf("send: %w", &daemon.UncertainOutcomeError{Operation: "submit", Cause: io.ErrUnexpectedEOF}), Err: fmt.Errorf("send: %w", &daemon.UncertainOutcomeError{Operation: "submit", Cause: io.ErrUnexpectedEOF, Handle: handle}), }) if cmd != nil || m.isSending || m.interruptDeferred { t.Fatal("uncertain acknowledgement must settle sending without another mutation") if cmd == nil || m.isSending || m.interruptDeferred { t.Fatal("uncertain acknowledgement must settle sending and schedule receipt recovery") } if m.TextArea.Value() != "" || m.AttachedImage != nil { t.Fatal("uncertain send restored a duplicate prompt or image") }4 unmodified lines t.Fatalf("missing uncertainty guidance: %+v", m.Notices) }
for _, event := range []daemon.StreamEvent{ {Type: daemon.EventUser, Text: "check this", Source: "chat", ClientID: m.client.ClientID()}, {Type: daemon.EventUser, Text: "check this", Source: "chat", ClientID: m.client.ClientID(), OperationID: m.pendingUsers[0].OperationID}, {Type: daemon.EventCommitted, Seq: 1}, } { m, _ = m.Update(ChatStreamEventMsg{SessionID: "s", Generation: m.Generation, Event: event}) }19 unmodified lines image := &daemon.ImageAttachment{} m.AttachedImage = image var commands []tea.Cmd m.submitInput("already accepted", &commands) handle := m.pendingUsers[0].Handle m.TextArea.Reset() for _, event := range []daemon.StreamEvent{ {Type: daemon.EventUser, Text: "already accepted", Source: "chat", ClientID: m.client.ClientID()}, {Type: daemon.EventUser, Text: "already accepted", Source: "chat", ClientID: m.client.ClientID(), OperationID: m.pendingUsers[0].OperationID}, {Type: daemon.EventCommitted, Seq: 1}, } { m, _ = m.Update(ChatStreamEventMsg{SessionID: "s", Generation: m.Generation, Event: event}) }
m, cmd := m.Update(ChatTurnSentMsg{ SessionID: "s", Generation: m.Generation, Prompt: "already accepted", Image: image, Handle: handle, OperationID: handle.ID, Err: &daemon.UncertainOutcomeError{Operation: "submit", Cause: io.EOF}, Err: &daemon.UncertainOutcomeError{Operation: "submit", Cause: io.EOF, Handle: handle}, }) if cmd != nil || m.isSending || len(m.pendingUsers) != 0 || m.TextArea.Value() != "" || m.AttachedImage != nil { t.Fatal("late uncertain acknowledgement restored or resubmitted an echoed message") }7 unmodified lines t.Fatalf("want one committed user message, got %d", users) }}
func TestUncertainContinueDoesNotChangePendingOrRestorePrompt(t *testing.T) { m := newTestChatModel(t, &daemon.Session{ID: "s"}) m.pendingUsers = []PendingUserTurn{{Text: "already waiting", Queued: true}} m.isSending = true m, cmd := m.Update(ChatTurnSentMsg{ SessionID: "s", Generation: m.Generation, Continue: true, Prompt: ".", Err: &daemon.UncertainOutcomeError{Operation: "continue", Cause: io.EOF}, }) if cmd != nil || m.isSending || m.TextArea.Value() != "" || len(m.pendingUsers) != 1 { t.Fatal("uncertain continue restored or removed a user submission") } if len(m.Notices) != 1 || m.Notices[0].Message == "" { t.Fatalf("continue reported definite rejection: %+v", m.Notices) }}
func TestDefiniteRejectionRestoresPromptAndImage(t *testing.T) { m := newTestChatModel(t, &daemon.Session{ID: "s"}) image := &daemon.ImageAttachment{} m.pendingUsers = []PendingUserTurn{{Text: "try this", Image: image}} m, _ = m.Update(ChatTurnSentMsg{ SessionID: "s", Generation: m.Generation, Prompt: "try this", Image: image, Err: errors.New("submission refused"), }) if len(m.pendingUsers) != 0 || m.TextArea.Value() != "try this" || m.AttachedImage != image { t.Fatal("definite rejection did not restore the user's input") }}
func TestUncertainCreationKeepsCurrentSession(t *testing.T) { conn := daemon.NewConnection(daemon.ConnectionSnapshot{Port: 1}, "") session := &daemon.Session{ID: "current"} m := NewAppModel(conn, config.Profiles{}, session, "", false, nil) t.Cleanup(m.Chat.Close) updated, cmd := m.Update(sessionCreatedMsg{Err: &daemon.UncertainOutcomeError{Operation: "create session", Cause: io.EOF}}) m = updated.(AppModel) if cmd != nil || m.ActiveSession != session || m.Chat.SessionID != session.ID || m.State != AppStateChat { t.Fatal("uncertain creation changed the active session or resubmitted creation") } if len(m.Chat.Notices) != 1 || m.Chat.Notices[0].Message == "" { t.Fatalf("missing creation guidance: %+v", m.Chat.Notices) }}
func TestUncertainCommandKeepsSessionAndEffort(t *testing.T) { conn := daemon.NewConnection(daemon.ConnectionSnapshot{Port: 1}, "") session := &daemon.Session{ID: "current", Effort: "low"} m := NewAppModel(conn, config.Profiles{}, session, "", false, nil)66 unmodified lines return &daemon.UncertainOutcomeError{Operation: operation, Cause: &daemon.ProtocolError{ Code: "invalid_response", Operation: operation, Field: field, Cause: cause, }}}
func TestIdenticalPendingTurnsReconcileByOperationID(t *testing.T) { m := newTestChatModel(t, &daemon.Session{ID: "s"}) var commands []tea.Cmd m.submitInput("same message", &commands) first := m.pendingUsers[0].OperationID m.submitInput("same message", &commands) second := m.pendingUsers[1].OperationID m, _ = m.Update(ChatTurnSentMsg{SessionID: "s", Generation: m.Generation, OperationID: second, Handle: m.pendingUsers[1].Handle, Queued: true, OK: true}) if m.pendingUsers[0].Queued || !m.pendingUsers[1].Queued { t.Fatal("admission result matched the wrong identical message") } m.handleStreamEvent(daemon.StreamEvent{Type: daemon.EventUser, Text: "same message", OperationID: second}) if len(m.pendingUsers) != 1 || m.pendingUsers[0].OperationID != first { t.Fatal("stream echo retired the wrong identical message") } m.handleStreamEvent(daemon.StreamEvent{Type: daemon.EventUser, Text: "same message", OperationID: first}) if len(m.pendingUsers) != 0 { t.Fatal("original pending message was not reconciled") }}
// A timer scheduled before expiry may arrive afterward; rejecting it in the// model prevents the timer from issuing a receipt query.func TestScheduledReceiptPollStopsAfterExpiry(t *testing.T) { for _, prompt := range []string{"waiting input", "."} { t.Run(prompt, func(t *testing.T) { m := newTestChatModel(t, &daemon.Session{ID: "s"}) image := &daemon.ImageAttachment{} if prompt != "." { m.AttachedImage = image } var commands []tea.Cmd m.submitInput(prompt, &commands) var handle *daemon.OperationHandle if prompt == "." { for _, pending := range m.pendingContinuations { handle = pending.Handle } } else { handle = m.pendingUsers[0].Handle } tick := ChatOperationPollMsg{SessionID: "s", Generation: m.Generation, Handle: handle} m, command := m.Update(ChatOperationResolvedMsg{ SessionID: "s", Generation: m.Generation, Handle: handle, Err: &daemon.APIError{StatusCode: 410, Code: "operation_expired"}, }) if command != nil { t.Fatal("expiry scheduled another receipt poll") } m, command = m.Update(tick) if command != nil || m.operationRecoverable(handle.ID) { t.Fatal("already scheduled tick restarted expired receipt recovery") } if prompt != "." && (len(m.pendingUsers) != 1 || m.pendingUsers[0].Handle != handle || m.pendingUsers[0].Image != image) { t.Fatal("expiry lost the original submission or attachment") } }) }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221//go:build unix
// Expired receipts stop recovery without losing the original intent. The real// daemon admits each request before the proxy discards its acknowledgement.package e2e
import ( "errors" "fmt" "net/http" "slices" "strings" "sync" "testing" "time"
"albedo/cli/internal/daemon" "albedo/cli/internal/tui" tea "charm.land/bubbletea/v2")
func operationRequests(proxy *acknowledgementProxy) []string { proxy.mu.Lock() defer proxy.mu.Unlock() return slices.DeleteFunc(slices.Clone(proxy.requests), func(request string) bool { receiptQuery := strings.HasPrefix(request, "GET /operations/") creation := request == "POST /sessions" submission := strings.HasPrefix(request, "POST /sessions/") && strings.HasSuffix(request, "/events") return !receiptQuery && !creation && !submission })}
func TestTUIExpiredSubmissionsRemainUnresolvedAcrossNavigation(t *testing.T) { entered, release := make(chan struct{}), make(chan struct{}) released := false defer func() { if !released { close(release) } }() var enteredOnce sync.Once profile := providerRoute(t, func(request map[string]any) string { if lastUserText(request) == "hold expired submissions" { enteredOnce.Do(func() { close(entered) }) <-release } return echoReply(request) }) session := daemonSession(t, newSession(t, t.TempDir())) other := daemonSession(t, newSession(t, t.TempDir())) if _, err := daemon.NewChatClient(conn(t), session.ID).Send(t.Context(), "hold expired submissions", nil); err != nil { t.Fatal(err) } select { case <-entered: case <-time.After(10 * time.Second): t.Fatal("active turn never started") } proxy := cutAcknowledgement(t, "/sessions/"+session.ID+"/events", expiredAcknowledgement) driver := driveTUIWithConnection(t, &session, proxy.connection) t.Cleanup(func() { driver.App.Chat.Close() }) driver.Update(tea.WindowSizeMsg{Width: 120, Height: 100}) driver.connected() var handles []*daemon.OperationHandle for index := 0; index < tui.MaxPendingUsers+1; index++ { prompt := fmt.Sprintf("unresolved user %d", index) driver.App.Chat.TextArea.SetValue(prompt) if index == 0 { driver.App.Chat.AttachedImage = &daemon.ImageAttachment{Data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC", ImageMetadata: daemon.ImageMetadata{MimeType: daemon.ImagePNG, Width: 1, Height: 1, Bytes: 69}} } var sent tui.ChatTurnSentMsg for _, message := range driver.results(driver.Update(tea.KeyPressMsg{Code: tea.KeyEnter})) { if result, ok := message.(tui.ChatTurnSentMsg); ok { sent = result } if cmd := driver.Update(message); cmd != nil { if _, ok := message.(tui.ChatTurnSentMsg); ok { t.Fatalf("expired submission scheduled recovery: %T", message) } } } uncertainty, ok := errors.AsType[*daemon.UncertainOutcomeError](sent.Err) if !ok || uncertainty.Handle != sent.Handle || !daemon.IsOperationExpired(sent.Err) || sent.Handle == nil { t.Fatalf("expired outcome lost original handle: %+v", sent) } handles = append(handles, sent.Handle) } driver.App.Chat.TextArea.SetValue(".") for _, message := range driver.results(driver.Update(tea.KeyPressMsg{Code: tea.KeyEnter})) { if sent, ok := message.(tui.ChatTurnSentMsg); ok { if sent.Handle == nil || !daemon.IsOperationExpired(sent.Err) { t.Fatalf("expired continuation: %+v", sent) } handles = append(handles, sent.Handle) } if cmd := driver.Update(message); cmd != nil { if _, ok := message.(tui.ChatTurnSentMsg); ok { t.Fatal("expired continuation scheduled recovery") } } } if len(handles) != tui.MaxPendingUsers+2 { t.Fatal("expired turns consumed pending capacity") } before := operationRequests(proxy) for _, handle := range handles { if cmd := driver.Update(tui.ChatOperationPollMsg{SessionID: session.ID, Generation: driver.App.Chat.Generation, Handle: handle}); cmd != nil { t.Fatal("expired operation resumed polling") } } driver.Update(tui.FolderOpenSessionMsg{Session: other}) for _, message := range driver.results(driver.Update(tui.FolderOpenSessionMsg{Session: session})) { driver.Update(message) } driver.Update(tea.WindowSizeMsg{Width: 120, Height: 100}) view := driver.View() for _, handle := range handles { if !strings.Contains(view, handle.ID) { t.Fatalf("navigation lost unresolved identity %s:\n%s", handle.ID, view) } } if after := operationRequests(proxy); !slices.Equal(before, after) { t.Fatalf("navigation resumed recovery: before=%v after=%v", before, after) } if len(before) != 2*len(handles) { t.Fatalf("expiry retried HTTP admission: %v", before) } // A durable event can still reconcile an outcome whose receipt has expired. close(release) released = true waitIdle(t, session.ID, profile, 2) history, err := daemon.NewChatClient(conn(t), session.ID).History(t.Context(), 0, 120) if err != nil { t.Fatal(err) } matched := false for _, event := range history.Events { if event.Type == daemon.EventUser && event.OperationID == handles[0].ID { matched = true if event.Image == nil || event.Image.Width != 1 || event.Image.Height != 1 { t.Fatalf("expired image metadata lost: %+v", event) } driver.Update(tui.ChatStreamEventMsg{SessionID: session.ID, Generation: driver.App.Chat.Generation, Event: event}) } } if !matched || strings.Contains(driver.View(), handles[0].ID) { t.Fatal("durable user echo did not reconcile expired row") }
}
func TestTUICreationExpiryStopsRecoveryWithoutSwitchingSession(t *testing.T) { providerRoute(t, echoReply) initial := daemonSession(t, newSession(t, t.TempDir())) other := daemonSession(t, newSession(t, t.TempDir())) proxy := cutAcknowledgement(t, "/sessions", unresolvedAcknowledgement) driver := driveTUIWithConnection(t, &initial, proxy.connection) t.Cleanup(func() { driver.App.Chat.Close() }) uncertain := driver.Send(tui.FolderNewSessionMsg{Workspace: t.TempDir()}) driver.Update(tui.FolderOpenSessionMsg{Session: other}) recovery := driver.Update(uncertain) if recovery == nil { t.Fatal("uncertain creation lost recovery") } proxy.allowReceipt.Store(true) proxy.expireReceipt.Store(true) var tick tea.Msg for _, message := range driver.results(recovery) { tick = message response := driver.Send(message) if cmd := driver.Update(response); cmd != nil { t.Fatal("expired creation scheduled further recovery") } } if driver.App.ActiveSession.ID != other.ID || !strings.Contains(driver.View(), "expired") { t.Fatalf("creation expiry switched session or lost notice:\n%s", driver.View()) } before := operationRequests(proxy) operationID := strings.TrimPrefix(before[len(before)-1], "GET /operations/") if !strings.Contains(driver.View(), operationID) { t.Fatal("expired creation notice lost its operation identity") } if tick == nil || driver.Update(tick) != nil { t.Fatal("scheduled creation tick restarted expired lookup") } if after := operationRequests(proxy); !slices.Equal(before, after) { t.Fatal("creation expiry resumed HTTP recovery") }}
func TestTUIRejectedSubmissionRestoresPromptAndImage(t *testing.T) { providerRoute(t, echoReply) driver := newTUIDriver(t) t.Cleanup(func() { driver.App.Chat.Close() }) driver.connected() image := &daemon.ImageAttachment{Data: "invalid", ImageMetadata: daemon.ImageMetadata{MimeType: daemon.ImagePNG, Width: 1, Height: 1, Bytes: 1}} driver.App.Chat.AttachedImage = image driver.App.Chat.TextArea.SetValue("restore this rejected image") var sent tui.ChatTurnSentMsg for _, message := range driver.results(driver.Update(tea.KeyPressMsg{Code: tea.KeyEnter})) { if result, ok := message.(tui.ChatTurnSentMsg); ok { sent = result } driver.Update(message) } api, ok := errors.AsType[*daemon.APIError](sent.Err) if !ok || api.StatusCode != http.StatusBadRequest || sent.Handle == nil { t.Fatalf("real daemon did not reject image: %+v", sent) } if driver.App.Chat.TextArea.Value() != sent.Prompt || driver.App.Chat.AttachedImage != image { t.Fatal("rejection lost prompt or attachment") } // A following valid submission demonstrates that the rejected row left no pending intent. driver.App.Chat.AttachedImage = nil driver.App.Chat.TextArea.SetValue("valid after rejection") for _, message := range driver.results(driver.Update(tea.KeyPressMsg{Code: tea.KeyEnter})) { if result, ok := message.(tui.ChatTurnSentMsg); ok && result.Err != nil { t.Fatal(result.Err) } }}4 unmodified lines567891011121314151617181961 unmodified lines81828384838485858687888990919287888990939495929394969798991001011021031041051061071081091101111121131141151161171181191201219712212312412510210312612710510610710812812913013113213313413513613713813914014114214314414514614786 unmodified lines4 unmodified lines// real directory, resolving a relative path, or keeping the new workspace.package e2e
import ( "albedo/cli/internal/daemon" "fmt" "os" "path/filepath" "strings" "testing" "time"
"albedo/cli/internal/tui"
tea "charm.land/bubbletea/v2"61 unmodified lines }
d.connected() d.App.Chat.TextArea.SetValue("hello from nowhere") d.Dispatch(tea.KeyPressMsg{Code: tea.KeyEnter}) if view := d.View(); d.App.State != tui.AppStateFolderPicker || !strings.Contains(view, "not found") { t.Fatalf("a missing workspace did not open the picker:\n%s", view) messages := d.results(d.Update(tea.KeyPressMsg{Code: tea.KeyEnter})) var admitted tui.ChatTurnSentMsg for _, message := range messages { if sent, ok := message.(tui.ChatTurnSentMsg); ok { admitted = sent d.Update(sent) } } // Leaving the picker gives the refused turn back to the composer. d.Dispatch(tea.KeyPressMsg{Code: tea.KeyEscape}) if got := d.App.Chat.TextArea.Value(); d.App.State != tui.AppStateChat || got != "hello from nowhere" { t.Fatalf("esc left state %v and composer %q", d.App.State, got) if admitted.Err != nil || admitted.Handle == nil || !admitted.OK { t.Fatalf("durable admission: %+v", admitted) } d.Dispatch(tea.KeyPressMsg{Code: tea.KeyEnter}) if d.App.State != tui.AppStateFolderPicker { t.Fatalf("sending again did not reopen the picker:\n%s", d.View()) var receipt daemon.OperationReceipt deadline := time.Now().Add(5 * time.Second) for { var err error receipt, err = daemon.ResolveOperation(t.Context(), conn(t), admitted.Handle) if err != nil { t.Fatal(err) } if receipt.BlockingReason != "" { break } if time.Now().After(deadline) { t.Fatal("missing workspace did not retain blocked pending input") } time.Sleep(25 * time.Millisecond) } if receipt.DeliveryStatus != "pending" { t.Fatalf("blocked input was lost: %+v", receipt) } d.Update(tui.ChatOperationResolvedMsg{SessionID: session.ID, Generation: d.App.Chat.Generation, Handle: admitted.Handle, Receipt: receipt}) if d.App.State != tui.AppStateChat || d.App.Chat.TextArea.Value() != "" || !strings.Contains(d.View(), "hello from nowhere") { t.Fatalf("blocked admission state=%v composer=%q view=%q", d.App.State, d.App.Chat.TextArea.Value(), d.View()) }
d.Dispatch(tui.ChatExecuteCommandMsg{Name: "/cd"}) d.Type(root + "/fou") // Settling would follow the turn's animation forever; one step sends it. moved, ok := d.Key(tea.KeyEnter).(tui.FolderMovedMsg) if !ok || moved.Err != nil { t.Fatalf("enter did not move the session: %#v", moved) } d.results(d.Update(moved))
d.Update(moved) waitIdle(t, session.ID, profile, 1) requests := suite.provider.requests(profile) body, _ := requests[len(requests)-1]["body"].(map[string]any) if got := lastUserText(body); !strings.Contains(got, "hello from nowhere") { t.Fatalf("the refused turn was not sent again, the provider saw %q", got) history, err := daemon.NewChatClient(conn(t), session.ID).History(t.Context(), 0, 120) if err != nil { t.Fatal(err) } users := 0 for _, event := range history.Events { if event.Type == daemon.EventUser { users++ if event.OperationID != admitted.OperationID { t.Fatal("workspace repair created another user intent") } } } if users != 1 { t.Fatalf("workspace repair delivered %d user rows", users) } if got, _ := filepath.EvalSymlinks(daemonSession(t, session.ID).Workspace); got != mustEval(t, found) { t.Fatalf("the daemon kept workspace %q", got) }}86 unmodified lines