tasks

Documentation

Overview

Package tasks runs work in the background for an agent session: start a command (or a function: a watch loop, a hook, later a sub-agent) and get an id at once, read its output while it runs, wait for it with a timeout, kill it. Output goes line by line (cleaned by the caller, e.g. decoded and redacted) to a file the host names per session. A finished task leaves a notice the host hands to the model before its next request (TakeNotices), unless the model already saw the end through Wait or Kill; a running function task may leave notices too (Reporter.Notify). A host that took notices for a turn the model never got them in hands them back with ReturnNotices. The whole process tree of every command task is killed by Kill, by the limits and by Close; a function task's context is cancelled. Nothing outlives the manager.

Index

• Variables
• func Describe(i Info) string
• func GrepLines(path, pattern string, limit int) ([]string, int, error)
• func ReadLines(path string, offset, limit int) (lines []string, total int, more bool, err error)
• func TailText(i Info, n int) string
• type Config
• type Info
  • func (i Info) Duration() time.Duration
• type Limits
• type Manager
  • func NewManager(cfg Config) *Manager
  • func (m *Manager) Adopt(ctx context.Context, spec Spec, running RunningCommand) (Info, error)
  • func (m *Manager) Close()
  • func (m *Manager) Get(sessionID, id string) (Info, error)
  • func (m *Manager) HasNotices(sessionID string) bool
  • func (m *Manager) Kill(sessionID, id string) (Info, error)
  • func (m *Manager) KillSession(sessionID string) []Info
  • func (m *Manager) Limits() Limits
  • func (m *Manager) List(sessionID string) []Info
  • func (m *Manager) ReturnNotices(sessionID string, infos []Info)
  • func (m *Manager) Running(sessionID string) []Info
  • func (m *Manager) SessionOf(ctx context.Context) string
  • func (m *Manager) SetOnEnd(fn func(Info))
  • func (m *Manager) Start(ctx context.Context, spec Spec) (Info, error)
  • func (m *Manager) TakeNotices(sessionID string) []Info
  • func (m *Manager) Wait(ctx context.Context, sessionID, id string, timeout time.Duration) (Info, WaitOutcome, error)
• type ProcessTree
• type Reporter
• type RunFunc
• type RunningCommand
• type Spec
• type Status
• type WaitOutcome

Variables

var DefaultLimits = Limits{MaxRunning: 4, MaxDuration: 2 * time.Hour, MaxOutputBytes: 64 << 20}

DefaultLimits: 4 running per session, 2 hours each, 64 MiB of output.

var ErrUnknownTask = errors.New("unknown task")

ErrUnknownTask is returned for an id the session does not have.

Functions

func Describe

func Describe(i Info) string

Describe is a one-line account of a task, shared by the task tool, the end notices and the trace: "t3 exited (exit 0) after 4m12s · 1.2 MB / 3400 lines · go test ./... · out:sess:ab12"; a function task adds its error and progress ("t4 running for 3m · run 5, exit 0 · ...").

func GrepLines

func GrepLines(path, pattern string, limit int) ([]string, int, error)

GrepLines returns the lines (numbered "n: text") of a task's output that match pattern (RE2), at most limit, and how many matched in all.

func ReadLines

func ReadLines(path string, offset, limit int) (lines []string, total int, more bool, err error)

ReadLines returns up to limit lines of a task's output from line offset (1-based; values below 1 read from the start), the total number of lines written so far and whether more follow. It works while the task runs.

func TailText

func TailText(i Info, n int) string

TailText is the last n lines of the task's tail, one per line.

Types

type Config

type Config struct {
	Limits Limits
	// Output names the output file of a task and its ref (a handle the
	// model can pass to other tools, e.g. "out:<session>:<hex>"). Required.
	Output func(sessionID, taskID string) (path, ref string, err error)
	// Session returns the session a call belongs to; nil uses
	// api.SessionIDFrom.
	Session func(ctx context.Context) string
	// OnNotice, when set, is called for every notice left pending: a task
	// that ended unseen, or a running task's Reporter.Notify (e.g. to wake an
	// idle chat). It runs in the task's goroutine and must not block.
	OnNotice func(Info)
}

Config sets up a Manager.

type Info

