workflow

Documentation

Overview

Package workflow is cc-fleet's deterministic orchestration runtime: it runs a JavaScript workflow script that fans out provider subagent leaves, in a cc-fleet process OFF the main Claude context. The script's plan lives in script variables (CPU, ~0 tokens); the model is invoked only at agent() leaves. The API mirrors the native Claude Code Workflow tool (const meta / agent / parallel / pipeline / phase / log / budget / args / workflow); the only shape difference is agent()'s required opts.provider.

Concurrency is a single-owner loop (see loop.go): one goroutine runs ALL script execution and engine-state mutation, builtins return Promises, and the blocking provider execs run on leaf goroutines — so the engine is -race clean while the slow leaves still overlap up to a bounded pool.

Index

• Variables
• func Execute(ctx context.Context, scriptPath, runID string, opts Options) (err error)
• func Launch(ctx context.Context, scriptPath string, opts Options, foreground bool) (string, error)
• func Prepare(scriptPath string) (subagent.WorkflowRun, error)
• func RenderEventLine(ev EventRecord) string
• func RenderWaitLine(res WaitResult) string
• func Restart(ctx context.Context, runID, journalKey string) error
• func RestartPhase(ctx context.Context, runID, phase string) ([]string, error)
• func SendLeafCommand(runID, op, leafID string) error
• func SendPhaseCommand(runID, op, phase string) error
• func WaitEngineStarted(runID string, childPID int) bool
• func Watch(ctx context.Context, runID string, out io.Writer, opts WatchOptions) (subagent.WorkflowRun, error)
• type EventRecord
• type EventTail
• type Options
• type WaitCounts
• type WaitHeldLeaf
• type WaitOptions
• type WaitOutcome
• type WaitResult
  • func Wait(ctx context.Context, runID string, opts WaitOptions) (WaitResult, error)
• type WatchOptions

Variables

var ErrEngineGone = errors.New("workflow: detached engine is gone but the run never finalized")

ErrEngineGone is returned by Watch when the manifest is still "running" but the run's DETACHED engine process is gone (crashed / killed) without finalizing — so waiting for a terminal status would block forever. The watcher surfaces it and stops.

Functions

func Execute

func Execute(ctx context.Context, scriptPath, runID string, opts Options) (err error)

Execute runs a prepared script's body to completion in the CURRENT process, tagging every leaf with runID, and flips the manifest to done/failed — or "stopped" when the run ctx was cancelled (a cooperative stop is not a failure) — on exit. It NEVER lets a panic escape: a panic is recovered into a failed status so a detached run always finalizes.

func Launch

func Launch(ctx context.Context, scriptPath string, opts Options, foreground bool) (string, error)

func Prepare

func Prepare(scriptPath string) (subagent.WorkflowRun, error)

Prepare parses a script, extracts + validates its `const meta` literal, and mints a run manifest with the name/description/declared phases — BEFORE any execution, so a bad script never mints a half-run and the board shows the named, phase-skeletoned run immediately. The normalize step full-parses the wrapped body, so any syntax error fails here with NO manifest left behind. Returns the new manifest (its RunID is handed to a detached child or printed to the caller).

func RenderEventLine

func RenderEventLine(ev EventRecord) string

RenderEventLine formats one event for the live stream. Every opaque field (status / label / provider / model / phase / msg / group type) is CleanTitle-scrubbed before it reaches the terminal; the event carries no key or answer by construction (events.go). It is the renderer for the `cc-fleet workflow watch` live status stream.

func RenderWaitLine

func RenderWaitLine(res WaitResult) string

RenderWaitLine formats Wait's single human exit line: every opaque string is CleanTitle-scrubbed (terminal-injection defense) and it prints Status only — never the raw WorkflowRun.Error, which a schema-reject can taint (parity with finalLine).

func Restart

func Restart(ctx context.Context, runID, journalKey string) error

Restart re-runs a workflow run, optionally scoped to a single leaf. With journalKey set it drops that leaf's cached result so the resume re-runs ONLY it — plus any downstream leaf whose input embedded the old answer (content-addressing recomputes those keys → cache miss); every other leaf replays from the journal instantly. With an empty journalKey it is a whole- run resume (re-runs only the un-journaled / failed leaves). cc-fleet runs ONE engine per run, so before touching the journal a still-"running" run is resolved: a verifiably-live detached engine is stopped + confirmed dead (its in-flight leaves then re-run on resume); a crashed/killed detached run (recorded pid now dead) is resumed as-is; a foreground run with no killable engine fails closed. The resume replays the run's original launch options (args / persistIO / budget) off the manifest so a leaf's key — and thus its cache validity — doesn't shift. The whole decision runs under the per-run execution lock so a concurrent restart / resume / stop never acts on stale state or races the pre-launch pid window; the lock releases when Launch returns (after the child registers) and is NEVER held around the engine's Execute. StopRun/Launch internals stay lock-free.

func RestartPhase

