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

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
 	}