okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petfeat(usage): time provider calls and show tok/s on the turn signoff0a3eddd2026-10-07

package 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,
			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
}

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 of a usage record that carries one.
func (t *openTurn) measure(usage *daemon.Usage) {
	if usage != nil && usage.TokensPerSecond != nil && *usage.TokensPerSecond > 0 {
		t.tokensPerSecond = *usage.TokensPerSecond
	}
}

// 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
	return []HistoryEntry{{
		Kind: EntryTurnEnd, Mood: outcome(turn.failed, stopped, elapsed),
		ElapsedMs: elapsed, Tools: turn.tools, Timestamp: end,
		TokensPerSecond: turn.tokensPerSecond,
	}}
}

func inputSpeaker(evt daemon.StreamEvent) string {
	if evt.Source == "mail" {
		return cmp.Or(evt.Speaker, "Agent")
	}
	if evt.Source == "" || evt.Source == "chat" {
		return "You"
	}
	return evt.Source
}