Documentation
¶
Index ¶
- Constants
- func Classify(event map[string]any) string
- func DigestLine(raw []byte) (line string, event map[string]any, err error)
- func ExtractLoopMarker(text string) (directive, label string, ok bool)
- func ExtractStatusAngle(text string) (phase, label string, ok bool)
- func ExtractStatusMarker(text string) (phase, label string, ok bool)
- func Stream(in io.Reader, out io.Writer, opts Options) error
- type Accumulator
- type ChunkBuffer
- type Options
Constants ¶
const ( TagMilestone = "MILESTONE" TagFinding = "FINDING" TagActivity = "ACTIVITY" )
Variables ¶
This section is empty.
Functions ¶
func Classify ¶
Classify returns a tag (MILESTONE, FINDING, or ACTIVITY) for a parsed event. The event map is what DigestLine has already unmarshaled.
Rules (in priority order):
MILESTONE: session.end, permission.request, and tool.call / tool.call_update where kind=="commit" or kind=="task" with status=="completed", or where the title/command contains "git commit".
FINDING: agent.message_chunk, agent.thought_chunk, user.message_chunk where the content text contains a deliberate finding signal (confidence ≥60, explicit [finding] prefix, or multi-word phrase: "reviewer flagged", "correction needed", "failed test").
ACTIVITY: everything else.
func DigestLine ¶
DigestLine parses a single raw JSON line and returns the formatted plain-text line, the parsed event map, and any error.
When the JSON has no "event" field the line should be skipped: the returned string is empty and event is nil (no error). Callers that only need the string can ignore the map; callers that also need classification (e.g. streamLine) can pass the returned map directly to Classify without re-parsing the bytes.
func ExtractLoopMarker ¶
func ExtractStatusAngle ¶
ExtractStatusAngle scans text for a full-line <|status: phase|> or <|status: phase | label|> angle-token. Returns the normalised phase, trimmed label, and ok=true when a known phase is found. Unlike the legacy bracket syntax, angle tokens must occupy their own line.
func ExtractStatusMarker ¶
ExtractStatusMarker scans text for the first [status: <phase>] or [status: <phase> | <label>] marker (inline, legacy). Returns the normalised phase, trimmed label, and ok=true when a known phase is found. Unknown phase words are ignored (ok=false) to avoid spurious events from unrelated bracket text.
Known phases: thinking, working, waiting, done.
Types ¶
type Accumulator ¶
type Accumulator struct {
// contains filtered or unexported fields
}
Accumulator buffers consecutive chunk events (agent.message_chunk, agent.thought_chunk, user.message_chunk) with the same session ID and event type, then flushes them as a single human-readable line when the event type or session changes.
func NewAccumulator ¶
func NewAccumulator() *Accumulator
func (*Accumulator) FlushRemaining ¶
func (a *Accumulator) FlushRemaining(classify bool) []string
FlushRemaining returns formatted lines for any buffered chunks that haven't been flushed yet. Call this at EOF to avoid losing trailing accumulated text.
func (*Accumulator) Process ¶
func (a *Accumulator) Process(raw []byte, classify bool) ([]string, error)
Process ingests a raw event line. It returns zero or more formatted lines that the caller should write to output. For chunk events that match the current buffer key, nothing is returned (the text is accumulated). When a chunk event triggers a flush (different key), the flushed lines are returned but the current event's text is buffered for next time. For non-chunk events, all accumulated buffers are flushed plus the current event is formatted.
Empty lines and events with no "event" field are silently ignored.
type ChunkBuffer ¶
type ChunkBuffer struct {
// contains filtered or unexported fields
}
ChunkBuffer accumulates text from streaming chunks and provides a sliding window for marker extraction. It prevents unbounded growth and deduplicates status markers so that the same status is not re-emitted on every subsequent chunk.
func NewChunkBuffer ¶
func NewChunkBuffer() *ChunkBuffer
NewChunkBuffer creates a buffer with the default maximum size.
func NewChunkBufferLen ¶
func NewChunkBufferLen(maxLen int) *ChunkBuffer
NewChunkBufferLen creates a buffer with a custom maximum size.
func (*ChunkBuffer) Append ¶
func (cb *ChunkBuffer) Append(text string)
Append adds text to the buffer and trims the oldest bytes when the total exceeds maxLen.
func (*ChunkBuffer) ScanLoopMarker ¶
func (cb *ChunkBuffer) ScanLoopMarker() (directive, label string, ok bool)
ScanLoopMarker runs ExtractLoopMarker over the accumulated buffer. Note: unlike ScanStatusMarker, this does NOT deduplicate. In the WaitForSession loop the severity guard (abort > exit > continue) prevents incorrect behaviour even if a loop marker is re-found while still in the buffer.
func (*ChunkBuffer) ScanStatusAngle ¶
func (cb *ChunkBuffer) ScanStatusAngle() (phase, label string, ok bool)
ScanStatusAngle runs ExtractStatusAngle over the accumulated buffer. It returns ok=false for a status angle token that was already returned by a previous call, preventing duplicate emissions.
func (*ChunkBuffer) ScanStatusMarker ¶
func (cb *ChunkBuffer) ScanStatusMarker() (phase, label string, ok bool)
ScanStatusMarker runs ExtractStatusMarker over the accumulated buffer. It returns ok=false for a status marker that was already returned by a previous call, preventing duplicate emissions when the same text remains in the buffer across multiple chunks.
func (*ChunkBuffer) Text ¶
func (cb *ChunkBuffer) Text() string
Text returns the accumulated buffer contents. Used in unit tests for assertions.
type Options ¶
type Options struct {
Follow bool
PollInterval time.Duration
Format string
// Classify, when true, prefixes each plain-format line with a tag:
// MILESTONE, FINDING, or ACTIVITY. For JSON format, a top-level
// "classify" field is injected into each emitted object.
Classify bool
// Accumulate, when true in plain format, buffers consecutive chunk
// events (thought/message chunks) with the same session ID and event
// type, flushing them as a single human-readable line when the event
// type or session changes. Ignored in json format.
Accumulate bool
// CursorPath, when non-empty, causes Stream to atomically rewrite the
// cursor file after processing. The caller is responsible for seeking the
// underlying reader to the offset recorded in the cursor file before
// calling Stream; CursorStartOffset must be set to that same offset so
// Stream can compute the absolute file position when writing the cursor.
//
// Design note: we keep Stream's signature as io.Reader (rather than
// *os.File) so tests can pass bytes.Buffer or strings.Reader without
// touching the filesystem. The caller in watch.go handles the actual
// os.File seek; we track bytes consumed here via a counting wrapper.
CursorPath string
CursorStartOffset int64
}
Source Files