From 467589d78afc4fc13b68a155f774b0fa0313d2a7 Mon Sep 17 00:00:00 2001 From: dtoro Date: Thu, 30 Jul 2026 22:40:56 +0200 Subject: [PATCH] fix(nomos): generation-relative plan seq + real activity timestamps The plan recorded false history after a re-plan and the activity panel showed fabricated, churning timestamps. Two bugs compounding on one event stream. Plan drift (P0.1): - proposePlan seq is now 1..N per generation; (session,generation,seq) is the addressing key. The model's 1-based update_plan_step calls always map to the CURRENT plan after a re-plan, instead of resurrecting a superseded `replaced` row as done while the live work went unrecorded. - updatePlanStep resolves against MAX(generation); a stale/out-of-range seq returns errPlanStepNotFound (never touches a superseded generation). - getPlanSteps returns only the current generation by default; ?all=true keeps the audit/eval view (plan_generations assertion). - completeTask auto-close scopes to the current gen, stamps started_at, and emits one plan.step.finished per closed step so the panel converges instead of freezing on "running" after completion (P1.1). - propose_plan result enumerates step seqs; writeback detector matches "write back"/"writeback"/"upsert_knowledge" so a natural-language final step isn't doubled (P1.2). - migration 029 renumbers existing seq per generation + unique index. Activity panel (P0.2 / P1.1, web): - computeActivityLog uses the real message created_at for tool calls; live entries fall back to wall-clock frozen on first sight, killing the 3s poll churn. Steps use real started_at. - dropped plan-step events warn + count instead of a silent no-op. Tests: TestProposePlan updated; + generation-relative-seq and auto-close event-emission regression tests; + web activity purity/timestamp tests. VERSION: 0.14.0 -> 0.14.1 --- VERSION | 2 +- cmd/nomos/eval/main.go | 6 +- cmd/nomos/main.go | 3 +- cmd/nomos/store.go | 152 +++++++++++++----- cmd/nomos/store_test.go | 144 ++++++++++++++++- cmd/nomos/tasks.go | 38 ++++- .../029_plan_generation_relative_seq.up.sql | 33 ++++ web/src/lib/stores/activity.test.ts | 103 ++++++++++++ web/src/lib/stores/activity.ts | 65 ++++++-- web/src/lib/stores/workspace.ts | 20 ++- 10 files changed, 499 insertions(+), 67 deletions(-) create mode 100644 migrations/029_plan_generation_relative_seq.up.sql create mode 100644 web/src/lib/stores/activity.test.ts diff --git a/VERSION b/VERSION index a803cc2..930e300 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.14.0 +0.14.1 diff --git a/cmd/nomos/eval/main.go b/cmd/nomos/eval/main.go index 8dfc631..b2a848d 100644 --- a/cmd/nomos/eval/main.go +++ b/cmd/nomos/eval/main.go @@ -319,8 +319,10 @@ func fetchTranscript(ctx context.Context, gateway, sid string) (transcript, sess } // Fetch the plan (steps with generation numbers) for the // plan_generations assertion. A 404 or empty response is fine — a - // pure-DB Q&A with no propose_plan has no plan. - if planResp, perr := http.Get(gateway + "/sessions/" + sid + "/plan"); perr == nil { + // pure-DB Q&A with no propose_plan has no plan. ?all=true returns every + // generation so the assertion can count them (the default view returns + // only the current generation). + if planResp, perr := http.Get(gateway + "/sessions/" + sid + "/plan?all=true"); perr == nil { if planResp.StatusCode == 200 { pb, _ := io.ReadAll(planResp.Body) _ = json.Unmarshal(pb, &t) // fills t.PlanSteps via "steps" field diff --git a/cmd/nomos/main.go b/cmd/nomos/main.go index 8707020..a8b8633 100644 --- a/cmd/nomos/main.go +++ b/cmd/nomos/main.go @@ -483,7 +483,8 @@ func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store, a *a if len(parts) == 2 && r.Method == http.MethodGet { switch parts[1] { case "plan": - steps, err := st.getPlanSteps(r.Context(), id) + all := r.URL.Query().Has("all") && r.URL.Query().Get("all") != "0" && r.URL.Query().Get("all") != "false" + steps, err := st.getPlanSteps(r.Context(), id, all) if err != nil { http.Error(w, err.Error(), 500) return diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index 6139742..093738e 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -13,6 +13,7 @@ import ( "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -25,6 +26,13 @@ const maxToolResultSize = 4096 // The caller translates this into a directive tool result. var errPlanInFlight = errors.New("plan already in flight") +// errPlanStepNotFound is returned by updatePlanStep when no step matches the +// given seq in the CURRENT (MAX) generation — either the seq is out of range, +// or (after a re-plan) the model addressed a stale 1-based number. seq is +// generation-relative, so this never resurrects a superseded generation's row. +// The caller translates it into a directive tool result (P0.1). +var errPlanStepNotFound = errors.New("plan step not found in current generation") + type store struct { pool *pgxpool.Pool } @@ -879,16 +887,15 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS } defer tx.Rollback(ctx) - var startSeq int var anyStarted bool // `replaced` steps (from a prior plan generation superseded by a - // follow-up sub-task — see reopenSession) are excluded: they prove a - // prior plan was completed and superseded, not that a plan is in flight. - // Without this exclusion, reopenSession's `replaced` marking would be + // follow-up sub-task — see setGoal/reopenSession) are excluded: they + // prove a prior plan was completed and superseded, not that a plan is in + // flight. Without this exclusion, setGoal's `replaced` marking would be // useless — propose_plan would still refuse on the follow-up. if err := tx.QueryRow(ctx, ` - SELECT COALESCE(max(seq), 0), COALESCE(bool_or(status NOT IN ('pending', 'replaced')), false) - FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&startSeq, &anyStarted); err != nil { + SELECT COALESCE(bool_or(status NOT IN ('pending', 'replaced')), false) + FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&anyStarted); err != nil { return nil, err } if anyStarted { @@ -898,35 +905,27 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS return nil, errPlanInFlight } // Fresh/revise: mark any prior PENDING steps as `replaced` (not DELETE). - // This preserves the rows for the generation counter (MAX(generation)+1 - // below) and the plan_generations eval assertion. Without this, a first - // plan that was proposed but never executed (all pending) would be - // wiped, resetting the counter to 1 — making a follow-up's plan look - // like generation 1 instead of 2. `replaced` steps are excluded from - // the anyStarted check above, so they don't block the fresh proposal. + // The rows are kept for the generation counter (MAX(generation)+1 below) + // and the plan_generations eval assertion. `replaced` steps are excluded + // from the anyStarted check above, so they don't block this proposal. if _, err := tx.Exec(ctx, `UPDATE session_plan_steps SET status = 'replaced', finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status = 'pending'`, sessionID); err != nil { return nil, err } - // startSeq keeps the max(seq) from the query above: if prior steps - // exist (replaced or done), the new generation's steps start after them - // (no seq collisions across generations). If no rows exist (first plan), - // startSeq is 0 and the first step is seq 1. - // Resolve the generation number for this plan. Generation 1 is the - // initial plan; a genuine revise (which currently goes through the same - // fresh-start path above because all steps were pending) resets to 1 - // since the DELETE wiped the prior rows. The column is wired here so a - // future explicit mid-flight revise path can increment it. + // nextGen: generation 1 for the first plan, MAX(generation)+1 for every + // revise/follow-up (prior rows were marked `replaced` above, not deleted, + // so the counter survives). seq is generation-relative — it resets to + // 1..N for this generation, so (session_id, generation, seq) is the + // addressing key and the model's 1-based update_plan_step always maps to + // the CURRENT plan after a re-plan (P0.1). var nextGen int if err := tx.QueryRow(ctx, ` SELECT COALESCE(MAX(generation), 0) + 1 FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&nextGen); err != nil { return nil, err } - // After the DELETE above, no rows remain, so MAX(generation) is NULL → - // nextGen = 1. (Keep the query for the future revise path; it's cheap.) out := make([]map[string]any, 0, len(steps)) for i, st := range steps { @@ -934,7 +933,7 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS if st.TargetSlug != "" { targetSlug = &st.TargetSlug } - seq := startSeq + i + 1 + seq := i + 1 var id uuid.UUID if err := tx.QueryRow(ctx, ` INSERT INTO session_plan_steps (session_id, seq, title, detail, target_slug, generation) @@ -977,6 +976,23 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s if s == nil || sessionID == "" || sessionID == "ephemeral" { return nil } + // Resolve the CURRENT generation: seq is generation-relative (1-based + // within the plan the model is working), so (session_id, MAX(generation), + // seq) is the addressing key. A re-plan's superseded generations have + // their own seq space and must never be touched by a follow-up's + // update_plan_step — that was the root cause of "the plan was off" + // (gen-1 `replaced` rows resurrected as `done` while gen-2 work went + // unrecorded). The MAX(generation) step is by construction the active + // plan, never `replaced`, so this can't resurrect a superseded row (P0.1). + var curGen int + if err := s.pool.QueryRow(ctx, + `SELECT COALESCE(MAX(generation), 0) FROM session_plan_steps WHERE session_id = $1`, + sessionID).Scan(&curGen); err != nil { + return err + } + if curGen == 0 { + return errPlanStepNotFound + } stamp := "" switch status { case "running": @@ -984,16 +1000,18 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s case "done", "failed", "skipped", "blocked", "replaced": stamp = ", finished_at = now()" } - // Completion ordering: for terminal states, check that no earlier step - // is still pending. Running steps can start out of order (the agent - // may dispatch parallel work), but completion must be sequential. + // Completion ordering, scoped to the CURRENT generation: for terminal + // states, no earlier step in THIS plan may still be pending. Running + // steps can start out of order (the agent may dispatch parallel work), + // but completion must be sequential. Earlier generations are superseded + // and irrelevant. if status == "done" || status == "failed" || status == "skipped" || status == "blocked" { var blockedBy int if err := s.pool.QueryRow(ctx, ` SELECT COALESCE(MIN(seq), 0) FROM session_plan_steps - WHERE session_id = $1 AND seq < $2 AND status = 'pending'`, - sessionID, seq).Scan(&blockedBy); err == nil && blockedBy > 0 { + WHERE session_id = $1 AND generation = $2 AND seq < $3 AND status = 'pending'`, + sessionID, curGen, seq).Scan(&blockedBy); err == nil && blockedBy > 0 { return fmt.Errorf("cannot complete step %d — step %d is still pending", seq, blockedBy) } } @@ -1004,11 +1022,18 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s var stepID uuid.UUID var targetSlug *string // stamp is a fixed literal from the switch above — never user input. - if err := s.pool.QueryRow(ctx, ` + // status <> 'replaced' is defense-in-depth: MAX(generation) can't hold a + // replaced row, but if it ever could, this refuses the write instead of + // resurrecting it. No matching row → errPlanStepNotFound (stale/out-of-range seq). + err := s.pool.QueryRow(ctx, ` UPDATE session_plan_steps - SET status = $3, execution_id = COALESCE($4, execution_id)`+stamp+` - WHERE session_id = $1 AND seq = $2 - RETURNING id, target_slug`, sessionID, seq, status, execPtr).Scan(&stepID, &targetSlug); err != nil { + SET status = $4, execution_id = COALESCE($5, execution_id)`+stamp+` + WHERE session_id = $1 AND generation = $2 AND seq = $3 AND status <> 'replaced' + RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr).Scan(&stepID, &targetSlug) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return errPlanStepNotFound + } return err } // Anchor the event to the step's target entity when it has one, else the task. @@ -1098,13 +1123,58 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st if outcome != "success" { closeStatus = "skipped" } + // Auto-close only the CURRENT generation's in-flight steps — superseded + // generations were already resolved when their plan was replaced. Stamp + // started_at so no `done` step is left with a NULL start time (P0.1 fix + // 5), and emit a plan.step.finished event per closed step so the panel + // converges instead of freezing on "running" after the task completes + // (P1.1: no bulk plan-step status write without a corresponding event). + type closingStep struct { + id uuid.UUID + seq int + targetSlug *string + } + var toClose []closingStep + if rows, qerr := s.pool.Query(ctx, ` + SELECT id, seq, target_slug FROM session_plan_steps + WHERE session_id = $1 + AND generation = (SELECT MAX(generation) FROM session_plan_steps WHERE session_id = $1) + AND status IN ('pending', 'running')`, sessionID); qerr == nil { + for rows.Next() { + var cs closingStep + if err := rows.Scan(&cs.id, &cs.seq, &cs.targetSlug); err == nil { + toClose = append(toClose, cs) + } + } + rows.Close() + } if _, err := s.pool.Exec(ctx, ` UPDATE session_plan_steps - SET status = $2, finished_at = COALESCE(finished_at, now()) - WHERE session_id = $1 AND status IN ('pending', 'running')`, + SET status = $2, + started_at = COALESCE(started_at, now()), + finished_at = COALESCE(finished_at, now()) + WHERE session_id = $1 + AND generation = (SELECT MAX(generation) FROM session_plan_steps WHERE session_id = $1) + AND status IN ('pending', 'running')`, sessionID, closeStatus); err != nil { slog.Warn("nomos: completeTask failed to auto-close in-flight steps", "session", sessionID, "error", err) } + // Emit one plan.step.finished per closed step so the live panel advances + // (mirrors updatePlanStep's event). A bulk UPDATE that skips the event + // bus guarantees a stale panel — the rule is: no plan-step status change + // without a corresponding event. + taskEnt := s.taskEntityPtr(ctx, sessionID) + for _, cs := range toClose { + evEnt := taskEnt + if cs.targetSlug != nil && *cs.targetSlug != "" { + var tid uuid.UUID + if s.pool.QueryRow(ctx, `SELECT id FROM entities WHERE slug = $1`, *cs.targetSlug).Scan(&tid) == nil { + evEnt = &tid + } + } + _ = observability.Event(ctx, sqlcgen.New(s.pool), "plan.step.finished", evEnt, "info", "nomos", sessionID, + map[string]any{"step_id": cs.id.String(), "seq": cs.seq, "status": closeStatus}) + } // Clean up assent and destructive window keys from autonomy_settings. s.pool.Exec(ctx, `DELETE FROM autonomy_settings @@ -1347,15 +1417,23 @@ type planStep struct { // getPlanSteps returns a task's plan in order — REST hydration for the context // panel when it first opens a task (live events only carry deltas from then on). -func (s *store) getPlanSteps(ctx context.Context, sessionID string) ([]planStep, error) { +// By default only the CURRENT (MAX) generation is returned — the panel shows the +// live plan, not an archaeological record of every superseded generation. Pass +// all=true for the audit/eval view that needs every generation (the +// plan_generations assertion counts distinct generations across the full set). +func (s *store) getPlanSteps(ctx context.Context, sessionID string, all bool) ([]planStep, error) { if s == nil { return nil, nil } + genFilter := "" + if !all { + genFilter = "AND generation = (SELECT MAX(generation) FROM session_plan_steps WHERE session_id = $1)" + } rows, err := s.pool.Query(ctx, ` SELECT id::text, seq, title, detail, status, execution_id::text, target_slug, started_at::text, finished_at::text, generation - FROM session_plan_steps WHERE session_id = $1 ORDER BY seq`, sessionID) + FROM session_plan_steps WHERE session_id = $1 `+genFilter+` ORDER BY generation, seq`, sessionID) if err != nil { return nil, err } diff --git a/cmd/nomos/store_test.go b/cmd/nomos/store_test.go index d12d99d..2ed6da5 100644 --- a/cmd/nomos/store_test.go +++ b/cmd/nomos/store_test.go @@ -205,7 +205,7 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { } // The original step 1 must be untouched — not erased, not appended to. - steps, err := s.getPlanSteps(ctx, sess.ID) + steps, err := s.getPlanSteps(ctx, sess.ID, false) if err != nil { t.Fatalf("getPlanSteps: %v", err) } @@ -217,7 +217,8 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { } // Third call BEFORE anything runs on a fresh session: every step is - // still pending, so this must REPLACE, not refuse. + // still pending, so this must REPLACE (mark the prior plan `replaced`), + // not refuse. The new plan becomes generation 2. sess2, err := s.createSession(ctx, "plan replace test") if err != nil { t.Fatalf("createSession: %v", err) @@ -228,15 +229,146 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { if _, err := s.proposePlan(ctx, sess2.ID, []planStepInput{{Title: "Revised"}}); err != nil { t.Fatalf("proposePlan (revise before execution): %v", err) } - revisedSteps, err := s.getPlanSteps(ctx, sess2.ID) + // Default (current generation) view: only the revised step. + revisedSteps, err := s.getPlanSteps(ctx, sess2.ID, false) if err != nil { t.Fatalf("getPlanSteps: %v", err) } if len(revisedSteps) != 1 || revisedSteps[0].Title != "Revised" { - t.Fatalf("got %+v, want a single 'Revised' step (pre-execution revise must replace, not refuse)", revisedSteps) + t.Fatalf("got %+v, want a single 'Revised' step (current-generation view)", revisedSteps) } - if revisedSteps[0].Generation != 1 { - t.Fatalf("revised step generation = %d, want 1 (fresh-start after DELETE resets generation)", revisedSteps[0].Generation) + if revisedSteps[0].Seq != 1 { + t.Fatalf("revised step seq = %d, want 1 (seq is generation-relative, resets to 1..N)", revisedSteps[0].Seq) + } + if revisedSteps[0].Generation != 2 { + t.Fatalf("revised step generation = %d, want 2 (prior pending plan is replaced, not deleted, so the counter increments)", revisedSteps[0].Generation) + } + // all=true audit view: both generations, the original marked `replaced`. + allSteps, err := s.getPlanSteps(ctx, sess2.ID, true) + if err != nil { + t.Fatalf("getPlanSteps(all): %v", err) + } + if len(allSteps) != 2 { + t.Fatalf("all=true got %d steps, want 2 (Original replaced gen1 + Revised gen2)", len(allSteps)) + } + if allSteps[0].Title != "Original" || allSteps[0].Status != "replaced" || allSteps[0].Generation != 1 { + t.Errorf("gen1 step = %+v, want Original/replaced/gen1", allSteps[0]) + } + if allSteps[1].Title != "Revised" || allSteps[1].Generation != 2 || allSteps[1].Seq != 1 { + t.Errorf("gen2 step = %+v, want Revised/gen2/seq1", allSteps[1]) + } +} + +// TestUpdatePlanStep_GenerationRelative is the P0.1 regression proof: after a +// re-plan, update_plan_step(seq=N) — using the 1-based number the model +// naturally carries — must address the CURRENT generation and never resurrect +// a superseded generation's `replaced` row. Before the fix, seq was globally +// increasing across generations, so seq=1 after a re-plan flipped the gen-1 +// `replaced` step back to `running`/`done` while the real gen-2 work went +// unrecorded. +func TestUpdatePlanStep_GenerationRelative(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + sess, err := s.createSession(ctx, "gen-relative seq test") + if err != nil { + t.Fatalf("createSession: %v", err) + } + // Generation 1: two steps. + if _, err := s.proposePlan(ctx, sess.ID, []planStepInput{{Title: "A"}, {Title: "B"}}); err != nil { + t.Fatalf("proposePlan #1: %v", err) + } + // Re-plan: setGoal marks the gen-1 plan `replaced`, proposePlan starts gen 2. + if err := s.setGoal(ctx, sess.ID, "follow-up sub-task"); err != nil { + t.Fatalf("setGoal: %v", err) + } + if _, err := s.proposePlan(ctx, sess.ID, []planStepInput{{Title: "C"}, {Title: "D"}}); err != nil { + t.Fatalf("proposePlan #2: %v", err) + } + + // The model addresses the new plan with 1-based seq. seq=1 must hit + // gen-2 "C", leaving gen-1 "A" (replaced) untouched. + if err := s.updatePlanStep(ctx, sess.ID, 1, "running", ""); err != nil { + t.Fatalf("updatePlanStep(seq=1, running): %v", err) + } + if err := s.updatePlanStep(ctx, sess.ID, 1, "done", ""); err != nil { + t.Fatalf("updatePlanStep(seq=1, done): %v", err) + } + + all, err := s.getPlanSteps(ctx, sess.ID, true) + if err != nil { + t.Fatalf("getPlanSteps(all): %v", err) + } + byTitle := map[string]planStep{} + for _, st := range all { + byTitle[st.Title] = st + } + // gen-1 steps stay `replaced` — NOT resurrected to running/done. + if byTitle["A"].Status != "replaced" || byTitle["A"].Generation != 1 { + t.Errorf("A = %+v, want replaced/gen1 (a superseded row must never be touched)", byTitle["A"]) + } + if byTitle["B"].Status != "replaced" || byTitle["B"].Generation != 1 { + t.Errorf("B = %+v, want replaced/gen1", byTitle["B"]) + } + // gen-2 seq=1 advanced; seq=2 untouched. + if byTitle["C"].Status != "done" || byTitle["C"].Generation != 2 || byTitle["C"].Seq != 1 { + t.Errorf("C = %+v, want done/gen2/seq1 (the 1-based update must address the current generation)", byTitle["C"]) + } + if byTitle["D"].Status != "pending" || byTitle["D"].Seq != 2 { + t.Errorf("D = %+v, want pending/seq2", byTitle["D"]) + } + + // Out-of-range seq must be refused (no current-gen step there). + if err := s.updatePlanStep(ctx, sess.ID, 99, "running", ""); !errors.Is(err, errPlanStepNotFound) { + t.Fatalf("updatePlanStep(seq=99) err = %v, want errPlanStepNotFound", err) + } +} + +// TestCompleteTask_AutoCloseEmitsEvents is the P1.1 regression proof: +// completeTask's bulk auto-close of in-flight steps must emit one +// plan.step.finished event per closed step (so the live panel converges +// instead of freezing on "running" after the task completes) and must stamp +// started_at so no closed step is left un-timestamped (P0.1 fix 5). +func TestCompleteTask_AutoCloseEmitsEvents(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + sess, err := s.createSession(ctx, "auto-close events test") + if err != nil { + t.Fatalf("createSession: %v", err) + } + if _, err := s.proposePlan(ctx, sess.ID, []planStepInput{{Title: "A"}, {Title: "B"}}); err != nil { + t.Fatalf("proposePlan: %v", err) + } + // A is running, B still pending at completion time. + if err := s.updatePlanStep(ctx, sess.ID, 1, "running", ""); err != nil { + t.Fatalf("updatePlanStep(1, running): %v", err) + } + if err := s.completeTask(ctx, sess.ID, "success", "done"); err != nil { + t.Fatalf("completeTask: %v", err) + } + + // Every auto-closed step should now carry both a started_at and a + // finished_at (no NULL-started `done` step). + steps, err := s.getPlanSteps(ctx, sess.ID, true) + if err != nil { + t.Fatalf("getPlanSteps: %v", err) + } + for _, st := range steps { + if st.Status == "done" && st.StartedAt == nil { + t.Errorf("step %q done but started_at is NULL (P0.1 fix 5: stamp it)", st.Title) + } + } + + // Exactly two plan.step.finished events — one per closed step (A and B). + var finished int + if err := s.pool.QueryRow(ctx, + `SELECT COUNT(*) FROM events WHERE type = 'plan.step.finished' AND correlation_id = $1`, + sess.ID).Scan(&finished); err != nil { + t.Fatalf("count events: %v", err) + } + if finished != 2 { + t.Fatalf("plan.step.finished events = %d, want 2 (one per auto-closed step)", finished) } } diff --git a/cmd/nomos/tasks.go b/cmd/nomos/tasks.go index 669d50c..ee6912f 100644 --- a/cmd/nomos/tasks.go +++ b/cmd/nomos/tasks.go @@ -245,13 +245,17 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args // the seq-order enforcement (5.6) require it to be completed last, // and D.1's complete_task gate enforces the actual calls. Together // they close the loop structurally — neither relies on the agent - // reading SOUL.md. + // reading SOUL.md. The match is broadened past the literal tool + // names so a natural-language step ("Write back: update entity + // attributes…") isn't doubled by an auto-appended duplicate (P1.2). hasWritebackStep := false for _, st := range steps { - if strings.Contains(st.Title, "update_entity_attributes") || - strings.Contains(st.Title, "create_relationship") || - strings.Contains(st.Detail, "update_entity_attributes") || - strings.Contains(st.Detail, "create_relationship") { + t := strings.ToLower(st.Title + " " + st.Detail) + if strings.Contains(t, "update_entity_attributes") || + strings.Contains(t, "create_relationship") || + strings.Contains(t, "upsert_knowledge") || + strings.Contains(t, "write back") || + strings.Contains(t, "writeback") { hasWritebackStep = true break } @@ -280,8 +284,19 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args // The writeback step is now always present (D.2 auto-appends it if // the agent forgot), so the old advisory nudge is replaced by the // structural gate: D.1 refuses complete_task without the actual - // update_entity_attributes/create_relationship calls. - result := fmt.Sprintf("Plan set (%d steps)%s. If all steps are read-only, execute now — call update_plan_step(running) + run for each step, no approval needed. If any step is config_mutation/destructive, STOP and wait for operator approval (\"approved\", \"yes\", \"go\", \"proceed\", \"continue\", \"ok\", \"go ahead\"). Do not call propose_plan again.", len(persisted), appendedNote) + // update_entity_attributes/create_relationship calls. Enumerate the + // step seqs so the model knows exactly which numbers to address with + // update_plan_step (seq is 1-based within this plan — the addressing + // key, not a global counter). + var seqs strings.Builder + for i, p := range persisted { + if i > 0 { + seqs.WriteString("; ") + } + title := fmt.Sprint(p["title"]) + fmt.Fprintf(&seqs, "%v=%s", p["seq"], title) + } + result := fmt.Sprintf("Plan set (%d steps): %s.%s Address them with update_plan_step(seq=N). If all steps are read-only, execute now — call update_plan_step(running) + run for each step, no approval needed. If any step is config_mutation/destructive, STOP and wait for operator approval (\"approved\", \"yes\", \"go\", \"proceed\", \"continue\", \"ok\", \"go ahead\"). Do not call propose_plan again.", len(persisted), seqs.String(), appendedNote) return result, true case "update_plan_step": @@ -292,6 +307,15 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args return "error: update_plan_step needs seq (>=1) and status", true } if err := a.store.updatePlanStep(ctx, sessionID, seq, status, execID); err != nil { + if errors.Is(err, errPlanStepNotFound) { + // The seq doesn't address a step in the CURRENT plan — most + // often a stale 1-based number the model carried across a + // re-plan, or an out-of-range seq. seq is generation-relative + // (1..N within the latest propose_plan), so a superseded + // generation's row is never touched (P0.1 fix 3). Direct the + // model instead of silently no-op'ing. + return fmt.Sprintf("Step %d is not in the current plan. seq is 1-based within your latest propose_plan (a re-plan resets it to 1..N, so an old step number no longer applies). The plan was not changed. Re-address with the correct 1-based seq, or if you've lost track, re-read the plan.", seq), true + } return fmt.Sprintf("error updating step %d: %v", seq, err), true } return fmt.Sprintf("Step %d → %s. (Advance with update_plan_step + run; do not re-propose.)", seq, status), true diff --git a/migrations/029_plan_generation_relative_seq.up.sql b/migrations/029_plan_generation_relative_seq.up.sql new file mode 100644 index 0000000..a36f513 --- /dev/null +++ b/migrations/029_plan_generation_relative_seq.up.sql @@ -0,0 +1,33 @@ +-- 029_plan_generation_relative_seq.up.sql +-- Make plan-step seq generation-relative: 1..N within each +-- (session_id, generation). Before this, seq was globally increasing across +-- generations (gen1: 1..6, gen2: 7..12), so the model's 1-based +-- update_plan_step calls — which the prompt and schema explicitly tell it to +-- use — landed on superseded gen-1 rows after a re-plan while the live gen-2 +-- work went unrecorded (or, worse, resurrected a `replaced` row as `done`). +-- The addressing key is now (session_id, generation, seq); updatePlanStep +-- resolves against MAX(generation), so a 1-based seq always maps to the +-- CURRENT plan. See plans/2026-07-30-session-review-plan-drift-and-dead- +-- activity-panel.md P0.1. + +-- Renumber existing rows so seq resets to 1..N per (session, generation), +-- preserving each generation's step order. +WITH ranked AS ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY session_id, generation + ORDER BY seq, created_at + ) AS new_seq + FROM session_plan_steps +) +UPDATE session_plan_steps s +SET seq = ranked.new_seq +FROM ranked +WHERE s.id = ranked.id AND s.seq <> ranked.new_seq; + +-- (session_id, seq) is no longer unique once seq resets per generation; the +-- store resolves via (session_id, generation, seq). Drop the old composite +-- index (it now collides on seq) and add the generation-scoped unique index. +DROP INDEX IF EXISTS idx_plan_steps_session; +CREATE UNIQUE INDEX IF NOT EXISTS idx_plan_steps_session_gen_seq + ON session_plan_steps (session_id, generation, seq); diff --git a/web/src/lib/stores/activity.test.ts b/web/src/lib/stores/activity.test.ts new file mode 100644 index 0000000..e861aaf --- /dev/null +++ b/web/src/lib/stores/activity.test.ts @@ -0,0 +1,103 @@ +import { describe, it, expect, vi } from 'vitest' +import { writable } from 'svelte/store' + +// computeActivityLog is a pure derivation; it only needs TYPES from the store +// modules, so mock their runtime exports to keep the test isolated from the +// real store graph (workspace.ts, e.g., starts a top-level setInterval). +vi.mock('./chat', () => ({ + messages: writable([]), + chatFor: vi.fn(() => ({ + messages: writable([]), + streaming: writable(false), + connectionState: writable('connected'), + error: writable(null), + notFound: writable(false) + })) +})) +vi.mock('./workspace', () => ({ + planSteps: writable([]), + currentTask: writable(null), + workspaceFor: vi.fn(() => ({})), + taskFor: vi.fn(() => writable(null)) +})) +vi.mock('./execstream', () => ({ + liveExecutionOutputFor: vi.fn(() => writable(null)) +})) + +import { computeActivityLog } from './activity' +import type { ChatMessage } from './chat' +import type { PlanStep, Session } from '$lib/api' + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) + +function msg(partial: Partial & { id: string }): ChatMessage { + return { + id: partial.id, + role: 'assistant', + text: '', + tools: partial.tools ?? [], + pendingApprovals: [], + created_at: partial.created_at + } +} + +function toolResult(name: string, id: string): NonNullable[number] { + return { type: 'tool_result', name, id, result: 'ok' } +} + +describe('computeActivityLog timestamps (P0.2)', () => { + it('uses the message created_at for persisted tool calls, not a fabricated spread', () => { + const created = '2026-07-29T20:08:10Z' + const m = msg({ + id: 'm1', + created_at: created, + tools: [toolResult('get_entity', 't1'), toolResult('run', 't2')] + }) + const entries = computeActivityLog([m], [], null, new Map()) + const want = new Date(created).getTime() + for (const id of ['t1', 't2']) { + const e = entries.find((x) => x.id === id) + expect(e, `entry ${id} should exist`).toBeDefined() + expect(e!.timestamp).toBe(want) // real time, shared per message — no now-(len-i)*1000 + } + }) + + it('freezes live entries (no created_at) so re-derivation never churns them', async () => { + const m = msg({ id: 'm1', tools: [toolResult('run', 't1')] }) // no created_at + const frozen = new Map() + const a = computeActivityLog([m], [], null, frozen) + const ts1 = a.find((e) => e.id === 't1')!.timestamp + await sleep(60) // the 3s poller re-derives on a later tick + const b = computeActivityLog([m], [], null, frozen) + const ts2 = b.find((e) => e.id === 't1')!.timestamp + expect(ts2).toBe(ts1) // frozen — the previous bug marched every entry forward + }) + + it('is pure w.r.t. wall-clock: two calls with identical inputs give identical output', async () => { + const created = '2026-07-29T20:08:10Z' + const msgs = [ + msg({ id: 'm1', created_at: created, tools: [toolResult('get_entity', 't1')] }), + msg({ id: 'm2', created_at: created, tools: [toolResult('run', 't2')] }) + ] + const steps: PlanStep[] = [ + { id: 's1', seq: 1, title: 'A', detail: '', status: 'done', started_at: created } + ] + const task = { id: 'sess', outcome: 'success', summary: 'done' } as unknown as Session + const frozen = new Map() + const a = computeActivityLog(msgs, steps, task, frozen) + await sleep(50) + const b = computeActivityLog(msgs, steps, task, frozen) + expect(b.map((e) => [e.id, e.timestamp, e.type])).toEqual( + a.map((e) => [e.id, e.timestamp, e.type]) + ) + }) + + it('uses a step started_at instead of falling back to now', () => { + const started = '2026-07-29T20:07:40Z' + const steps: PlanStep[] = [ + { id: 's1', seq: 1, title: 'A', detail: '', status: 'done', started_at: started } + ] + const entries = computeActivityLog([], steps, null, new Map()) + expect(entries.find((e) => e.id === 's1')!.timestamp).toBe(new Date(started).getTime()) + }) +}) diff --git a/web/src/lib/stores/activity.ts b/web/src/lib/stores/activity.ts index f50cbed..c583469 100644 --- a/web/src/lib/stores/activity.ts +++ b/web/src/lib/stores/activity.ts @@ -58,13 +58,37 @@ function stringifyResult(result: unknown): string { // Pure derivation, parameterized so it can back both the global "current // session" activityLog below and a per-session activityLogFor(sessionId) for // a floating task window. -function computeActivityLog( +// +// Timestamps are REAL where the data has them and FROZEN where it doesn't: +// - persisted tool calls use their message's created_at (a true time); +// - steps use their started_at when present; +// - only genuinely-live entries (a tool call on the in-flight message that +// has no created_at yet) fall back to wall-clock, and that value is frozen +// into `frozen` on FIRST sight so a re-derive (the 3s poller re-sets +// `messages` every tick) reads the same value instead of marching every +// entry forward. Before this, every entry's time was +// `now - (len - index) * 1000` — fabricated at render time and churning +// every poll (P0.2). `frozen` is owned by the caller and lives across +// re-derivations; pass a fresh Map for a purity test. +export function computeActivityLog( $msgs: ChatMessage[], $steps: PlanStep[], - $task: Session | null + $task: Session | null, + frozen: Map ): ActivityEntry[] { const entries: ActivityEntry[] = [] const now = Date.now() + // freeze: prefer a real persisted time; else reuse a value already pinned + // for this id; else pin wall-clock now and remember it. + const freeze = (id: string, real?: number | null): number => { + if (real && real > 0) return real + const hit = frozen.get(id) + if (hit !== undefined) return hit + frozen.set(id, now) + return now + } + const tsOf = (iso?: string): number | undefined => + iso ? new Date(iso).getTime() || undefined : undefined // Goal if ($task?.goal) { @@ -87,7 +111,7 @@ function computeActivityLog( s.status === 'running' ? 'step_running' : s.status === 'done' ? 'step_done' : 'step_failed', description: `Step ${s.seq}: ${stepLabel}`, detail: s.detail || undefined, - timestamp: s.started_at ? new Date(s.started_at).getTime() : now, + timestamp: freeze(s.id, tsOf(s.started_at)), status: s.status === 'running' ? 'running' : s.status === 'done' ? 'done' : 'failed' }) } @@ -97,6 +121,7 @@ function computeActivityLog( let currentStepSeq = 0 let entryIdx = 0 for (let mi = 0; mi < $msgs.length; mi++) { + const msgTs = tsOf($msgs[mi].created_at) for (const t of $msgs[mi].tools) { // Track current step from update_plan_step calls if (t.type === 'tool_use' && t.name === 'update_plan_step') { @@ -109,13 +134,14 @@ function computeActivityLog( const label = toolActivityLabel(t) const stepTag = currentStepSeq > 0 ? currentStepSeq : undefined + const id = t.id ?? `tool_${mi}_${entryIdx++}` if (t.type === 'tool_use') { entries.push({ - id: t.id ?? `tool_${mi}_${entryIdx++}`, + id, type: 'tool_running', description: label, args: summarizeArgs(t.args), - timestamp: now - ($msgs.length - mi) * 1000, + timestamp: freeze(id, msgTs), toolName: t.name, stepSeq: stepTag, indent: stepTag != null, @@ -142,12 +168,12 @@ function computeActivityLog( // full entry itself. It used to fall back to the raw tool name // (e.g. "get_entity") instead of the humanized label here. entries.push({ - id: t.id ?? `tool_${mi}_${entryIdx++}`, + id, type: t.error ? 'tool_error' : 'tool_done', description: t.error ? `${label}: ${t.error.slice(0, 80)}` : label, detail: t.error ? t.error : stringifyResult(t.result), args: summarizeArgs(t.args), - timestamp: now - ($msgs.length - mi) * 1000, + timestamp: freeze(id, msgTs), toolName: t.name, stepSeq: stepTag, indent: stepTag != null, @@ -160,14 +186,16 @@ function computeActivityLog( // Knowledge recorded — detect from upsert_knowledge tool results for (let mi = 0; mi < $msgs.length; mi++) { + const msgTs = tsOf($msgs[mi].created_at) for (const t of $msgs[mi].tools) { if (t.type === 'tool_result' && t.name === 'upsert_knowledge' && !t.error) { const title = typeof t.args?.title === 'string' ? t.args.title : '' + const kid = `knowledge_${mi}_${t.id ?? ''}` entries.push({ - id: `knowledge_${mi}`, + id: kid, type: 'knowledge', description: title ? `Recorded: ${title.slice(0, 60)}` : 'Recorded knowledge', - timestamp: now - ($msgs.length - mi) * 1000, + timestamp: freeze(kid, msgTs), status: 'done' }) } @@ -180,7 +208,7 @@ function computeActivityLog( id: 'complete', type: 'complete', description: $task.summary || `Task ${$task.outcome}`, - timestamp: now, + timestamp: freeze('complete'), status: $task.outcome === 'failure' ? 'failed' : 'done' }) } @@ -200,8 +228,13 @@ function computeActivityLog( return entries } +// A per-store freeze map: the first time a live entry (no real timestamp +// yet) is seen, its wall-clock time is pinned here so the 3s poller's +// re-derivation can't march it forward. Owned here, outside the derivation, +// so it survives re-runs. The per-session path has its own Map keyed by id. +const frozenTimestamps = new Map() export const activityLog = derived([messages, planSteps, currentTask], ([$msgs, $steps, $task]) => - computeActivityLog($msgs, $steps, $task) + computeActivityLog($msgs, $steps, $task, frozenTimestamps) ) // Attach streaming output to the `run` entry that is currently executing. @@ -222,13 +255,21 @@ function withLiveOutput( return entries } +// One freeze map per session window (entry ids are UUIDs, but the synthetic +// 'goal'/'complete' ids collide across sessions, so each window keeps its own). +const sessionFrozenTimestamps = new Map>() export function activityLogFor(sessionId: string): Readable { const chat = chatFor(sessionId) const ws = workspaceFor(sessionId) const task = taskFor(sessionId) const live = liveExecutionOutputFor(sessionId) + let frozen = sessionFrozenTimestamps.get(sessionId) + if (!frozen) { + frozen = new Map() + sessionFrozenTimestamps.set(sessionId, frozen) + } return derived([chat.messages, ws.planSteps, task, live], ([$msgs, $steps, $task, $live]) => - withLiveOutput(computeActivityLog($msgs, $steps, $task), $live) + withLiveOutput(computeActivityLog($msgs, $steps, $task, frozen), $live) ) } diff --git a/web/src/lib/stores/workspace.ts b/web/src/lib/stores/workspace.ts index 23e262e..1e63efe 100644 --- a/web/src/lib/stores/workspace.ts +++ b/web/src/lib/stores/workspace.ts @@ -93,12 +93,30 @@ function scheduleSessionsRefresh() { refreshTimer = setTimeout(() => loadSessions(), 300) } +// A dropped plan-step event is one that matched no step on screen — +// historically a silent return, which made a disagreeing backend look like a +// dead UI (the panel froze showing pending steps while work happened +// elsewhere). Surface it so the next divergence is visible. Exported so a +// debug surface (or a test) can read the count. +export let droppedPlanStepEvents = 0 +export function resetDroppedPlanStepEvents(): void { + droppedPlanStepEvents = 0 +} + function applyPlanStepEventTo(ws: WorkspaceState, data: PlanStepEventData) { const stepID = data?.step_id const seq = data?.seq ws.planSteps.update((steps) => { const i = steps.findIndex((s) => (stepID && s.id === stepID) || (seq != null && s.seq === seq)) - if (i === -1) return steps + if (i === -1) { + droppedPlanStepEvents++ + console.warn('plan step event matched no step on screen', { + stepID, + seq, + status: data?.status + }) + return steps + } const next = [...steps] next[i] = { ...next[i],