Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413package tui
import ( "albedo/cli/internal/daemon" "cmp" "slices" "strings" "time")
// transcriptState buffers streamed entries and tracks the turn in progress.type transcriptState struct { // Live thoughts are timed here; replayed thoughts carry daemon elapsed time. thinkingSince time.Time // Summarized thinking arrives after it is written, so its start is the last // live event other than thinking rather than its first received delta. lastEvent time.Time turn *openTurn activeKind ActiveStreamKind // Only the current update path appends to activeBuffer. Retained model // copies read activeText, whose bytes stay unchanged after each append. activeBuffer *strings.Builder activeText string thoughtMs int64 activeMessageID string liveIDs map[string]bool pendingCanonical *BoundedHistory}
func newTranscriptState() transcriptState { return transcriptState{}}
func (t *transcriptState) resetStream() { t.activeKind, t.activeText = StreamKindNone, "" t.activeBuffer = nil t.activeMessageID = "" t.liveIDs = nil t.pendingCanonical = nil}
func (t transcriptState) activeEntryKind() EntryKind { if t.activeKind == StreamKindThinking { return EntryThinking } return EntryAssistant}
func (t *transcriptState) settle(agentName string) []HistoryEntry { var entries []HistoryEntry if t.activeKind != StreamKindNone && t.activeText != "" { entry := HistoryEntry{ MessageID: t.activeMessageID, Kind: t.activeEntryKind(), Speaker: agentName, Text: strings.Clone(t.activeText), Timestamp: time.Now().UnixMilli(), } if entry.Kind == EntryThinking { entry.ElapsedMs = t.thoughtMs if !t.thinkingSince.IsZero() { entry.ElapsedMs = time.Since(t.thinkingSince).Milliseconds() } } entries = append(entries, entry) } t.activeKind, t.activeText = StreamKindNone, "" t.activeBuffer = nil t.activeMessageID = "" t.thinkingSince, t.thoughtMs = time.Time{}, 0 return entries}
func (t *transcriptState) streamDelta(kind ActiveStreamKind, text, agentName, messageID string) []HistoryEntry { if text == "" { return nil } t.turnIsLive() t.turn.answered = true var entries []HistoryEntry if t.activeKind != kind || t.activeMessageID != messageID { entries = t.settle(agentName) t.activeKind = kind t.activeMessageID = messageID } if t.activeBuffer == nil { t.activeBuffer = new(strings.Builder) } if messageID != "" { if t.liveIDs == nil { t.liveIDs = map[string]bool{} } t.liveIDs[messageID] = true } t.activeBuffer.WriteString(text) t.activeText = t.activeBuffer.String() if len(t.activeText) > MaxLiveStreamBytes { // One long thought splits into entries, each timed from its own start. watched := !t.thinkingSince.IsZero() entries = append(entries, t.settle(agentName)...) t.activeKind = kind t.activeMessageID = messageID if watched { t.thinkingSince = time.Now() } } return entries}
// apply returns newly settled entries. The history owner stamps committed rows.func (t *transcriptState) apply(evt daemon.StreamEvent, agentName string) []HistoryEntry { // Empty provider records and agent progress are observations, not replies. if evt.Type == daemon.EventMessage && evt.Text == "" || evt.Type == daemon.EventNote && evt.Source == "agent" { return nil } if evt.EntryID != "" && !evt.Replayed && len(t.liveIDs) > 0 && (evt.Type == daemon.EventMessage || evt.Type == daemon.EventThinking) { if t.pendingCanonical == nil { t.pendingCanonical = NewBoundedHistory(500, 2*1024*1024) } kind := EntryAssistant if evt.Type == daemon.EventThinking { kind = EntryThinking } entry := HistoryEntry{ID: evt.EntryID, Seq: evt.Position, Kind: kind, Speaker: agentName, Text: evt.Text, ElapsedMs: evt.ElapsedMs} if evt.Timestamp != nil { entry.Timestamp = *evt.Timestamp } if !t.pendingCanonical.Replace(entry) { t.pendingCanonical.Append(entry) } return nil } thoughtStart := cmp.Or(t.lastEvent, time.Now()) if !evt.Replayed && evt.Type != daemon.EventThinking { t.lastEvent = time.Now() } var entries []HistoryEntry switch evt.Type { case daemon.EventReset: t.resetStream() t.thinkingSince, t.thoughtMs = time.Time{}, 0 t.turn = nil case daemon.EventRetry: t.resetStream() case daemon.EventCommitted: if evt.ReplacesAllLive || slices.Contains(evt.ReplacesLiveIDs, t.activeMessageID) { t.activeKind = StreamKindNone t.activeText = "" t.activeBuffer = nil t.activeMessageID = "" t.thinkingSince = time.Time{} } if evt.ReplacesAllLive { t.liveIDs = nil } for _, id := range evt.ReplacesLiveIDs { delete(t.liveIDs, id) } if t.pendingCanonical != nil { entries = append(entries, t.pendingCanonical.Entries()...) t.pendingCanonical = nil } case daemon.EventUser: entries = t.settle(agentName) speaker := inputSpeaker(evt) var ts int64 if evt.Timestamp != nil && *evt.Timestamp > 0 { ts = *evt.Timestamp } // Your message closes the previous turn. Other sources open a turn // only when none is in flight. opens := t.turn == nil || evt.Source == "" || evt.Source == "chat" if opens { entries = append(entries, t.closeTurn(false)...) } entries = append(entries, HistoryEntry{ Kind: EntryUser, ID: evt.EntryID, Seq: evt.Position, Source: evt.Source, MailKind: evt.MailKind, SenderSessionID: evt.SenderSessionID, Speaker: speaker, Text: evt.Text, Timestamp: ts, }) if opens { t.turn = newOpenTurn(ts) } case daemon.EventText: entries = t.streamDelta(StreamKindText, evt.Text, agentName, evt.MessageID) case daemon.EventThinking: if evt.EntryID != "" { entries = t.settle(agentName) entries = append(entries, HistoryEntry{ID: evt.EntryID, Seq: evt.Position, Kind: EntryThinking, Speaker: agentName, Text: evt.Text, ElapsedMs: evt.ElapsedMs}) break } if t.activeKind != StreamKindThinking || t.activeMessageID != evt.MessageID { entries = t.settle(agentName) t.activeKind = StreamKindThinking t.activeMessageID = evt.MessageID if !evt.Replayed { t.thinkingSince = thoughtStart } } if evt.ElapsedObserved && !evt.Replayed { t.thinkingSince = time.Time{} t.thoughtMs = evt.ElapsedMs } else { t.thoughtMs += evt.ElapsedMs } entries = append(entries, t.streamDelta(StreamKindThinking, evt.Text, agentName, evt.MessageID)...) case daemon.EventToolProgress: if !evt.Replayed { t.turnIsLive() if evt.Progress != nil { entries = t.settle(agentName) } } case daemon.EventTool: entries = t.settle(agentName) entries = append(entries, HistoryEntry{ Kind: EntryTool, ID: evt.EntryID, Seq: evt.Position, ToolName: evt.ToolName, ToolArgs: evt.ToolArgs, ToolResult: evt.ToolResult, ToolTrace: evt.ToolTrace, ToolTitle: evt.ToolTitle, Timestamp: time.Now().UnixMilli(), }) if t.turn == nil { t.turn = newOpenTurn(0) } t.turn.tools++ t.turn.touch(evt.Timestamp) case daemon.EventMessage: entries = t.settle(agentName) var ts int64 if evt.Timestamp != nil && *evt.Timestamp > 0 { ts = *evt.Timestamp } entries = append(entries, HistoryEntry{Kind: EntryAssistant, ID: evt.EntryID, Seq: evt.Position, Speaker: agentName, Text: evt.Text, Timestamp: ts}) if t.turn == nil { t.turn = newOpenTurn(ts) } t.turn.answered = true t.turn.touch(evt.Timestamp) case daemon.EventNote: entries = t.settle(agentName) entries = append(entries, HistoryEntry{Kind: EntryNote, ID: evt.EntryID, Seq: evt.Position, Text: evt.Text, Timestamp: time.Now().UnixMilli()}) case daemon.EventError: entries = t.settle(agentName) entries = append(entries, HistoryEntry{Kind: EntryError, Text: evt.Text, Timestamp: time.Now().UnixMilli()}) if t.turn != nil { t.turn.failed = true entries = append(entries, t.closeTurn(false)...) } case daemon.EventCompacted: entries = t.settle(agentName) entries = append(entries, HistoryEntry{ Kind: EntryCompacted, ID: evt.EntryID, Seq: evt.Position, Text: evt.Summary, Evicted: evt.Evicted, Strategy: evt.Strategy, Timestamp: time.Now().UnixMilli(), }) case daemon.EventUsage: entries = t.settle(agentName) if t.turn != nil { t.turn.measure(evt.Usage) } case daemon.EventTurnCompleted: entries = t.settle(agentName) if t.turn != nil { t.turn.failed = t.turn.failed || evt.Source == "failed" entries = append(entries, t.closeTurn(evt.Source == "interrupted")...) } case daemon.EventInterrupted: entries = t.settle(agentName) if t.turn != nil { entries = append(entries, t.closeTurn(true)...) } else { entries = append(entries, HistoryEntry{Kind: EntryNote, Text: "stopped by you", Timestamp: time.Now().UnixMilli()}) } } return entries}
// A page boundary settles buffered text but does not finish the turn.// History retention is applied by the caller.func replayTranscript(events []daemon.StreamEvent, agentName string) []HistoryEntry { state := newTranscriptState() var entries []HistoryEntry for _, evt := range events { entries = append(entries, state.apply(evt, agentName)...) switch evt.Type { case daemon.EventReset: entries = nil case daemon.EventCommitted: for i := len(entries) - 1; i >= 0 && entries[i].Seq == 0; i-- { if entries[i].MessageID == "" { entries[i].Seq = evt.Seq } } } } return append(entries, state.settle(agentName)...)}
// openTurn follows the turn in flight until its signoff.type openTurn struct { // Unix timestamps in milliseconds. start, last int64 tools int // A live turn ends now, rather than at its last replayed event. live bool failed bool // answered is whether the agent wrote anything; a turn opened by a note // it never answered has nothing to sign off. answered bool // tokensPerSecond is the rate of the latest call that reported its timing. // A usage record can arrive again, so it replaces the rate, never adds to it. tokensPerSecond float64 // cacheLosses is what each call of the turn lost from its cache, by the // call's record time, so a usage record that arrives again counts once. cacheLosses map[int64]cacheLoss}
// cacheLoss is what a call read uncached that the call before it left cached:// Miss when the provider let it go, Drift when albedo's own prefix changed.type cacheLoss struct{ Miss, Drift int }
func newOpenTurn(ts int64) *openTurn { if ts == 0 { ts = time.Now().UnixMilli() } return &openTurn{start: ts, last: ts}}
func (t *openTurn) touch(ts *int64) { if ts != nil && *ts > t.last { t.last = *ts }}
// measure takes the rate and the cache miss of a usage record that carries them.func (t *openTurn) measure(usage *daemon.Usage) { if usage == nil { return } if usage.TokensPerSecond != nil && *usage.TokensPerSecond > 0 { t.tokensPerSecond = *usage.TokensPerSecond } loss := cacheLoss{reportedTokens(usage.CacheMissTokens), reportedTokens(usage.CacheDriftTokens)} if loss != (cacheLoss{}) && usage.RecordedAt != nil { if t.cacheLosses == nil { t.cacheLosses = map[int64]cacheLoss{} } t.cacheLosses[*usage.RecordedAt] = loss }}
func (t *openTurn) cacheLoss() cacheLoss { var total cacheLoss for _, loss := range t.cacheLosses { total.Miss += loss.Miss total.Drift += loss.Drift } return total}
// reportedTokens is a usage count, 0 when the record has none.func reportedTokens(count *int) int { if count == nil || *count < 0 { return 0 } return *count}
// An idle status racing the first event must not sign a turn off early.func (t *openTurn) begun() bool { return t.live || t.tools > 0 || t.last > t.start}
func (t *transcriptState) turnIsLive() { if t.turn == nil { t.turn = newOpenTurn(0) } t.turn.live = true}
func (t *transcriptState) closeTurn(stopped bool) []HistoryEntry { turn := t.turn if turn == nil { return nil } t.turn, t.lastEvent = nil, time.Time{} if !turn.answered && turn.tools == 0 && !turn.failed && !stopped { return nil } end := turn.last if turn.live { end = max(end, time.Now().UnixMilli()) } elapsed := end - turn.start loss := turn.cacheLoss() return []HistoryEntry{{ Kind: EntryTurnEnd, Mood: outcome(turn.failed, stopped, elapsed), ElapsedMs: elapsed, Tools: turn.tools, Timestamp: end, TokensPerSecond: turn.tokensPerSecond, CacheMissTokens: loss.Miss, CacheDriftTokens: loss.Drift, }}}
func inputSpeaker(evt daemon.StreamEvent) string { switch evt.Source { case "mail": return cmp.Or(evt.Speaker, "Agent") case "event": // Named by its producer, such as discord. return cmp.Or(evt.Speaker, "event") } if evt.Source == "" || evt.Source == "chat" { return "You" } return evt.Source}