Skip to content
Merged
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
54 changes: 52 additions & 2 deletions cmd/flynn/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -888,7 +888,7 @@ func resumeRun(modelSpec, runID, dataDir string, verbose bool, shell governed.Sh
if err != nil {
return err
}
_, _, _, err = driveRequest{
req, err := driveRequest{
out: os.Stdout,
model: model,
plan: plan,
Expand All @@ -899,6 +899,56 @@ func resumeRun(modelSpec, runID, dataDir string, verbose bool, shell governed.Sh
log: store.Log(),
verbose: verbose,
resumeID: runID,
}.withOptions(withShellPolicy(shell)).execute(ctx)
}.resuming(ctx, dataDir, shell)
if err != nil {
return err
}
_, _, _, err = req.execute(ctx)
return err
}

// resuming completes a request that continues run r.resumeID: the guards its first
// assembly recorded (see [resumeGuards]), its fan-out if it was one, and the operator's
// shell policy for this invocation.
func (r driveRequest) resuming(ctx context.Context, dataDir string, shell governed.ShellPolicy) (driveRequest, error) {
posture, _, err := governed.ReadPosture(ctx, r.log, r.resumeID)
if err != nil {
return r, fmt.Errorf("resume: read the run's posture: %w", err)
}
guards, fanout, err := resumeGuards(posture)
if err != nil {
return r, err
}
if fanout {
r.fanout = &fanoutConfig{resolveModel: childModelResolver(ctx, dataDir)}
}
return r.withOptions(append(guards, withShellPolicy(shell))...), nil
}

// resumeGuards turns the posture a run was first assembled with into the options that
// assemble it the same way again, and reports whether it was a fan-out. A resume reads
// only the shell flags from its own command line, so without these a resumed goal had
// no item verifier (its first verify step failed the run), no spend cap, no approval
// gate and no outside-action gate. A zero posture, from a run recorded before postures
// were, yields none of them, as before. An external agent's run is refused: resume
// drives a native model, and continuing that run on one would change who is driving it.
func resumeGuards(p governed.Posture) (opts []driveOption, fanout bool, err error) {
if p.Mode == governed.ModeExternal {
return nil, false, errors.New("resume: this run was driven by an external agent CLI, and resume can only continue a native run")
}
opts = []driveOption{
withBudget(p.BudgetLimits),
// With nobody at a prompt the gate refuses a listed action, as on a one-shot goal.
withApproval(p.ApprovalActions, nil),
// The run's declared allowances are on its goal, which a resume keeps.
withAllowance(p.AllowanceActions, nil),
}
// A fan-out always wires the verifier, so only the planned single run needs telling.
if p.Verifier && p.Mode != governed.ModeFanout {
opts = append(opts, withPlanning())
}
if p.LedgerProof {
opts = append(opts, withLedgerProof())
}
return opts, p.Mode == governed.ModeFanout, nil
}
64 changes: 64 additions & 0 deletions cmd/flynn/resume_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,12 @@ package main
import (
"bytes"
"context"
"reflect"
"strings"
"testing"
"time"

budgetpkg "github.com/ionalpha/flynn/budget"
"github.com/ionalpha/flynn/governed"

"github.com/ionalpha/flynn/harness"
Expand Down Expand Up @@ -85,3 +87,65 @@ func TestResumeContinuesInterruptedRun(t *testing.T) {
t.Fatalf("resume did not drive the run to completion; result = %q\nout:\n%s", result, out2.String())
}
}

// TestResumeRebuildsTheRunsGuards assembles a goal run with every guard the goal command
// can set, then assembles its resumption the way `flynn resume` does, and requires the two
// postures to match. A resume used to come back with none of them: no item verifier, so
// its first verify step failed the run with goal_no_verifier, and no spend cap, approval
// gate or outside-action gate.
func TestResumeRebuildsTheRunsGuards(t *testing.T) {
ctx := t.Context()
store := memStore(t)
reg, err := governed.Registry()
if err != nil {
t.Fatal(err)
}
base := driveRequest{
out: &bytes.Buffer{}, model: llmtest.NewScripted(llmtest.SayText("done")), workdir: t.TempDir(),
objective: "do the task", system: defaultSystemPrompt, rstore: store.Resources(reg), jq: store.Jobs(), log: store.Log(),
}
first, err := base.withOptions(
withShellPolicy(checkShell), withPlanning(), withLedgerProof(),
withBudget(budgetpkg.Limits{Tokens: 400_000, Cost: 0.5}),
withApproval([]string{"bash"}, nil), withAllowance([]string{"write"}, nil),
).assembly().assemble(ctx)
if err != nil {
t.Fatal(err)
}
want := first.Posture()
_ = first.Close()

resumed := base
resumed.resumeID = first.ID()
resumed, err = resumed.resuming(ctx, t.TempDir(), checkShell)
if err != nil {
t.Fatal(err)
}
second, err := resumed.assembly().assemble(ctx)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = second.Close() })
if got := second.Posture(); !reflect.DeepEqual(got, want) {
t.Fatalf("the resumed run's posture\n%+v\nwant the original run's\n%+v", got, want)
}
}

