Documentation
¶
Overview ¶
Package daemon implements zero's long-running daemon/server mode: a local control server that supervises a pool of headless `zero exec` worker processes and routes multiple agent sessions to them over an owner-only local socket.
It reuses zero's existing building blocks rather than inventing new ones:
- internal/background : child-process group setup + cross-platform terminate.
- internal/streamjson : the line-based agent event protocol on worker stdio.
- internal/cli (exec) : a worker is a `zero exec -i/-o stream-json` process.
The control plane (this file) is a small framed codec mirrored from the reference daemon's protocol.js: a 4-byte big-endian length prefix, a 1-byte frame kind, then a JSON control payload, with a 1 MiB frame cap and a version-negotiation handshake. The agent event stream itself stays stream-json.
Index ¶
- Constants
- Variables
- func BackoffWithJitter(attempt int) time.Duration
- func DefaultDir() (string, error)
- func NegotiateVersion(clientVersion int) (int, bool)
- func WriteControl(w io.Writer, msg Ctrl) error
- func WriteFrame(w io.Writer, kind FrameKind, payload []byte) error
- type Client
- type Ctrl
- type CtrlType
- type ExecLauncherConfig
- type FrameKind
- type Launcher
- type Lines
- type Paths
- type Pool
- type PoolOptions
- type Server
- type ServerOptions
- type Session
- type SessionManager
- func (m *SessionManager) Attach(id string) (buffered []string, live <-chan string, cancel func(), err error)
- func (m *SessionManager) Get(id string) (*Session, bool)
- func (m *SessionManager) Start(ctx context.Context, spec WorkerSpec) (*Session, error)
- func (m *SessionManager) Statuses() []SessionStatus
- type SessionManagerOptions
- type SessionState
- type SessionStatus
- type Sink
- type StatusFile
- type StatusReport
- type WorkerHandle
- type WorkerSpec
- type WorkerStatus
Constants ¶
const ( // ExitTempfail asks for a retry after a fixed cool-off (transient failure). ExitTempfail = 75 // ExitPermanent asks the daemon to STOP retrying this work (fatal). ExitPermanent = 76 )
Exit codes a worker may use to signal restart policy. Mirrors reference-daemon-code-agent-js/worker-manager.js EXIT_TEMPFAIL/EXIT_PERMANENT.
const ( // ProtoVersion is the control-protocol version advertised in the handshake. ProtoVersion = 1 // MaxFrameSize caps a single control frame payload at 1 MiB. A larger // advertised length is rejected before any allocation (fail closed). MaxFrameSize = 1 << 20 )
Wire constants. Mirrors reference-daemon-code-agent-js/protocol.js (FRAME_DATA/FRAME_CTRL, HEADER_SIZE, MAX_FRAME_SIZE, PROTO_VERSION).
Variables ¶
var ErrAlreadyRunning = errors.New("daemon: another instance is already running")
ErrAlreadyRunning is returned when a live daemon already holds the lock.
var ErrFrameTooLarge = errors.New("daemon: control frame exceeds size cap")
ErrFrameTooLarge is returned when a frame's advertised payload length exceeds MaxFrameSize. The reader rejects it without allocating the buffer.
var ErrPermanent = errors.New("daemon: worker failed permanently")
ErrPermanent is returned when a worker exits ExitPermanent or the attempt cap is exhausted; the request must not be retried.
var ErrPoolDraining = errors.New("daemon: pool is draining")
ErrPoolDraining is returned by Run once the pool is shutting down.
var ErrSessionExists = errors.New("daemon: session already exists")
ErrSessionExists is returned when starting a session ID already in use.
var ErrSessionNotFound = errors.New("daemon: session not found")
ErrSessionNotFound is returned by Get/Attach for an unknown session ID.
var ErrUnknownFrameKind = errors.New("daemon: unknown control frame kind")
ErrUnknownFrameKind is returned for a frame whose kind byte is not recognized.
Functions ¶
func BackoffWithJitter ¶
BackoffWithJitter returns backoffBase(attempt) scaled by a uniform random factor in [0.5, 1.5), matching worker-manager.js jitter(base) = base*(0.5 + random()). The jitter de-synchronizes simultaneous restarts so a fleet of workers does not thunder back at the same instant.
func DefaultDir ¶
DefaultDir returns the per-user directory for the daemon's runtime files: $XDG_RUNTIME_DIR/zero when set (a tmpfs owned by the user on Linux), otherwise ~/.zero. Mirrors supervisor.js / config.js choosing a per-user runtime dir.
func NegotiateVersion ¶
NegotiateVersion picks the protocol version both ends can speak: the lower of the client's advertised version and ProtoVersion. A non-positive client version is invalid and yields ok=false so the daemon can reject the connection.
func WriteControl ¶
WriteControl marshals a control message into a KindCtrl frame.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is a control-socket client used by the `zero daemon run|attach|status| stop` subcommands. It performs the version handshake on Dial.
func NewClientConn ¶
NewClientConn completes the control handshake on an already-connected transport (e.g. a TLS connection the remote bridge has authenticated) and returns a Client ready for Run/Attach/Status. It takes ownership of conn and closes it on handshake failure.
func (*Client) Attach ¶
Attach streams a running session's output (buffered history + live) to onLine.
func (*Client) Run ¶
Run starts/routes a session and streams its stream-json output lines to onLine until the session ends. It returns the session's terminal error, if any.
func (*Client) Status ¶
func (c *Client) Status() (*StatusReport, error)
Status requests the daemon/worker/session status report.
type Ctrl ¶
type Ctrl struct {
Type CtrlType `json:"type"`
Version int `json:"version,omitempty"`
Session string `json:"session,omitempty"`
Cwd string `json:"cwd,omitempty"`
Args []string `json:"args,omitempty"`
Prompt string `json:"prompt,omitempty"`
// Line is one stream-json event line (for CtrlData).
Line string `json:"line,omitempty"`
// Message is a human-readable detail (for CtrlError / CtrlAck).
Message string `json:"message,omitempty"`
// Status is populated on CtrlStatusResult.
Status *StatusReport `json:"status,omitempty"`
}
Ctrl is a control message carried in a KindCtrl frame as JSON. Unused fields are omitted so frames stay small.
type CtrlType ¶
type CtrlType string
CtrlType is the discriminator for a control message.
const ( // CtrlHello is the client's opening handshake; Version is the highest control // version it speaks. CtrlHello CtrlType = "hello" // CtrlHelloOK is the daemon's handshake reply with the negotiated Version. CtrlHelloOK CtrlType = "hello_ok" // CtrlRun asks the daemon to create/route a session and stream its output. CtrlRun CtrlType = "run" // CtrlAttach asks the daemon to attach the client to a running session's stream. CtrlAttach CtrlType = "attach" // CtrlStatus requests daemon/worker/session status. CtrlStatus CtrlType = "status" // CtrlStatusResult carries the status payload (see StatusReport in Status). CtrlStatusResult CtrlType = "status_result" // CtrlData carries one stream-json line of agent output back to the client. CtrlData CtrlType = "data" // CtrlAck acknowledges receipt/dispatch of a request (at-least-once dispatch). CtrlAck CtrlType = "ack" // CtrlEnd marks the end of a session's stream. CtrlEnd CtrlType = "end" // CtrlShutdown asks the daemon to drain and stop. CtrlShutdown CtrlType = "shutdown" // CtrlError carries a human-readable error for the client. CtrlError CtrlType = "error" )
type ExecLauncherConfig ¶
type ExecLauncherConfig struct {
// Executable is the zero binary to spawn (defaults to os.Executable()).
Executable string
// BaseArgs are prepended before the per-session args (defaults to the headless
// stream-json exec invocation `exec --output-format stream-json`).
BaseArgs []string
// Env is the base environment for workers (defaults to os.Environ()). The
// re-entrancy markers are always scrubbed from it.
Env []string
}
ExecLauncherConfig configures the production worker launcher.
type FrameKind ¶
type FrameKind uint8
FrameKind tags a frame's payload. Mirrors protocol.js FRAME_DATA/FRAME_CTRL.
func ReadFrame ¶
ReadFrame reads a single framed message. It validates the advertised length against MaxFrameSize BEFORE allocating, so an oversize or hostile frame is rejected (ErrFrameTooLarge) rather than triggering a huge allocation. An unrecognized kind byte yields ErrUnknownFrameKind after the payload is drained so the stream stays in sync.
type Launcher ¶
type Launcher func(ctx context.Context, spec WorkerSpec) (WorkerHandle, error)
Launcher spawns a worker for a single attempt. It must apply the sandbox/env policy (the production launcher scrubs the re-entrancy markers so the worker re-establishes its own sandbox — see newExecLauncher).
func NewExecLauncher ¶
func NewExecLauncher(cfg ExecLauncherConfig) (Launcher, error)
NewExecLauncher builds a Launcher that spawns `zero exec` workers which speak stream-json on stdout. Each worker:
- runs with the re-entrancy markers SCRUBBED from its env (so it establishes its own sandbox; the daemon never bypasses the sandbox), while the rest of the policy config/env is propagated;
- is placed in its own process group (background.ConfigureChildProcessGroup) so Kill / drain can terminate it and any children cleanly;
- has its per-session working directory (spec.Cwd) and flags (spec.Args) applied. The pool does the os/exec itself and uses internal/background only for process-group setup + cross-platform terminate.
type Lines ¶
Lines is a stream of stream-json output lines from a worker. Next returns the next line and io-EOF-style ok=false when the stream ends.
type Paths ¶
Paths bundles the daemon's socket, lock, and status file paths.
func DefaultPaths ¶
DefaultPaths returns the daemon control-socket, lock, and status file paths under DefaultDir.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool is a bounded, self-healing worker pool. A crashed worker never takes down the pool: Run retries on a fresh worker with backoff up to MaxAttempts, and an ExitPermanent worker stops retries. Drain stops new work and kills stragglers.
func NewPool ¶
func NewPool(opts PoolOptions) (*Pool, error)
NewPool builds a pool from opts, filling defaults.
func (*Pool) Drain ¶
func (p *Pool) Drain()
Drain stops accepting new work, gives in-flight workers a grace window (KillTimeout) to finish on their own, then force-kills any straggler. It returns as soon as the pool is idle (graceful) or the window elapses. Safe to call once; subsequent calls are no-ops.
func (*Pool) QueueDepth ¶
QueueDepth reports how many lease slots are currently in use.
func (*Pool) Run ¶
Run leases a worker slot and dispatches spec to a worker, streaming its stream-json lines to sink. It is at-least-once with bounded retries: a worker that crashes (non-zero, non-permanent) is retried on a fresh worker after a backoff; ExitPermanent or exhausting MaxAttempts returns ErrPermanent. Run queues when all slots are busy. The returned int is the final worker exit code.
func (*Pool) WorkerStats ¶
func (p *Pool) WorkerStats() []WorkerStatus
WorkerStats returns a snapshot of the currently-active (busy) workers, bounded by the pool size. The on-demand pool keeps no idle workers, so an empty result means the pool is idle.
type PoolOptions ¶
type PoolOptions struct {
// Size is the max number of concurrent workers (lease slots). Sessions beyond
// this queue until a slot frees.
Size int
// Launcher spawns a worker. Required.
Launcher Launcher
// MaxAttempts caps the total attempts per request (first + retries). When
// exhausted, Run returns ErrPermanent (mirrors capping consecutive restarts).
MaxAttempts int
// KillTimeout bounds graceful termination before SIGKILL-equivalent on drain.
KillTimeout time.Duration
// Backoff returns the delay before retry attempt n (1-based crash count).
// Defaults to BackoffWithJitter; tests inject a deterministic function.
Backoff func(attempt int) time.Duration
// TempfailDelay is the cool-off after an ExitTempfail exit. Defaults to 30s.
TempfailDelay time.Duration
// Log, when set, receives one line per lifecycle event.
Log func(string)
}
PoolOptions configures a Pool. Zero fields take documented defaults.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is the daemon control plane. Mirrors reference-daemon-code-agent-js/ supervisor.js (single-instance lock, status file, lifecycle) + the accept loop that routes framed control requests to the SessionManager/Pool. It listens on an owner-only local Unix socket and NEVER binds a TCP port.
func NewServer ¶
func NewServer(opts ServerOptions) (*Server, error)
NewServer validates options and builds a Server.
func (*Server) Serve ¶
Serve acquires the single-instance lock, binds the owner-only control socket, writes the status file, and serves connections until Shutdown. It blocks. On return it has released the lock and removed the socket/status files.
func (*Server) ServeConn ¶
ServeConn runs the control protocol (handshake + one command) on an already-established connection, reusing the exact local dispatch path. The remote bridge calls it AFTER authenticating a TLS connection, so a remote session is handled identically to a local one (same SessionManager/Pool, same sandbox/risk model) — remote never bypasses the local controls. It closes conn.
type ServerOptions ¶
type ServerOptions struct {
Paths Paths
Manager *SessionManager
Pool *Pool
Version int
Now func() time.Time
Log func(string)
// contains filtered or unexported fields
}
ServerOptions configures a Server.
type Session ¶
type Session struct {
// contains filtered or unexported fields
}
Session is one routed agent run. It buffers recent output (so a late `attach` sees history) and fans live lines out to subscribers. It implements the pool's Sink (Line) and the lease "started" hook (Started).
func (*Session) Done ¶
func (s *Session) Done() <-chan struct{}
Done is closed when the session finishes (success or failure).
func (*Session) Started ¶
func (s *Session) Started()
Started marks the session running once the pool leases a worker (lease hook).
func (*Session) State ¶
func (s *Session) State() SessionState
State returns the current lifecycle phase.
type SessionManager ¶
type SessionManager struct {
// contains filtered or unexported fields
}
SessionManager owns the live session registry and routes each session to a pool worker. Mirrors session-manager.js + roster.js. Finished sessions are retained (so a late `attach` sees history) up to MaxSessions, past which the oldest FINISHED ones are evicted; running/queued sessions are never evicted.
func NewSessionManager ¶
func NewSessionManager(opts SessionManagerOptions) (*SessionManager, error)
NewSessionManager builds a manager over pool.
func (*SessionManager) Attach ¶
func (m *SessionManager) Attach(id string) (buffered []string, live <-chan string, cancel func(), err error)
Attach returns the buffered history + live stream for an existing session.
func (*SessionManager) Get ¶
func (m *SessionManager) Get(id string) (*Session, bool)
Get returns a session by ID.
func (*SessionManager) Start ¶
func (m *SessionManager) Start(ctx context.Context, spec WorkerSpec) (*Session, error)
Start creates a session for spec and dispatches it to the pool. It returns the Session immediately (non-blocking); the run proceeds in the background and the session transitions queued -> running -> done/failed. A duplicate ID is rejected with ErrSessionExists.
func (*SessionManager) Statuses ¶
func (m *SessionManager) Statuses() []SessionStatus
Statuses returns a snapshot of all sessions, sorted by ID for stable output.
type SessionManagerOptions ¶
type SessionManagerOptions struct {
Pool *Pool
MaxBuffer int // per-session output ring size; 0 => default
// MaxSessions caps the retained session registry; 0 => default. The oldest
// FINISHED sessions are evicted once the cap is exceeded.
MaxSessions int
}
SessionManagerOptions configures a SessionManager.
type SessionState ¶
type SessionState string
SessionState is a session's lifecycle phase.
const ( // SessionQueued: created, waiting for a free pool slot. SessionQueued SessionState = "queued" // SessionRunning: a worker is leased and producing output. SessionRunning SessionState = "running" // SessionDone: the worker finished cleanly. SessionDone SessionState = "done" // SessionFailed: the worker failed permanently / the run errored. SessionFailed SessionState = "failed" )
type SessionStatus ¶
type SessionStatus struct {
ID string `json:"id"`
State string `json:"state"`
Lines int `json:"lines"`
}
SessionStatus is one session's routing/metrics snapshot.
type Sink ¶
type Sink interface {
Line(line string)
}
Sink receives a worker's stream-json output lines for one request.
type StatusFile ¶
type StatusFile struct {
PID int `json:"pid"`
Socket string `json:"socket"`
Version int `json:"version"`
StartedAt time.Time `json:"startedAt"`
}
StatusFile is the on-disk daemon status document (pid, socket, version, started-at), written next to the lock so `status`/`stop` work without a live connection. Mirrors supervisor.js's status file.
type StatusReport ¶
type StatusReport struct {
PID int `json:"pid"`
Version int `json:"version"`
Socket string `json:"socket"`
StartedAt time.Time `json:"startedAt"`
PoolSize int `json:"poolSize"`
Workers []WorkerStatus `json:"workers"`
Sessions []SessionStatus `json:"sessions"`
QueueDepth int `json:"queueDepth"`
}
StatusReport is the daemon/worker/session snapshot returned by `zero daemon status` (CtrlStatusResult). Mirrors reference-daemon-code-agent-js/status.js.
type WorkerHandle ¶
type WorkerHandle interface {
Stdout() Lines
// Wait blocks until the worker exits and returns its process exit code.
Wait() (int, error)
// Kill force-terminates the worker (used on drain timeout).
Kill() error
Pid() int
}
WorkerHandle is a launched worker process the pool supervises. The production launcher wraps an exec.Cmd running `zero exec -i/-o stream-json`; tests inject a fake. Stdout yields the worker's stream-json event lines.
type WorkerSpec ¶
type WorkerSpec struct {
Session string
Cwd string
// Args are the per-session exec flags (e.g. --prompt ...), parsed by the CLI
// and appended after `exec` by the production launcher.
Args []string
}
WorkerSpec describes the worker to launch for one session attempt.