type Info struct {
	ID        string    `json:"id"`
	SessionID string    `json:"session_id"`
	Kind      string    `json:"kind"`
	Summary   string    `json:"summary"`
	Status    Status    `json:"status"`
	ExitCode  int       `json:"exit_code"`
	Started   time.Time `json:"started"`
	Ended     time.Time `json:"ended,omitzero"`
	// Bytes and Lines count what was written to the output file.
	Bytes     int64  `json:"bytes"`
	Lines     int    `json:"lines"`
	Truncated bool   `json:"truncated,omitempty"` // output past MaxOutputBytes was dropped
	Path      string `json:"path"`
	Ref       string `json:"ref"`
	// Tail is the last lines of the output (up to tailLines / tailBytes).
	Tail []string `json:"tail,omitempty"`
	// Error is why a function task failed.
	Error string `json:"error,omitempty"`
	// Progress and Counters are what a function task reported (Reporter).
	Progress string            `json:"progress,omitempty"`
	Counters map[string]int    `json:"counters,omitempty"`
	Labels   map[string]string `json:"labels,omitempty"`
	// Notice is set on a notice of a running task (TakeNotices): its text.
	// Empty on an end notice.
	Notice string `json:"notice,omitempty"`
}

Info is a snapshot of a task.

func (Info) Duration

func (i Info) Duration() time.Duration

Duration is how long the task ran (so far, while running).

type Limits

type Limits struct {
	// MaxRunning is how many tasks of one session may run at once.
	MaxRunning int
	// MaxDuration ends a task that runs longer.
	MaxDuration time.Duration
	// MaxOutputBytes stops writing a task's output file past this size.
	MaxOutputBytes int64
}

Limits bound the tasks of a manager.

type Manager

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

Manager owns the background tasks of one host process.

func NewManager

func NewManager(cfg Config) *Manager

NewManager returns a Manager; cfg.Output is required.

func (*Manager) Adopt

func (m *Manager) Adopt(ctx context.Context, spec Spec, running RunningCommand) (Info, error)

Adopt takes ownership of a command that is already running. It is used when a foreground shell reaches its handoff threshold; the caller's process tree and output stream are transferred before the manager starts waiting for it.

func (*Manager) Close

func (m *Manager) Close()

Close kills every running task and refuses new ones.

func (*Manager) Get

func (m *Manager) Get(sessionID, id string) (Info, error)

Get returns one task's snapshot.

func (*Manager) HasNotices

func (m *Manager) HasNotices(sessionID string) bool

HasNotices reports whether the session has pending notices.

func (*Manager) Kill

func (m *Manager) Kill(sessionID, id string) (Info, error)

Kill ends the task's process tree and waits (briefly) for it to be reaped.

func (*Manager) KillSession

func (m *Manager) KillSession(sessionID string) []Info

KillSession kills the session's running tasks and returns them.

func (*Manager) Limits

func (m *Manager) Limits() Limits

Limits returns the effective limits.

func (*Manager) List

func (m *Manager) List(sessionID string) []Info

List returns the session's tasks, in start order.

func (*Manager) ReturnNotices

func (m *Manager) ReturnNotices(sessionID string, infos []Info)

ReturnNotices puts notices back into the session's pending queue (see TakeNotices): a host that took them for a turn which then did not deliver them to the model - a turn that ends cancelled or failed, whose entries only replay up to its last complete tool exchange - hands them back here instead of losing them. They keep their original order and go before the notices that arrived in the meantime, so the next TakeNotices returns each of them once more. A notice of a task that left a newer notice meanwhile (a running task reporting again, or ending) is dropped: the same "a newer notice replaces the older one" rule Reporter.Notify applies.

func (*Manager) Running

func (m *Manager) Running(sessionID string) []Info

Running returns the session's running tasks.

func (*Manager) SessionOf

func (m *Manager) SessionOf(ctx context.Context) string

SessionOf returns the session of ctx (Config.Session).

func (*Manager) SetOnEnd

func (m *Manager) SetOnEnd(fn func(Info))

