stream

README

Stream

The stream package writes run events as versioned JSONL for uagent --stream. It is the contract for programs that drive uagent from outside Go.

1. Read the stream
2. Events
3. Summary
4. Compatibility

uagent --stream writes one JSON object per line to stdout and nothing else to stdout. Diagnostics and preflight errors still go to stderr. The process exit code follows the same table as a normal run.

Read the stream

Every line has three common fields:

Field	Meaning
v	Schema version. This document describes version 1.
type	Event type from the table below.
at	RFC 3339 timestamp in UTC. Runner events carry the runner's own recording time.

A consumer reads lines until EOF. Once the runner has started, the stream ends with exactly one run_finished line, even when saving summary.json fails. When preflight blocks the run, no line is written, stderr carries the reason, and the exit code is 2. When the runner cannot start, the stream stops after run_started, stderr carries the reason, and the exit code is 1. Ignore unknown type values and unknown fields so that additive changes do not break the reader.

Events

Type	Fields	Meaning
run_started	run_id, session_id, provider, model, effort, workspace	Always first. run_id names the run directory.
preflight_warning	code, message	A non-blocking preflight finding, such as auth_expiring.
user_message	id, text	The runner accepted a user message. id is the message ID it deduplicates on, so it acknowledges delivery of a message sent with that ID.
developer_message	id, text	The runner accepted a developer message: the harness's, not the user's, such as context it prepared. It is not counted in user_messages.
control_input	id, mode, effort, reason	The runner accepted a control message: settings (with effort), when_idle, hard, or heartbeat (with reason).
turn_started	turn, turn_id	The runner sent a request to the model. Turns count from 1; turn_id is the runner's ID for the turn.
model_responded	turn, turn_id, duration_ms, usage, stop, failure	The model answered. usage has input, cached_input, cache_write_input, output, and reasoning token counts. stop and failure are omitted when normal.
tool_called	call_id, name, label, arguments	The model asked for a tool. label is the Bash command or the raw arguments, on one line; arguments is the raw JSON.
tool_started	call_id, op_id, name, label, op_type	The runner started the operation for a call. op_type is the runner's operation type, such as shell.
tool_finished	call_id, op_id, name, label, ok, detail, duration_ms, op_type, out_path, err_path	The operation ended. detail is exit N, a status, or an error. op_id is omitted when the call failed before an operation started. out_path and err_path name the files holding a shell command's full output.
assistant_message	turn, text, final	Assistant text from a turn. final: true marks the final answer.
reasoning	turn, text	One reasoning summary line.
runner_error	message	The runner reported an error or printed a line that is not JSON.
run_finished	summary	Always last. See the summary below.

A tool that is still running when the run is killed has a tool_started line and no tool_finished line. The runner reports output files once an operation has finished, so out_path and err_path appear on tool_finished and are usually absent from tool_started.

Summary

summary in run_finished has the same shape as summary.json in the run directory:

Field	Meaning
status	ok, error, timeout, interrupted, or disk_limit; running in summary.json while the run is in progress
runner_exit_code	Exit code of uah-core-runner, or -1 when it was killed
run_id, session_id, provider, model, effort, workspace	The run's identity and settings
started_at, wall_ms	Start time (UTC) and wall-clock duration
stats	Aggregated statistics, described below
answer	The final answer, or the last assistant text when there was none

stats has event_span_ms, model_ms, tool_busy_ms, tool_model_overlap_ms, turns, user_messages, model_responses, tool_calls, failed_tool_calls, max_parallel_tools, tools_by_name, tokens, and final_answer. It also has stop_reasons, failures, errors, and warnings, which are omitted when empty. tool_model_overlap_ms is the time tools ran while the model was generating, which is where the runner's asynchronous tools save time.

Compatibility

The schema version changes only for a breaking change: a removed or renamed field, a changed type, or a changed meaning. New event types and new fields are additive and keep version 1. uagent --version reports the build that produced a stream.

