session reliability: reconnect, knowledge loop, retire request_execution

Phase 1 — crash recovery: SSE auto-reconnect + backoff, polling gate during disconnect, connection banner with retry button, empty-response retry 3x, non-terminal resume on empty response, persistent error cards. Phase 2/4 — visibility + continuation: custom ExecutionStatus renderer, approvals extracted on every tool_result (not just done), activity bar with status/goal, SessionDigest live polling, Continue button. Phase 3 — cleanup: complete_task auto-cancels orphaned approvals, deletes assent/destructive window keys, propose_plan marks pending steps as replaced, plan step seq-order enforcement. Phase 5 — knowledge loop: list_lxcs state filter (active/destroyed), SOUL.md unmissable writeback section, propose_plan validation nudge, complete_task writeback check, upsert_knowledge about array support, plan generation grouping in frontend, session approval count badge. Retire request_execution — all mutations now route through run. Updated SOUL.md, AGENTS.md, CLIENTS.md, skills, and agent system notes. Migration 020: plan step generation column, audit_log session_id index, nomos_plan_executions pending-approval index.
2026-07-14 11:03:23 +02:00
parent b446909ea5
commit 60effcb2fe
25 changed files with 1894 additions and 323 deletions
--- a/cmd/nomos/agent.go
+++ b/cmd/nomos/agent.go
@@ -17,12 +17,12 @@ import (
 )

 // maxIterations bounds one chat turn's tool-calling loop. Provisioning a
-// service is a long chain (research → plan → request_execution → per-step
+// service is a long chain (research → plan → run → per-step
 // install/verify run calls), so this must be generous; a full deploy with the
 // decomposed pct_create flow can legitimately need many steps. On exhaustion
 // the loop now produces a real summary (finalSummary) rather than a dead end.
 const maxIterations = 40
-const maxLLMRetries = 1
+const maxLLMRetries = 2

 // historyWindowSize bounds how many of a session's most recent persisted
 // messages are replayed into the LLM's context on each turn — see