// TestResumeGuardsRefusesAnExternalAgentRun: resume drives a native model, so continuing a
// run an external agent CLI drove would hand it to a different driver.
func TestResumeGuardsRefusesAnExternalAgentRun(t *testing.T) {
if _, _, err := resumeGuards(governed.Posture{Mode: governed.ModeExternal}); err == nil {
t.Fatal("resumeGuards accepted an external agent's run")
}
opts, fanout, err := resumeGuards(governed.Posture{Mode: governed.ModeFanout, Verifier: true})
if err != nil || !fanout {
t.Fatalf("a fan-out run: fanout %v, err %v", fanout, err)
}
var c driveConfig
for _, o := range opts {
o(&c)
}
if c.planning {
t.Error("a fan-out resume turned on the single run's planning phase")
}
}
1 change: 1 addition & 0 deletions docs/releases/v0.2.0.md
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ Run `flynn version` afterwards; it should name a `v0.2.0-rc` release. v0.1.3 can
- Two Flynn processes opening the same new database at once (a goal starting while you run `flynn runs`, or two commands after an upgrade that adds migrations) could fail with `no such column: "uses"`, because both applied every migration. Each migration is now applied under the database write lock after checking it was not already applied, and the first connection waits out a competing one.
- `go test`, `npm install` and `pip install` failed inside the sandbox: a confined command could not write the caches these tools keep under the home directory, on macOS it could not write its own `TMPDIR`, and on Windows its `TEMP` was the workspace, where Go ignores a `go.mod`. Each sandbox now has a private scratch directory outside the workspace, writable by its commands and deleted when the sandbox closes. The temp variables, `GOCACHE`, `GOMODCACHE`, `npm_config_cache` and `PIP_CACHE_DIR` point into it unless you set them yourself. On Linux it is under `/var/tmp`, so the caches last from one command to the next.
- On Windows a command the agent ran had no memory cap unless you passed `--max-memory`, and its process cap (4096) did not bound memory. A command that kept starting copies of itself, here a `go` shim on `PATH` that resolved back to itself, used all of a 95 GB machine's memory at a few hundred processes. With no `--max-memory`, a command and everything it starts may now commit at most half the machine's physical memory, in `flynn goal`, the interactive session, `flynn serve` and the MCP server. Pass `--max-memory` to set another cap. Local model servers are not affected.
- `flynn resume` rebuilt a run without the guards it was started with. A planned goal resumed after a crash or kill then stalled at its first item check with `goal_no_verifier`, and a resumed run had no `--max-cost` or `--max-tokens` cap, no `--require-approval` gate, no `--irreversible` gate and no `--require-proof`. A resume now reads how the run was first assembled from its record and wires the same guards, with the spend already recorded still counted against the cap. A run driven by an external agent CLI is refused, since resume drives a native model.
- A confined signer on Windows can read its sealed key.
- The interactive session could hang reading input when a multi-byte character arrived with no room left for it in the read buffer.
- Markdown in the terminal could render a line one cell wider than the screen, so it wrapped.
Expand Down
5 changes: 5 additions & 0 deletions governed/assemble.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@ import (
"errors"
"fmt"
"io"
"maps"
"os"
"slices"
"time"

"github.com/ionalpha/flynn/brakes"
Expand Down Expand Up @@ -376,6 +378,9 @@ func (c *config) build(model llm.Model, b Backends, p *parts, mode Mode) (*Run,
if g.stack != nil {
post.ApprovalActions = g.stack.actions
}
if outside := outsidePolicy(c.outside); len(outside) > 0 {
post.AllowanceActions = slices.Sorted(maps.Keys(outside))
}
if mode == ModeExternal {
// An external agent loop runs its own tools beyond the approval gate, so the gate is
// built and never consulted. Claiming it would tell a reader of the record that a
Expand Down
148 changes: 125 additions & 23 deletions governed/guard.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@ package governed

import (
"context"
"encoding/json"
"fmt"
"time"

"github.com/ionalpha/flynn/brakes"
budgetpkg "github.com/ionalpha/flynn/budget"
Expand Down Expand Up @@ -82,6 +85,10 @@ func recordPosture(ctx context.Context, log spine.Log, stream string, p Posture)
for _, a := range p.ApprovalActions {
actions = append(actions, a)
}
outside := make([]any, 0, len(p.AllowanceActions))
for _, a := range p.AllowanceActions {
outside = append(outside, a)
}
waived := make(map[string]any, len(p.Waived))
for g, reason := range p.Waived {
waived[string(g)] = reason
Expand All @@ -91,34 +98,126 @@ func recordPosture(ctx context.Context, log spine.Log, stream string, p Posture)
Type: PostureEventType,
Actor: spine.ActorSystem,
Payload: map[string]any{
"mode": string(p.Mode),
"sandbox": p.Sandbox,
"hostSandbox": p.HostSandbox,
"containment": p.Containment,
"approval": p.Approval,
"approvalActions": actions,
"allowance": p.Allowance,
"brakes": p.Brakes,
"budget": p.Budget,
"budgetTokens": p.BudgetLimits.Tokens,
"budgetCost": p.BudgetLimits.Cost,
"brakeMaxActions": p.BrakeLimits.MaxActions,
"brakeWindow": p.BrakeLimits.Window.String(),
"halt": p.Halt,
"eventSink": p.EventSink,
"verifier": p.Verifier,
"evidence": p.Evidence,
"ledgerProof": p.LedgerProof,
"auditor": p.Auditor,
"steerJudge": p.SteerJudge,
"progress": p.Progress,
"refusals": p.Refusals,
"waived": waived,
"mode": string(p.Mode),
"sandbox": p.Sandbox,
"hostSandbox": p.HostSandbox,
"containment": p.Containment,
"approval": p.Approval,
"approvalActions": actions,
"allowance": p.Allowance,
"allowanceActions": outside,
"brakes": p.Brakes,
"budget": p.Budget,
"budgetTokens": p.BudgetLimits.Tokens,
"budgetCost": p.BudgetLimits.Cost,
"brakeMaxActions": p.BrakeLimits.MaxActions,
"brakeWindow": p.BrakeLimits.Window.String(),
"halt": p.Halt,
"eventSink": p.EventSink,
"verifier": p.Verifier,
"evidence": p.Evidence,
"ledgerProof": p.LedgerProof,
"auditor": p.Auditor,
"steerJudge": p.SteerJudge,
"progress": p.Progress,
"refusals": p.Refusals,
"waived": waived,
},
})
return err
}

// ReadPosture returns the posture the first assembly of stream recorded, and false when
// the stream has none (a run recorded before postures were). A resume reads it to wire
// the guards the run was started with, so continuing a run cannot drop its spend cap,
// its approvals, its outside-action gate or its item verifier. It reads the first and
// not the latest because a resume records its own posture, and one that had lost a
// guard would otherwise pass the loss to the next resume.
func ReadPosture(ctx context.Context, log spine.Log, stream string) (Posture, bool, error) {
const page = 64
for after := int64(0); ; {
events, err := log.Read(ctx, spine.Query{Stream: stream, AfterSeq: after, Limit: page})
if err != nil {
return Posture{}, false, err
}
for _, e := range events {
if e.Type == PostureEventType {
p, err := decodePosture(e.Payload)
return p, err == nil, err
}
}
if len(events) < page {
return Posture{}, false, nil
}
after = events[len(events)-1].Seq
}
}

// decodePosture is the inverse of the payload [recordPosture] writes. It goes through
// JSON because a payload read back from a durable log holds JSON's types (a number is a
// float64), not the ones it was appended with.
func decodePosture(payload map[string]any) (Posture, error) {
raw, err := json.Marshal(payload)
if err != nil {
return Posture{}, fmt.Errorf("governed: posture payload: %w", err)
}
var w struct {
Mode string `json:"mode"`
Sandbox bool `json:"sandbox"`
HostSandbox bool `json:"hostSandbox"`
Containment bool `json:"containment"`
Approval bool `json:"approval"`
ApprovalActions []string `json:"approvalActions"`
Allowance bool `json:"allowance"`
AllowanceActions []string `json:"allowanceActions"`
Brakes bool `json:"brakes"`
Budget bool `json:"budget"`
BudgetTokens int64 `json:"budgetTokens"`
BudgetCost float64 `json:"budgetCost"`
BrakeMaxActions int `json:"brakeMaxActions"`
BrakeWindow string `json:"brakeWindow"`
Halt bool `json:"halt"`
EventSink bool `json:"eventSink"`
Verifier bool `json:"verifier"`
Evidence bool `json:"evidence"`
LedgerProof bool `json:"ledgerProof"`
Auditor bool `json:"auditor"`
SteerJudge bool `json:"steerJudge"`
Progress bool `json:"progress"`
Refusals bool `json:"refusals"`
Waived map[string]string `json:"waived"`
}
if err := json.Unmarshal(raw, &w); err != nil {
return Posture{}, fmt.Errorf("governed: posture payload: %w", err)
}
window, err := time.ParseDuration(w.BrakeWindow)
if err != nil && w.BrakeWindow != "" {
return Posture{}, fmt.Errorf("governed: posture brake window: %w", err)
}
p := Posture{
Mode: Mode(w.Mode), Sandbox: w.Sandbox, HostSandbox: w.HostSandbox, Containment: w.Containment,
Approval: w.Approval, ApprovalActions: w.ApprovalActions,
Allowance: w.Allowance, AllowanceActions: w.AllowanceActions,
Brakes: w.Brakes, Budget: w.Budget, Halt: w.Halt,
BudgetLimits: budgetpkg.Limits{Tokens: w.BudgetTokens, Cost: w.BudgetCost},
BrakeLimits: brakes.Limits{MaxActions: w.BrakeMaxActions, Window: window},
EventSink: w.EventSink, Verifier: w.Verifier, Evidence: w.Evidence, LedgerProof: w.LedgerProof,
Auditor: w.Auditor, SteerJudge: w.SteerJudge, Progress: w.Progress, Refusals: w.Refusals,
Waived: make(map[Guard]string, len(w.Waived)),
}
for g, reason := range w.Waived {
p.Waived[Guard(g)] = reason
}
// recordPosture writes an empty list for none; the run reports none as nil.
if len(p.ApprovalActions) == 0 {
p.ApprovalActions = nil
}
if len(p.AllowanceActions) == 0 {
p.AllowanceActions = nil
}
return p, nil
}

// Mode is the shape of the run's loop.
type Mode string

Expand Down Expand Up @@ -148,6 +247,9 @@ type Posture struct {
Approval bool
ApprovalActions []string
Allowance bool
// AllowanceActions is the actions marked as reaching outside the workspace, sorted:
// the ones [Posture.Allowance] gates.
AllowanceActions []string

Brakes bool
Budget bool
Expand Down
Loading
Loading