Documentation

Overview

Package stream writes run events as versioned JSONL, one object per line. It is the contract for programs that drive uagent, such as rs-uagent-tui.

Index

• Constants
• func EventToDTO(event core.Event) (dto any, ok bool)
• func SummaryFromDTO(d SummaryDTO) core.Result
• type AssistantMessageDTO
• type ControlInputDTO
• type DeveloperMessageDTO
• type ModelRespondedDTO
• type PreflightWarningDTO
• type ReasoningDTO
• type RunFinishedDTO
• type RunStartedDTO
• type RunnerErrorDTO
• type Sink
  • func NewSink(w io.Writer) *Sink
  • func (s *Sink) Emit(event core.Event)
  • func (s *Sink) Err() error
• type StatsDTO
• type SummaryDTO
  • func SummaryToDTO(r core.Result) SummaryDTO
• type TokensDTO
• type ToolCalledDTO
• type ToolFinishedDTO
• type ToolStartedDTO
• type TurnStartedDTO
• type UserMessageDTO

Constants

const SchemaVersion = 1

SchemaVersion is bumped on any breaking change to the event or summary shape.

Functions

func EventToDTO

func EventToDTO(event core.Event) (dto any, ok bool)

EventToDTO maps a domain event to its wire form. ok is false for unknown events.

func SummaryFromDTO

func SummaryFromDTO(d SummaryDTO) core.Result

SummaryFromDTO restores run metadata from a saved summary. Stats are not restored: callers recompute them from events.jsonl.

Types

type AssistantMessageDTO

type AssistantMessageDTO struct {
	Turn  int    `json:"turn,omitempty"`
	Text  string `json:"text"`
	Final bool   `json:"final"`
	// contains filtered or unexported fields
}

type ControlInputDTO

type ControlInputDTO struct {
	ID     string `json:"id"`
	Mode   string `json:"mode"`
	Effort string `json:"effort,omitempty"`
	Reason string `json:"reason,omitempty"`
	// contains filtered or unexported fields
}

type DeveloperMessageDTO

type DeveloperMessageDTO struct {
	ID   string `json:"id"`
	Text string `json:"text"`
	// contains filtered or unexported fields
}

type ModelRespondedDTO

type ModelRespondedDTO struct {
	Turn       int       `json:"turn"`
	TurnID     string    `json:"turn_id,omitempty"`
	DurationMS int64     `json:"duration_ms"`
	Usage      TokensDTO `json:"usage"`
	Stop       string    `json:"stop,omitempty"`
	Failure    string    `json:"failure,omitempty"`
	// contains filtered or unexported fields
}

type PreflightWarningDTO

type PreflightWarningDTO struct {
	Code    string `json:"code"`
	Message string `json:"message"`
	// contains filtered or unexported fields
}

type ReasoningDTO

type ReasoningDTO struct {
	Turn int    `json:"turn,omitempty"`
	Text string `json:"text"`
	// contains filtered or unexported fields
}

type RunFinishedDTO

type RunFinishedDTO struct {
	Summary SummaryDTO `json:"summary"`
	// contains filtered or unexported fields
}

type RunStartedDTO

type RunStartedDTO struct {
	RunID     string `json:"run_id"`
	SessionID string `json:"session_id"`
	Provider  string `json:"provider"`
	Model     string `json:"model"`
	Effort    string `json:"effort"`
	Workspace string `json:"workspace"`
	// contains filtered or unexported fields
}

type RunnerErrorDTO

type RunnerErrorDTO struct {
	Message string `json:"message"`
	// contains filtered or unexported fields
}

type Sink

type Sink struct {
	// contains filtered or unexported fields
}

Sink writes each event as one JSON line.

func NewSink

func NewSink(w io.Writer) *Sink

func (*Sink) Emit

func (s *Sink) Emit(event core.Event)