@@ -129,7 +129,7 @@ func loadSoul() string {
 	}
 	return `You are Nomos, the steward of the oikos — the AI agent for the hubris homelab.
 You have access to MCP tools to query topology, health, knowledge, and request
-gated mutations through request_execution. Be concise. Prefer tools over guessing.`
+gated mutations through run. Be concise. Prefer tools over guessing.`
 }

 // assentWindowDuration is how long after an operator approves a plan that
@@ -294,7 +294,7 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
 		}
 		if len(granted) > 0 {
 			a.openAssentWindow(ctx, sessionID)
-			note := fmt.Sprintf("[System: the operator's message approved pending execution(s) %s via chat assent — they are now running. An assent window is now active for 30 minutes: config_mutation commands will auto-run without re-approval. Do not re-request or call request_execution/run again for these; check get_execution_status if you need the outcome. CONTINUE executing the full plan — do not stop and wait for 'continue' after each step. Only surface to the operator for destructive actions (need typed confirmation) or if you're genuinely stuck after trying alternatives.]", strings.Join(granted, ", "))
+			note := fmt.Sprintf("[System: the operator's message approved pending execution(s) %s via chat assent — they are now running. An assent window is now active for 30 minutes: config_mutation commands will auto-run without re-approval. Do not re-request or call run again for these; check get_execution_status if you need the outcome. CONTINUE executing the full plan — do not stop and wait for 'continue' after each step. Only surface to the operator for destructive actions (need typed confirmation) or if you're genuinely stuck after trying alternatives.]", strings.Join(granted, ", "))
 			messages = append(messages, openai.SystemMessage(note))
 		}
 		if len(blocked) > 0 {
@@ -305,9 +305,9 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
 		// The operator said "proceed"/"go ahead"/"yes" but the preceding
 		// assistant turn had NO pending approvals — meaning the agent
 		// proposed a plan in text and asked "shall I?" without calling
-		// request_execution yet. Inject a system note telling the agent
+		// run yet. Inject a system note telling the agent
 		// the operator approved — go execute the plan now.
-		note := "[System: The operator approved your proposed plan. Execute it now — call request_execution or run to carry out the steps you described. Do not re-describe the plan or ask for confirmation again. The assent window is active: config_mutation commands will auto-run once you create them.]"
+		note := "[System: The operator approved your proposed plan. Execute it now — call run to carry out the steps you described. Do not re-describe the plan or ask for confirmation again. The assent window is active: config_mutation commands will auto-run once you create them.]"
 		messages = append(messages, openai.SystemMessage(note))
 		a.openAssentWindow(ctx, sessionID)
 	}
--- a/cmd/nomos/continue.go
+++ b/cmd/nomos/continue.go
@@ -52,7 +52,7 @@ func (a *agent) runIdleSweepWorker(ctx context.Context) {
 		return
 	}
 	slog.Info("nomos: idle sweep worker started")
-	ticker := time.NewTicker(5 * time.Minute)
+	ticker := time.NewTicker(2 * time.Minute)
 	defer ticker.Stop()
 	for {
 		select {
@@ -143,11 +143,24 @@ func (a *agent) processContinuations(ctx context.Context) {
 		// multiple tasks in flight, one task's open window must never cover a
 		// pending continuation belonging to a different task.
 		if !a.store.assentWindowActive(ctx, a.agentID, p.SessionID) {
-			// A finished one-off execution with no window is left as-is
-			// (marked continued so we don't re-check it forever) — the
-			// operator decides what happens next, as today.
-			a.store.markContinued(ctx, p.ExecID)
-			continue
+			// Re-open the assent window if this session is genuinely
+			// executing (plan was approved, work is in progress) — the
+			// window may have expired while the execution ran. Don't
+			// penalize timing: the plan was approved, the work happened,
+			// the result should flow back.
+			sesh, seshErr := a.store.getSession(ctx, p.SessionID)
+			if seshErr == nil && sesh.Goal != "" && (sesh.Status == "executing" || sesh.Status == "planning") {
+				a.openAssentWindow(ctx, p.SessionID)
+				slog.Info("nomos: re-opened assent window for continuing session", "session", p.SessionID, "execution", p.ExecID)
+			} else {
+				// Genuinely no plan — inject a visible note so the
+				// operator knows WHY the agent didn't auto-continue.
+				note := fmt.Sprintf("[System: execution %s finished with status=%s, but the assent window for this session is not active. The agent will not auto-continue. Reply 'continue' or re-approve the plan to resume.]", p.ExecID, p.Status)
+				body, _ := json.Marshal(map[string]any{"role": "assistant", "text": note, "auto": true})
+				a.store.saveMessage(context.Background(), p.SessionID, "assistant", body)
+				a.store.markContinued(ctx, p.ExecID)
+				continue
+			}
 		}
 		a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop
 		safego.Go("nomos:continue-session:"+p.SessionID, func() { a.continueSession(ctx, p) })
@@ -213,7 +226,7 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
 	// without this outer retry the operator would see nothing at all.
 	cctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
 	defer cancel()
-	for attempt := 0; attempt < 2; attempt++ {
+	for attempt := 0; attempt < 3; attempt++ {
 		toolCalls, finalText, errText = nil, "", ""
 		emit := func(ev agentEvent) {
 			if ev.Type == "tool_use" || ev.Type == "tool_result" {
@@ -241,21 +254,24 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {

 	if errText != "" && finalText == "" {
 		slog.Error("nomos: resume produced no response after retry", "session", sessionID, "error", errText)
-		// Give the task a real, operator-visible terminal state instead of
-		// leaving it silently stuck at whatever status it was in (typically
-		// 'executing' or 'awaiting_input') forever. Before this, a
-		// permanently-failed resume was invisible beyond a log line — the
-		// task board just showed a task that never changed, with nothing
-		// telling the operator it needed attention. Marking it failed here
-		// doesn't prevent the operator from continuing to work the task via
-		// a fresh chat message afterward; it just stops the silent hang.
-		summary := fmt.Sprintf("Auto-resume failed after retrying: %s", errText)
-		if len(summary) > 200 {
-			summary = summary[:200] + "…"
-		}
-		if cerr := a.store.completeTask(context.Background(), sessionID, "failure", summary); cerr != nil {
-			slog.Error("nomos: failed to mark task failed after resume gave up", "session", sessionID, "error", cerr)
+		// Persist a visible system note in the transcript so the
+		// operator sees what happened, but do NOT auto-complete the
+		// task — leave it in 'executing' so a follow-up chat message
+		// can resume it. Before this fix, the task was marked 'failed'
+		// here, which ended it permanently and required starting over.
+		resumeFailedNote := fmt.Sprintf("[System: auto-resume failed after retrying: %s. The task is paused — send another message to continue.]", errText)
+		body, _ := json.Marshal(map[string]any{
+			"role": "assistant",
+			"text": resumeFailedNote,
+			"auto": true,
+		})
+		if msgID != uuid.Nil {
+			a.store.updateMessage(context.Background(), msgID, body)
+		} else {
+			// No placeholder was inserted (rare), save directly.
+			a.store.saveMessage(context.Background(), sessionID, "assistant", body)
 		}
+		return // do not call persist() again — already persisted above
 	}
 	persist() // final state — same row, updated one last time with the concluding text
 }
--- a/cmd/nomos/main.go
+++ b/cmd/nomos/main.go
@@ -164,11 +164,29 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
 		http.Error(w, "bad request: "+err.Error(), 400)
 		return
 	}
-	if req.Message == "" {
+	if req.Message == "" && req.SessionID == "" {
 		http.Error(w, "message is required", 400)
 		return
 	}

+	// Empty message with an existing session = reconnect/resume. The
+	// frontend sends this after a dropped SSE stream to re-establish the
+	// connection and catch up on any auto-continuation work that happened
+	// while disconnected. Route into resumeSession so the agent sees a
+	// system note and reports current state.
+	if req.Message == "" && req.SessionID != "" {
+		slog.Info("nomos: reconnect", "session", req.SessionID)
+		safego.Go("nomos:reconnect:"+req.SessionID, func() {
+			note := "[System: the operator's connection was re-established. The task may have progressed in the background — report your current state and progress.]"
+			a.resumeSession(context.Background(), req.SessionID, note)
+		})
+		// Return 202 so the frontend doesn't try to consume an SSE stream
+		// from this POST — resumeSession writes to the DB directly and
+		// the poller (already running from handleDisconnect) picks it up.
+		w.WriteHeader(202)
+		return
+	}
+
 	flusher, ok := w.(http.Flusher)
 	if !ok {
 		http.Error(w, "streaming not supported", 500)
@@ -318,6 +336,14 @@ func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store, a *a
 		return
 	}

+	// POST /sessions/{id}/resume — the operator asks the agent to continue.
+	if len(parts) == 2 && parts[1] == "resume" && r.Method == http.MethodPost {
+		note := "[System: the operator wants you to continue. Pick up where you left off — execute the next step of the plan, diagnose and fix any failures, or report progress if everything is done.]"
+		safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), id, note) })
+		w.WriteHeader(202)
+		return
+	}
+
 	// GET /sessions/{id}/plan and /sessions/{id}/questions — REST hydration for
 	// the context panel when it first opens a task; live events carry deltas
 	// from there.
--- a/cmd/nomos/store.go
+++ b/cmd/nomos/store.go
@@ -46,16 +46,17 @@ func (s *store) close() {
 // lifecycle status and an outcome (see migration 018 / the task-board plan).
 // Outcome/Summary/EntityID are empty until set, hence omitempty.
 type session struct {
-	ID           string    `json:"id"`
-	Title        string    `json:"title"`
-	Actor        string    `json:"actor"`
-	Goal         string    `json:"goal"`
-	Status       string    `json:"status"`
-	Outcome      string    `json:"outcome,omitempty"`
-	Summary      string    `json:"summary,omitempty"`
-	EntityID     string    `json:"entity_id,omitempty"`
-	CreatedAt    time.Time `json:"created_at"`
-	LastActiveAt time.Time `json:"last_active_at"`
+	ID               string    `json:"id"`
+	Title            string    `json:"title"`
+	Actor            string    `json:"actor"`
+	Goal             string    `json:"goal"`
+	Status           string    `json:"status"`
+	Outcome          string    `json:"outcome,omitempty"`
+	Summary          string    `json:"summary,omitempty"`
+	EntityID         string    `json:"entity_id,omitempty"`
+	PendingApprovals int       `json:"pending_approvals"`
+	CreatedAt        time.Time `json:"created_at"`
+	LastActiveAt     time.Time `json:"last_active_at"`
 }

 type message struct {
@@ -195,9 +196,19 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
 		return nil, nil
 	}
 	rows, err := s.pool.Query(ctx,
-		`SELECT id, title, actor, goal, status, COALESCE(outcome, ''), summary,
-		        COALESCE(entity_id::text, ''), created_at, last_active_at
-		 FROM agent_sessions ORDER BY last_active_at DESC LIMIT 50`)
+		`SELECT s.id, s.title, s.actor, s.goal, s.status, COALESCE(s.outcome, ''), s.summary,
+		        COALESCE(s.entity_id::text, ''),
+		        COALESCE(pa.cnt, 0),
+		        s.created_at, s.last_active_at
+		 FROM agent_sessions s
+		 LEFT JOIN (
+		   SELECT l.session_id, COUNT(*) AS cnt
+		   FROM nomos_plan_executions l
+		   JOIN executions e ON e.entity_id = l.execution_id
+		   WHERE e.status = 'pending_approval'
+		   GROUP BY l.session_id
+		 ) pa ON pa.session_id = s.id
+		 ORDER BY s.last_active_at DESC LIMIT 50`)
 	if err != nil {
 		return nil, err
 	}
@@ -207,7 +218,8 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
 	for rows.Next() {
 		var sess session
 		if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status,
-			&sess.Outcome, &sess.Summary, &sess.EntityID, &sess.CreatedAt, &sess.LastActiveAt); err != nil {
+			&sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals,
+			&sess.CreatedAt, &sess.LastActiveAt); err != nil {
 			return nil, err
 		}
 		out = append(out, sess)
@@ -215,6 +227,24 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
 	return out, rows.Err()
 }

+func (s *store) getSession(ctx context.Context, id string) (*session, error) {
+	if s == nil {
+		return nil, nil
+	}
+	var sess session
+	err := s.pool.QueryRow(ctx,
+		`SELECT id, title, actor, goal, status, COALESCE(outcome, ''), summary,
+		        COALESCE(entity_id::text, ''), 0, created_at, last_active_at
+		 FROM agent_sessions WHERE id = $1`, id).
+		Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status,
+			&sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals,
+			&sess.CreatedAt, &sess.LastActiveAt)
+	if err != nil {
+		return nil, err
+	}
+	return &sess, nil
+}
+
 // getMessages returns a session's ENTIRE message history, unbounded — used
 // for the UI's own transcript view (GET /sessions/{id}), where the operator
 // should be able to see everything a task has done regardless of how long
@@ -387,6 +417,17 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
 			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
+		}
 	}

 	out := make([]map[string]any, 0, len(steps))
@@ -428,6 +469,11 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
 // plan.step.finished (terminal) so the panel advances live. The execution link
 // is also what lets the api auto-close the step when the execution finishes
 // (see closePlanStepForExecution).
+//
+// Completion ordering (done/failed/skipped/blocked) is enforced: a step cannot
+// be marked complete while an earlier step is still pending, preventing the
+// agent from marking step 5 done before step 4 (observed in production: the
+// agent rushed to close all steps in a final turn, in reverse order).
 func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, status, execID string) error {
 	if s == nil || sessionID == "" || sessionID == "ephemeral" {
 		return nil
@@ -436,9 +482,22 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s
 	switch status {
 	case "running":
 		stamp = ", started_at = COALESCE(started_at, now())"
-	case "done", "failed", "skipped", "blocked":
+	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.
+	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 {
+			return fmt.Errorf("cannot complete step %d — step %d is still pending", seq, blockedBy)
+		}
+	}
 	var execPtr *uuid.UUID
 	if id, err := uuid.Parse(execID); err == nil {
 		execPtr = &id
@@ -481,6 +540,33 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st
 	if s == nil || sessionID == "" || sessionID == "ephemeral" {
 		return nil
 	}
+
+	// Auto-cancel any executions still in pending_approval/approved/queued
+	// state for this session — preventing orphaned approvals (observed in
+	// production: 4 approvals left open after session completed).
+	var cancelledCount int
+	if err := s.pool.QueryRow(ctx, `
+		WITH cancelled AS (
+			UPDATE executions SET status = 'cancelled',
+				result = '{"message": "task completed — auto-cancelled"}'::jsonb
+			WHERE entity_id IN (
+				SELECT execution_id FROM nomos_plan_executions WHERE session_id = $1
+			) AND status IN ('pending_approval', 'approved', 'queued')
+			RETURNING entity_id
+		)
+		SELECT COUNT(*) FROM cancelled
+	`, sessionID).Scan(&cancelledCount); err != nil {
+		slog.Warn("nomos: completeTask failed to cancel orphaned executions", "session", sessionID, "error", err)
+	}
+
+	// Mark all continuations done so the worker won't try to feed them back.
+	s.pool.Exec(ctx, `UPDATE nomos_plan_executions SET continued_at = now()
+		WHERE session_id = $1 AND continued_at IS NULL`, sessionID)
+
+	// Clean up assent and destructive window keys from autonomy_settings.
+	s.pool.Exec(ctx, `DELETE FROM autonomy_settings
+		WHERE key LIKE '%:' || $1`, sessionID)
+
 	status := "done"
 	if outcome == "failure" {
 		status = "failed"
@@ -504,10 +590,27 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st
 		severity = "warning"
 	}
 	_ = observability.Event(ctx, sqlcgen.New(s.pool), "task.status", entPtr, severity, "nomos", sessionID,
-		map[string]any{"status": status, "outcome": outcome, "summary": summary})
+		map[string]any{"status": status, "outcome": outcome, "summary": summary,
+			"cancelled_executions": cancelledCount})
 	return nil
 }

+// hadEntityWriteback checks whether this session called update_entity_attributes
+// or create_relationship — used by complete_task to warn the agent when it
+// forgot to persist entity facts (the #1 cause of knowledge graph drift).
+func (s *store) hadEntityWriteback(ctx context.Context, sessionID string) bool {
+	if s == nil || sessionID == "" {
+		return true // fail safe: don't warn when we can't check
+	}
+	var count int
+	s.pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM agent_activity
+		WHERE session_id = $1
+		  AND tool_name IN ('update_entity_attributes', 'create_relationship')
+		  AND success = true`, sessionID).Scan(&count)
+	return count > 0
+}
+
 // staleGoalSession is a goal-bearing task that's gone idle without reaching
 // a terminal state — the idle-sweep worker's work list (fix 2+3 of
 // plans/2026-07-11-task-completion-safety-net.md).
