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 ¶ added in v0.6.0
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 ¶ added in v0.9.0
func ExtractStatusAngle ¶ added in v0.19.0
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 ¶ added in v0.7.2
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 ¶ added in v0.13.1
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 ¶ added in v0.13.1
func NewAccumulator() *Accumulator
func (*Accumulator) FlushRemaining ¶ added in v0.13.1
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 ¶ added in v0.13.1
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 ¶ added in v0.11.0
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 ¶ added in v0.11.0
func NewChunkBuffer() *ChunkBuffer
NewChunkBuffer creates a buffer with the default maximum size.
func NewChunkBufferLen ¶ added in v0.11.0
func NewChunkBufferLen(maxLen int) *ChunkBuffer
NewChunkBufferLen creates a buffer with a custom maximum size.
func (*ChunkBuffer) Append ¶ added in v0.11.0
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 ¶ added in v0.11.0
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 ¶ added in v0.19.0
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 ¶ added in v0.11.0
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 ¶ added in v0.11.0
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