func (*Sink) Err

func (s *Sink) Err() error

Err returns the first write error, if any.

type StatsDTO

type StatsDTO struct {
	EventSpanMS        int64          `json:"event_span_ms"`
	ModelMS            int64          `json:"model_ms"`
	ToolBusyMS         int64          `json:"tool_busy_ms"`
	ToolModelOverlapMS int64          `json:"tool_model_overlap_ms"`
	Turns              int            `json:"turns"`
	UserMessages       int            `json:"user_messages"`
	ModelResponses     int            `json:"model_responses"`
	ToolCalls          int            `json:"tool_calls"`
	FailedToolCalls    int            `json:"failed_tool_calls"`
	MaxParallelTools   int            `json:"max_parallel_tools"`
	ToolsByName        map[string]int `json:"tools_by_name"`
	Tokens             TokensDTO      `json:"tokens"`
	StopReasons        map[string]int `json:"stop_reasons,omitempty"`
	Failures           []string       `json:"failures,omitempty"`
	Errors             []string       `json:"errors,omitempty"`
	Warnings           []string       `json:"warnings,omitempty"`
	FinalAnswer        bool           `json:"final_answer"`
}

type SummaryDTO

type SummaryDTO struct {
	V              int       `json:"v"`
	Status         string    `json:"status"`
	RunnerExitCode int       `json:"runner_exit_code"`
	RunID          string    `json:"run_id"`
	SessionID      string    `json:"session_id"`
	Provider       string    `json:"provider"`
	Model          string    `json:"model"`
	Effort         string    `json:"effort"`
	Workspace      string    `json:"workspace"`
	StartedAt      time.Time `json:"started_at"`
	WallMS         int64     `json:"wall_ms"`
	Stats          StatsDTO  `json:"stats"`
	Answer         string    `json:"answer"`
}

SummaryDTO is a finished run. It is the run_finished payload and the content of summary.json.

func SummaryToDTO

func SummaryToDTO(r core.Result) SummaryDTO

type TokensDTO

type TokensDTO struct {
	Input           int64 `json:"input"`
	CachedInput     int64 `json:"cached_input"`
	CacheWriteInput int64 `json:"cache_write_input"`
	Output          int64 `json:"output"`
	Reasoning       int64 `json:"reasoning"`
}

type ToolCalledDTO

type ToolCalledDTO struct {
	CallID    string `json:"call_id"`
	Name      string `json:"name"`
	Label     string `json:"label"`
	Arguments string `json:"arguments,omitempty"`
	// contains filtered or unexported fields
}

type ToolFinishedDTO

type ToolFinishedDTO struct {
	CallID     string `json:"call_id"`
	OpID       string `json:"op_id,omitempty"`
	Name       string `json:"name"`
	Label      string `json:"label"`
	OK         bool   `json:"ok"`
	Detail     string `json:"detail"`
	DurationMS int64  `json:"duration_ms"`
	OpType     string `json:"op_type,omitempty"`
	OutPath    string `json:"out_path,omitempty"`
	ErrPath    string `json:"err_path,omitempty"`
	// contains filtered or unexported fields
}

type ToolStartedDTO

type ToolStartedDTO struct {
	CallID  string `json:"call_id"`
	OpID    string `json:"op_id"`
	Name    string `json:"name"`
	Label   string `json:"label"`
	OpType  string `json:"op_type,omitempty"`
	OutPath string `json:"out_path,omitempty"`
	ErrPath string `json:"err_path,omitempty"`
	// contains filtered or unexported fields
}

type TurnStartedDTO

type TurnStartedDTO struct {
	Turn   int    `json:"turn"`
	TurnID string `json:"turn_id,omitempty"`
	// contains filtered or unexported fields
}

type UserMessageDTO

type UserMessageDTO struct {
	ID   string `json:"id"`
	Text string `json:"text"`
	// contains filtered or unexported fields
}