--- a/cmd/nomos/tasks.go
+++ b/cmd/nomos/tasks.go
@@ -202,7 +202,16 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args
 		if err != nil {
 			return fmt.Sprintf("error proposing plan: %v", err), true
 		}
-		return fmt.Sprintf("Plan set: %d step(s). Execute them now, marking each with update_plan_step as you go.", len(persisted)), true
+		// Nudge: if the last step doesn't mention entity writeback tools,
+		// the graph will keep drifting — discovered facts won't be persisted.
+		lastStep := steps[len(steps)-1]
+		hasWriteback := strings.Contains(lastStep.Title+lastStep.Detail, "update_entity_attributes") ||
+			strings.Contains(lastStep.Title+lastStep.Detail, "create_relationship")
+		result := fmt.Sprintf("Plan set: %d step(s). Execute them now, marking each with update_plan_step as you go.", len(persisted))
+		if !hasWriteback {
+			result += "\n\n⚠️ The final step doesn't mention update_entity_attributes or create_relationship. Without those, any facts you discovered about entities (IPs, versions, hosts, states) will be LOST — the next session starts from scratch. Consider revising the last step to include entity writeback BEFORE completing the task."
+		}
+		return result, true

 	case "update_plan_step":
 		seq := toInt(args["seq"])
@@ -261,7 +270,11 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args
 		if err := a.store.completeTask(ctx, sessionID, outcome, summary); err != nil {
 			return fmt.Sprintf("error completing task: %v", err), true
 		}
-		return fmt.Sprintf("Task marked %s: %s", outcome, summary), true
+		result := fmt.Sprintf("Task marked %s: %s", outcome, summary)
+		if !a.store.hadEntityWriteback(ctx, sessionID) {
+			result += "\n\n⚠️ No entity attributes or relationships were updated in this session. Call update_entity_attributes and create_relationship to persist what you learned about entities before the next session starts from scratch."
+		}
+		return result, true
 	default:
 		return nil, false
 	}