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
This commit is contained in:
@@ -319,8 +319,10 @@ func fetchTranscript(ctx context.Context, gateway, sid string) (transcript, sess
|
|||||||
}
|
}
|
||||||
// Fetch the plan (steps with generation numbers) for the
|
// Fetch the plan (steps with generation numbers) for the
|
||||||
// plan_generations assertion. A 404 or empty response is fine — a
|
// plan_generations assertion. A 404 or empty response is fine — a
|
||||||
// pure-DB Q&A with no propose_plan has no plan.
|
// pure-DB Q&A with no propose_plan has no plan. ?all=true returns every
|
||||||
if planResp, perr := http.Get(gateway + "/sessions/" + sid + "/plan"); perr == nil {
|
// 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 {
|
if planResp.StatusCode == 200 {
|
||||||
pb, _ := io.ReadAll(planResp.Body)
|
pb, _ := io.ReadAll(planResp.Body)
|
||||||
_ = json.Unmarshal(pb, &t) // fills t.PlanSteps via "steps" field
|
_ = json.Unmarshal(pb, &t) // fills t.PlanSteps via "steps" field
|
||||||
|
|||||||
@@ -483,7 +483,8 @@ func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store, a *a
|
|||||||
if len(parts) == 2 && r.Method == http.MethodGet {
|
if len(parts) == 2 && r.Method == http.MethodGet {
|
||||||
switch parts[1] {
|
switch parts[1] {
|
||||||
case "plan":
|
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 {
|
if err != nil {
|
||||||
http.Error(w, err.Error(), 500)
|
http.Error(w, err.Error(), 500)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
||||||
"github.com/dtoro/oikos/internal/observability"
|
"github.com/dtoro/oikos/internal/observability"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -25,6 +26,13 @@ const maxToolResultSize = 4096
|
|||||||
// The caller translates this into a directive tool result.
|
// The caller translates this into a directive tool result.
|
||||||
var errPlanInFlight = errors.New("plan already in flight")
|
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 {
|
type store struct {
|
||||||
pool *pgxpool.Pool
|
pool *pgxpool.Pool
|
||||||
}
|
}
|
||||||
@@ -879,16 +887,15 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
|||||||
}
|
}
|
||||||
defer tx.Rollback(ctx)
|
defer tx.Rollback(ctx)
|
||||||
|
|
||||||
var startSeq int
|
|
||||||
var anyStarted bool
|
var anyStarted bool
|
||||||
// `replaced` steps (from a prior plan generation superseded by a
|
// `replaced` steps (from a prior plan generation superseded by a
|
||||||
// follow-up sub-task — see reopenSession) are excluded: they prove a
|
// follow-up sub-task — see setGoal/reopenSession) are excluded: they
|
||||||
// prior plan was completed and superseded, not that a plan is in flight.
|
// prove a prior plan was completed and superseded, not that a plan is in
|
||||||
// Without this exclusion, reopenSession's `replaced` marking would be
|
// flight. Without this exclusion, setGoal's `replaced` marking would be
|
||||||
// useless — propose_plan would still refuse on the follow-up.
|
// useless — propose_plan would still refuse on the follow-up.
|
||||||
if err := tx.QueryRow(ctx, `
|
if err := tx.QueryRow(ctx, `
|
||||||
SELECT COALESCE(max(seq), 0), COALESCE(bool_or(status NOT IN ('pending', 'replaced')), false)
|
SELECT COALESCE(bool_or(status NOT IN ('pending', 'replaced')), false)
|
||||||
FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&startSeq, &anyStarted); err != nil {
|
FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&anyStarted); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if anyStarted {
|
if anyStarted {
|
||||||
@@ -898,35 +905,27 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
|||||||
return nil, errPlanInFlight
|
return nil, errPlanInFlight
|
||||||
}
|
}
|
||||||
// Fresh/revise: mark any prior PENDING steps as `replaced` (not DELETE).
|
// Fresh/revise: mark any prior PENDING steps as `replaced` (not DELETE).
|
||||||
// This preserves the rows for the generation counter (MAX(generation)+1
|
// The rows are kept for the generation counter (MAX(generation)+1 below)
|
||||||
// below) and the plan_generations eval assertion. Without this, a first
|
// and the plan_generations eval assertion. `replaced` steps are excluded
|
||||||
// plan that was proposed but never executed (all pending) would be
|
// from the anyStarted check above, so they don't block this proposal.
|
||||||
// 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.
|
|
||||||
if _, err := tx.Exec(ctx,
|
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'`,
|
`UPDATE session_plan_steps SET status = 'replaced', finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status = 'pending'`,
|
||||||
sessionID); err != nil {
|
sessionID); err != nil {
|
||||||
return nil, err
|
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
|
// nextGen: generation 1 for the first plan, MAX(generation)+1 for every
|
||||||
// initial plan; a genuine revise (which currently goes through the same
|
// revise/follow-up (prior rows were marked `replaced` above, not deleted,
|
||||||
// fresh-start path above because all steps were pending) resets to 1
|
// so the counter survives). seq is generation-relative — it resets to
|
||||||
// since the DELETE wiped the prior rows. The column is wired here so a
|
// 1..N for this generation, so (session_id, generation, seq) is the
|
||||||
// future explicit mid-flight revise path can increment it.
|
// 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
|
var nextGen int
|
||||||
if err := tx.QueryRow(ctx, `
|
if err := tx.QueryRow(ctx, `
|
||||||
SELECT COALESCE(MAX(generation), 0) + 1
|
SELECT COALESCE(MAX(generation), 0) + 1
|
||||||
FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&nextGen); err != nil {
|
FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&nextGen); err != nil {
|
||||||
return nil, err
|
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))
|
out := make([]map[string]any, 0, len(steps))
|
||||||
for i, st := range steps {
|
for i, st := range steps {
|
||||||
@@ -934,7 +933,7 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
|||||||
if st.TargetSlug != "" {
|
if st.TargetSlug != "" {
|
||||||
targetSlug = &st.TargetSlug
|
targetSlug = &st.TargetSlug
|
||||||
}
|
}
|
||||||
seq := startSeq + i + 1
|
seq := i + 1
|
||||||
var id uuid.UUID
|
var id uuid.UUID
|
||||||
if err := tx.QueryRow(ctx, `
|
if err := tx.QueryRow(ctx, `
|
||||||
INSERT INTO session_plan_steps (session_id, seq, title, detail, target_slug, generation)
|
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" {
|
if s == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||||
return nil
|
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 := ""
|
stamp := ""
|
||||||
switch status {
|
switch status {
|
||||||
case "running":
|
case "running":
|
||||||
@@ -984,16 +1000,18 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s
|
|||||||
case "done", "failed", "skipped", "blocked", "replaced":
|
case "done", "failed", "skipped", "blocked", "replaced":
|
||||||
stamp = ", finished_at = now()"
|
stamp = ", finished_at = now()"
|
||||||
}
|
}
|
||||||
// Completion ordering: for terminal states, check that no earlier step
|
// Completion ordering, scoped to the CURRENT generation: for terminal
|
||||||
// is still pending. Running steps can start out of order (the agent
|
// states, no earlier step in THIS plan may still be pending. Running
|
||||||
// may dispatch parallel work), but completion must be sequential.
|
// 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" {
|
if status == "done" || status == "failed" || status == "skipped" || status == "blocked" {
|
||||||
var blockedBy int
|
var blockedBy int
|
||||||
if err := s.pool.QueryRow(ctx, `
|
if err := s.pool.QueryRow(ctx, `
|
||||||
SELECT COALESCE(MIN(seq), 0)
|
SELECT COALESCE(MIN(seq), 0)
|
||||||
FROM session_plan_steps
|
FROM session_plan_steps
|
||||||
WHERE session_id = $1 AND seq < $2 AND status = 'pending'`,
|
WHERE session_id = $1 AND generation = $2 AND seq < $3 AND status = 'pending'`,
|
||||||
sessionID, seq).Scan(&blockedBy); err == nil && blockedBy > 0 {
|
sessionID, curGen, seq).Scan(&blockedBy); err == nil && blockedBy > 0 {
|
||||||
return fmt.Errorf("cannot complete step %d — step %d is still pending", seq, blockedBy)
|
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 stepID uuid.UUID
|
||||||
var targetSlug *string
|
var targetSlug *string
|
||||||
// stamp is a fixed literal from the switch above — never user input.
|
// 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
|
UPDATE session_plan_steps
|
||||||
SET status = $3, execution_id = COALESCE($4, execution_id)`+stamp+`
|
SET status = $4, execution_id = COALESCE($5, execution_id)`+stamp+`
|
||||||
WHERE session_id = $1 AND seq = $2
|
WHERE session_id = $1 AND generation = $2 AND seq = $3 AND status <> 'replaced'
|
||||||
RETURNING id, target_slug`, sessionID, seq, status, execPtr).Scan(&stepID, &targetSlug); err != nil {
|
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
|
return err
|
||||||
}
|
}
|
||||||
// Anchor the event to the step's target entity when it has one, else the task.
|
// 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" {
|
if outcome != "success" {
|
||||||
closeStatus = "skipped"
|
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, `
|
if _, err := s.pool.Exec(ctx, `
|
||||||
UPDATE session_plan_steps
|
UPDATE session_plan_steps
|
||||||
SET status = $2, finished_at = COALESCE(finished_at, now())
|
SET status = $2,
|
||||||
WHERE session_id = $1 AND status IN ('pending', 'running')`,
|
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 {
|
sessionID, closeStatus); err != nil {
|
||||||
slog.Warn("nomos: completeTask failed to auto-close in-flight steps", "session", sessionID, "error", err)
|
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.
|
// Clean up assent and destructive window keys from autonomy_settings.
|
||||||
s.pool.Exec(ctx, `DELETE 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
|
// 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).
|
// 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 {
|
if s == nil {
|
||||||
return nil, 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, `
|
rows, err := s.pool.Query(ctx, `
|
||||||
SELECT id::text, seq, title, detail, status,
|
SELECT id::text, seq, title, detail, status,
|
||||||
execution_id::text, target_slug,
|
execution_id::text, target_slug,
|
||||||
started_at::text, finished_at::text, generation
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -205,7 +205,7 @@ func TestProposePlan_RefuseInFlight(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// The original step 1 must be untouched — not erased, not appended to.
|
// 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 {
|
if err != nil {
|
||||||
t.Fatalf("getPlanSteps: %v", err)
|
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
|
// 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")
|
sess2, err := s.createSession(ctx, "plan replace test")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("createSession: %v", err)
|
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 {
|
if _, err := s.proposePlan(ctx, sess2.ID, []planStepInput{{Title: "Revised"}}); err != nil {
|
||||||
t.Fatalf("proposePlan (revise before execution): %v", err)
|
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 {
|
if err != nil {
|
||||||
t.Fatalf("getPlanSteps: %v", err)
|
t.Fatalf("getPlanSteps: %v", err)
|
||||||
}
|
}
|
||||||
if len(revisedSteps) != 1 || revisedSteps[0].Title != "Revised" {
|
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 {
|
if revisedSteps[0].Seq != 1 {
|
||||||
t.Fatalf("revised step generation = %d, want 1 (fresh-start after DELETE resets generation)", revisedSteps[0].Generation)
|
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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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,
|
// the seq-order enforcement (5.6) require it to be completed last,
|
||||||
// and D.1's complete_task gate enforces the actual calls. Together
|
// and D.1's complete_task gate enforces the actual calls. Together
|
||||||
// they close the loop structurally — neither relies on the agent
|
// 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
|
hasWritebackStep := false
|
||||||
for _, st := range steps {
|
for _, st := range steps {
|
||||||
if strings.Contains(st.Title, "update_entity_attributes") ||
|
t := strings.ToLower(st.Title + " " + st.Detail)
|
||||||
strings.Contains(st.Title, "create_relationship") ||
|
if strings.Contains(t, "update_entity_attributes") ||
|
||||||
strings.Contains(st.Detail, "update_entity_attributes") ||
|
strings.Contains(t, "create_relationship") ||
|
||||||
strings.Contains(st.Detail, "create_relationship") {
|
strings.Contains(t, "upsert_knowledge") ||
|
||||||
|
strings.Contains(t, "write back") ||
|
||||||
|
strings.Contains(t, "writeback") {
|
||||||
hasWritebackStep = true
|
hasWritebackStep = true
|
||||||
break
|
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 writeback step is now always present (D.2 auto-appends it if
|
||||||
// the agent forgot), so the old advisory nudge is replaced by the
|
// the agent forgot), so the old advisory nudge is replaced by the
|
||||||
// structural gate: D.1 refuses complete_task without the actual
|
// structural gate: D.1 refuses complete_task without the actual
|
||||||
// update_entity_attributes/create_relationship calls.
|
// update_entity_attributes/create_relationship calls. Enumerate the
|
||||||
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)
|
// 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
|
return result, true
|
||||||
|
|
||||||
case "update_plan_step":
|
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
|
return "error: update_plan_step needs seq (>=1) and status", true
|
||||||
}
|
}
|
||||||
if err := a.store.updatePlanStep(ctx, sessionID, seq, status, execID); err != nil {
|
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("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
|
return fmt.Sprintf("Step %d → %s. (Advance with update_plan_step + run; do not re-propose.)", seq, status), true
|
||||||
|
|||||||
33
migrations/029_plan_generation_relative_seq.up.sql
Normal file
33
migrations/029_plan_generation_relative_seq.up.sql
Normal file
@@ -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);
|
||||||
103
web/src/lib/stores/activity.test.ts
Normal file
103
web/src/lib/stores/activity.test.ts
Normal file
@@ -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<ChatMessage> & { 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<ChatMessage['tools']>[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<string, number>()
|
||||||
|
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<string, number>()
|
||||||
|
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())
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -58,13 +58,37 @@ function stringifyResult(result: unknown): string {
|
|||||||
// Pure derivation, parameterized so it can back both the global "current
|
// Pure derivation, parameterized so it can back both the global "current
|
||||||
// session" activityLog below and a per-session activityLogFor(sessionId) for
|
// session" activityLog below and a per-session activityLogFor(sessionId) for
|
||||||
// a floating task window.
|
// 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[],
|
$msgs: ChatMessage[],
|
||||||
$steps: PlanStep[],
|
$steps: PlanStep[],
|
||||||
$task: Session | null
|
$task: Session | null,
|
||||||
|
frozen: Map<string, number>
|
||||||
): ActivityEntry[] {
|
): ActivityEntry[] {
|
||||||
const entries: ActivityEntry[] = []
|
const entries: ActivityEntry[] = []
|
||||||
const now = Date.now()
|
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
|
// Goal
|
||||||
if ($task?.goal) {
|
if ($task?.goal) {
|
||||||
@@ -87,7 +111,7 @@ function computeActivityLog(
|
|||||||
s.status === 'running' ? 'step_running' : s.status === 'done' ? 'step_done' : 'step_failed',
|
s.status === 'running' ? 'step_running' : s.status === 'done' ? 'step_done' : 'step_failed',
|
||||||
description: `Step ${s.seq}: ${stepLabel}`,
|
description: `Step ${s.seq}: ${stepLabel}`,
|
||||||
detail: s.detail || undefined,
|
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'
|
status: s.status === 'running' ? 'running' : s.status === 'done' ? 'done' : 'failed'
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -97,6 +121,7 @@ function computeActivityLog(
|
|||||||
let currentStepSeq = 0
|
let currentStepSeq = 0
|
||||||
let entryIdx = 0
|
let entryIdx = 0
|
||||||
for (let mi = 0; mi < $msgs.length; mi++) {
|
for (let mi = 0; mi < $msgs.length; mi++) {
|
||||||
|
const msgTs = tsOf($msgs[mi].created_at)
|
||||||
for (const t of $msgs[mi].tools) {
|
for (const t of $msgs[mi].tools) {
|
||||||
// Track current step from update_plan_step calls
|
// Track current step from update_plan_step calls
|
||||||
if (t.type === 'tool_use' && t.name === 'update_plan_step') {
|
if (t.type === 'tool_use' && t.name === 'update_plan_step') {
|
||||||
@@ -109,13 +134,14 @@ function computeActivityLog(
|
|||||||
|
|
||||||
const label = toolActivityLabel(t)
|
const label = toolActivityLabel(t)
|
||||||
const stepTag = currentStepSeq > 0 ? currentStepSeq : undefined
|
const stepTag = currentStepSeq > 0 ? currentStepSeq : undefined
|
||||||
|
const id = t.id ?? `tool_${mi}_${entryIdx++}`
|
||||||
if (t.type === 'tool_use') {
|
if (t.type === 'tool_use') {
|
||||||
entries.push({
|
entries.push({
|
||||||
id: t.id ?? `tool_${mi}_${entryIdx++}`,
|
id,
|
||||||
type: 'tool_running',
|
type: 'tool_running',
|
||||||
description: label,
|
description: label,
|
||||||
args: summarizeArgs(t.args),
|
args: summarizeArgs(t.args),
|
||||||
timestamp: now - ($msgs.length - mi) * 1000,
|
timestamp: freeze(id, msgTs),
|
||||||
toolName: t.name,
|
toolName: t.name,
|
||||||
stepSeq: stepTag,
|
stepSeq: stepTag,
|
||||||
indent: stepTag != null,
|
indent: stepTag != null,
|
||||||
@@ -142,12 +168,12 @@ function computeActivityLog(
|
|||||||
// full entry itself. It used to fall back to the raw tool name
|
// full entry itself. It used to fall back to the raw tool name
|
||||||
// (e.g. "get_entity") instead of the humanized label here.
|
// (e.g. "get_entity") instead of the humanized label here.
|
||||||
entries.push({
|
entries.push({
|
||||||
id: t.id ?? `tool_${mi}_${entryIdx++}`,
|
id,
|
||||||
type: t.error ? 'tool_error' : 'tool_done',
|
type: t.error ? 'tool_error' : 'tool_done',
|
||||||
description: t.error ? `${label}: ${t.error.slice(0, 80)}` : label,
|
description: t.error ? `${label}: ${t.error.slice(0, 80)}` : label,
|
||||||
detail: t.error ? t.error : stringifyResult(t.result),
|
detail: t.error ? t.error : stringifyResult(t.result),
|
||||||
args: summarizeArgs(t.args),
|
args: summarizeArgs(t.args),
|
||||||
timestamp: now - ($msgs.length - mi) * 1000,
|
timestamp: freeze(id, msgTs),
|
||||||
toolName: t.name,
|
toolName: t.name,
|
||||||
stepSeq: stepTag,
|
stepSeq: stepTag,
|
||||||
indent: stepTag != null,
|
indent: stepTag != null,
|
||||||
@@ -160,14 +186,16 @@ function computeActivityLog(
|
|||||||
|
|
||||||
// Knowledge recorded — detect from upsert_knowledge tool results
|
// Knowledge recorded — detect from upsert_knowledge tool results
|
||||||
for (let mi = 0; mi < $msgs.length; mi++) {
|
for (let mi = 0; mi < $msgs.length; mi++) {
|
||||||
|
const msgTs = tsOf($msgs[mi].created_at)
|
||||||
for (const t of $msgs[mi].tools) {
|
for (const t of $msgs[mi].tools) {
|
||||||
if (t.type === 'tool_result' && t.name === 'upsert_knowledge' && !t.error) {
|
if (t.type === 'tool_result' && t.name === 'upsert_knowledge' && !t.error) {
|
||||||
const title = typeof t.args?.title === 'string' ? t.args.title : ''
|
const title = typeof t.args?.title === 'string' ? t.args.title : ''
|
||||||
|
const kid = `knowledge_${mi}_${t.id ?? ''}`
|
||||||
entries.push({
|
entries.push({
|
||||||
id: `knowledge_${mi}`,
|
id: kid,
|
||||||
type: 'knowledge',
|
type: 'knowledge',
|
||||||
description: title ? `Recorded: ${title.slice(0, 60)}` : 'Recorded knowledge',
|
description: title ? `Recorded: ${title.slice(0, 60)}` : 'Recorded knowledge',
|
||||||
timestamp: now - ($msgs.length - mi) * 1000,
|
timestamp: freeze(kid, msgTs),
|
||||||
status: 'done'
|
status: 'done'
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -180,7 +208,7 @@ function computeActivityLog(
|
|||||||
id: 'complete',
|
id: 'complete',
|
||||||
type: 'complete',
|
type: 'complete',
|
||||||
description: $task.summary || `Task ${$task.outcome}`,
|
description: $task.summary || `Task ${$task.outcome}`,
|
||||||
timestamp: now,
|
timestamp: freeze('complete'),
|
||||||
status: $task.outcome === 'failure' ? 'failed' : 'done'
|
status: $task.outcome === 'failure' ? 'failed' : 'done'
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -200,8 +228,13 @@ function computeActivityLog(
|
|||||||
return entries
|
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<string, number>()
|
||||||
export const activityLog = derived([messages, planSteps, currentTask], ([$msgs, $steps, $task]) =>
|
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.
|
// Attach streaming output to the `run` entry that is currently executing.
|
||||||
@@ -222,13 +255,21 @@ function withLiveOutput(
|
|||||||
return entries
|
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<string, Map<string, number>>()
|
||||||
export function activityLogFor(sessionId: string): Readable<ActivityEntry[]> {
|
export function activityLogFor(sessionId: string): Readable<ActivityEntry[]> {
|
||||||
const chat = chatFor(sessionId)
|
const chat = chatFor(sessionId)
|
||||||
const ws = workspaceFor(sessionId)
|
const ws = workspaceFor(sessionId)
|
||||||
const task = taskFor(sessionId)
|
const task = taskFor(sessionId)
|
||||||
const live = liveExecutionOutputFor(sessionId)
|
const live = liveExecutionOutputFor(sessionId)
|
||||||
|
let frozen = sessionFrozenTimestamps.get(sessionId)
|
||||||
|
if (!frozen) {
|
||||||
|
frozen = new Map<string, number>()
|
||||||
|
sessionFrozenTimestamps.set(sessionId, frozen)
|
||||||
|
}
|
||||||
return derived([chat.messages, ws.planSteps, task, live], ([$msgs, $steps, $task, $live]) =>
|
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)
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -93,12 +93,30 @@ function scheduleSessionsRefresh() {
|
|||||||
refreshTimer = setTimeout(() => loadSessions(), 300)
|
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) {
|
function applyPlanStepEventTo(ws: WorkspaceState, data: PlanStepEventData) {
|
||||||
const stepID = data?.step_id
|
const stepID = data?.step_id
|
||||||
const seq = data?.seq
|
const seq = data?.seq
|
||||||
ws.planSteps.update((steps) => {
|
ws.planSteps.update((steps) => {
|
||||||
const i = steps.findIndex((s) => (stepID && s.id === stepID) || (seq != null && s.seq === seq))
|
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]
|
const next = [...steps]
|
||||||
next[i] = {
|
next[i] = {
|
||||||
...next[i],
|
...next[i],
|
||||||
|
|||||||
Reference in New Issue
Block a user