func RestartPhase(ctx context.Context, runID, phase string) ([]string, error)

RestartPhase re-runs a TERMINAL run's phase: under the per-run lock + stop barrier it collects the phase's journal-key SET from the member jobs, whole-key drops the set, and resumes (un-journaled members — failed/stopped leaves — re-run regardless). Returns the OTHER phase titles the restart widens into: identical agent() calls share one content key, so a key whose jobs span more than one phase re-runs everywhere it appears — meta-derived per-phase counts would over-remove, the whole-key drop is the honest scope and the caller names it.

func SendLeafCommand

func SendLeafCommand(runID, op, leafID string) error

SendLeafCommand appends a leaf directive to a LIVE run's control file. It validates both ids (they become path components / file content) and requires the manifest to be `running` — a dead run's control file has no poller, so the append would silently rot.

func SendPhaseCommand

func SendPhaseCommand(runID, op, phase string) error

SendPhaseCommand appends a phase directive to a LIVE run's control file: the leaf atom fanned out over every member of the EXACT title (a reused title is one merged target; "" addresses unphased leaves), with future members of a held phase parking before their first exec.

func WaitEngineStarted

func WaitEngineStarted(runID string, childPID int) bool

WaitEngineStarted blocks until the detached child has self-stamped EnginePID == childPID (its first manifest write in Execute's pre-run saveManifest) or the startup budget elapses. The launcher holds the per-run execution lock across this wait, so a serialized second restart/stop/resume always observes a fully-registered engine — never the pre-stamp EnginePID==0 window. Returns false on timeout; the caller must then kill+reap the child before failing the run.

func Watch

func Watch(ctx context.Context, runID string, out io.Writer, opts WatchOptions) (subagent.WorkflowRun, error)

Watch streams a run's live events to out, one scrubbed line per event, and blocks until the run reaches a terminal status (then a final status line + return nil), the detached engine dies without finalizing (ErrEngineGone), or ctx is cancelled/timed-out (a "still running" line + ctx.Err()). It is INSPECTION-ONLY: no spawn/stop/teardown — the only writes are the dead-job result memoization any status read (the board, `workflow status`) already performs.

Key-safety rests on the field SOURCE, not on scrubbing: the events stream carries no provider KEY by construction (a key never enters the engine's string space — it flows only via the apiKeyHelper), and the final line prints only Status, never the raw WorkflowRun.Error (which a schema-reject can taint with a provider reply fragment). Like the board and `workflow status`, the stream DOES surface script-authored text (phase / label / log), so a script that deliberately logs a provider reply will show it — that is the author's choice, not an autonomous leak. CleanTitle is layered on every opaque string only as a terminal-injection defense.

Types

type EventRecord

type EventRecord struct {
	Seq      int64  `json:"seq"`
	Kind     string `json:"kind"`             // phase | log | leaf | group-open | group-close
	Status   string `json:"status,omitempty"` // leaf: launch | done | failed | cached | held | stopped
	Phase    string `json:"phase,omitempty"`
	Label    string `json:"label,omitempty"`
	Provider string `json:"provider,omitempty"`
	Model    string `json:"model,omitempty"`
	// Group fields: a parallel/pipeline/workflow group's id and (on group-open) its kind.
	// `workflow watch` brackets the group by seq order (open…close), so no explicit parent
	// id is needed. Empty for plain leaves.
	GroupID string `json:"group_id,omitempty"`
	GroupTy string `json:"group_type,omitempty"`
	Msg     string `json:"msg,omitempty"` // phase title / log narrator line
}

EventRecord is one line of a run's live-event channel (runs/<id>.events). It is PURE OBSERVABILITY: `cc-fleet workflow watch` tails it for a scrubbed live status stream — the engine NEVER reads it back and it NEVER feeds journalKey, so it cannot perturb resume determinism (the load-bearing rule). It is key-safe BY CONSTRUCTION: there is no prompt or answer field, so a provider reply (and the never-present provider key) cannot be written here; a leaf's prompt/answer live in their own 0600 io files. Msg is author-supplied script text (phase title / log line), never provider output.

type EventTail

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

EventTail is the incremental-tail state for a run's events file: the byte offset already consumed, a torn trailing partial line carried across reads, the identity of the generation being tailed, the seq of the last consumed event (contiguity check + reattach hint), and a one-time skip threshold (SinceSeq, for a clean reattach). Watch holds a long-lived *EventTail across its loop and calls read on it.

type Options

type Options struct {
	// RunID, when set, names an EXISTING manifest to execute (the detached child /
	// foreground re-exec path); empty means Prepare mints a fresh one.
	RunID       string
	Concurrency int    // 0 → defaultConcurrency()
	ArgsJSON    string // optional; the script's `args` value
	// Resume names an EXISTING run to re-execute against its journal (cross-invocation
	// replay): Launch reuses this id instead of minting a fresh one, and the engine's
	// unconditional journal load then serves the leaves that already completed. Empty =
	// a fresh run. Used only at Launch — the detached child carries the id via RunID and
	// resumes transparently (the journal load keys on the id, not this flag).
	Resume string
	// NoPersistIO opts OUT of the board's inline prompt/answer detail (persistence is DEFAULT ON).
	// When set, leaves persist no <jobID>.prompt/.answer side files; the result cache is
	// answer-stripped either way, so the board table is unaffected. Propagated to the
	// detached child so the whole run honors it.
	NoPersistIO bool
	// BudgetUSD caps total provider spend in USD (a sum of leaf CostUSD — an Anthropic list-price
	// estimate, not the third-party provider's actual charge); agent() raises once the cap would be
	// breached. <=0 is uncapped; a -1 sentinel from --no-budget is normalized to 0 (explicit uncap)
	// in Launch before any persist. Propagated to the detached child.
	BudgetUSD float64
	// BudgetTokens caps total provider token spend (Usage.InputTokens+OutputTokens summed across
	// leaves, cache-read excluded — the exact provider-neutral ceiling); <=0 uncapped, -1 normalized
	// to 0 like BudgetUSD. The first cap to trip aborts. Propagated to the detached child.
	BudgetTokens int64
	// LeadSessionID is the parent Claude session this run was launched from (detected at
	// `workflow run`). Stored on the manifest at mint so the board groups runs by session.
	// Used only by Launch's fresh-mint path; a resume reads it back off the manifest.
	LeadSessionID string
}

Options configures a workflow run.

type WaitCounts

type WaitCounts struct {
	Running int `json:"running,omitempty"`
	Queued  int `json:"queued,omitempty"`
	Held    int `json:"held,omitempty"`
	Done    int `json:"done,omitempty"`
	Failed  int `json:"failed,omitempty"`
	Stopped int `json:"stopped,omitempty"`
}

WaitCounts tallies a run's leaves by status at the moment Wait returned.

type WaitHeldLeaf

type WaitHeldLeaf struct {
	JobID string `json:"job_id"`
	Label string `json:"label,omitempty"`
}

WaitHeldLeaf identifies one held leaf in a parked snapshot, so the caller can name what needs an operator without a second status call.

type WaitOptions

type WaitOptions struct {
	// Interval is the poll cadence; <=0 uses waitDefaultInterval.
	Interval time.Duration
}

WaitOptions configures a workflow-run wait.

type WaitOutcome

type WaitOutcome string

WaitOutcome classifies why Wait returned.

const (
	// WaitTerminal — the run reached done / failed / stopped.
	WaitTerminal WaitOutcome = "terminal"
	// WaitEngineGone — the manifest still says running but the detached engine is dead.
	WaitEngineGone WaitOutcome = "engine_gone"
	// WaitParked — every remaining leaf is held: the run cannot progress until an
	// operator restarts one (`workflow restart --leaf/--phase`) and waits indefinitely.
	WaitParked WaitOutcome = "parked"
	// WaitTimeout — ctx ended while the run was still progressing; the snapshot is a
	// heartbeat, not a final state.
	WaitTimeout WaitOutcome = "timeout"
)

type WaitResult

type WaitResult struct {
	Run     subagent.WorkflowRun
	Outcome WaitOutcome
	Counts  WaitCounts
	Held    []WaitHeldLeaf
}

WaitResult is Wait's exit snapshot: the manifest as last read, why Wait returned, the leaf tally, and (parked) the held leaves.

func Wait

func Wait(ctx context.Context, runID string, opts WaitOptions) (WaitResult, error)

Wait blocks until a run needs its caller's attention, polling the manifest + tagged jobs (never the events stream — every exit condition is a manifest/jobs fact). It is the quiet sibling of Watch, built to run in a backgrounded shell whose process exit IS the completion signal. INSPECTION-ONLY: the only writes are the dead-job result memoization any status read already performs.

Returns nil with Outcome terminal or parked; ErrEngineGone with Outcome engine_gone; ctx.Err() with Outcome timeout (the snapshot is then a heartbeat, not a final state).

Parked = 0 running ∧ 0 queued ∧ ≥1 held, stable for waitParkedTicks consecutive polls. The engine has no timers — leaf completion is its only async event source, and a pending script with zero in-flight leaves fails the run outright — so a running manifest whose only live leaves are held cannot progress until an operator restarts one. The check order (terminal → engine-gone → parked) needs no engine-alive conjunct on parked, so it fires for foreground runs (EnginePID 0) too. One window survives the debounce: a long synchronous JS continuation after a leaf settles, before the next agent() mints a job — the caller treats parked as "re-check, then act", not as a verdict.

type WatchOptions

type WatchOptions struct {
	// Interval is the poll cadence; <=0 uses watchDefaultInterval.
	Interval time.Duration
	// SinceSeq skips events whose Seq <= SinceSeq, for a clean reattach (the agent passes the
	// last seq from a prior "still running (seq=N)" line so a re-run doesn't replay history).
	SinceSeq int64
}

WatchOptions configures a workflow-run watch.