diff --git a/cmd/flynn/fanout_e2e_test.go b/cmd/flynn/fanout_e2e_test.go new file mode 100644 index 0000000..70974e0 --- /dev/null +++ b/cmd/flynn/fanout_e2e_test.go @@ -0,0 +1,162 @@ +package main + +import ( + "bytes" + "context" + "crypto/ed25519" + "encoding/json" + "io" + "strings" + "testing" + + "github.com/ionalpha/flynn/chain" + "github.com/ionalpha/flynn/goal" + "github.com/ionalpha/flynn/harness" + "github.com/ionalpha/flynn/llm" + "github.com/ionalpha/flynn/mission" +) + +// fanoutChildMarker is the token the parent puts in a delegated sub-goal's objective +// so the test model recognises a child run and answers it directly. +const fanoutChildMarker = "CHILD-TASK" + +// fanoutTestModel is a deterministic model for the fan-out e2e: it answers from the +// request content rather than a fixed sequence, so it behaves correctly whether the +// parent run or a delegated child calls it, in any interleaving (the two run +// concurrently and share this one model). The parent delegates once through the spawn +// tool, then folds the child's answer and finishes; a child, recognised by the marker +// its parent put in the objective, answers directly. +type fanoutTestModel struct{} + +var _ llm.Model = fanoutTestModel{} + +func (fanoutTestModel) Generate(_ context.Context, req llm.Request) (llm.Response, error) { + // Once a tool result is in the transcript the parent has folded its child: finish. + for _, m := range req.Messages { + for _, b := range m.Blocks { + if b.Kind == llm.KindToolResult { + return llm.Response{Message: llm.Text(llm.RoleAssistant, "folded the child and finished"), StopReason: llm.StopEndTurn}, nil + } + } + } + // A child recognises itself by the marker in its objective and answers. + if strings.Contains(messageText(req.Messages), fanoutChildMarker) { + return llm.Response{Message: llm.Text(llm.RoleAssistant, "child finished "+fanoutChildMarker), StopReason: llm.StopEndTurn}, nil + } + // The parent's first turn: delegate one self-contained sub-goal, requesting only + // the write tool. The child still gets the model call (the spawner always grants + // it), so this exercises that path too. + return llm.Response{ + Message: llm.Message{Role: llm.RoleAssistant, Blocks: []llm.Block{{ + Kind: llm.KindToolUse, + ToolUse: &llm.ToolUse{ + ID: "spawn-1", + Name: mission.ActionSpawn, + Input: json.RawMessage(`{"objective":"` + fanoutChildMarker + `: produce the answer","actions":["write"]}`), + }, + }}}, + StopReason: llm.StopToolUse, + }, nil +} + +// messageText concatenates the text of a request's messages, the signal the test +// model branches on. +func messageText(msgs []llm.Message) string { + var b strings.Builder + for _, m := range msgs { + b.WriteString(m.TextContent()) + b.WriteByte(' ') + } + return b.String() +} + +// TestFanoutRunSealsOneVerifiableRecord is the end-to-end proof of the fan-out path: +// a run with delegation enabled drives the goals engine over the durable store, the +// model delegates a sub-goal to a concurrent child agent, and the parent and child +// fold into one event stream that seals into a single signed record a third party can +// verify. It asserts the delegation actually happened (a child goal owned by the +// parent exists) and that a single-byte tamper of the sealed record fails to verify. +func TestFanoutRunSealsOneVerifiableRecord(t *testing.T) { + ctx := context.Background() + store, err := openStore(ctx, "") + if err != nil { + t.Fatal(err) + } + defer func() { _ = store.Close() }() + reg, err := missionRegistry() + if err != nil { + t.Fatal(err) + } + + // Record every event the run appends, without changing how it produces them. + rec := chain.NewRecordingLog(store.Log(), nil) + + // Fan-out enabled. Children name no model, so the resolver is never consulted; a + // nil resolver is the n=0 case of per-goal model routing. + fc := &fanoutConfig{} + + result, runID, _, err := drive(ctx, io.Discard, fanoutTestModel{}, harness.Plan{}, t.TempDir(), + "delegate the work to a child agent and then finish", defaultSystemPrompt, + store.Resources(reg), store.Jobs(), rec, false, "", fc) + if err != nil { + t.Fatalf("drive: %v", err) + } + if result == "" { + t.Fatal("run produced no result") + } + + // The delegation really happened: a child goal owned by the parent run exists. + goals, err := store.Resources(reg).ListAll(ctx, goal.Kind, nil) + if err != nil { + t.Fatal(err) + } + children := 0 + for _, g := range goals { + for _, o := range g.OwnerReferences { + if o.Kind == goal.Kind && o.Controller { + children++ + } + } + } + if children == 0 { + t.Fatalf("fan-out enabled but no child goal was spawned (have %d goals)", len(goals)) + } + + // Seal the run's checkpoint with a fixed test key (no randomness needed) and verify + // it as a third party would, trusting only the signing key. + seed := make([]byte, ed25519.SeedSize) + priv := ed25519.NewKeyFromSeed(seed) + pub := priv.Public().(ed25519.PublicKey) + signer, err := chain.NewEd25519RootSigner("inst", priv) + if err != nil { + t.Fatal(err) + } + sealed, err := rec.Seal(runID, signer) + if err != nil { + t.Fatalf("seal run %q: %v", runID, err) + } + record, err := sealed.Marshal() + if err != nil { + t.Fatal(err) + } + + ring := chain.NewRootKeyring() + if err := ring.Add("inst", pub); err != nil { + t.Fatal(err) + } + events, err := chain.VerifyRun(record, ring) + if err != nil { + t.Fatalf("verifying the fan-out run's record failed: %v", err) + } + if len(events) == 0 { + t.Fatal("verified record carried no events") + } + + bad := append([]byte{}, record...) + bad[len(bad)/2] ^= 0xff + if !bytes.Equal(bad, record) { + if _, err := chain.VerifyRun(bad, ring); err == nil { + t.Fatal("a tampered fan-out record still verified") + } + } +} diff --git a/cmd/flynn/main.go b/cmd/flynn/main.go index c943cf9..9f0453c 100644 --- a/cmd/flynn/main.go +++ b/cmd/flynn/main.go @@ -54,6 +54,7 @@ func main() { verboseLong = flag.Bool("verbose", false, "alias for -v") plain = flag.Bool("plain", false, "interactive session: use the line-based interface, not the full-screen one") verify = flag.String("verify", "", "a command that independently checks the goal succeeded; run after the agent stops, its result grounds the run's success in the verifiable record") + fanout = flag.Bool("fanout", false, "let the goal delegate sub-tasks to concurrent child agents (each routed to the model its archetype pins), all folded into one verifiable record") showVersion = flag.Bool("version", false, "print version and exit") ) flag.Parse() @@ -70,7 +71,7 @@ func main() { fmt.Fprintln(os.Stderr, `usage: flynn goal ""`) os.Exit(2) } - if err := runGoal(*model, objective, *verify, *dataDir, !*noLearn, vrb); err != nil { + if err := runGoal(*model, objective, *verify, *dataDir, !*noLearn, vrb, *fanout); err != nil { fmt.Fprintln(os.Stderr, "error:", err) os.Exit(1) } @@ -186,7 +187,7 @@ func printUsage(w io.Writer) { flynn regrade re-grade learned skills against the working directory flynn serve [--telegram-token T] [--signal-tcp ADDR] [--api-addr ADDR] run as a service: answer chat messages (Telegram, Signal) and/or expose the read-only monitor API flynn --version print the version -Flags: --model, --data-dir, --no-learn, --verify "", -v/--verbose, --plain (run with --help for details).`) +Flags: --model, --data-dir, --no-learn, --verify "", --fanout, -v/--verbose, --plain (run with --help for details).`) } // defaultDataDir is where durable state lives unless overridden: a per-user @@ -203,7 +204,7 @@ func defaultDataDir() string { // completion in the current directory, recalling past learning into the prompt and // (unless disabled) distilling the result back out. Progress and the final result // are printed; Ctrl-C cancels the run. -func runGoal(modelSpec, objective, verify, dataDir string, learnEnabled, verbose bool) error { +func runGoal(modelSpec, objective, verify, dataDir string, learnEnabled, verbose, fanout bool) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() @@ -227,6 +228,15 @@ func runGoal(modelSpec, objective, verify, dataDir string, learnEnabled, verbose distiller = governedDistiller(model) } + // With fan-out enabled the run drives the full goals engine: the model may + // delegate sub-goals to concurrent child agents, each routed to the model its + // bound archetype pins, all folded into this run's one sealed record. A child that + // names a model resolves it through the same credential chain as the root. + var fc *fanoutConfig + if fanout { + fc = &fanoutConfig{resolveModel: childModelResolver(ctx, dataDir)} + } + // Load the instance signer so the run is sealed into a verifiable record. This is // best effort: if the identity cannot be loaded, the run proceeds unsigned rather // than failing. @@ -238,7 +248,7 @@ func runGoal(modelSpec, objective, verify, dataDir string, learnEnabled, verbose // The objective and the final answer are rendered from the run's own events // (session.started and session.converged), so the live transcript and a later // `flynn inspect` of the same run read identically. - if _, err := runLearningMission(ctx, os.Stdout, model, plan, distiller, cwd, objective, verify, store, signer, verbose); err != nil { + if _, err := runLearningMission(ctx, os.Stdout, model, plan, distiller, cwd, objective, verify, store, signer, verbose, fc); err != nil { return err } return nil @@ -305,6 +315,6 @@ func resumeRun(modelSpec, runID, dataDir string, verbose bool) error { if err != nil { return err } - _, _, _, err = drive(ctx, os.Stdout, model, plan, cwd, "", defaultSystemPrompt, store.Resources(reg), store.Jobs(), store.Log(), verbose, runID) + _, _, _, err = drive(ctx, os.Stdout, model, plan, cwd, "", defaultSystemPrompt, store.Resources(reg), store.Jobs(), store.Log(), verbose, runID, nil) return err } diff --git a/cmd/flynn/record_e2e_test.go b/cmd/flynn/record_e2e_test.go index 4ccd804..55994b6 100644 --- a/cmd/flynn/record_e2e_test.go +++ b/cmd/flynn/record_e2e_test.go @@ -40,7 +40,7 @@ func TestRunEmitsVerifiableRecord(t *testing.T) { result, runID, _, err := drive(ctx, io.Discard, model, harness.Plan{}, t.TempDir(), "reply done and stop", defaultSystemPrompt, - store.Resources(reg), store.Jobs(), rec, false, "") + store.Resources(reg), store.Jobs(), rec, false, "", nil) if err != nil { t.Fatalf("drive: %v", err) } diff --git a/cmd/flynn/run.go b/cmd/flynn/run.go index c379b14..274ea26 100644 --- a/cmd/flynn/run.go +++ b/cmd/flynn/run.go @@ -13,12 +13,14 @@ import ( "time" "github.com/ionalpha/flynn/archetype" + "github.com/ionalpha/flynn/brakes" "github.com/ionalpha/flynn/bus" "github.com/ionalpha/flynn/capability" "github.com/ionalpha/flynn/chain" "github.com/ionalpha/flynn/credential" "github.com/ionalpha/flynn/dependency" "github.com/ionalpha/flynn/dispatch" + "github.com/ionalpha/flynn/driver" "github.com/ionalpha/flynn/extension" "github.com/ionalpha/flynn/goal" "github.com/ionalpha/flynn/harness" @@ -28,8 +30,10 @@ import ( "github.com/ionalpha/flynn/learn" "github.com/ionalpha/flynn/llm" "github.com/ionalpha/flynn/mission" + "github.com/ionalpha/flynn/orchestration" "github.com/ionalpha/flynn/playbook" "github.com/ionalpha/flynn/profilestore" + "github.com/ionalpha/flynn/provider" "github.com/ionalpha/flynn/resource" "github.com/ionalpha/flynn/runtime" "github.com/ionalpha/flynn/sandbox" @@ -241,7 +245,7 @@ func missionRegistry() (*resource.Registry, error) { // sandboxed toolset, and (when a distiller is supplied) distills the converged run // back into skills and memory so the next run starts ahead. Progress is written to // out; the model's final summary is returned. -func runLearningMission(ctx context.Context, out io.Writer, model llm.Model, plan harness.Plan, distiller learn.Distiller, workdir, objective, verify string, store *sqlite.Store, signer chain.RootSigner, verbose bool) (string, error) { +func runLearningMission(ctx context.Context, out io.Writer, model llm.Model, plan harness.Plan, distiller learn.Distiller, workdir, objective, verify string, store *sqlite.Store, signer chain.RootSigner, verbose bool, fanout *fanoutConfig) (string, error) { reg, err := missionRegistry() if err != nil { return "", err @@ -266,7 +270,7 @@ func runLearningMission(ctx context.Context, out io.Writer, model llm.Model, pla log = rec } - result, source, transcript, err := drive(ctx, out, model, plan, workdir, objective, system, store.Resources(reg), store.Jobs(), log, verbose, "") + result, source, transcript, err := drive(ctx, out, model, plan, workdir, objective, system, store.Resources(reg), store.Jobs(), log, verbose, "", fanout) // Reinforce the recalled skills by the run's outcome: a skill present in a run // that converged earns a win; one in a run that failed earns only a use. This is @@ -411,6 +415,23 @@ func governedDistiller(model llm.Model) learn.Distiller { return learn.NewGovernedDistiller(learn.NewModelDistiller(model), dispatch.WithAdmitter(capability.Admitter{})) } +// childModelResolver builds the model resolver the Router consults for a delegated +// child goal that names a model other than the run's default. A local catalog model +// is provisioned and served on demand; a hosted model resolves through the same +// credential chain (vault then environment) as the root model, so a child runs on +// whatever its archetype pins without any extra setup. The child's scaffolding plan +// is not threaded through (the Router shares one base plan across loops), so a local +// child runs on the shared defaults. +func childModelResolver(ctx context.Context, dataDir string) driver.ModelResolver { + return func(id string) (llm.Model, error) { + if isLocalModelID(id) { + m, _, err := resolveLocalModel(ctx, id, dataDir) + return m, err + } + return provider.ResolveWith(ctx, id, credentialSource(dataDir)) + } +} + // recallContext queries the durable skills and memory for what is relevant to the // objective and renders a compact, bounded block to prepend to the system prompt. // It returns "" when nothing is on file, so a fresh agent's prompt is unchanged. @@ -614,9 +635,18 @@ func truncate(s string, n int) string { // id (used as learning provenance), and the conversation transcript (so the // distiller can learn from how the goal was reached, not just the final summary). // The system prompt is supplied so the caller can fold recalled knowledge into it. -func drive(ctx context.Context, out io.Writer, model llm.Model, plan harness.Plan, workdir, objective, system string, rstore resource.Store, jq jobs.Queue, log spine.Log, verbose bool, resumeID string) (result, source string, transcript []llm.Message, err error) { +func drive(ctx context.Context, out io.Writer, model llm.Model, plan harness.Plan, workdir, objective, system string, rstore resource.Store, jq jobs.Queue, log spine.Log, verbose bool, resumeID string, fanout *fanoutConfig) (result, source string, transcript []llm.Message, err error) { w := &syncWriter{w: out} - run, err := assembleMission(model, plan, workdir, system, rstore, jq, log, resumeID) + // A run with fan-out enabled drives the full goals engine (the Router plus a + // delegation spawner); otherwise it is a single governed conversation. Both seal + // into the same verifiable record, so fan-out adds delegation without changing how + // a run is recorded or checked. + var run *missionRun + if fanout != nil { + run, err = assembleFanoutMission(model, plan, workdir, system, rstore, jq, log, resumeID, fanout.resolveModel) + } else { + run, err = assembleMission(model, plan, workdir, system, rstore, jq, log, resumeID) + } if err != nil { return "", "", nil, err } @@ -643,6 +673,13 @@ func drive(ctx context.Context, out io.Writer, model llm.Model, plan harness.Pla } else if _, err := run.sess.Submit(runCtx, run.rt, goal.Spec{ Objective: objective, StopCondition: "the objective is fully accomplished", + // A fan-out parent spends steps it would not as a single conversation: a step + // dispatching each delegation, and a step per poll while it waits for the + // children to finish (a wait makes no model call, but the reconciler still + // counts it). Give it a larger budget so a legitimate delegation that waits on + // a few children is not cut off mid-fold; a single conversation keeps the + // default. The safety brake and fan-out width still bound a runaway. + MaxSteps: fanoutMaxSteps(fanout), }); err != nil { return "", "", nil, err } @@ -732,6 +769,135 @@ func assembleMission(model llm.Model, plan harness.Plan, workdir, system string, return &missionRun{rt: rt, sess: sess}, nil } +// defaultFanoutWidth caps how many child runs a fan-out may have outstanding at +// once, bounding the blast radius of delegation alongside the depth guard. +const defaultFanoutWidth = 8 + +// fanoutRootMaxSteps is the step budget a fan-out parent runs under. A fan-out adds +// orchestration steps a single conversation never takes (a dispatch per delegation, +// and one poll per reconcile while it waits for the children), so the default budget +// that suits a single loop would cut a legitimate delegation off mid-fold. Zero +// (single conversation) keeps the reconciler's default. +const fanoutRootMaxSteps = 200 + +// fanoutMaxSteps returns the step budget for a run's root goal: the larger fan-out +// budget when delegation is enabled, or zero (the reconciler's default) otherwise. +func fanoutMaxSteps(fanout *fanoutConfig) int { + if fanout != nil { + return fanoutRootMaxSteps + } + return 0 +} + +// defaultMaxActionsPerMinute is the rate the default safety brake halts a run at. +// It is set well above any real run's pace so it catches only a degenerate tight +// loop, not legitimate tool use. +const defaultMaxActionsPerMinute = 600 + +// fanoutConfig enables the goals engine on a one-shot run: the model may delegate +// self-contained sub-goals to concurrent, governed child agents, and each child is +// routed to the model and loop its bound Agent archetype pins (resolveModel turns a +// named model into a client). Every child runs under a grant narrowed from the +// parent's, shares the run's budget, and folds back into the parent's single sealed +// record, so a multi-goal, multi-model fan-out stays one verifiable run. A nil +// *fanoutConfig leaves a run as a single conversation, which is the n=1 case of the +// same mechanism. +type fanoutConfig struct { + resolveModel driver.ModelResolver +} + +// assembleFanoutMission wires a one-shot run that drives the full goals engine over +// the durable store: a Router that builds one loop per (driver, model) a goal +// selects, and a fan-out spawner that creates governed child goals. It mirrors +// assembleMission (same sandbox, session, toolset, grant, governance recording, and +// compaction) but adds the spawn action to the grant and routes each goal through +// the Router, so a delegated child runs as the agent its archetype names, on that +// agent's model, while the root and every child fold into one recorded, sealable +// stream. The shared store backs the child goals a fan-out spawns, so they land +// where the runtime reconciles them. +func assembleFanoutMission(model llm.Model, plan harness.Plan, workdir, system string, rstore resource.Store, jq jobs.Queue, log spine.Log, runID string, resolveModel driver.ModelResolver) (*missionRun, error) { + sb, err := sandbox.NewLocal(workdir, sandbox.WithDefaultConfinement()) + if err != nil { + return nil, err + } + + var sopts []session.Option + if runID != "" { + sopts = append(sopts, session.WithID(runID)) + } + sess := session.New(log, bus.NewMemory(), sopts...) + + toolset := tools.New(sb).Tools() + // The grant lists every action the run may take: the tools, the model call, the + // distillation, and the spawn that delegates a sub-goal. A child narrows from this + // set, so a delegation can never widen authority; a run whose grant omitted spawn + // could not fan out at all. + names := make([]string, 0, len(toolset)+3) + for _, t := range toolset { + names = append(names, t.Def().Name) + } + names = append(names, mission.ActionModelGenerate, learn.DistillAction, mission.ActionSpawn) + + // The spawner is the run's fan-out: it creates governed child goals (owned by the + // parent, grant narrowed, depth- and concurrency-bounded) and hands them to the + // runtime. Its enqueue hook is bound once the runtime exists (below). + spawner := orchestration.NewSpawner(rstore, nil, orchestration.WithConcurrency(defaultFanoutWidth)) + + // The Router drives each goal through the loop and model its spec selects: the + // default loop and host model for the root, and the bound Agent's loop and model + // for a delegated child. The shared ingredients (tools, default prompt and grant, + // sandbox gate, governance recording, compaction, brake, fan-out) apply to every + // loop; the per-goal prompt, grant, and model are applied from the goal. + router := driver.NewRouter(driver.RouterConfig{ + Registry: driver.Default(), + DefaultModel: model, + ResolveModel: resolveModel, + Base: driver.Spec{ + Tools: toolset, + System: system, + Grant: capability.NewGrant(names...), + HasGrant: true, + Sandbox: sb, + Reporter: sess.Reporter(), + Fanout: spawner, + // Record every governed action's lifecycle onto the run's own stream, so the + // admission decisions (including each delegation) are part of the sealed record. + EventSink: spinesink.New(log, sess.ID()), + CompactionBudget: defaultCompactionBudget, + // Halt a runaway from outside the model loop. The default is a generous rate + // backstop: a real run dispatches far fewer than this per minute, so the breaker + // fires only on a degenerate tight loop, never on legitimate tool use. + Brakes: brakes.NewHook(brakes.Limits{MaxActions: defaultMaxActionsPerMinute, Window: time.Minute}, nil), + // Apply the model's scaffolding plan so a weaker model is driven with the + // support it needs; the zero plan of a strong model adds nothing. + Plan: plan, + }, + }) + + rt, err := runtime.New(runtime.Config{ + Executor: router, + Stop: router, + Store: rstore, + Jobs: jq, + PollInterval: 200 * time.Millisecond, + WorkerPoll: 50 * time.Millisecond, + // A CLI run drives only its own goal and the children it spawns (enqueued + // explicitly below), never a parked goal an earlier run left non-terminal. + DriveSubmittedOnly: true, + }) + if err != nil { + return nil, err + } + // Bind the spawner to the runtime so a spawned child is enqueued for + // reconciliation. Binding here (rather than at construction) breaks the cycle: the + // executor holds the spawner, and the runtime holds the executor. + spawner.SetEnqueue(func(ctx context.Context, key resource.Key) error { + _, rerr := rt.Resume(ctx, key.Name) + return rerr + }) + return &missionRun{rt: rt, sess: sess}, nil +} + // renderStream prints the session's events as they arrive and accumulates the // conversation transcript (the model's text and the tools it called), returning // once the session reaches a terminal event: the model's summary on convergence, diff --git a/cmd/flynn/run_test.go b/cmd/flynn/run_test.go index dcacf91..00d4edc 100644 --- a/cmd/flynn/run_test.go +++ b/cmd/flynn/run_test.go @@ -54,7 +54,7 @@ func TestRunWritesFileThroughSandbox(t *testing.T) { defer cancel() var out bytes.Buffer - result, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, dir, "create hello.txt with a greeting", "", memStore(t), nil, false) + result, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, dir, "create hello.txt with a greeting", "", memStore(t), nil, false, nil) if err != nil { t.Fatalf("run: %v", err) } @@ -82,7 +82,7 @@ func TestRunRejectsSandboxEscape(t *testing.T) { defer cancel() var out bytes.Buffer - if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, dir, "try to escape", "", memStore(t), nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, dir, "try to escape", "", memStore(t), nil, false, nil); err != nil { t.Fatalf("run: %v", err) } if _, err := os.Stat(filepath.Join(filepath.Dir(dir), "escape.txt")); !os.IsNotExist(err) { @@ -129,13 +129,13 @@ func TestRunRemembersAcrossRuns(t *testing.T) { distiller := &fakeDistiller{lessons: []learn.Lesson{ {Kind: learn.LessonMemory, Body: "the project uses pnpm for installs"}, }} - if _, err := runLearningMission(ctx, &out, run1, harness.Plan{}, distiller, dir, "set up the project", "", store, nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, run1, harness.Plan{}, distiller, dir, "set up the project", "", store, nil, false, nil); err != nil { t.Fatalf("run 1: %v", err) } // Run 2: shares a keyword ("pnpm") with the stored memory, so recall injects it. run2 := llmtest.NewScripted(llmtest.SayText("installed deps")) - if _, err := runLearningMission(ctx, &out, run2, harness.Plan{}, nil, dir, "install deps with pnpm", "", store, nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, run2, harness.Plan{}, nil, dir, "install deps with pnpm", "", store, nil, false, nil); err != nil { t.Fatalf("run 2: %v", err) } @@ -196,7 +196,7 @@ func TestRunFeedsTranscriptToDistiller(t *testing.T) { llmtest.SayText("wrote x.txt"), ) rec := &recordingDistiller{} - if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, rec, dir, "write x.txt", "", memStore(t), nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, rec, dir, "write x.txt", "", memStore(t), nil, false, nil); err != nil { t.Fatalf("run: %v", err) } @@ -263,7 +263,7 @@ func TestRunReinforcesRecalledSkill(t *testing.T) { t.Fatal(err) } model := llmtest.NewScripted(llmtest.SayText("done")) - if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, &fakeDistiller{}, dir, "deploy with docker", "", store, nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, &fakeDistiller{}, dir, "deploy with docker", "", store, nil, false, nil); err != nil { t.Fatalf("run: %v", err) } @@ -291,7 +291,7 @@ func TestRunVerifiesCapturedSkill(t *testing.T) { {Kind: learn.LessonSkill, Title: "Broken skill", Body: "does not work", Check: "exit 1"}, {Kind: learn.LessonSkill, Title: "Good skill", Body: "works", Check: "exit 0"}, }} - if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, distiller, dir, "do the work", "", store, nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, distiller, dir, "do the work", "", store, nil, false, nil); err != nil { t.Fatalf("run: %v", err) } @@ -344,7 +344,7 @@ func TestRunSpineIsDurableAndAddressable(t *testing.T) { if err != nil { t.Fatalf("open store: %v", err) } - if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, workdir, "do the thing", "", store1, nil, false); err != nil { + if _, err := runLearningMission(ctx, &out, model, harness.Plan{}, nil, workdir, "do the thing", "", store1, nil, false, nil); err != nil { t.Fatalf("run: %v", err) } runID := runIDFromOutput(t, out.String()) diff --git a/cmd/flynn/spine_test.go b/cmd/flynn/spine_test.go index 160b5d2..fd41a59 100644 --- a/cmd/flynn/spine_test.go +++ b/cmd/flynn/spine_test.go @@ -34,7 +34,7 @@ func TestRunRecordsGovernanceOnSpine(t *testing.T) { model := llmtest.NewScripted(llmtest.SayText("done")) _, runID, _, err := drive(ctx, io.Discard, model, harness.Plan{}, t.TempDir(), - "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), store.Log(), false, "") + "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), store.Log(), false, "", nil) if err != nil { t.Fatalf("drive: %v", err) } @@ -113,7 +113,7 @@ func TestSpineVerifyRoundTrip(t *testing.T) { rec := chain.NewRecordingLog(store.Log(), nil) model := llmtest.NewScripted(llmtest.SayText("done")) _, runID, _, err := drive(ctx, io.Discard, model, harness.Plan{}, t.TempDir(), - "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), rec, false, "") + "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), rec, false, "", nil) if err != nil { t.Fatalf("drive: %v", err) } @@ -201,7 +201,7 @@ func TestRunGroundTruthEndToEnd(t *testing.T) { rec := chain.NewRecordingLog(store.Log(), nil) model := llmtest.NewScripted(llmtest.SayText("done")) _, runID, _, err := drive(ctx, io.Discard, model, harness.Plan{}, t.TempDir(), - "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), rec, false, "") + "reply done and stop", defaultSystemPrompt, store.Resources(reg), store.Jobs(), rec, false, "", nil) if err != nil { t.Fatalf("drive: %v", err) } diff --git a/driver/default.go b/driver/default.go index a09df5b..c13e68a 100644 --- a/driver/default.go +++ b/driver/default.go @@ -45,6 +45,12 @@ func (defaultDriver) Build(s Spec) (goal.StepExecutor, goal.StopEvaluator, error if s.Fanout != nil { opts = append(opts, mission.WithFanout(s.Fanout)) } + if s.EventSink != nil { + opts = append(opts, mission.WithEventSink(s.EventSink)) + } + if s.CompactionBudget > 0 { + opts = append(opts, mission.WithCompactionBudget(s.CompactionBudget)) + } if s.Plan.SimplifyToolSchemas { opts = append(opts, mission.WithSimplifiedSchemas()) } diff --git a/driver/driver.go b/driver/driver.go index 92414a4..e100f0f 100644 --- a/driver/driver.go +++ b/driver/driver.go @@ -26,6 +26,7 @@ import ( "github.com/ionalpha/flynn/brakes" "github.com/ionalpha/flynn/capability" + "github.com/ionalpha/flynn/dispatch" "github.com/ionalpha/flynn/fault" "github.com/ionalpha/flynn/goal" "github.com/ionalpha/flynn/harness" @@ -62,6 +63,15 @@ type Spec struct { // (the model is offered a spawn tool). A loop that does not fan out (a single-shot // responder) ignores it. The default is nil: a goal runs as a single conversation. Fanout mission.Fanout + // EventSink, when set, records every governed action's lifecycle (admitted, + // completed, or rejected) onto the event spine, so the admission decisions are + // part of the run's recorded and sealed history rather than only the live trace. + // A nil sink records nothing extra, leaving the live trace unchanged. + EventSink dispatch.EventSink + // CompactionBudget, when greater than zero, is the input-token budget past which + // the loop elides the oldest middle turns to keep a long session within the + // model's window. Zero leaves compaction at its default. + CompactionBudget int // Plan is the capability-scaffolding for a weaker or more quantized model: how // hard the loop should work to keep it reliable. The zero Plan adds nothing, so a // capable model runs leanly. diff --git a/orchestration/spawner.go b/orchestration/spawner.go index 11636e2..0459be1 100644 --- a/orchestration/spawner.go +++ b/orchestration/spawner.go @@ -180,14 +180,14 @@ func (s *Spawner) create(ctx context.Context, parent resource.Resource, parentSp // (intersected with the parent's authority, never widened) and its system prompt. // An ad-hoc child takes its authority from the requested actions. An unknown Agent // fails the spawn closed. - childGrant := narrowGrant(parentSpec.Grant, sub.Actions) + childGrant := narrowGrant(parentSpec.Grant, withModelGenerate(sub.Actions)) childSystem, childDriver, childModel := "", "", "" if sub.Agent != "" { resolved, err := archetype.Resolve(ctx, s.store, parent.Scope, sub.Agent) if err != nil { return resource.Resource{}, fault.Wrap(fault.Forbidden, "spawn_agent_resolve", err) } - childGrant = narrowGrant(parentSpec.Grant, resolved.Capabilities) + childGrant = narrowGrant(parentSpec.Grant, withModelGenerate(resolved.Capabilities)) childSystem, childDriver, childModel = resolved.System, resolved.Driver, resolved.Model } childSpec := goal.Spec{ @@ -308,3 +308,19 @@ func narrowGrant(parentGrant, requested []string) []string { } return capability.NewGrant(parentGrant...).Narrow(requested...).Actions() } + +// withModelGenerate ensures the model-call action is in a child's requested +// authority. Every loop must call the model to make progress, so a child that +// requested only tool actions (or a bound Agent whose capabilities omit it) would +// otherwise be admitted for its tools but refused the model call, leaving it unable +// to think. The action is added to the request before it is narrowed against the +// parent, so it is still intersected with the parent's grant and never widens it; +// this mirrors how the host grants the root run its own model call. +func withModelGenerate(actions []string) []string { + for _, a := range actions { + if a == mission.ActionModelGenerate { + return actions + } + } + return append(append(make([]string, 0, len(actions)+1), actions...), mission.ActionModelGenerate) +} diff --git a/orchestration/spawner_test.go b/orchestration/spawner_test.go index 8bdf764..3988aab 100644 --- a/orchestration/spawner_test.go +++ b/orchestration/spawner_test.go @@ -232,8 +232,8 @@ func TestPollReportsOutcomes(t *testing.T) { } // TestNarrowGrantProperty is the rigor property: a child's grant is always a subset -// of both what its parent holds and what the sub-goal requested, so a delegation can -// never widen authority. +// of both what its parent holds and what the sub-goal requested (plus the model call, +// which every child gets so it can think), so a delegation can never widen authority. func TestNarrowGrantProperty(t *testing.T) { rapid.Check(t, func(rt *rapid.T) { actions := rapid.SliceOfNDistinct(rapid.StringMatching(`[a-z]{1,4}`), 0, 8, func(s string) string { return s }).Draw(rt, "actions") @@ -251,11 +251,13 @@ func TestNarrowGrantProperty(t *testing.T) { } got := childSpec(t, s, id).Grant pset := toSet(parentGrant) - rset := toSet(requested) + // The model call is always added to the request so a child can think; it is still + // intersected with the parent, so it never widens authority. + rset := toSet(append(requested, mission.ActionModelGenerate)) for _, a := range got { // When the parent is unconstrained (empty grant), the child is scoped to its // request; otherwise it is the intersection. - if len(parentGrant) > 0 && !pset[a] { + if len(parentGrant) > 0 && !pset[a] && a != mission.ActionModelGenerate { rt.Fatalf("child action %q not in parent grant %v", a, parentGrant) } if !rset[a] { @@ -265,6 +267,29 @@ func TestNarrowGrantProperty(t *testing.T) { }) } +// TestChildAlwaysGrantedModelCall checks that a child spawned with only tool actions +// can still call the model: every loop must generate to make progress, so the model +// call is added to a child's grant even when the delegation requested only tools. The +// parent here holds the model call, so the addition is within its authority. +func TestChildAlwaysGrantedModelCall(t *testing.T) { + s := newStore(t) + sp := orchestration.NewSpawner(s, nil) + sp.SetEnqueue((&recordingEnqueue{}).fn) + parent := putParent(t, s, "root", goal.Spec{ + Objective: "o", + StopCondition: "c", + Grant: []string{"write", mission.ActionModelGenerate}, + }) + + id, err := sp.Spawn(context.Background(), parent, mission.SubGoal{Objective: "x", Actions: []string{"write"}}) + if err != nil { + t.Fatalf("spawn: %v", err) + } + if !toSet(childSpec(t, s, id).Grant)[mission.ActionModelGenerate] { + t.Fatalf("child spawned with only tool actions cannot call the model: grant = %v", childSpec(t, s, id).Grant) + } +} + func mustSpawn(t *testing.T, sp *orchestration.Spawner, parent resource.Resource) string { t.Helper() id, err := sp.Spawn(context.Background(), parent, mission.SubGoal{Objective: "x", Actions: []string{"read"}})