Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,32 @@ interleaved with harness internals.
side land with the rest of #732). `pipeline.TerminalStatus.IsPaused()` covers it and
`paused_billing`; audit/list put both in the `paused` status class, and
`RunManager` reports both as `RunPaused`.
- **Pause at a human gate (library seam, #732).** `Config.GateMode: "pause"`
makes every human gate pause the run in `paused_gate` instead of blocking:
`handlers.PausingInterviewer` records exactly what the gate asked
(`pipeline.PausedGate` — prompt, choices, default, interview questions, a
fingerprint, a nonce) on the checkpoint in the secure state dir, mirrors it
to the run dir's `gate.json`, and `Result.PausedGate` names the gate.
`tracker.Answer` validates an answer (`tracker.ValidateAnswer`) against the
checkpoint record — never `gate.json` — and resumes the run holding the answer
in memory; `tracker.WaitingRuns` lists waiting runs and `tracker.PendingGates`
reads one. A gate-paused resume keeps its context (no resume compaction) and
its budget spans every answer. Parallel branch gates are all recorded (lowest
branch first; they are answered together — `AnswerInput.Gates` — because
answering re-runs the parallel node); a gate inside a subgraph or
`manager_loop` child fails closed in pause mode. Every resume of an existing
checkpoint (`tracker -r`, `Config.ResumeRunID`, `tracker.Answer`) now holds a
per-run OS file lock in the secure state dir, released when the process
exits however it exits, so a second resume of the same run fails with
`pipeline.ErrRunLocked`. `gate_paused` names the waiting gate, its scope,
how many wait, and the `tracker answer` command with the run's `-w` /
`--artifact-dir` (`gate_node`, `gate_scope_path`, `gate_waiting`,
`gate_answer_command`). Only the secure checkpoint counts: a checkpoint copy
in the run dir is never listed or answered. Gate events gain
`resolution` / `fingerprint` (activity log `gate_resolution` /
`gate_fingerprint`, NDJSON likewise), and `GateQuestion` gains `context`.
`handlers.GateInfoFrom(ctx)` gives an interviewer the gate's identity, full
prompt and scope on every `Ask*Context` call.

### Fixed

Expand Down
9 changes: 5 additions & 4 deletions docs/api-stability.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,11 @@ unrecognized value fail-closed (as "not a success/state I understand" — not as
error to report).

- **`Result.Status`** — `success`, `fail`, `budget_exceeded`,
`validation_overridden`, `paused_billing` today; more may be added. Classify
with `pipeline.TerminalStatus(r.Status).IsSuccess()`. Note `paused_billing` is
*recoverable* (resume via `Config.ResumeRunID`), and `Run` returns both the
result and the billing error — `err != nil` alone is not "dead".
`validation_overridden`, `paused_billing`, `paused_gate` today; more may be
added. Classify with `pipeline.TerminalStatus(r.Status).IsSuccess()` and
`.IsPaused()`. Both pauses are *recoverable* — `paused_billing` resumes via
`Config.ResumeRunID`, `paused_gate` via `tracker.Answer` — and `Run` returns
both the result and an error, so `err != nil` alone is not "dead".
- **`RunManager` `RunState`** — `RunStarting`, `RunRunning`, `RunSucceeded`,
`RunFailed`, `RunPaused`, `RunCanceled` today. Gate on `RunState.Terminal()`
rather than enumerating states.
Expand Down
20 changes: 19 additions & 1 deletion docs/architecture/embedding.md
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ Semantics:
## 4. Terminal status & events

- **Status is an open enum.** `Result.Status` is one of `success`, `fail`,
`budget_exceeded`, `validation_overridden`, `paused_billing` today; future
`budget_exceeded`, `validation_overridden`, `paused_billing`, `paused_gate` today; future
minors may add more. **Never switch exhaustively on the raw string** — classify
with `pipeline.TerminalStatus(r.Status).IsSuccess()` (fail-closed) and treat an
unknown value as "not a success I understand", not as a failure to report.
Expand All @@ -143,6 +143,24 @@ Semantics:
on `Result.Status` — `err != nil` alone is not "dead". Under `RunManager` this
surfaces as `RunPaused` with `ManagedRun.ResumeRunID()` (§ Concurrency in
[`transport-boundary.md`](transport-boundary.md)).
- **`paused_gate` waits for an answer.** With `Config.GateMode: "pause"` (needs
`Config.ArtifactDir`) a human gate stops the run in the resumable
`paused_gate` terminal instead of blocking; `Result.PausedGate` names the gate
and `Run` returns the result with a `*pipeline.PauseError`. To answer, call
`tracker.Answer(ctx, source, cfg, tracker.AnswerInput{RunID, Label | Text |
Default | InterviewAnswers | Cancel})` with the run's own source and
`Config`: it reads the gate from the secure checkpoint (`tracker.PendingGates`),
validates the answer (`tracker.ValidateAnswer` — case-insensitive labels, an
error listing the valid ones), takes the run lock, and resumes the run with
the answer held in memory. When a parallel node's branches wait at several
gates, answer them all at once with `AnswerInput.Gates` (keyed by gate node
ID or `<scope>/<node>`): the resume re-runs the parallel node, so an
unanswered sibling gate would pause it again. `tracker.WaitingRuns(artifactDir)`
lists runs waiting at a gate. Every resume of an existing checkpoint takes the
same OS file lock, so a second resume of a run in progress fails with
`pipeline.ErrRunLocked`. A
gate-paused resume keeps its context as-is (no resume compaction) and its
budget runs across answers. See [`handlers/human.md`](handlers/human.md#pause-mode).
- **Events:** `Config.EventHandler pipeline.PipelineEventHandler` receives the
`pipeline.PipelineEvent` stream (types in `pipeline/events.go`);
`Config.AgentEvents agent.EventHandler` receives per-session `agent.Event`s.
Expand Down
24 changes: 24 additions & 0 deletions docs/architecture/engine.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ is in this file tree.
[declared inputs binding](#declared-inputs-binding-553-555-556),
[checkpoint semantics](#checkpoint-semantics) (restart budgets, fail-routing
provenance, resume rewind)
— [pausing](#pausing-paused_billing-paused_gate)
3. [Run-state integrity: activity log and checkpoint](#run-state-integrity-activity-log-and-checkpoint-213-559)
4. [Outcomes and routing](#outcomes-and-routing) —
[strict failure edges and the failure cascade](#strict-failure-edges-and-the-failure-cascade)
Expand Down Expand Up @@ -445,6 +446,29 @@ route at some point.
(#697). See `pipeline/fidelity.go` and
[`context-flow.md`](./context-flow.md).

### Pausing (`paused_billing`, `paused_gate`)

A handler returns a `*pipeline.PauseError` to stop the run in a resumable
terminal instead of failing: `paused_billing` (credit/quota exhausted, #487) or
`paused_gate` (a human gate in pause mode). The engine saves the checkpoint with
the paused node **not** completed, preserves WIP, and emits the status's own
terminal event — `billing_paused` or `gate_paused` — with no `stage_failed` (the
trace entry reads `paused`). `Engine.Run` returns the result and the
`PauseError` itself, which unwraps to the underlying error; so a pause inside a
subgraph or `manager_loop` child pauses the parent at that node.
`TerminalStatus.IsPaused()` covers both.

For a gate pause the checkpoint also carries `PausedGates` (the authoritative
waiting-gate records, cleared when the paused node completes) and `PriorUsage`
(the run's cumulative usage). Resuming a run with `PausedGates` set skips resume
compaction — the gate must re-render the question it recorded — and seeds the
budget guard's baseline from `PriorUsage`, so cost and token caps bound the
whole run across answers. Wall time restarts per segment.

Each resume appends a segment to the activity log; audit and `ListRuns` take a
run's status from its latest segment (entries after the last
`pipeline_started`), while overrides count from every segment.

## Run-state integrity: activity log and checkpoint (#213, #559)

The operational contract (path resolution, sentinel, absolute-env rule,
Expand Down
42 changes: 42 additions & 0 deletions docs/architecture/handlers/human.md
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,48 @@ autopilot (LLM judge) and webhook (human over a transport).
This is deterministic — no LLM call. It's the fastest way to smoke-test a
pipeline end-to-end.

## Pause mode

`handlers.PausingInterviewer` (CLI: no terminal on stdin, `--json`, or
`--gates pause`; library: `Config.GateMode: "pause"`) answers no gate itself.

- **At a gate with no answer** it returns `pipeline.NewPausedGateError(record)`:
a `PauseError` with status `paused_gate` carrying a `pipeline.PausedGate` — the
gate's node, `ScopePath` (from `GateInfoFrom(ctx)`), mode, label, full prompt,
the exact `choices` / `default` the handler passed, interview questions, a
fingerprint (`pipeline.GateFingerprint`, over what was asked) and a random
nonce. The record's mode is the call kind, so an interview that parsed zero
questions records `freeform`. The handler emits `gate_resolved` with
`resolution: "paused"` and no error, and the engine stores the record on
`Checkpoint.PausedGates`, writes `<run dir>/gate.json` (0600, never read back)
and ends the run with `gate_paused`. No `stage_failed` is emitted.
- **On resume with an answer** (`tracker answer` / `tracker.Answer`) the answer
is held in memory only. It applies once, to the gate with the same node, scope
and fingerprint; a gate whose question changed drops it and pauses again with
the note "the question changed since you answered — answer again". The gate's
`timeout:` does not run while paused — the process has exited — and
`--default` takes the declared default explicitly.
- **Parallel branches:** every branch gate is recorded (lowest branch first)
and they are **answered together** (`AnswerInput.Gates`; CLI `--answer
<gate>=<answer>`, repeated): answering re-runs the parallel node, siblings
and their side effects included, so a gate left unanswered would pause it
again and discard the others. Each gate's fingerprint decides whether its
answer still fits; when one gate's question changed and it pauses again, the
node's other gates are recorded as waiting again too (their answers were
used in that pass, and the node re-runs). A branch gate whose prompt changes
on every run (e.g. it shows `${ctx.parallel.results}`) can never match its
answer — keep branch gate prompts to the pre-fan-out context. A billing pause in any branch
outranks the gates.
- **Subgraphs and `manager_loop`:** a gate inside a child fails closed — a
routable, non-retryable `OutcomeFail` with "a gate inside a subgraph can't
pause yet" — because re-running the subgraph would regenerate the question.

The record lives in the checkpoint in the secure state dir, out of the
tool-reachable workdir. That does not make it tamper-proof: a same-UID,
unjailed process that knows the run ID and `$HOME` can still change it (see
`engine.md` § Run-state integrity); the `writable_paths` jail is the real
boundary.

## Outcomes produced

| Mode | Status | PreferredLabel | ContextUpdates |
Expand Down
37 changes: 30 additions & 7 deletions docs/architecture/transport-boundary.md
Original file line number Diff line number Diff line change
Expand Up @@ -148,9 +148,32 @@ existing gate wait) so gate timeouts stay gate-scoped; this ships with a
tracker-runner migration ticket. Until it does, it keeps working on the legacy
fallback, with the pre-#599 coarse-teardown-on-timeout behaviour.

Every `Ask*Context` call carries the gate's identity on its context:
`handlers.GateInfoFrom(ctx)` returns the same `GateInfo` `BeginGate` receives,
plus the full prompt, the `ScopePath` (enclosing subgraph node IDs, then the
parallel branch) and the branch index. An interviewer serving concurrent branch
gates should read the gate from the call, not from the last `BeginGate`.

#### Pause mode

`handlers.PausingInterviewer` is the interviewer for a front-end that is not
there while the run waits (an agent's shell, CI). Each gate pauses the run in
the resumable `paused_gate` terminal with a `pipeline.PausedGate` record of
exactly what it asked (`Checkpoint.PausedGates`, mirrored to the run dir's
`gate.json`), and the front-end answers later with `tracker.Answer`, which
resumes the run holding the answer in memory (`AnswerInput.Gates` answers
every gate of a paused parallel node at once). The `gate_paused` event names
the gate (`Gate.Node`, `Gate.ScopePath`), how many wait (`Gate.Waiting`) and the
command that answers them (`Gate.AnswerCommand`). Select it with
`Config.GateMode: "pause"` (requires `Config.ArtifactDir`); list waiting runs with
`tracker.WaitingRuns` and read a run's gate with `tracker.PendingGates` — both
read the checkpoint, never `gate.json`. A gate inside a subgraph or
`manager_loop` child fails closed in pause mode (routable `OutcomeFail`). Full
contract: [`handlers/human.md`](handlers/human.md#pause-mode).

Reference implementations prove the seam is transport-neutral: `ConsoleInterviewer`,
`tui.BubbleteaInterviewer`, `WebhookInterviewer`, the autopilot interviewers, and
`trackerbot.SlackInterviewer`.
`tui.BubbleteaInterviewer`, `WebhookInterviewer`, `PausingInterviewer`, the
autopilot interviewers, and `trackerbot.SlackInterviewer`.

## 3. Observe progress

Expand All @@ -161,10 +184,10 @@ Two Config-wired streams (a transport merges them):
its model), `stage_*`, `cost_updated` (carries a `NodeID` for per-node
attribution + a `CostSnapshot`), `budget_exceeded`, `validation_overridden`,
and the terminal event (carries an authoritative `TerminalStatus` — `success` /
`validation_overridden` / `fail` / `budget_exceeded` / `paused_billing`; the set
is an **open enum**, so classify with
`pipeline.TerminalStatus(s).IsSuccess()` rather than switching exhaustively, and
note `paused_billing` is a *resumable* stop, not a failure). The **top-level** engine
`validation_overridden` / `fail` / `budget_exceeded` / `paused_billing` /
`paused_gate`; the set is an **open enum**, so classify with
`pipeline.TerminalStatus(s).IsSuccess()` / `.IsPaused()` rather than switching
exhaustively, and note both pauses are *resumable* stops, not failures). The **top-level** engine
emits exactly one terminal event carrying `TerminalStatus` for its own run —
including panic and invariant-error exits, which the backstop covers — so "any
event with a non-empty `TerminalStatus` **and an unscoped `NodeID`**" is the
Expand Down Expand Up @@ -320,7 +343,7 @@ once from one process:
- `RunState` is an **open enum** (`starting`, `running`, `succeeded`, `failed`,
`canceled`, `paused`) — use `State().Terminal()` plus the states you care about
rather than an exhaustive switch. `RunPaused` maps from the engine's
`paused_billing` terminal: the run is *finished-but-resumable*, so `Terminal()`
`paused_billing` and `paused_gate` terminals: the run is *finished-but-resumable*, so `Terminal()`
is **true** (its goroutine exited, `Done()` is closed, the key is free for the
resume attempt) while `ResumeRunID()` returns the id to feed back as
`Config.ResumeRunID`. A control plane must distinguish `paused` from `failed` —
Expand Down
32 changes: 32 additions & 0 deletions pipeline/budget.go
Original file line number Diff line number Diff line change
Expand Up @@ -361,3 +361,35 @@ func (g *BudgetGuard) checkUsage(usage *UsageSummary) BudgetBreach {
}
return BudgetBreach{Kind: BudgetOK}
}

// combinedUsageForBudget returns the usage snapshot that BudgetGuard sees.
// Child engines run with a baseline loaded from the parent's trace so the
// guard enforces against combined parent+child spend. When no baseline is
// set (top-level runs, or subgraph runs without an attached guard), the
// local trace aggregate is returned unchanged.
func (e *Engine) combinedUsageForBudget(s *runState) *UsageSummary {
local := s.trace.AggregateUsage()
if e.baselineUsage == nil {
return local
}
merged := cloneUsageSummary(e.baselineUsage)
if local != nil {
foldChildUsageIntoSummary(merged, local)
}
return merged
}

// cloneUsageSummary returns a deep-enough copy that mutations on the result
// do not affect the input. Used so combinedUsageForBudget can fold a trace
// aggregate into a baseline without corrupting the baseline on repeat calls.
func cloneUsageSummary(u *UsageSummary) *UsageSummary {
if u == nil {
return &UsageSummary{ProviderTotals: make(map[string]ProviderUsage)}
}
clone := *u
clone.ProviderTotals = make(map[string]ProviderUsage, len(u.ProviderTotals))
for k, v := range u.ProviderTotals {
clone.ProviderTotals[k] = v
}
return &clone
}
10 changes: 10 additions & 0 deletions pipeline/checkpoint.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,16 @@ type Checkpoint struct {
// paused, or is still in flight.
HaltedAt string `json:"halted_at,omitempty"`

// PausedGates is the authoritative record of the human gate(s) a
// paused_gate run is waiting at; `tracker answer` validates against it.
// Cleared when the paused node completes.
PausedGates []PausedGate `json:"paused_gates,omitempty"`

// PriorUsage is the run's cumulative usage at its last pause. A resume of
// a gate-paused run seeds its budget from it, so the budget bounds the
// whole run rather than each answer.
PriorUsage *UsageSummary `json:"prior_usage,omitempty"`

// completedSet provides O(1) lookup for IsCompleted. It is rebuilt from
// CompletedNodes on deserialization and kept in sync by MarkCompleted.
completedSet map[string]bool `json:"-"`
Expand Down
Loading