Documentation
¶
Overview ¶
Package buffer provides the per-chat assistant turn buffering subsystem.
Buffer accumulates streaming deltas per chat until turn_ended writes the finalized assistant message to the chat file. Store is a concurrency-safe map of per-chat buffers.
Index ¶
- Constants
- func PartialPath(configDir string, chatID api.ChatID) string
- type Buffer
- func (buf *Buffer) AppendCodeReferences(refs []api.CodeReference) []api.CodeReference
- func (buf *Buffer) AppendTextDelta(delta, subtaskID string) (idx int, seq int64)
- func (buf *Buffer) AppendThinkingDelta(delta, subtaskID string) (idx int, seq int64)
- func (buf *Buffer) AppendToolUseBlock(toolCallID, subtaskID string) int
- func (buf *Buffer) ClosePartial(ctx context.Context, path string)
- func (buf *Buffer) ComputeDuration(toolCallID string) int
- func (buf *Buffer) MarkCancelledToolsFailed() []api.ToolCall
- func (buf *Buffer) OpenPartial(ctx context.Context, path string)
- func (buf *Buffer) RecordToolStart(toolCallID string)
- func (buf *Buffer) SetRefusal(r *api.RefusalInfo)
- func (buf *Buffer) TrackFileChanges(diffs []api.ToolDiff, isNewFile bool)
- func (buf *Buffer) WritePartial(ctx context.Context)
- type Lifecycle
- func (l *Lifecycle) CloseAndRemovePartial(ctx context.Context, chatID api.ChatID, buf *Buffer)
- func (l *Lifecycle) OpenPartialFile(ctx context.Context, chatID api.ChatID, buf *Buffer)
- func (l *Lifecycle) PartialPathFor(chatID api.ChatID) string
- func (l *Lifecycle) RecoverPartials() []RecoveredPartial
- type LineRange
- type LineTracker
- func (lt *LineTracker) Clear(chatID api.ChatID)
- func (lt *LineTracker) Get(chatID api.ChatID, filePath string) []LineRange
- func (lt *LineTracker) Record(chatID api.ChatID, filePath string, startLine, endLine, turn int, kind string)
- func (lt *LineTracker) RecordFromDiffs(chatID api.ChatID, diffs []api.ToolDiff, turn int, kind string)
- type PartialSnapshot
- type RecoveredPartial
- type Store
- type WritingPartial
Constants ¶
const DefaultOutputCap = 64 * 1024
DefaultOutputCap is the shared byte budget for subprocess output buffers (agent terminals, stderr line caps, whoami output, MCP prewarm logs). 64 KiB covers a full terminal screen at 200 cols × 50 rows with generous ANSI escapes and is well below container memory limits.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Buffer ¶
type Buffer struct {
ToolStartTimes map[string]int64
ToolCallIndex map[string]int
ChangedFiles map[string]*api.FileChange
Partial *WritingPartial
MessageID string
Content strings.Builder
Reasoning strings.Builder
// Refusal marks the in-flight turn as a model refusal (kiro-cli 2.13):
// set once from the refusal explanation chunk's _meta.kiro.refusal,
// persisted onto the final assistant message at turn end. Guarded by mu.
Refusal *api.RefusalInfo
ToolCalls []api.ToolCall
Blocks []api.Block
CodeReferences []api.CodeReference
Started bool
// contains filtered or unexported fields
}
Buffer accumulates streaming deltas per chat until turn_ended writes the finalized assistant message to the chat file.
Content and Reasoning are sibling streams that both fill during a single turn — extended-thinking models emit reasoning ("Thinking…") deltas alongside the regular response. The translator routes each chunk to the appropriate builder based on the upstream IsReasoning flag.
SAFETY: Buffer is not goroutine-safe by design — the single-writer invariant is enforced by the hub's per-chat dispatch loop. The mu field guards against silent corruption if the invariant is ever violated by a future refactor.
func (*Buffer) AppendCodeReferences ¶ added in v0.1.165
func (buf *Buffer) AppendCodeReferences(refs []api.CodeReference) []api.CodeReference
AppendCodeReferences merges refs into the turn's licensed-code attributions, deduping by (licenseName, repository, url) so the KAS fan-out (the same references broadcast under every live session id) and a completion that reproduces the same snippet twice don't produce duplicate chips. Returns the full deduped list so the caller can broadcast it idempotently (the client replaces its list rather than appending). Empty entries (no license name) are dropped by the caller before this point.
func (*Buffer) AppendTextDelta ¶
AppendTextDelta extends the last text block with a delta, or starts a new text block if the trailing block isn't text. Returns the index of the (possibly new) text block — broadcast on MessageChunkPayload.BlockIndex so the client knows which block the delta belongs to — and the delta's sequence number (broadcast as MessageChunkPayload.Seq; see the chunkSeq field).
subtaskID attributes the block to the agent that produced it ("" = top-level agent). A trailing text block is only extended when it belongs to the SAME subtask; a differing subtask starts a new block so a subagent's text never merges into the parent's trailing block.
func (*Buffer) AppendThinkingDelta ¶
AppendThinkingDelta is the BlockThinking analogue of AppendTextDelta. Reasoning chunks share a block until a non-thinking block (tool_use or text) breaks the run — or until the subtask id differs, so a subagent's reasoning starts its own block rather than merging into the parent's trailing thinking block. subtaskID attributes the block to the producing agent ("" = top-level).
func (*Buffer) AppendToolUseBlock ¶
AppendToolUseBlock records a new tool_use block referencing the given tool call id. Always allocates a new block (one per tool call, even if back-to-back tool calls would coalesce into a single "thinking" run for text). Returns the new block's index. subtaskID attributes the block to the agent that produced it ("" = top-level).
func (*Buffer) ClosePartial ¶
ClosePartial flushes the final state, closes the partial file fd, and removes the file at path.
func (*Buffer) ComputeDuration ¶
ComputeDuration returns the elapsed time since the tool started, or 0 if the start time was not recorded.
func (*Buffer) MarkCancelledToolsFailed ¶
MarkCancelledToolsFailed sets all in-progress tool calls to failed. Called on cancel so the client doesn't show stuck spinners.
func (*Buffer) OpenPartial ¶
OpenPartial opens the partial recovery file for a chat. Idempotent: a no-op once the file is already open, so it can be called on every content/reasoning chunk to open the .partial lazily (the first call wins). If opening fails, Partial remains nil (degraded mode).
func (*Buffer) RecordToolStart ¶
RecordToolStart records the start time for a tool call so we can compute DurationMs on completion.
func (*Buffer) SetRefusal ¶ added in v0.2.8
func (buf *Buffer) SetRefusal(r *api.RefusalInfo)
SetRefusal records the turn's model-refusal metadata (first write wins — KAS emits at most one refusal chunk per turn, so a duplicate is a replay and keeps the original).
func (*Buffer) TrackFileChanges ¶
TrackFileChanges accumulates per-file change stats from tool call diffs.
func (*Buffer) WritePartial ¶
WritePartial rewrites the partial file with the current buffer state. No-op if the partial writer was not opened or is disabled.
type Lifecycle ¶
Lifecycle encapsulates buffer partial-file orchestration: path resolution, open/close/remove, and crash-recovery scanning. The hub delegates to this type for all partial-file coordination.
func (*Lifecycle) CloseAndRemovePartial ¶
CloseAndRemovePartial closes the partial file fd and deletes the file.
func (*Lifecycle) OpenPartialFile ¶
OpenPartialFile opens (or creates) the partial recovery file for a chat.
func (*Lifecycle) PartialPathFor ¶
PartialPathFor returns the path for a chat's partial recovery file.
func (*Lifecycle) RecoverPartials ¶
func (l *Lifecycle) RecoverPartials() []RecoveredPartial
RecoverPartials scans the chats directory for orphaned .partial files left by a crash. Returns recovered snapshots paired with their chat IDs for the caller to persist. The caller is responsible for appending messages and broadcasting events.
type LineRange ¶
type LineRange struct {
Kind string `json:"kind"`
StartLine int `json:"start_line"`
EndLine int `json:"end_line"`
Turn int `json:"turn"`
}
LineRange is a range of lines modified by the agent.
type LineTracker ¶
type LineTracker struct {
// contains filtered or unexported fields
}
LineTracker tracks per-file line changes across all chats.
func (*LineTracker) Clear ¶
func (lt *LineTracker) Clear(chatID api.ChatID)
Clear removes all tracking data for a chat.
func (*LineTracker) Get ¶
func (lt *LineTracker) Get(chatID api.ChatID, filePath string) []LineRange
Get returns the line ranges for a file in a chat.
func (*LineTracker) Record ¶
func (lt *LineTracker) Record(chatID api.ChatID, filePath string, startLine, endLine, turn int, kind string)
Record adds a line range for a file change.
func (*LineTracker) RecordFromDiffs ¶
func (lt *LineTracker) RecordFromDiffs(chatID api.ChatID, diffs []api.ToolDiff, turn int, kind string)
RecordFromDiffs extracts line ranges from tool call diffs.
type PartialSnapshot ¶
type PartialSnapshot struct {
// Refusal preserves the model-refusal metadata (kiro-cli 2.13) so a
// crash between the refusal chunk and the turn commit still recovers
// the refusal callout onto the interrupted assistant message.
Refusal *api.RefusalInfo `json:"refusal,omitempty"`
MessageID string `json:"message_id"`
Content string `json:"content"`
Reasoning string `json:"reasoning,omitempty"`
ToolCalls []api.ToolCall `json:"tool_calls,omitempty"`
// Blocks mirrors Buffer.Blocks so a crash mid-turn preserves the
// chronological order. Without this, recovery would reconstruct
// the message from Content+ToolCalls only and fall back to the
// legacy "all text, then all tools" rendering.
Blocks []api.Block `json:"blocks,omitempty"`
// CodeReferences preserves licensed-code attributions accumulated
// during the turn so a crash mid-turn recovers them onto the
// interrupted assistant message.
CodeReferences []api.CodeReference `json:"code_references,omitempty"`
Ts int64 `json:"ts"`
}
PartialSnapshot is the JSON shape written to .partial files.
func RecoverPartialSnapshot ¶
func RecoverPartialSnapshot(data []byte) (PartialSnapshot, bool)
RecoverPartialSnapshot parses a partial file's bytes into a snapshot. A snapshot with empty Content is still recoverable when Reasoning is non-empty (a turn that was entirely reasoning at crash time).
type RecoveredPartial ¶
type RecoveredPartial struct {
ChatID api.ChatID
Path string
Snapshot PartialSnapshot
}
RecoveredPartial holds a recovered partial snapshot and its metadata.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a concurrency-safe store for per-chat assistant buffers. It owns its own mutex so buffer operations don't contend with unrelated Hub state.
type WritingPartial ¶
type WritingPartial struct {
// contains filtered or unexported fields
}
WritingPartial is the active-state handle for the crash-recovery partial file. It is only obtainable via OpenPartial, making it impossible to call Write on an idle or disabled writer at compile time.
Each flush replaces the file atomically (temp → fsync → rename via atomicfile) instead of truncate-and-rewrite on one inode: a crash mid-rewrite previously left truncated JSON that RecoverPartialSnapshot dropped as unrecoverable, silently losing the crash-recovery copy of the in-flight turn.
func OpenPartial ¶
func OpenPartial(ctx context.Context, path string) *WritingPartial
OpenPartial transitions from idle to writing for the partial file at path. Returns nil if path is empty or the context is cancelled — callers check for nil to detect degraded mode. The file itself is created by the first flush (atomically); an open failure surfaces there and disables further writes for the turn.
func (*WritingPartial) CloseAndRemove ¶
func (wp *WritingPartial) CloseAndRemove(ctx context.Context, path string)
CloseAndRemove removes the file at path. (No fd is held between flushes — each Write is an atomic replacement.)
func (*WritingPartial) Write ¶
func (wp *WritingPartial) Write(ctx context.Context, snap *PartialSnapshot)
Write rewrites the partial file with the given snapshot data. Writes are throttled to at most once per writeThrottleInterval to reduce fsync pressure during fast model output.
Source Files