diff --git a/chain/conformance/conformance.go b/chain/conformance/conformance.go index 7f6478c..b52415a 100644 --- a/chain/conformance/conformance.go +++ b/chain/conformance/conformance.go @@ -354,9 +354,9 @@ func largeIntEvent(seq, timeNanos int64) []byte { func unicodePayloadEvent() []byte { e := baseEvent() e.Payload = map[string]any{ - "Z": "latin", - "é": "e-acute", - "☃": "snowman", + "Z": "latin", + "é": "e-acute", + "☃": "snowman", "🎯": "target", } return mustCanonical(e) diff --git a/cmd/flynn/serve.go b/cmd/flynn/serve.go index a5f3949..675f482 100644 --- a/cmd/flynn/serve.go +++ b/cmd/flynn/serve.go @@ -289,7 +289,7 @@ func instanceReporter(store resource.Store, instanceID string) instance.Reporter continue } switch st.Phase { - case goal.PhasePending, goal.PhaseRunning: + case goal.PhasePending, goal.PhasePlanning, goal.PhaseRunning: active = append(active, r.Name) case goal.PhaseConverged, goal.PhaseStalled: // Terminal: the goal is finished, so it is not active work. diff --git a/goal/goal.go b/goal/goal.go index 4009eb4..e8f8fa2 100644 --- a/goal/goal.go +++ b/goal/goal.go @@ -25,6 +25,11 @@ const ( Finalizer = "goal.ionagent.io/cleanup" // StepJobKind is the job kind a dispatched goal step is enqueued under. StepJobKind = "goal.step" + // PlanJobKind is the job kind a goal's planning step is enqueued under. It rides + // the same queue and worker as a build step and is distinguished by kind, so + // planning inherits the lease, crash-resume and retry ladder a step already has + // rather than growing a second execution path. + PlanJobKind = "goal.plan" ) // Phase is a coarse, human-facing lifecycle summary of a goal, derived from its @@ -35,6 +40,7 @@ type Phase string // stalled state. const ( PhasePending Phase = "Pending" // accepted, no step run yet + PhasePlanning Phase = "Planning" // expanding the objective into a ledger, before any building PhaseRunning Phase = "Running" // a step is in flight or more are queued PhaseConverged Phase = "Converged" // the stop condition is satisfied PhaseStalled Phase = "Stalled" // out of budget or a step failed terminally @@ -58,6 +64,19 @@ var specSchema = json.RawMessage(`{ "grant": {"type": "array", "items": {"type": "string"}}, "depth": {"type": "integer", "minimum": 0}, "budgetPool": {"type": "string"}, + "ledger": { + "type": "array", + "items": { + "type": "object", + "required": ["id", "item", "verify"], + "properties": { + "id": {"type": "string", "minLength": 1}, + "item": {"type": "string", "minLength": 1}, + "verify": {"type": "string", "minLength": 1} + }, + "additionalProperties": false + } + }, "system": {"type": "string"}, "driver": {"type": "string"}, "model": {"type": "string"}, @@ -122,6 +141,14 @@ type Spec struct { // turn. Empty on every text-only goal, so the text path serializes // identically and is unchanged. Attachments []llm.Image `json:"attachments,omitempty"` + // Ledger is the goal's objective expanded into the units of work it implies, + // each carrying its own declared way to verify it. It is desired state, which is + // why it lives here and not in status: it is what the goal is committed to + // doing, and the per-item proven/unproven observation of it is Status.Ledger. + // The planner phase writes it before any building starts, and it is + // append-and-mark-only from then on (see ledger.go). Empty on a goal that runs + // without a planner, which is every goal composed before planning is wired. + Ledger []LedgerItem `json:"ledger,omitempty"` } // InFlight records a dispatched step not yet observed complete, so a re-reconcile @@ -129,6 +156,12 @@ type Spec struct { type InFlight struct { JobID string `json:"jobID"` StartedAt time.Time `json:"startedAt"` + // Kind is the job kind dispatched (StepJobKind or PlanJobKind), so the observing + // pass knows what it is observing. A planning step is not building, so it does + // not spend the build budget; without this the reconciler cannot tell the two + // apart once the job is done and the reservation is being cleared. Empty on a + // reservation written before planning existed, which reads as a build step. + Kind string `json:"kind,omitempty"` } // Status is a goal's observed state. @@ -152,6 +185,15 @@ type Status struct { // step budget. A settling child clears it (the prompt wake); the reconciler's // recheck fallback clears it after a bounded delay if that wake is lost. WaitingSince *time.Time `json:"waitingSince,omitempty"` + // Planned records that the planning phase has run to completion. It is separate + // from a non-empty Ledger because the two answer different questions: a planner + // that ran and produced nothing is a planned goal with an empty ledger, which is + // a stall, not an invitation to plan again on the next pass. + Planned bool `json:"planned,omitempty"` + // Ledger is the observed state of Spec.Ledger, one entry per planned item and in + // the same order. Every entry starts unproven, so a goal begins from a record + // that says nothing is done. Completion is a transition here. + Ledger []LedgerState `json:"ledger,omitempty"` } // Condition is one standard status condition (the shared resource.Condition). diff --git a/goal/ledger.go b/goal/ledger.go new file mode 100644 index 0000000..f0b3357 --- /dev/null +++ b/goal/ledger.go @@ -0,0 +1,261 @@ +package goal + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "strings" + "time" +) + +// The ledger is a goal's expansion of its objective into the units of work that +// objective actually implies: what to do, and for each one the declared way to +// prove it was done. It is written by a planner phase before any building starts, +// it lives on the goal resource so it survives a crash and rehydrates on lease +// recovery, and it is the record completion is decided against. +// +// Completion being a state transition on this record, rather than a phrase matched +// in a transcript, is the point. Prose completion fails in both directions: a run +// declares victory on step one having written nothing, and a genuinely finished run +// goes undetected because its wording drifted. Neither is recoverable by parsing +// harder. A goal that cannot mark an item proven without saying which item and how +// has no way to express either failure. +// +// Two structural rules protect the record from the agent writing it: +// +// Append-and-mark-only. An item may be added, and an item's state may be marked, +// but no item is ever removed or rewritten. The failure this forecloses is the one +// where an agent under completion pressure edits the definition of done: dropping a +// test, narrowing an item until the thing it required is no longer required. A +// prompt asking the model not to do that is advisory and gets ignored under +// pressure; a ledger whose prefix cannot change is not. +// +// Content-addressed items. An item's ID is a hash of its text and its verify +// clause, so a rewrite is not a modified item, it is a different ID, and the +// extension check below sees it as the removal it is. + +// ledgerIDLen is how many hex characters of an item's content hash form its ID. +// Sixteen (64 bits) is far past collision range for the handful of items one goal +// carries, and short enough to stay readable in a status dump. +const ledgerIDLen = 16 + +// Ledger errors. They are distinct because they mean different things about the +// writer: a regression is the definition of done being edited, an unknown item is a +// claim about work that was never planned, and an incomplete item is a plan entry +// with no way to check it. +var ( + // ErrLedgerRegressed reports that a proposed ledger is not an extension of the + // one already recorded: an item was removed, reordered, or rewritten. + ErrLedgerRegressed = errors.New("goal: ledger is not an append-only extension") + // ErrLedgerUnknownItem reports a state change naming an item the ledger does not + // carry. + ErrLedgerUnknownItem = errors.New("goal: ledger item not found") + // ErrLedgerIncomplete reports a ledger item missing its text or its verify clause. + // An item with no declared way to check it cannot be proven, only asserted. + ErrLedgerIncomplete = errors.New("goal: ledger item needs both an item and a verify") + // ErrLedgerDuplicate reports the same item appended twice. + ErrLedgerDuplicate = errors.New("goal: ledger item already present") +) + +// LedgerItem is one planned unit of work. Verify is what makes the item checkable: +// the item's own declared way to prove it, recorded when the work is planned rather +// than argued for once it is time to claim the item is done. +type LedgerItem struct { + // ID is the content address of Item and Verify (see ItemID). It is assigned on + // append, not by the planner, so an item's identity is its content and a rewrite + // cannot masquerade as an edit of the same item. + ID string `json:"id"` + // Item is what the work is, in the planner's words. + Item string `json:"item"` + // Verify is the declared way to prove the item: the command to run, the check to + // make, the observation that would settle it. + Verify string `json:"verify"` +} + +// LedgerState is the observed state of one ledger item, keyed to it by ID. It +// starts unproven for every item, which is why the planner writes the whole ledger +// before step one: the run begins from a record that says nothing is done yet. +type LedgerState struct { + ID string `json:"id"` + Proven bool `json:"proven,omitempty"` + // ProvenAt is when the item was marked, and Evidence is the reference to what + // was actually observed. Evidence is free-form here and stays that way until the + // evidence gate lands, which is what turns it from a note into a requirement. + ProvenAt *time.Time `json:"provenAt,omitempty"` + Evidence string `json:"evidence,omitempty"` +} + +// ItemID returns the content address of an item: the first ledgerIDLen hex +// characters of a SHA-256 over its text and verify clause. The two are separated by +// a NUL so no pair of fields can be re-split into a different pair with the same +// concatenation. +func ItemID(item, verify string) string { + sum := sha256.Sum256([]byte(item + "\x00" + verify)) + return hex.EncodeToString(sum[:])[:ledgerIDLen] +} + +// AppendItems returns ledger with items appended, stamping each new item's ID from +// its content. It rejects an item missing its text or verify clause, and an item +// already on the ledger. The returned slice is new, so a caller that fails +// validation still holds an unmodified ledger. +func AppendItems(ledger []LedgerItem, items ...LedgerItem) ([]LedgerItem, error) { + out := make([]LedgerItem, len(ledger), len(ledger)+len(items)) + copy(out, ledger) + seen := make(map[string]struct{}, len(out)+len(items)) + for _, it := range out { + seen[it.ID] = struct{}{} + } + for _, it := range items { + it.Item = strings.TrimSpace(it.Item) + it.Verify = strings.TrimSpace(it.Verify) + if it.Item == "" || it.Verify == "" { + return nil, fmt.Errorf("%w: %q", ErrLedgerIncomplete, it.Item) + } + it.ID = ItemID(it.Item, it.Verify) + if _, dup := seen[it.ID]; dup { + return nil, fmt.Errorf("%w: %q", ErrLedgerDuplicate, it.Item) + } + seen[it.ID] = struct{}{} + out = append(out, it) + } + return out, nil +} + +// ValidateExtension reports whether next is a legal successor to prev: prev must be +// a prefix of next, item for item and in order, and every item in next must carry +// the ID its own content addresses to. Anything else is a removal, a reordering, or +// a rewrite, and all three are the same failure wearing different clothes. +// +// This is the check that makes append-and-mark-only enforceable rather than +// requested. It runs on the write path, so an edited ledger is refused at the point +// of the write, and it runs again on reconcile, so a ledger edited around the write +// path settles the goal as stalled instead of being silently accepted. +func ValidateExtension(prev, next []LedgerItem) error { + if len(next) < len(prev) { + return fmt.Errorf("%w: %d items dropped", ErrLedgerRegressed, len(prev)-len(next)) + } + seen := make(map[string]struct{}, len(next)) + for i, it := range next { + if i < len(prev) { + if was := prev[i]; it.ID != was.ID || it.Item != was.Item || it.Verify != was.Verify { + return fmt.Errorf("%w: item %d (%s) was rewritten or reordered", ErrLedgerRegressed, i, was.ID) + } + } + if it.Item == "" || it.Verify == "" { + return fmt.Errorf("%w: item %d", ErrLedgerIncomplete, i) + } + if want := ItemID(it.Item, it.Verify); it.ID != want { + return fmt.Errorf("%w: item %d carries id %q but addresses to %q", ErrLedgerRegressed, i, it.ID, want) + } + if _, dup := seen[it.ID]; dup { + return fmt.Errorf("%w: item %d (%s)", ErrLedgerDuplicate, i, it.ID) + } + seen[it.ID] = struct{}{} + } + return nil +} + +// ValidateLedger checks a goal's spec ledger against the state the status already +// records for it: the recorded entries must be a prefix of the ledger, in order and +// by id, and every ledger item must still address to its own id. It is the +// reconcile-time half of the extension rule. The write path refuses an edited +// ledger, and this refuses one that was edited around the write path, so a +// tampered ledger settles the goal as stalled rather than quietly becoming the new +// definition of done. +func (s Status) ValidateLedger(ledger []LedgerItem) error { + if len(s.Ledger) > len(ledger) { + return fmt.Errorf("%w: status records %d items, the ledger carries %d", ErrLedgerRegressed, len(s.Ledger), len(ledger)) + } + for i, st := range s.Ledger { + if ledger[i].ID != st.ID { + return fmt.Errorf("%w: item %d was %s, the ledger now carries %s", ErrLedgerRegressed, i, st.ID, ledger[i].ID) + } + } + for i, it := range ledger { + if it.Item == "" || it.Verify == "" { + return fmt.Errorf("%w: item %d", ErrLedgerIncomplete, i) + } + if want := ItemID(it.Item, it.Verify); it.ID != want { + return fmt.Errorf("%w: item %d carries id %q but addresses to %q", ErrLedgerRegressed, i, it.ID, want) + } + } + return nil +} + +// SyncLedger brings the status's per-item state into line with the spec's ledger: +// every planned item gains an unproven entry, in ledger order, and existing state +// is carried across untouched. It is called before the state is read or written, so +// a ledger that grew mid-run has state for its new items without any of them +// arriving pre-proven. +// +// State for an item no longer on the ledger is dropped rather than kept. That is not +// a way to lose a proof: the extension rule means an item cannot leave the ledger in +// the first place, so reaching this case at all means the ledger was already +// refused, and carrying orphan state forward would only let a proof survive the +// removal of the thing it proved. +func (s *Status) SyncLedger(ledger []LedgerItem) { + if len(ledger) == 0 { + s.Ledger = nil + return + } + by := make(map[string]LedgerState, len(s.Ledger)) + for _, st := range s.Ledger { + by[st.ID] = st + } + out := make([]LedgerState, 0, len(ledger)) + for _, it := range ledger { + if st, ok := by[it.ID]; ok { + out = append(out, st) + continue + } + out = append(out, LedgerState{ID: it.ID}) + } + s.Ledger = out +} + +// MarkProven records an item as proven, with the evidence reference and the time it +// was marked. It reports ErrLedgerUnknownItem for an id the status carries no state +// for, so a claim about work that was never planned is a failure rather than a +// silently accepted new item. Marking an already-proven item again is a no-op that +// keeps the first proof: the record says when the item was first settled, not when +// it was last re-asserted. +func (s *Status) MarkProven(id, evidence string, now time.Time) error { + for i := range s.Ledger { + if s.Ledger[i].ID != id { + continue + } + if s.Ledger[i].Proven { + return nil + } + at := now + s.Ledger[i].Proven = true + s.Ledger[i].ProvenAt = &at + s.Ledger[i].Evidence = evidence + return nil + } + return fmt.Errorf("%w: %q", ErrLedgerUnknownItem, id) +} + +// Unproven returns the ids of the items still unproven, in ledger order. An empty +// result on a non-empty ledger is what "every planned item is settled" looks like as +// a fact about a record rather than a sentence in a transcript. +func (s Status) Unproven() []string { + var out []string + for _, st := range s.Ledger { + if !st.Proven { + out = append(out, st.ID) + } + } + return out +} + +// LedgerSettled reports whether the ledger has items and all of them are proven. An +// empty ledger is not settled: a goal with nothing planned has not finished, it has +// not started. +func (s Status) LedgerSettled() bool { + if len(s.Ledger) == 0 { + return false + } + return len(s.Unproven()) == 0 +} diff --git a/goal/ledger_test.go b/goal/ledger_test.go new file mode 100644 index 0000000..b706ea6 --- /dev/null +++ b/goal/ledger_test.go @@ -0,0 +1,509 @@ +package goal + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/ionalpha/flynn/clock" + "github.com/ionalpha/flynn/jobs" + "github.com/ionalpha/flynn/reconcile" + "github.com/ionalpha/flynn/resource" +) + +// testNow is the fixed instant the pure ledger tests stamp proofs with. The record +// tests do not run a clock, and a real one would make the assertions about when an +// item was proven meaningless. +var testNow = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + +// --- fakes ------------------------------------------------------------------ + +// fakePlanner returns a fixed set of items, or an error. +type fakePlanner struct { + items []LedgerItem + err error + calls int +} + +func (p *fakePlanner) Plan(context.Context, resource.Resource) ([]LedgerItem, error) { + p.calls++ + return p.items, p.err +} + +// neverStop never reports the stop condition met, so a test drives the phases +// itself rather than racing a convergence. +type neverStop struct{} + +func (neverStop) Met(context.Context, Spec, Status) (bool, string, error) { return false, "", nil } + +// planHarness wires a reconciler gated on planning to a worker that can plan, which +// is the pairing WithPlanning documents. +type planHarness struct { + *harness + w *Worker + p *fakePlanner +} + +func newPlanHarness(t *testing.T, stop StopEvaluator, planner *fakePlanner) *planHarness { + t.Helper() + m := clock.NewManual(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) + reg := resource.NewRegistry() + if err := resource.RegisterCoreKinds(reg); err != nil { + t.Fatal(err) + } + if err := RegisterKind(reg); err != nil { + t.Fatal(err) + } + store := resource.NewMemory(reg, resource.WithClock(m)) + q := jobs.NewMemory(jobs.WithClock(m)) + gr := NewReconciler(store, q, m, stop, WithPlanning()) + w := NewWorker(store, q, m, &fakeExec{}, WithPlanner(planner), WithLease(time.Minute)) + h := &harness{ctx: context.Background(), store: store, jobs: q, gr: gr, clk: m} + return &planHarness{harness: h, w: w, p: planner} +} + +func (h *planHarness) spec(t *testing.T, ref reconcile.Ref) Spec { + t.Helper() + r, err := h.store.Get(h.ctx, ref.Kind, ref.Scope, ref.Name) + if err != nil { + t.Fatalf("get goal: %v", err) + } + sp, err := DecodeSpec(r) + if err != nil { + t.Fatal(err) + } + return sp +} + +func item(text, verify string) LedgerItem { + return LedgerItem{ID: ItemID(text, verify), Item: text, Verify: verify} +} + +// --- the record itself ------------------------------------------------------ + +// TestAppendItemsStampsContentAddresses pins that an item's identity is its +// content, which is what makes a rewrite detectable as a rewrite: the planner does +// not get to choose an id, so it cannot give changed text the id of the text it +// replaced. +func TestAppendItemsStampsContentAddresses(t *testing.T) { + got, err := AppendItems(nil, LedgerItem{Item: "add the parser", Verify: "go test ./parser"}) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Fatalf("appended %d items, want 1", len(got)) + } + if want := ItemID("add the parser", "go test ./parser"); got[0].ID != want { + t.Fatalf("id = %q, want the content address %q", got[0].ID, want) + } + // A different verify clause is a different item, not the same item restated. + other, err := AppendItems(got, LedgerItem{Item: "add the parser", Verify: "eyeball it"}) + if err != nil { + t.Fatal(err) + } + if other[1].ID == other[0].ID { + t.Fatal("items differing only in their verify clause share an id") + } +} + +// TestAppendItemsRejectsAnItemWithNoWayToCheckIt is the rule that an item must +// arrive with its own verification. An item with no check can only ever be +// asserted, and an asserted item is exactly what the ledger exists to replace. +func TestAppendItemsRejectsAnItemWithNoWayToCheckIt(t *testing.T) { + for _, tc := range []struct{ name, text, verify string }{ + {"no verify", "add the parser", ""}, + {"blank verify", "add the parser", " "}, + {"no item", "", "go test ./parser"}, + } { + t.Run(tc.name, func(t *testing.T) { + if _, err := AppendItems(nil, LedgerItem{Item: tc.text, Verify: tc.verify}); !errors.Is(err, ErrLedgerIncomplete) { + t.Fatalf("err = %v, want ErrLedgerIncomplete", err) + } + }) + } +} + +func TestAppendItemsRejectsADuplicate(t *testing.T) { + base, err := AppendItems(nil, LedgerItem{Item: "a", Verify: "check a"}) + if err != nil { + t.Fatal(err) + } + if _, err := AppendItems(base, LedgerItem{Item: "a", Verify: "check a"}); !errors.Is(err, ErrLedgerDuplicate) { + t.Fatalf("err = %v, want ErrLedgerDuplicate", err) + } +} + +// TestAppendItemsLeavesTheInputLedgerAlone matters because a rejected append must +// not half-apply: the caller still holds the ledger it had. +func TestAppendItemsLeavesTheInputLedgerAlone(t *testing.T) { + base := []LedgerItem{item("a", "check a")} + if _, err := AppendItems(base, LedgerItem{Item: "b", Verify: ""}); err == nil { + t.Fatal("expected the incomplete item to be refused") + } + if len(base) != 1 || base[0].Item != "a" { + t.Fatalf("input ledger was mutated: %+v", base) + } +} + +// TestValidateExtensionAcceptsOnlyAppends is the append-and-mark-only rule stated +// as a test. Every rejected case here is a way an agent under completion pressure +// could otherwise edit the definition of done: drop an item it did not do, reorder +// so a prefix check passes, or narrow an item's verify clause until the check it +// promised is no longer the check it faces. +func TestValidateExtensionAcceptsOnlyAppends(t *testing.T) { + a, b := item("a", "check a"), item("b", "check b") + prev := []LedgerItem{a, b} + + t.Run("append is allowed", func(t *testing.T) { + next := []LedgerItem{a, b, item("c", "check c")} + if err := ValidateExtension(prev, next); err != nil { + t.Fatalf("append refused: %v", err) + } + }) + t.Run("unchanged is allowed", func(t *testing.T) { + if err := ValidateExtension(prev, []LedgerItem{a, b}); err != nil { + t.Fatalf("unchanged ledger refused: %v", err) + } + }) + t.Run("removal is refused", func(t *testing.T) { + if err := ValidateExtension(prev, []LedgerItem{a}); !errors.Is(err, ErrLedgerRegressed) { + t.Fatalf("err = %v, want ErrLedgerRegressed", err) + } + }) + t.Run("reorder is refused", func(t *testing.T) { + if err := ValidateExtension(prev, []LedgerItem{b, a}); !errors.Is(err, ErrLedgerRegressed) { + t.Fatalf("err = %v, want ErrLedgerRegressed", err) + } + }) + t.Run("rewrite is refused", func(t *testing.T) { + next := []LedgerItem{a, item("b", "just look at it")} + if err := ValidateExtension(prev, next); !errors.Is(err, ErrLedgerRegressed) { + t.Fatalf("err = %v, want ErrLedgerRegressed", err) + } + }) + t.Run("a rewrite wearing the old id is refused", func(t *testing.T) { + // The most direct attack on the rule: keep the id, change what it stands + // for. The self-addressing check is what catches it. + forged := LedgerItem{ID: b.ID, Item: "b", Verify: "just look at it"} + if err := ValidateExtension(prev, []LedgerItem{a, forged}); !errors.Is(err, ErrLedgerRegressed) { + t.Fatalf("err = %v, want ErrLedgerRegressed", err) + } + }) +} + +// TestSyncLedgerStartsEveryItemUnproven is the property that a run begins from a +// record saying nothing is done, and that a ledger which grows mid-run does not +// bring proven items with it. +func TestSyncLedgerStartsEveryItemUnproven(t *testing.T) { + a, b := item("a", "check a"), item("b", "check b") + var st Status + st.SyncLedger([]LedgerItem{a, b}) + if len(st.Ledger) != 2 { + t.Fatalf("synced %d entries, want 2", len(st.Ledger)) + } + for _, e := range st.Ledger { + if e.Proven { + t.Fatalf("item %s arrived proven", e.ID) + } + } + if err := st.MarkProven(a.ID, "go test ./a: ok", testNow); err != nil { + t.Fatal(err) + } + + c := item("c", "check c") + st.SyncLedger([]LedgerItem{a, b, c}) + if len(st.Ledger) != 3 { + t.Fatalf("synced %d entries, want 3", len(st.Ledger)) + } + if !st.Ledger[0].Proven { + t.Fatal("the existing proof was lost by a re-sync") + } + if st.Ledger[2].Proven { + t.Fatal("a newly planned item arrived proven") + } +} + +func TestMarkProvenRefusesAnItemThatWasNeverPlanned(t *testing.T) { + var st Status + st.SyncLedger([]LedgerItem{item("a", "check a")}) + if err := st.MarkProven("deadbeefdeadbeef", "trust me", testNow); !errors.Is(err, ErrLedgerUnknownItem) { + t.Fatalf("err = %v, want ErrLedgerUnknownItem", err) + } +} + +// TestMarkProvenKeepsTheFirstProof: the record says when an item was settled, not +// when it was most recently re-asserted. +func TestMarkProvenKeepsTheFirstProof(t *testing.T) { + a := item("a", "check a") + var st Status + st.SyncLedger([]LedgerItem{a}) + first := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + if err := st.MarkProven(a.ID, "the real run", first); err != nil { + t.Fatal(err) + } + if err := st.MarkProven(a.ID, "a later claim", first.Add(time.Hour)); err != nil { + t.Fatal(err) + } + if got := st.Ledger[0].Evidence; got != "the real run" { + t.Fatalf("evidence = %q, want the first proof", got) + } + if !st.Ledger[0].ProvenAt.Equal(first) { + t.Fatalf("provenAt = %v, want %v", st.Ledger[0].ProvenAt, first) + } +} + +// TestLedgerSettledIsNotTrueForAnEmptyLedger: a goal with nothing planned has not +// finished, it has not started. Getting this backwards is the false-complete on +// iteration one. +func TestLedgerSettledIsNotTrueForAnEmptyLedger(t *testing.T) { + var st Status + if st.LedgerSettled() { + t.Fatal("an empty ledger reported settled") + } + a := item("a", "check a") + st.SyncLedger([]LedgerItem{a}) + if st.LedgerSettled() { + t.Fatal("an unproven ledger reported settled") + } + if err := st.MarkProven(a.ID, "ok", testNow); err != nil { + t.Fatal(err) + } + if !st.LedgerSettled() { + t.Fatal("a fully proven ledger did not report settled") + } + if got := st.Unproven(); len(got) != 0 { + t.Fatalf("unproven = %v, want none", got) + } +} + +// --- the phase -------------------------------------------------------------- + +// TestGoalPlansBeforeItBuilds is the gate: the first job a planning goal dispatches +// is a plan, not a step, and the ledger exists before any build step is claimed. +func TestGoalPlansBeforeItBuilds(t *testing.T) { + p := &fakePlanner{items: []LedgerItem{ + {Item: "add the parser", Verify: "go test ./parser"}, + {Item: "wire the CLI flag", Verify: "flynn parse --help mentions it"}, + }} + h := newPlanHarness(t, neverStop{}, p) + ref := h.createGoal(t, "g", Spec{Objective: "ship the parser", StopCondition: "it parses"}) + + h.reconcile(t, ref) + if st := h.status(t, ref); st.Phase != PhasePlanning { + t.Fatalf("phase = %q, want %q", st.Phase, PhasePlanning) + } + claimed, err := h.jobs.Claim(h.ctx, jobs.ClaimParams{Queue: StepQueue, Limit: 1, LeaseFor: int64(time.Minute)}) + if err != nil || len(claimed) != 1 { + t.Fatalf("no job dispatched (err=%v)", err) + } + if claimed[0].Kind != PlanJobKind { + t.Fatalf("first dispatched job kind = %q, want %q", claimed[0].Kind, PlanJobKind) + } + // Hand it back so the worker can claim and run it for real. + if err := h.jobs.Fail(h.ctx, claimed[0].ID, "released for the worker", h.clk.Now().UnixNano()); err != nil { + t.Fatal(err) + } + if _, err := h.w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + + sp := h.spec(t, ref) + if len(sp.Ledger) != 2 { + t.Fatalf("spec ledger has %d items, want 2: %+v", len(sp.Ledger), sp.Ledger) + } + st := h.status(t, ref) + if !st.Planned { + t.Fatal("goal is not marked planned after the plan step") + } + if len(st.Ledger) != 2 { + t.Fatalf("status ledger has %d entries, want 2", len(st.Ledger)) + } + if got := st.Unproven(); len(got) != 2 { + t.Fatalf("%d items unproven, want all 2", len(got)) + } + + // Only now does a build step get dispatched. + h.reconcile(t, ref) + claimed, err = h.jobs.Claim(h.ctx, jobs.ClaimParams{Queue: StepQueue, Limit: 1, LeaseFor: int64(time.Minute)}) + if err != nil || len(claimed) != 1 { + t.Fatalf("no build job dispatched (err=%v)", err) + } + if claimed[0].Kind != StepJobKind { + t.Fatalf("second dispatched job kind = %q, want %q", claimed[0].Kind, StepJobKind) + } +} + +// TestPlanningDoesNotSpendTheBuildBudget: planning decides what the budget is spent +// on, so charging the budget for it would make a goal that plans strictly poorer +// than one that does not, and a tight MaxSteps would leave nothing to build with. +func TestPlanningDoesNotSpendTheBuildBudget(t *testing.T) { + p := &fakePlanner{items: []LedgerItem{{Item: "a", Verify: "check a"}}} + h := newPlanHarness(t, neverStop{}, p) + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c", MaxSteps: 1}) + + h.reconcile(t, ref) // dispatch the plan + if _, err := h.w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + h.reconcile(t, ref) // observe the plan, then dispatch the first build step + + st := h.status(t, ref) + if st.Steps != 0 { + t.Fatalf("steps = %d after planning, want 0", st.Steps) + } + if st.Phase != PhaseRunning { + t.Fatalf("phase = %q, want %q: the one step of budget was spent on planning", st.Phase, PhaseRunning) + } +} + +// TestAnEmptyPlanStallsTheGoal: a planner that produced nothing leaves no +// definition of done. Building anyway is how a run claims success against a record +// that never said what success was. +func TestAnEmptyPlanStallsTheGoal(t *testing.T) { + h := newPlanHarness(t, neverStop{}, &fakePlanner{}) + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c"}) + + h.reconcile(t, ref) + if _, err := h.w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + h.reconcile(t, ref) + + st := h.status(t, ref) + if st.Phase != PhaseStalled { + t.Fatalf("phase = %q, want %q", st.Phase, PhaseStalled) + } + if st.Message != "planning produced an empty ledger" { + t.Fatalf("message = %q", st.Message) + } +} + +// TestPlanningRunsOnce pins that the planning mark, not the ledger's emptiness, is +// what ends the phase, and that a re-reconcile does not plan the goal again. +func TestPlanningRunsOnce(t *testing.T) { + p := &fakePlanner{items: []LedgerItem{{Item: "a", Verify: "check a"}}} + h := newPlanHarness(t, neverStop{}, p) + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c"}) + + for range 4 { + h.reconcile(t, ref) + if _, err := h.w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + } + if p.calls != 1 { + t.Fatalf("planner ran %d times, want 1", p.calls) + } +} + +// TestAnEditedLedgerStallsTheGoal is the reconcile-time half of the extension rule: +// a ledger edited around the write path is refused rather than adopted as the new +// definition of done. The goal settles as stalled with the cause, because a run +// whose record was tampered with has failed, not paused. +func TestAnEditedLedgerStallsTheGoal(t *testing.T) { + p := &fakePlanner{items: []LedgerItem{ + {Item: "a", Verify: "check a"}, + {Item: "b", Verify: "check b"}, + }} + h := newPlanHarness(t, neverStop{}, p) + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c"}) + h.reconcile(t, ref) + if _, err := h.w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + h.reconcile(t, ref) + + // Drop the second item straight onto the store, the way an agent with a write + // tool and a deadline would. + r, err := h.store.Get(h.ctx, ref.Kind, ref.Scope, ref.Name) + if err != nil { + t.Fatal(err) + } + sp, err := DecodeSpec(r) + if err != nil { + t.Fatal(err) + } + sp.Ledger = sp.Ledger[:1] + raw, err := json.Marshal(sp) + if err != nil { + t.Fatal(err) + } + r.Spec = raw + if _, err := h.store.Put(h.ctx, r); err != nil { + t.Fatal(err) + } + + h.reconcile(t, ref) + st := h.status(t, ref) + if st.Phase != PhaseStalled { + t.Fatalf("phase = %q, want %q", st.Phase, PhaseStalled) + } + if !containsAll(st.Message, "ledger", "regressed") { + t.Fatalf("message does not name the cause: %q", st.Message) + } +} + +// TestAPlanJobWithNoPlannerFailsTerminally: a reconciler gated on planning wired to +// a worker that cannot plan is a misconfiguration no retry fixes, and the goal would +// otherwise sit unplanned and undispatched with nothing saying why. +func TestAPlanJobWithNoPlannerFailsTerminally(t *testing.T) { + m := clock.NewManual(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) + reg := resource.NewRegistry() + if err := resource.RegisterCoreKinds(reg); err != nil { + t.Fatal(err) + } + if err := RegisterKind(reg); err != nil { + t.Fatal(err) + } + store := resource.NewMemory(reg, resource.WithClock(m)) + q := jobs.NewMemory(jobs.WithClock(m)) + gr := NewReconciler(store, q, m, neverStop{}, WithPlanning()) + w := NewWorker(store, q, m, &fakeExec{}, WithLease(time.Minute)) // no planner + h := &harness{ctx: context.Background(), store: store, jobs: q, gr: gr, clk: m} + + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c"}) + h.reconcile(t, ref) + if _, err := w.ProcessOnce(h.ctx); err != nil { + t.Fatalf("worker: %v", err) + } + h.reconcile(t, ref) + + st := h.status(t, ref) + if st.Phase != PhaseStalled { + t.Fatalf("phase = %q, want %q", st.Phase, PhaseStalled) + } + if !containsAll(st.Message, "planner") { + t.Fatalf("message does not name the missing planner: %q", st.Message) + } +} + +// TestAGoalWithoutPlanningIsUnchanged: the gate is opt-in, so every goal composed +// before planning existed still dispatches a build step first and never waits on a +// ledger it has no way to produce. +func TestAGoalWithoutPlanningIsUnchanged(t *testing.T) { + h := newHarness(t, stopAfter{at: 1}) + ref := h.createGoal(t, "g", Spec{Objective: "o", StopCondition: "c"}) + h.reconcile(t, ref) + if st := h.status(t, ref); st.Phase != PhaseRunning { + t.Fatalf("phase = %q, want %q", st.Phase, PhaseRunning) + } + h.completeStep(t) // asserts the job kind is a build step +} + +func containsAll(s string, subs ...string) bool { + for _, sub := range subs { + found := false + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + found = true + break + } + } + if !found { + return false + } + } + return true +} diff --git a/goal/reconciler.go b/goal/reconciler.go index 43053da..f8c790a 100644 --- a/goal/reconciler.go +++ b/goal/reconciler.go @@ -57,6 +57,7 @@ type Reconciler struct { stop StopEvaluator cleaner Cleaner bus bus.Bus // optional; nil disables owner wake signals + planning bool poll time.Duration waitRecheck time.Duration // 0 derives DefaultWaitRecheckFactor * poll stepTries int @@ -87,6 +88,17 @@ func WithWaitRecheck(d time.Duration) Option { } } +// WithPlanning makes the goal plan before it builds: the first thing a goal does is +// a planning step that expands its objective into a ledger, and no build step is +// dispatched until that ledger exists. It pairs with a Worker configured with a +// Planner, which is what actually runs the planning step. +// +// It is an option rather than the unconditional behaviour because a goal composed +// without a planner has no way to produce a ledger, and gating those goals would +// park every one of them forever. Wiring the planner and turning this on are the +// same decision, made in the same place. +func WithPlanning() Option { return func(g *Reconciler) { g.planning = true } } + // WithWakeBus sets the bus the reconciler signals a parked owner on when one of // its children settles, so a fan-out parent re-checks on child state-change // instead of waiting out the recheck fallback. @@ -238,9 +250,17 @@ func (g *Reconciler) reconcile(ctx context.Context, ref reconcile.Ref) (reconcil return reconcile.Result{}, fault.Wrap(fault.Terminal, "goal_status_decode", err) } + // A ledger that lost or rewrote an item between reconciles is the definition of + // done being edited mid-run, so the goal fails rather than adopting the edit. + if err := status.ValidateLedger(spec.Ledger); err != nil { + return reconcile.Result{}, fault.Wrap(fault.Terminal, "goal_ledger_regressed", err) + } + status.SyncLedger(spec.Ledger) + // Observe an in-flight step. observed := false if status.InFlight != nil { + inFlightKind := status.InFlight.Kind job, err := g.jobs.Get(ctx, status.InFlight.JobID) switch { case err != nil: @@ -261,7 +281,10 @@ func (g *Reconciler) reconcile(ctx context.Context, ref reconcile.Ref) (reconcil // A step that parked the goal (ErrWaiting) made no progress, so it // does not count against the step budget: a fan-out whose children // outlast the budget's worth of re-checks must wait, not false-stall. - if status.WaitingSince == nil { + // A planning step is not building either: it is the phase that decides + // what the build budget will be spent on, so charging the budget for it + // would make a goal that plans strictly poorer than one that does not. + if status.WaitingSince == nil && inFlightKind != PlanJobKind { status.Steps++ } } @@ -286,6 +309,23 @@ func (g *Reconciler) reconcile(ctx context.Context, ref reconcile.Ref) (reconcil status.WaitingSince = nil // fallback elapsed with no wake: re-check now } + // Planning gate. A goal that plans expands its objective into a ledger before it + // builds anything, so the first dispatch is a planning step and the stop + // condition is not evaluated until there is a record to evaluate it against. + if g.planning && !status.Planned { + return g.dispatch(ctx, r, status, specHash, PlanJobKind, PhasePlanning, "PlanDispatched") + } + // A planner that ran and produced nothing leaves a goal with no definition of + // done, which is a stall. Letting it build anyway is how a run ends up claiming + // success against a record that never said what success was. + if g.planning && len(spec.Ledger) == 0 { + status.Phase = PhaseStalled + status.Message = "planning produced an empty ledger" + status.SetCondition(Condition{Type: CondStalled, Status: "True", Reason: "EmptyLedger", Message: status.Message}, g.clk.Now()) + status.SetCondition(Condition{Type: CondReconciling, Status: "False", Reason: "Stalled"}, g.clk.Now()) + return g.terminal(ctx, r, status, specHash) + } + // Converged? met, reason, err := g.stop.Met(ctx, spec, status) if err != nil { @@ -309,9 +349,18 @@ func (g *Reconciler) reconcile(ctx context.Context, ref reconcile.Ref) (reconcil } // Dispatch the next step and record it in flight. + return g.dispatch(ctx, r, status, specHash, StepJobKind, PhaseRunning, "StepDispatched") +} + +// dispatch enqueues one job of the given kind against the goal and records it in +// flight under the phase that job puts the goal in. Planning and building share it +// so a planning step gets the same reservation, lease, crash-resume and retry +// behaviour a build step has, and the only thing that differs between them is what +// the executor is asked to do. +func (g *Reconciler) dispatch(ctx context.Context, r resource.Resource, status Status, specHash, kind string, phase Phase, reason string) (reconcile.Result, error) { job, err := g.jobs.Enqueue(ctx, jobs.EnqueueParams{ Queue: StepQueue, - Kind: StepJobKind, + Kind: kind, Payload: []byte(r.ID), Scope: state.Scope(r.Scope), MaxAttempts: g.stepTries, @@ -319,9 +368,9 @@ func (g *Reconciler) reconcile(ctx context.Context, ref reconcile.Ref) (reconcil if err != nil { return reconcile.Result{}, putErr(err) } - status.Phase = PhaseRunning - status.InFlight = &InFlight{JobID: job.ID, StartedAt: g.clk.Now()} - status.SetCondition(Condition{Type: CondReconciling, Status: "True", Reason: "StepDispatched"}, g.clk.Now()) + status.Phase = phase + status.InFlight = &InFlight{JobID: job.ID, StartedAt: g.clk.Now(), Kind: kind} + status.SetCondition(Condition{Type: CondReconciling, Status: "True", Reason: reason}, g.clk.Now()) if err := g.recordDispatch(ctx, r, status, specHash); err != nil { return reconcile.Result{}, err } @@ -375,6 +424,13 @@ func (g *Reconciler) recordDispatch(ctx context.Context, r resource.Resource, st } status.Checkpoint = cur.Checkpoint status.WaitingSince = cur.WaitingSince + // The planning mark and the per-item state are the worker's too, and the + // planning step is the one most likely to land inside this window: the plan + // job is enqueued before this reservation is written, so a worker that + // claims and finishes it first would otherwise have its ledger erased here + // and be asked to plan the same goal all over again. + status.Planned = cur.Planned + status.Ledger = cur.Ledger enc, err := status.Encode() if err != nil { return err diff --git a/goal/worker.go b/goal/worker.go index 185a8e0..eba268d 100644 --- a/goal/worker.go +++ b/goal/worker.go @@ -47,6 +47,21 @@ type StepExecutor interface { Execute(ctx context.Context, goal resource.Resource) (checkpoint json.RawMessage, err error) } +// Planner expands a goal's objective into the ledger of work that objective +// implies, each item carrying its own declared way to verify it. It runs once, +// before any building, and it is a genuinely different prompt from a build step +// rather than the build prompt with a plan instruction attached: the whole first +// context window goes on working out what the objective actually requires, which is +// not something a step also trying to make progress will do well. +// +// A planner that returns no items is not an error here. It is recorded as planned +// with an empty ledger, and the reconciler stalls the goal, so "the objective could +// not be expanded" surfaces as a settled goal with a reason rather than a retry +// loop. +type Planner interface { + Plan(ctx context.Context, goal resource.Resource) ([]LedgerItem, error) +} + // ErrWaiting is returned by a StepExecutor whose step made no progress because it // is waiting on external state, such as a fan-out whose children are still // running. The worker completes the job without persisting a checkpoint and stamps @@ -65,6 +80,7 @@ type Worker struct { store resource.Store jobs jobs.Queue exec StepExecutor + planner Planner // optional; nil means a plan job is refused rather than guessed at clk clock.Timing bus bus.Bus // optional; nil disables completion signals lease time.Duration @@ -78,6 +94,10 @@ type WorkerOption func(*Worker) // WithBus sets the bus a worker publishes step-completion signals on. func WithBus(b bus.Bus) WorkerOption { return func(w *Worker) { w.bus = b } } +// WithPlanner sets the planner that runs a goal's planning step. It pairs with the +// reconciler's WithPlanning, which is what makes a goal plan before it builds. +func WithPlanner(p Planner) WorkerOption { return func(w *Worker) { w.planner = p } } + // WithLease overrides the step lease duration. func WithLease(d time.Duration) WorkerOption { return func(w *Worker) { @@ -189,6 +209,9 @@ func (w *Worker) runStep(ctx context.Context, job jobs.Job) error { if r.DeletionTimestamp != nil { return w.jobs.Complete(ctx, job.ID) // terminating; stop working on it } + if job.Kind == PlanJobKind { + return w.runPlan(ctx, job, r) + } checkpoint, err := w.exec.Execute(ctx, r) if errors.Is(err, ErrWaiting) { @@ -212,6 +235,76 @@ func (w *Worker) runStep(ctx context.Context, job jobs.Job) error { return nil } +// runPlan executes a goal's planning step: expand the objective into a ledger and +// record it on the goal before any building is dispatched. +// +// The ledger write is not best-effort the way a checkpoint is. A checkpoint that +// fails to land costs one repeated turn; a ledger that fails to land leaves the goal +// unplanned, and the reconciler would dispatch planning again on the next pass, +// forever. So a failed write fails the job instead, which puts it on the retry +// ladder and, if it keeps failing, stalls the goal with the cause attached. +func (w *Worker) runPlan(ctx context.Context, job jobs.Job, r resource.Resource) error { + if w.planner == nil { + // A goal was gated on planning by a reconciler wired to a worker that cannot + // plan. No number of retries fixes a missing port, and the goal would + // otherwise sit unplanned and undispatched with nothing saying why. + return w.fail(ctx, job, fault.Wrap(fault.Terminal, "goal_no_planner", errors.New("goal: plan step dispatched to a worker with no planner"))) + } + items, err := w.planner.Plan(ctx, r) + if err != nil { + return w.fail(ctx, job, err) + } + if err := w.recordPlan(ctx, r, items); err != nil { + return w.fail(ctx, job, err) + } + if err := w.jobs.Complete(ctx, job.ID); err != nil { + return err // crashed before completing: the lease lapses and planning re-runs + } + w.signal(ctx, r) + return nil +} + +// recordPlan writes the planned items onto the goal's spec and marks the goal +// planned, against a fresh read under the shared conflict-retry policy. The append +// rule is enforced here at the point of the write: the planner's items are appended +// to whatever ledger the goal already carries, so re-running a planning step after a +// crash adds to the record rather than replacing it, and a planner that tries to +// restate an existing item differently is refused instead of quietly redefining it. +func (w *Worker) recordPlan(ctx context.Context, r resource.Resource, items []LedgerItem) error { + _, err := resource.UpdateByID(ctx, w.store, r.ID, func(fresh *resource.Resource) error { + spec, err := DecodeSpec(*fresh) + if err != nil { + return err + } + status, err := DecodeStatus(*fresh) + if err != nil { + return err + } + ledger, err := AppendItems(spec.Ledger, items...) + if err != nil { + return fault.Wrap(fault.Terminal, "goal_plan_invalid", err) + } + if err := ValidateExtension(spec.Ledger, ledger); err != nil { + return fault.Wrap(fault.Terminal, "goal_plan_invalid", err) + } + spec.Ledger = ledger + encSpec, err := json.Marshal(spec) + if err != nil { + return err + } + status.Planned = true + status.SyncLedger(ledger) + encStatus, err := status.Encode() + if err != nil { + return err + } + fresh.Spec = encSpec + fresh.Status = encStatus + return nil + }) + return err +} + // persistCheckpoint records the step's progress on the goal's status so the next // step resumes from it. A version conflict does NOT mean the checkpoint may be // dropped: the reconciler dispatches the job before it persists the in-flight