diff --git a/cmd/flynn/main.go b/cmd/flynn/main.go index f169d6d3..be51a235 100644 --- a/cmd/flynn/main.go +++ b/cmd/flynn/main.go @@ -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, @@ -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 +} diff --git a/cmd/flynn/resume_test.go b/cmd/flynn/resume_test.go index 24a4e4e5..3970a29b 100644 --- a/cmd/flynn/resume_test.go +++ b/cmd/flynn/resume_test.go @@ -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" @@ -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") + } +} diff --git a/docs/releases/v0.2.0.md b/docs/releases/v0.2.0.md index 34c7e906..e4105bf9 100644 --- a/docs/releases/v0.2.0.md +++ b/docs/releases/v0.2.0.md @@ -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. diff --git a/governed/assemble.go b/governed/assemble.go index d1175de9..bdd0e377 100644 --- a/governed/assemble.go +++ b/governed/assemble.go @@ -5,7 +5,9 @@ import ( "errors" "fmt" "io" + "maps" "os" + "slices" "time" "github.com/ionalpha/flynn/brakes" @@ -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 diff --git a/governed/guard.go b/governed/guard.go index fad20322..08ef01c9 100644 --- a/governed/guard.go +++ b/governed/guard.go @@ -2,6 +2,9 @@ package governed import ( "context" + "encoding/json" + "fmt" + "time" "github.com/ionalpha/flynn/brakes" budgetpkg "github.com/ionalpha/flynn/budget" @@ -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 @@ -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 @@ -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 diff --git a/governed/security_test.go b/governed/security_test.go index 006f802d..bbfa7c2f 100644 --- a/governed/security_test.go +++ b/governed/security_test.go @@ -9,10 +9,12 @@ import ( "go/token" "os" "path/filepath" + "reflect" "slices" "strings" "testing" + budgetpkg "github.com/ionalpha/flynn/budget" "github.com/ionalpha/flynn/capability" "github.com/ionalpha/flynn/dispatch" "github.com/ionalpha/flynn/goal" @@ -370,6 +372,43 @@ func TestResumeStatesThePostureAgain(t *testing.T) { } } +// TestReadPostureReturnsTheFirstAssembly: what [ReadPosture] reads back is the posture the +// run reported when it was first assembled, every field of it, and a later assembly with +// fewer guards does not replace it. A resume rebuilds from this, so a field that fails to +// round-trip is a guard a resumed run silently loses. +func TestReadPostureReturnsTheFirstAssembly(t *testing.T) { + b, _ := memBackends(t) + if _, ok, err := ReadPosture(ctxT(t), b.Log, "no-such-run"); err != nil || ok { + t.Fatalf("ReadPosture on an empty stream = ok %v, err %v; want false, nil", ok, err) + } + + first, err := Assemble(ctxT(t), say("done"), b, WithWorkDir(t.TempDir()), WithPlanning(), WithLedgerProof(), + WithApproval([]string{"write", "bash"}, nil), WithAllowance([]string{"write", "bash"}), + WithBudget(budgetpkg.Limits{Tokens: 400_000, Cost: 0.5})) + if err != nil { + t.Fatalf("assemble: %v", err) + } + defer func() { _ = first.Close() }() + second, err := Assemble(ctxT(t), say("done"), b, WithWorkDir(t.TempDir()), WithRunID(first.ID()), + Waive(GuardBudget, "the host meters spend")) + if err != nil { + t.Fatalf("assemble the resumed run: %v", err) + } + defer func() { _ = second.Close() }() + + got, ok, err := ReadPosture(ctxT(t), b.Log, first.ID()) + if err != nil || !ok { + t.Fatalf("ReadPosture = ok %v, err %v", ok, err) + } + want := first.Posture() + if len(want.AllowanceActions) != 2 || !want.Verifier || !want.LedgerProof || want.BudgetLimits.Cost != 0.5 { + t.Fatalf("the first run was not assembled with the guards this test reads back: %+v", want) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("ReadPosture =\n%+v\nwant the first assembly's\n%+v", got, want) + } +} + // bareSandbox hides the containment declaration of the sandbox it wraps, as a host // sandbox that declares no tier would, and counts closes. type bareSandbox struct {