fix(agent): refuse plan re-proposal + emit done on error (close divergence chain)
Operator-reported bug: on 'proceed with the rest' the agent re-proposed the
plan, duplicating it in the sidebar. Root cause was a three-bug chain, not
one bug:
1. Trigger — model empty-response on 'proceed' (approval vocabulary didn't
list 'proceed', so the agent wasn't sure it was approved and no-op'd).
2. Amplifier — chatWith emitted 'error' without 'done' on empty response
(agent.go:370). The frontend's onComplete saw !receivedDone and
misclassified the model failure as a network disconnect, calling
handleDisconnect -> resumeSession.
3. Divergence — the reconnect note was generic ('report your state'), so
the agent re-proposed + re-executed instead of advancing the plan.
Fixes (shipped, e2e-validated against the live agent on oikos-nomos-1):
- A.2: proposePlan refuses re-proposal once a step has started (returns
errPlanInFlight). Drops the append-mode safety net (commit 5384499) that
was the direct source of the sidebar duplication. The agent must advance
with update_plan_step + run; the tool result directs it.
- A.1: proposePlan sets the 'generation' column on INSERT (migration 020
added the column + frontend grouping, but the INSERT never wired it).
- A.3: propose_plan tool description restated as a crisp contract (ONCE,
STOP and wait, REFUSES once a step started, advance with update_plan_step).
- F.3: approval vocabulary expanded to approved/yes/go/proceed/continue/ok/
go ahead; propose_plan result string tightened to an imperative.
- B.1: chatWith emits 'done' after 'error' on every terminal path via a new
emitError helper. The frontend now treats model errors as ended (not
disconnected), so no auto-reconnect -> resumeSession fires.
- B.2: reconnect/resume note carries the operator's last message + an
explicit 'advance the plan, do NOT call propose_plan again' directive when
a plan is in flight. Wired into all 4 resume entry points (reconnect,
/resume, idle-sweep, question-answer) via enrichResumeNote.
- B.3: resumeSession escalates the recovery note across its 3 attempts (final
retry: 'pick the lowest-pending step, mark it running, call run — do that
now') instead of 3 identical notes -> 3 identical empties.
Verification: TestProposePlan_RefuseInFlight replaces TestProposePlan_
AppendVsReplace. e2e conversations against the rebuilt container:
conv2 ('proceed with the rest') -> 0 propose_plan calls, plan stayed at
3 steps (was 6+ before), update_plan_step x5 + run x2 + complete_task.
conv3 (full plan, 'go ahead') -> apt-get update on lxc:dns auto-ran under
the plan window, update_entity_attributes writeback, clean complete_task.
nomos logs show zero reconnect/resume entries for the plan-proposing
sessions (the three-bug chain is closed).
Remaining (not in this commit): D.1 refuse complete_task without writeback
(next blocker), C.1/C.2, F.1/F.2 SOUL.md consolidation, B.4-B.6, E.1/E.2.
See plans/2026-07-14-post-fix-session-remainders.md.
Also: re-audit 2026-07-10-general-gated-execution.md — request_execution enum
retirement (60effcb) closes item 9; only auto-act revival (item 10) remains.
Version 0.4.1 -> 0.5.0 (minor: new structural behavior, not a bugfix).
This commit is contained in:
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"regexp"
|
||||
@@ -17,6 +18,13 @@ import (
|
||||
|
||||
const maxToolResultSize = 4096
|
||||
|
||||
// errPlanInFlight is returned by proposePlan when called again after a step
|
||||
// has already started. The agent must advance the existing plan with
|
||||
// update_plan_step + run instead of re-proposing — re-proposing was the
|
||||
// source of duplicate plans in the sidebar (operator-reported 2026-07-14).
|
||||
// The caller translates this into a directive tool result.
|
||||
var errPlanInFlight = errors.New("plan already in flight")
|
||||
|
||||
type store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
@@ -150,6 +158,76 @@ func (s *store) updateMessage(ctx context.Context, id uuid.UUID, content json.Ra
|
||||
return err
|
||||
}
|
||||
|
||||
// lastUserMessage returns the most recent user message text for a session,
|
||||
// or "" if none. Used to build a context-rich reconnect/resume note: instead
|
||||
// of a generic "report your state," the note can say "the operator's last
|
||||
// message was X — advance the plan" so the agent doesn't re-propose or
|
||||
// re-execute on a reconnect (the operator-reported 2026-07-14 divergence).
|
||||
func (s *store) lastUserMessage(ctx context.Context, sessionID string) string {
|
||||
if s == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||
return ""
|
||||
}
|
||||
var content json.RawMessage
|
||||
if err := s.pool.QueryRow(ctx,
|
||||
`SELECT content FROM agent_messages
|
||||
WHERE session_id = $1 AND role = 'user'
|
||||
ORDER BY created_at DESC LIMIT 1`, sessionID).Scan(&content); err != nil {
|
||||
return ""
|
||||
}
|
||||
var m struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal(content, &m); err != nil {
|
||||
return ""
|
||||
}
|
||||
return m.Text
|
||||
}
|
||||
|
||||
// hasPlanInFlight reports whether a session has a plan with at least one
|
||||
// step in a non-terminal state (pending/running). Used to direct the
|
||||
// reconnect/resume note: if a plan is in flight, the note says "advance
|
||||
// the plan with update_plan_step + run" instead of the generic "report
|
||||
// your state" (which caused the agent to re-propose and duplicate the plan
|
||||
// in the sidebar — operator-reported 2026-07-14).
|
||||
func (s *store) hasPlanInFlight(ctx context.Context, sessionID string) bool {
|
||||
if s == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||
return false
|
||||
}
|
||||
var exists bool
|
||||
if err := s.pool.QueryRow(ctx,
|
||||
`SELECT EXISTS(SELECT 1 FROM session_plan_steps
|
||||
WHERE session_id = $1 AND status IN ('pending', 'running'))`, sessionID).Scan(&exists); err != nil {
|
||||
return false
|
||||
}
|
||||
return exists
|
||||
}
|
||||
|
||||
// enrichResumeNote appends session context to a base resume/reconnect note:
|
||||
// the operator's last user message and, if a plan is in flight, an explicit
|
||||
// directive to advance it with update_plan_step + run (not re-propose). The
|
||||
// generic "report your state" note caused the agent to re-propose and
|
||||
// duplicate the plan on a reconnect (operator-reported 2026-07-14); this
|
||||
// enrichment gives the agent enough context to do the right thing even
|
||||
// through the reconnect path.
|
||||
func (s *store) enrichResumeNote(ctx context.Context, sessionID, base string) string {
|
||||
if s == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||
return base
|
||||
}
|
||||
last := s.lastUserMessage(ctx, sessionID)
|
||||
inFlight := s.hasPlanInFlight(ctx, sessionID)
|
||||
if last == "" && !inFlight {
|
||||
return base
|
||||
}
|
||||
note := base
|
||||
if last != "" {
|
||||
note += fmt.Sprintf(" The operator's last message was: %q.", last)
|
||||
}
|
||||
if inFlight {
|
||||
note += " A plan is in flight — advance it with update_plan_step (status=running) + run for the next step's target. Do NOT call propose_plan again."
|
||||
}
|
||||
return note
|
||||
}
|
||||
|
||||
func truncateToolResults(content json.RawMessage) json.RawMessage {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(content, &m); err != nil {
|
||||
@@ -404,14 +482,14 @@ type planStepInput struct {
|
||||
// Two modes, chosen by whether any existing step has left 'pending':
|
||||
// - Fresh/revise (no step started yet): full replace (delete + insert). This
|
||||
// covers the first call, and a genuine re-plan before any work began.
|
||||
// - Mid-flight (some step is running/done/failed/…): APPEND the new steps
|
||||
// after the current max seq instead of wiping. The model is instructed to
|
||||
// propose the whole plan in one call, but nothing stops it from calling
|
||||
// propose_plan again per-step as it goes — a destructive replace in that
|
||||
// case would erase every already-completed step, leaving the operator
|
||||
// seeing only the most recent single step ("1/1") instead of real
|
||||
// progress. Appending makes the panel's step history correct regardless
|
||||
// of how the model chooses to call the tool.
|
||||
// - Mid-flight (some step is running/done/failed/…): REFUSE the call.
|
||||
// The agent must advance the existing plan with update_plan_step + run
|
||||
// instead of re-proposing. The previous append-mode safety net (commit
|
||||
// 5384499) preserved history but produced a confusing duplicate sidebar
|
||||
// when the agent re-proposed on "proceed" (operator-reported 2026-07-14).
|
||||
// Refusing is the correct default — the tool result tells the agent how
|
||||
// to advance, and the generation column tracks revisions if a genuine
|
||||
// re-plan is ever allowed.
|
||||
func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planStepInput) ([]map[string]any, error) {
|
||||
if s == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||
return nil, nil
|
||||
@@ -429,23 +507,33 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
||||
FROM session_plan_steps WHERE session_id = $1`, sessionID).Scan(&startSeq, &anyStarted); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !anyStarted {
|
||||
if _, err := tx.Exec(ctx, `DELETE FROM session_plan_steps WHERE session_id = $1`, sessionID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
startSeq = 0
|
||||
} else {
|
||||
// Mid-flight plan revision: mark any still-pending steps from the
|
||||
// previous generation as 'replaced' so the panel doesn't show them
|
||||
// as incomplete forever. Only pending steps — already-running or
|
||||
// done steps from the prior plan are preserved as history.
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE session_plan_steps
|
||||
SET status = 'replaced', finished_at = now()
|
||||
WHERE session_id = $1 AND status = 'pending'`, sessionID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if anyStarted {
|
||||
// A plan is already in flight (a step is running/done/failed/...).
|
||||
// Refuse the re-proposal — the agent must advance with
|
||||
// update_plan_step + run. The caller surfaces a directive.
|
||||
return nil, errPlanInFlight
|
||||
}
|
||||
// Fresh/revise: delete any prior pending steps and start a new
|
||||
// generation. The DELETE covers the genuine pre-execution revision
|
||||
// case (operator asked to revise before any step started).
|
||||
if _, err := tx.Exec(ctx, `DELETE FROM session_plan_steps WHERE session_id = $1`, sessionID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
startSeq = 0
|
||||
|
||||
// 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.
|
||||
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 {
|
||||
@@ -456,14 +544,15 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
||||
seq := startSeq + i + 1
|
||||
var id uuid.UUID
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO session_plan_steps (session_id, seq, title, detail, target_slug)
|
||||
VALUES ($1, $2, $3, $4, $5) RETURNING id`,
|
||||
sessionID, seq, st.Title, st.Detail, targetSlug).Scan(&id); err != nil {
|
||||
INSERT INTO session_plan_steps (session_id, seq, title, detail, target_slug, generation)
|
||||
VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
|
||||
sessionID, seq, st.Title, st.Detail, targetSlug, nextGen).Scan(&id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, map[string]any{
|
||||
"id": id.String(), "seq": seq, "title": st.Title,
|
||||
"detail": st.Detail, "target_slug": st.TargetSlug,
|
||||
"generation": nextGen,
|
||||
})
|
||||
}
|
||||
if _, err := tx.Exec(ctx,
|
||||
@@ -474,10 +563,10 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
|
||||
return nil, err
|
||||
}
|
||||
// Event after commit so subscribers only ever see a persisted plan.
|
||||
// appended=true tells the panel to add these steps to its existing list
|
||||
// rather than replace it (mirrors the mid-flight append above).
|
||||
// appended=false (always now — we refuse mid-flight re-proposals) tells
|
||||
// the panel to replace its list with these steps.
|
||||
_ = observability.Event(ctx, sqlcgen.New(s.pool), "plan.proposed", s.taskEntityPtr(ctx, sessionID),
|
||||
"info", "nomos", sessionID, map[string]any{"steps": out, "appended": anyStarted})
|
||||
"info", "nomos", sessionID, map[string]any{"steps": out, "appended": false, "generation": nextGen})
|
||||
// Record a plan-proposed window so the `run` handler knows a plan is
|
||||
// pending approval and can skip per-action approval for config_mutation
|
||||
// commands within the plan. Transitions to 'active' when the operator
|
||||
@@ -703,6 +792,7 @@ type planStep struct {
|
||||
TargetSlug *string `json:"target_slug,omitempty"`
|
||||
StartedAt *string `json:"started_at,omitempty"`
|
||||
FinishedAt *string `json:"finished_at,omitempty"`
|
||||
Generation int `json:"generation"`
|
||||
}
|
||||
|
||||
// getPlanSteps returns a task's plan in order — REST hydration for the context
|
||||
@@ -714,7 +804,7 @@ func (s *store) getPlanSteps(ctx context.Context, sessionID string) ([]planStep,
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id::text, seq, title, detail, status,
|
||||
execution_id::text, target_slug,
|
||||
started_at::text, finished_at::text
|
||||
started_at::text, finished_at::text, generation
|
||||
FROM session_plan_steps WHERE session_id = $1 ORDER BY seq`, sessionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -725,7 +815,7 @@ func (s *store) getPlanSteps(ctx context.Context, sessionID string) ([]planStep,
|
||||
var st planStep
|
||||
var execID, target, started, finished *string
|
||||
if err := rows.Scan(&st.ID, &st.Seq, &st.Title, &st.Detail, &st.Status,
|
||||
&execID, &target, &started, &finished); err != nil {
|
||||
&execID, &target, &started, &finished, &st.Generation); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
st.ExecutionID, st.TargetSlug, st.StartedAt, st.FinishedAt = execID, target, started, finished
|
||||
|
||||
Reference in New Issue
Block a user