SetOnEnd sets a function told (in the task's goroutine) of every task that ends, seen or not, e.g. to record it in the trace; the latest call wins.

func (*Manager) Start

func (m *Manager) Start(ctx context.Context, spec Spec) (Info, error)

Start runs spec in the background for the session of ctx and returns its snapshot at once.

func (*Manager) TakeNotices

func (m *Manager) TakeNotices(sessionID string) []Info

TakeNotices returns the session's ended tasks whose notice is pending and clears them.

func (*Manager) Wait

func (m *Manager) Wait(ctx context.Context, sessionID, id string, timeout time.Duration) (Info, WaitOutcome, error)

Wait waits up to timeout for the task to end or, for a function task, to report progress (Reporter.Progress): a watch never ends by itself, so waiting for it means waiting for its next run. What the caller saw is not noticed again.

type ProcessTree

type ProcessTree interface {
	Kill() error
	Close()
	// Started assigns a start suspended process to the tree (Windows) and
	// resumes it; it returns an error if the process could not join, and the
	// caller then ends it and reports a failed command.
	Started() error
	// Exceeded reports whether a resource limit ended a process of the tree.
	Exceeded() bool
}

ProcessTree owns a command's process group or Windows job object. It is passed when a foreground command is adopted so ownership of the already started process tree moves to the manager without opening a kill window.

type Reporter

type Reporter interface {
	// Notify leaves a notice for the model now, while the task keeps running
	// (e.g. a watch whose result changed). A newer notice of the same task
	// replaces one the host has not taken yet.
	Notify(text string)
	// Progress sets a short state shown with the task (e.g. "run 5, exit 0").
	Progress(text string)
	// Count sets a named counter reported with the task (Info.Counters).
	Count(name string, n int)
	// Label sets a named text reported with the task (Info.Labels), e.g. the
	// session a sub-agent ran in. Unlike Progress it returns no Wait.
	Label(name, value string)
}

Reporter is what a function task (Spec.Run) tells the manager while it runs. Progress also returns a Wait on the task (WaitUpdated), taking a notice left before it: report a step as Count, Notify, then Progress.

type RunFunc

type RunFunc func(ctx context.Context, out io.Writer, r Reporter) (exitCode int, err error)

RunFunc is a function task: it writes its output to out (line by line, cleaned like a command's) and returns its exit code. ctx is cancelled by Kill, by Limits.MaxDuration and by Close; a non-nil error marks it failed.

type RunningCommand

type RunningCommand struct {
	Cmd          *exec.Cmd
	Tree         ProcessTree
	AttachOutput func(io.Writer)
	Wait         <-chan error
	Started      time.Time
	// MaxDuration is measured from Started; zero uses the manager limit.
	MaxDuration time.Duration
}

RunningCommand is a command that has already been started by its caller. AttachOutput must atomically switch the command's output to the supplied writer, including bytes written before the handoff.

type Spec

type Spec struct {
	Kind    string // "shell", "watch", "hook"
	Summary string // what it runs, shown in lists and notices
	// Cmd is ready to Start (path, args, dir, env); its Stdout, Stderr and
	// SysProcAttr are set by the manager.
	Cmd *exec.Cmd
	// Limits caps the memory, process count and CPU time of Cmd and its whole
	// process tree (tools.shell_limits); the zero value is no limit.
	Limits proctree.Limits
	// LimitNotice, when set, is asked for the line to write to the task output
	// when the command ended after a limit: it gets whether the system ended a
	// process of the tree, the exit code and the last lines of the output (what a
	// refused allocation printed). An empty answer writes nothing.
	LimitNotice func(terminated bool, exitCode int, output string) string
	// Run is a function task, instead of Cmd.
	Run RunFunc
	// NoEndNotice: the task reports what matters through Reporter.Notify,
	// so its end leaves no notice (a Stop hook run).
	NoEndNotice bool
	// Clean turns raw output into text for the file and the tail, one line
	// at a time (no trailing newline); nil keeps the bytes as they are.
	Clean func(line []byte) string
}

Spec is one task to start: a command (Cmd) or a function (Run).

type Status

type Status string

Status is where a task is in its life.

const (
	Running  Status = "running"
	Exited   Status = "exited"    // the command ended by itself (any exit code)
	Killed   Status = "killed"    // Kill or Close ended it
	TimedOut Status = "timed_out" // it ran past Limits.MaxDuration
	Failed   Status = "failed"    // it could not be waited for (rare)
)

type WaitOutcome

type WaitOutcome int

WaitOutcome is how a Wait returned.

const (
	WaitTimedOut WaitOutcome = iota // still running after the timeout
	WaitEnded                       // the task ended; its notices are taken (Info.Notice: its last report)
	// WaitUpdated: a running function task reported progress (a watch
	// finished a run); a notice it left is taken into Info.Notice.
	WaitUpdated
)