Documentation
¶
Overview ¶
Package round implements the protocol-neutral "message round" aggregator shared by every agent path. A Builder consumes chunk-level events via the agentstream.StreamHandler contract, forwards pre-aggregated UI events via agui.EventHandler, and emits complete rounds (assistant message + tool results) via a persistence callback.
Concurrency ¶
Builder methods are internally synchronised with a single mutex; callers do not need to add their own locking. Streaming events may arrive on a different goroutine from the cancel / finalize path — the builder serialises them.
onFlush context ¶
The onFlush callback installed via Reset is invoked to persist a completed round. Implementations MUST use a detached ctx (e.g. context.Background()) rather than the run ctx: rounds can be flushed after a cancel, and persisting them is precisely the behaviour we want to preserve in that case.
Flush policy ¶
There are two eager-flush triggers:
- A new message / thought chunk arrives while tool calls have already accumulated — the previous round is complete and is flushed before the new round begins.
- The tool-result count catches up with the tool-call count — the round is complete and flushed.
Any flush path (eager + explicit CollectMessages) goes through the same half-complete-round handling: each tool call that has not received a result is filled with a [placeholder] role=tool message, so the on-disk history is always tool_use / tool_result paired. Placeholders are synthesised before flushing, never after — the invariant "any flushed output is paired" must hold at every disk write.
The role=tool messages (real + synthetic) in a flushed round are reordered to match the declared order of the assistant's ToolCalls, regardless of the order in which results arrived. This keeps summary index bookkeeping and LLM context reconstruction deterministic.
Index ¶
- Constants
- func FlushPending(b *Builder, persist func([]*schema.Message) error, label string)
- func PersistContext(parent context.Context) (context.Context, context.CancelFunc)
- type Builder
- func (b *Builder) ClearHandler()
- func (b *Builder) ClearOnFlush()
- func (b *Builder) CollectMessages(reason PlaceholderReason) []*schema.Message
- func (b *Builder) EmitPendingEnds(reason PlaceholderReason)
- func (b *Builder) FinalizeStreaming() (inMsg bool, inThought bool)
- func (b *Builder) HasAccumulatedContent() bool
- func (b *Builder) HasOnFlush() bool
- func (b *Builder) OnMessageChunk(text string)
- func (b *Builder) OnThoughtChunk(text string)
- func (b *Builder) OnTokenUsage(totalTokens int)
- func (b *Builder) OnToolCall(id, title string)
- func (b *Builder) OnToolCallArgsSnapshot(id, args string)
- func (b *Builder) OnToolCallUpdate(id, content string, status agentstream.ToolCallStatus)
- func (b *Builder) PendingToolCallIDs() []string
- func (b *Builder) Reset(h agui.EventHandler, onFlush func([]*schema.Message))
- func (b *Builder) SawTokenUsage() bool
- func (b *Builder) SetLogLabel(label string)
- func (b *Builder) SetStitcher(fn func(toolCallID string, real *schema.Message))
- type PersistErr
- type PlaceholderReason
Constants ¶
const PersistTimeout = 10 * time.Second
PersistTimeout is the deadline attached to onFlush / onStitch / final post-Run sync persist contexts. It bounds how long any one persist step can wait on the per-session lock or on a ctx-aware repo operation before returning ctx.DeadlineExceeded, so callers can reason about a "this should not have blocked beyond N seconds" budget when surfacing warnings.
Coverage today: the per-session lock in repository/session_locks.go honours ctx via ctxRWMutex, and every ChatContextRepo / ChatContextManager method threads ctx through to the lock acquire, so a contended lock or hung competing writer surfaces as ctx.Err() within this budget. Underlying file I/O (FileRead / JSONLAppendLine / AtomicWriteFile) is still blocking-only — once the goroutine has the lock it can stall arbitrarily on a sick disk; only the lock-wait phase is bounded today. 10s sized to comfortably cover normal write latency while keeping a stuck disk from pinning the deferred chain (and through it runSem) indefinitely.
Variables ¶
This section is empty.
Functions ¶
func FlushPending ¶
FlushPending drains any in-flight round from the builder and persists the collected messages when no onFlush callback is still installed. This is the shared cleanup primitive used by every agent implementation.
Ordering is load-bearing: EmitPendingEnds MUST run before CollectMessages. CollectMessages clears the tool-call accumulators and synthesises [placeholder] role=tool messages for any unresolved ids; if UI end events fire after, the Run's deferred cleanup sees no pending ids and never emits OnToolCallEnd for them, leaving the UI bubbles stuck in "pending" until history reload.
If onFlush is still wired (a Run is active), CollectMessages simply drains — the callback has already persisted everything. Otherwise persist is invoked inline with the drained buffer.
label identifies the agent path in warning logs (e.g. "acp acpSession=…"). FlushPending is typically called from a cleanup path that has no live ctx (external Cancel()), so log fallback uses context.Background(); callers with a live ctx can log around it.
func PersistContext ¶
PersistContext returns a detached context with the canonical persist timeout applied. Detached so caller-side Stop / DeadlineExceeded do NOT cancel the persist write (we want flush after Stop to land on disk), and bounded by PersistTimeout so the ctx-aware repo / chatctx layer observes a single canonical deadline. See PersistTimeout for the current scope of what the deadline covers.
Use as:
persistCtx, cancel := round.PersistContext(ctx)
defer cancel()
if err := ctxMgr.AppendMessages(persistCtx, msgs...); err != nil { ... }
Retains the caller's ctx values (logger / trace attrs) via context.WithoutCancel so post-cancel writes are still attributable in logs.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder aggregates streaming events into complete rounds. One Builder per agent Run; call Reset before each Run.
func New ¶
func New() *Builder
New returns a fresh Builder. Call Reset to wire a handler and onFlush callback before feeding events.
func (*Builder) ClearHandler ¶
func (b *Builder) ClearHandler()
ClearHandler removes the agui handler reference without touching the accumulated message state.
func (*Builder) ClearOnFlush ¶
func (b *Builder) ClearOnFlush()
ClearOnFlush removes the onFlush and onStitch callbacks so late events do not try to persist via handlers that have already returned. Kept under the old name to avoid churn at the callsites.
func (*Builder) CollectMessages ¶
func (b *Builder) CollectMessages(reason PlaceholderReason) []*schema.Message
CollectMessages flushes the current round (synthesising placeholders for any missing tool results with the given reason) and returns every completed round message accumulated since the last call.
Typical use: on Run end use ReasonCanceled when ctx was cancelled, ReasonInterrupted when the Run errored, or pass ReasonInterrupted unconditionally — the reason only affects placeholder content.
func (*Builder) EmitPendingEnds ¶
func (b *Builder) EmitPendingEnds(reason PlaceholderReason)
EmitPendingEnds finalises any in-flight message/thought streaming and emits matching OnMessageEnd / OnThoughtEnd / OnToolCallEnd events via the stored handler. Invariant: callers MUST invoke this BEFORE CollectMessages on any flush-cleanup path, otherwise CollectMessages will clear the tool-call accumulators and synthesise [placeholder] results on disk without the UI ever seeing OnToolCallEnd — the tool bubble would stay "pending" until history reload.
reason is the same PlaceholderReason the caller will pass to the immediately-following CollectMessages. It governs the live OnToolCallInterrupted payload so the tooltip the user sees during stop/cancel matches the reason baked into the on-disk placeholder — otherwise the live UI says one thing (e.g. "interrupted") and after reload the same bubble says another ("canceled").
Duration bookkeeping: the flip from inMessage/inThought=true to false ALSO records the matching *FinishedAt timestamp, and every pending tool-call gets a toolFinishedAt entry. Without this, the subsequent CollectMessages flush would persist assistant finished_at as the moment thinking ended (or 0 → fallback to flush time) for a round that actually ended with streaming text, making history reload report a near-zero assistant-text duration.
Reading the handler + state under b.mu then releasing the lock before invoking handler methods follows the same pattern as OnMessageChunk etc., so handler callbacks that re-enter the builder do not deadlock.
func (*Builder) FinalizeStreaming ¶
FinalizeStreaming reports and clears the inMessage / inThought flags so the caller can emit matching OnMessageEnd / OnThoughtEnd events.
func (*Builder) HasAccumulatedContent ¶
HasAccumulatedContent reports whether the builder has received any meaningful content (message chunks, thought chunks, or tool calls) since the last Reset. Used for diagnostics: if a Run completes successfully but the builder is empty, the subprocess likely returned a vacuous response.
Rounds already flushed mid-run count too via completedMessages — a pure-tool round (every tool_call carrying its terminal status, no trailing message chunks) leaves the live accumulators empty after the result-catch-up flush, yet the run did produce real output. Reset clears completedMessages, so this never leaks state across runs.
func (*Builder) HasOnFlush ¶
HasOnFlush reports whether an onFlush callback is currently installed.
func (*Builder) OnMessageChunk ¶
OnMessageChunk appends a text fragment to the current assistant message. If tool calls have accumulated, this triggers an eager flush of the prior round before the new chunk starts (previous tool calls without results become [placeholder]superseded).
func (*Builder) OnThoughtChunk ¶
OnThoughtChunk appends a text fragment to the current reasoning segment. Eager-flush semantics mirror OnMessageChunk.
func (*Builder) OnTokenUsage ¶
OnTokenUsage forwards the Agent CLI's token count to the UI handler and records that the upstream path has surfaced at least one usage event.
func (*Builder) OnToolCall ¶
OnToolCall declares a new tool call. The assistant message (if any) and reasoning segment are closed before the tool-call UI event fires. Re-declarations of an id already known to this Run are ignored — see toolCallDeclaredLocked.
func (*Builder) OnToolCallArgsSnapshot ¶
OnToolCallArgsSnapshot replaces the arguments accumulated for a tool call. ACP rawInput and replacement content collections use snapshot semantics; they must never be concatenated like model token deltas.
func (*Builder) OnToolCallUpdate ¶
func (b *Builder) OnToolCallUpdate(id, content string, status agentstream.ToolCallStatus)
OnToolCallUpdate appends argument deltas for non-terminal updates and records the final tool result for terminal updates. ACP full argument snapshots enter through OnToolCallArgsSnapshot instead. When the result count catches up with the tool-call count, the round is flushed.
func (*Builder) PendingToolCallIDs ¶
PendingToolCallIDs returns tool-call IDs that were started but never received a terminal (completed/failed) status, so the caller can emit OnToolCallEnd for each.
func (*Builder) Reset ¶
func (b *Builder) Reset(h agui.EventHandler, onFlush func([]*schema.Message))
Reset clears all accumulators and installs the active handler + onFlush for a new Run. onStitch is also cleared — production callers that care about late-terminal stitch must call SetStitcher after Reset (the ACP runner does). Safe to call between Runs on the same Builder.
func (*Builder) SawTokenUsage ¶
SawTokenUsage reports whether the stream has surfaced a token-usage event at least once. Consumers use this as a fallback toggle: if the underlying path never reports usage, they can compute a local estimate.
func (*Builder) SetLogLabel ¶
SetLogLabel attaches a tracing label (typically a session / job id) to the builder. It is included in the builder's anomaly warnings so log consumers can aggregate "upstream stream ordering" signals per caller.
func (*Builder) SetStitcher ¶
SetStitcher installs the callback invoked when a late tool terminal arrives AFTER an eager superseded flush has already persisted a [placeholder] superseded row. The callback's job is to replace that placeholder with the real tool result on disk — the builder also fixes up its in-memory completedMessages before invoking this. Call AFTER Reset and before feeding any events. nil disables stitching (tests that don't care can skip it).
type PersistErr ¶
type PersistErr struct {
// contains filtered or unexported fields
}
PersistErr accumulates the first persistence error observed across a Run's onFlush / onStitch callbacks. Multiple agent backends used to hand-roll this exact pattern (mutex-guarded "first error wins", surfaced to retErr only AFTER the final flush has had a chance to fail too); centralising it stops the "is the CapturePersistErrTo defer placed correctly?" question from being a footgun that has to be re-answered per backend on every change.
The zero value is ready to use. Methods are goroutine-safe — onFlush and onStitch can fire from a detached cleanup goroutine concurrently with the main Run loop.
func (*PersistErr) CapturePersistErrTo ¶
func (p *PersistErr) CapturePersistErrTo(retErr *error, prefix string)
CapturePersistErrTo writes the captured persist error into *retErr, wrapping it with prefix. No-op if *retErr already holds an error (stream / cancel error takes precedence — a persist failure on top of a stream error is noise; the stream error is the more actionable signal).
Use as `defer p.CapturePersistErrTo(&retErr, "<backend> stream completed but persist failed")`. Defers are LIFO, so this MUST be declared BEFORE the final-flush defer so it fires LAST and observes the post-flush state.
func (*PersistErr) Err ¶
func (p *PersistErr) Err() error
Err returns the first recorded error, or nil if none.
func (*PersistErr) Record ¶
func (p *PersistErr) Record(err error)
Record stores err iff no earlier error was already captured. nil is a no-op so callers can pipe ordinary err returns straight in without a defensive `if err != nil` at every site.
type PlaceholderReason ¶
type PlaceholderReason string
PlaceholderReason signals why a tool-call placeholder was synthesised at flush time. It affects only the placeholder's content prefix; builder behaviour is otherwise identical across reasons.
const ( // ReasonCanceled: run cancelled before the tool produced a result. ReasonCanceled PlaceholderReason = msgextra.PlaceholderReasonCanceled // ReasonInterrupted: run exited with an error / panic before the // tool produced a result. ReasonInterrupted PlaceholderReason = msgextra.PlaceholderReasonInterrupted // ReasonSuperseded: eager flush triggered by the next round while // tool calls were still pending (rare). ReasonSuperseded PlaceholderReason = msgextra.PlaceholderReasonSuperseded )
Source Files