v0.21.0: agent reliability overhaul — plan integrity, target validation, observability pipelines, learning loop

P0 — stop the bleeding: - prevent premature complete_task(success) when goal involves reachability - validate run targets: block host-only commands (qm/pct/pvesh) on LXC/VM - bump MCP client timeout 30s→120s to stop 'context deadline exceeded' P1 — fix the plan system: - add replaced_reason column to session_plan_steps (migration 030) - track WHY steps are replaced (wrong_diagnosis/scope_change/superseded/etc) - force fresh propose_plan on session resume (reopenSession marks old plan) P2 — cognitive guardrails: - SOUL.md scope-gate rule: ask before chasing unrelated subsystems - auto-upsert knowledge entry on every session close P3 — observability (all were empty/NULL): - populate agent_activity.token_count from LLM usage (was always NULL) - populate nomos_plan_executions linking executions to sessions - write plan_completion_rate metric on task close P4 — learning loop (all were empty/NULL): - auto-classify every run call → classifications table (was 0 rows) - auto-feedback on session close (was 0 rows)
2026-08-04 23:15:47 +02:00
parent 1aaedf498a
commit c3f478b8f8
10 changed files with 773 additions and 42 deletions
--- a/cmd/nomos/store.go
+++ b/cmd/nomos/store.go
@@ -807,10 +807,11 @@ func (s *store) setGoal(ctx context.Context, sessionID, goal string) error {
 	// Replace any prior plan steps (done/running/pending/...) as `replaced`.
 	// The rows are kept for the generation counter + audit trail; proposePlan
 	// excludes `replaced` from its in-flight check, so the next propose_plan
-	// takes the fresh-generation path.
+	// takes the fresh-generation path. replaced_reason records the cause
+	// (2026-08-04 plan-step integrity audit).
 	s.pool.Exec(ctx,
-		`UPDATE session_plan_steps SET status = 'replaced', finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status <> 'replaced'`,
-		sessionID)
+		`UPDATE session_plan_steps SET status = 'replaced', replaced_reason = $2, finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status <> 'replaced'`,
+		sessionID, "goal superseded")
 	if _, err := s.pool.Exec(ctx,
 		`UPDATE agent_sessions SET goal = $2, status = 'executing', title = $2, last_active_at = now() WHERE id = $1`,
 		sessionID, goal); err != nil {
@@ -850,6 +851,14 @@ func (s *store) reopenSession(ctx context.Context, sessionID string) bool {
 	s.pool.Exec(ctx,
 		`UPDATE agent_sessions SET status = 'executing', outcome = NULL, summary = NULL, last_active_at = now() WHERE id = $1`,
 		sessionID)
+	// Mark the prior plan's steps as replaced so the P1 plan-first gate in
+	// classifyAndGate forces a fresh propose_plan before any run. Without
+	// this, the agent could resume a session and call run against the old
+	// (completed) plan — exactly what caused the ZimaOS continuation to
+	// have 81 ad-hoc tool calls with zero plan structure (2026-08-04).
+	s.pool.Exec(ctx,
+		`UPDATE session_plan_steps SET status = 'replaced', replaced_reason = $2, finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status <> 'replaced'`,
+		sessionID, "session reopened — awaiting new plan")
 	_ = observability.Event(ctx, sqlcgen.New(s.pool), "task.reopened", s.taskEntityPtr(ctx, sessionID),
 		"info", "nomos", sessionID, map[string]any{"prior_status": currentStatus})
 	return true
@@ -908,9 +917,11 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
 	// The rows are kept for the generation counter (MAX(generation)+1 below)
 	// and the plan_generations eval assertion. `replaced` steps are excluded
 	// from the anyStarted check above, so they don't block this proposal.
+	// replaced_reason records the cause — required by the plan-step integrity
+	// gate (2026-08-04 session audit).
 	if _, err := tx.Exec(ctx,
-		`UPDATE session_plan_steps SET status = 'replaced', finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status = 'pending'`,
-		sessionID); err != nil {
+		`UPDATE session_plan_steps SET status = 'replaced', replaced_reason = $2, finished_at = COALESCE(finished_at, now()) WHERE session_id = $1 AND status = 'pending'`,
+		sessionID, "superseded by new plan generation"); err != nil {
 		return nil, err
 	}

@@ -972,7 +983,7 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
 // 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 {
+func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, status, execID, replacedReason string) error {
 	if s == nil || sessionID == "" || sessionID == "ephemeral" {
 		return nil
 	}
@@ -1025,16 +1036,30 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s
 	// status <> 'replaced' is defense-in-depth: MAX(generation) can't hold a
 	// replaced row, but if it ever could, this refuses the write instead of
 	// resurrecting it. No matching row → errPlanStepNotFound (stale/out-of-range seq).
-	err := s.pool.QueryRow(ctx, `
-		UPDATE session_plan_steps
-		SET status = $4, execution_id = COALESCE($5, execution_id)`+stamp+`
-		WHERE session_id = $1 AND generation = $2 AND seq = $3 AND status <> 'replaced'
-		RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr).Scan(&stepID, &targetSlug)
-	if err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return errPlanStepNotFound
+	if status == "replaced" && replacedReason != "" {
+		err := s.pool.QueryRow(ctx, `
+			UPDATE session_plan_steps
+			SET status = $4, execution_id = COALESCE($5, execution_id), replaced_reason = $6`+stamp+`
+			WHERE session_id = $1 AND generation = $2 AND seq = $3 AND status <> 'replaced'
+			RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr, replacedReason).Scan(&stepID, &targetSlug)
+		if err != nil {
+			if errors.Is(err, pgx.ErrNoRows) {
+				return errPlanStepNotFound
+			}
+			return err
+		}
+	} else {
+		err := s.pool.QueryRow(ctx, `
+			UPDATE session_plan_steps
+			SET status = $4, execution_id = COALESCE($5, execution_id)`+stamp+`
+			WHERE session_id = $1 AND generation = $2 AND seq = $3 AND status <> 'replaced'
+			RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr).Scan(&stepID, &targetSlug)
+		if err != nil {
+			if errors.Is(err, pgx.ErrNoRows) {
+				return errPlanStepNotFound
+			}
+			return err
 		}
-		return err
 	}
 	// Anchor the event to the step's target entity when it has one, else the task.
 	entPtr := s.taskEntityPtr(ctx, sessionID)
@@ -1218,9 +1243,141 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st
 	_ = observability.Event(ctx, sqlcgen.New(s.pool), "task.status", entPtr, severity, "nomos", sessionID,
 		map[string]any{"status": status, "outcome": outcome, "summary": summary,
 			"cancelled_executions": cancelledCount, "blocker": blocker})
+	// Auto-persist knowledge so the graph learns from this session regardless
+	// of whether the agent remembered to call upsert_knowledge (2026-08-04
+	// session audit: only 2.4% of sessions called upsert_knowledge manually).
+	if outcome == "success" || outcome == "partial" {
+		autoUpsertKnowledge(ctx, s, sessionID, outcome, summary)
+	}
+	// Plan quality metric: compute step completion rate for the session's
+	// current plan generation. Tracked as a task attribute so the trend
+	// can be monitored over time (2026-08-04 session audit: 38% baseline).
+	writePlanCompletionRate(ctx, s, sessionID)
+	// Auto-feedback: create a feedback entry linking the session's outcome
+	// to its last execution, feeding the pattern-extraction pipeline that
+	// has been empty since launch (2026-08-04 session audit: 0 feedback rows).
+	if outcome == "success" || outcome == "partial" {
+		autoFeedback(ctx, s, sessionID, outcome, summary)
+	}
 	return nil
 }

+// autoUpsertKnowledge creates a knowledge entry for a completed session,
+// capturing what was done and linking it to the entities involved. Called
+// automatically from completeTask so every session leaves a trace, even if
+// the agent forgot to call upsert_knowledge. Only fired for success/partial
+// outcomes (failures don't have actionable discoveries).
+func autoUpsertKnowledge(ctx context.Context, s *store, sessionID, outcome, summary string) {
+	var goal string
+	if err := s.pool.QueryRow(ctx,
+		`SELECT COALESCE(goal, '') FROM agent_sessions WHERE id = $1`,
+		sessionID).Scan(&goal); err != nil || goal == "" {
+		return
+	}
+	title := "Session " + sessionID[:8] + ": " + goal
+	if len(title) > 200 {
+		title = title[:200]
+	}
+	content := "## Outcome\n" + outcome + "\n\n## Summary\n" + summary
+	kind := "investigation"
+	slug := "investigation:nomos/" + sessionID
+	tags := []string{"nomos-session", "auto-generated"}
+
+	// Upsert the knowledge entity.
+	docID, _ := uuid.NewV7()
+	if err := s.pool.QueryRow(ctx, `
+		INSERT INTO entities (id, slug, type, name, attributes)
+		VALUES ($1, $2, $3, $4, '{}')
+		ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name, updated_at = now()
+		RETURNING id`, docID, slug, kind, title).Scan(&docID); err != nil {
+		slog.Warn("nomos: autoUpsertKnowledge entity insert", "session", sessionID, "error", err)
+		return
+	}
+
+	// Upsert the knowledge content.
+	if _, err := s.pool.Exec(ctx, `
+		INSERT INTO knowledge_entities (entity_id, title, content, source, tags, updated_at)
+		VALUES ($1, $2, $3, 'nomos-agent', $4, now())
+		ON CONFLICT (entity_id) DO UPDATE
+		  SET title = EXCLUDED.title, content = EXCLUDED.content,
+		      tags = EXCLUDED.tags, updated_at = now()`,
+		docID, title, content, tags); err != nil {
+		slog.Warn("nomos: autoUpsertKnowledge content insert", "session", sessionID, "error", err)
+		return
+	}
+
+	// Link to the task entity.
+	var taskEntID uuid.UUID
+	if s.pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`,
+		sessionID).Scan(&taskEntID) == nil && taskEntID != uuid.Nil {
+		s.pool.Exec(ctx, `
+			INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
+			SELECT $1, $2, 'involves', '{"by":"nomos","auto":true}'::jsonb, now()
+			WHERE NOT EXISTS (
+				SELECT 1 FROM relationships
+				WHERE source_id = $1 AND target_id = $2 AND type = 'involves' AND valid_to IS NULL)`,
+			taskEntID, docID)
+	}
+
+	slog.Info("nomos: auto-upserted knowledge for session",
+		"session", sessionID, "outcome", outcome, "slug", slug)
+}
+
+// writePlanCompletionRate computes the step completion rate for the current
+// plan generation and writes it as a task entity attribute so the trend can
+// be tracked. Baseline from 2026-08-04 audit: 38% (15/39 steps reached done).
+func writePlanCompletionRate(ctx context.Context, s *store, sessionID string) {
+	var total, completed int
+	s.pool.QueryRow(ctx, `
+		SELECT COUNT(*), COALESCE(SUM(CASE WHEN status = 'done' THEN 1 ELSE 0 END), 0)
+		FROM session_plan_steps
+		WHERE session_id = $1
+		  AND generation = (SELECT MAX(generation) FROM session_plan_steps WHERE session_id = $1)
+		  AND status <> 'replaced'`, sessionID).Scan(&total, &completed)
+	if total > 0 {
+		rate := float64(completed) / float64(total)
+		attrs, _ := json.Marshal(map[string]any{"plan_completion_rate": rate, "plan_steps_total": total, "plan_steps_completed": completed})
+		s.pool.Exec(ctx, `
+			UPDATE entities SET attributes = attributes || $2::jsonb, updated_at = now()
+			WHERE id = (SELECT entity_id FROM agent_sessions WHERE id = $1)`,
+			sessionID, string(attrs))
+		slog.Info("nomos: plan completion rate", "session", sessionID, "rate", fmt.Sprintf("%.0f%%", rate*100),
+			"completed", completed, "total", total)
+	}
+}
+
+// autoFeedback creates a feedback entry linking the session's outcome to its
+// last execution, feeding the pattern-extraction pipeline that has been empty
+// since launch. Only created for success/partial outcomes (failures don't
+// have a specific execution to tie to).
+func autoFeedback(ctx context.Context, s *store, sessionID, outcome, summary string) {
+	// Find the last execution linked to this session.
+	var execID uuid.UUID
+	if err := s.pool.QueryRow(ctx, `
+		SELECT pe.execution_id FROM nomos_plan_executions pe
+		WHERE pe.session_id = $1::uuid
+		ORDER BY pe.created_at DESC LIMIT 1`, sessionID).Scan(&execID); err != nil || execID == uuid.Nil {
+		return
+	}
+	fbID, _ := uuid.NewV7()
+	slug := "feedback:" + fbID.String()
+	if _, err := s.pool.Exec(ctx, `
+		INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'feedback', $3, '{}')`,
+		fbID, slug, "feedback for "+sessionID[:8]); err != nil {
+		slog.Warn("nomos: autoFeedback entity insert", "session", sessionID, "error", err)
+		return
+	}
+	_, err := s.pool.Exec(ctx, `
+		INSERT INTO feedback (entity_id, execution_id, outcome, observation, lesson, tags, created_at)
+		VALUES ($1, $2, $3, $4, $5, $6, now())`,
+		fbID, execID, outcome, summary, summary, []string{"nomos-session", "auto-generated", "session:" + sessionID[:8]})
+	if err != nil {
+		slog.Warn("nomos: autoFeedback insert", "session", sessionID, "error", err)
+		return
+	}
+	slog.Info("nomos: auto-feedback created for session", "session", sessionID, "outcome", outcome)
+}
+
 // blockerPatterns maps a substring (case-insensitive) to a structured blocker
 // reason. Order matters — earlier patterns take precedence. These are the
 // recurring failure signatures from the 2026-07-20 session audit. A
@@ -1310,6 +1467,53 @@ func (s *store) hadDiscovery(ctx context.Context, sessionID string) bool {
 	return count > 0
 }

+// sessionGoal returns the session's goal text, empty string if not found.
+// Used by complete_task to check whether the goal involved a reachability
+// verification before marking success.
+func (s *store) sessionGoal(ctx context.Context, sessionID string) string {
+	if s == nil || sessionID == "" {
+		return ""
+	}
+	var goal string
+	s.pool.QueryRow(ctx,
+		`SELECT COALESCE(goal, '') FROM agent_sessions WHERE id = $1`,
+		sessionID).Scan(&goal)
+	return goal
+}
+
+// hadRecentVerification checks whether the session successfully verified
+// reachability in recent turns — ping_service, or a run with curl/wget that
+// returned successfully. Used by complete_task as a soft warning when the
+// goal involved a reachability check but no recent verification occurred.
+func (s *store) hadRecentVerification(ctx context.Context, sessionID string) bool {
+	if s == nil || sessionID == "" {
+		return true // fail safe: don't warn when we can't check
+	}
+	// Check for ping_service calls in the last 5 activity entries for this session.
+	var pingCount int
+	s.pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM (
+			SELECT 1 FROM agent_activity
+			WHERE session_id = $1 AND tool_name = 'ping_service' AND success = true
+			ORDER BY ts DESC LIMIT 5
+		) sub`, sessionID).Scan(&pingCount)
+	if pingCount > 0 {
+		return true
+	}
+	// Check for run calls with curl/wget that returned successfully.
+	var curlCount int
+	s.pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM (
+			SELECT 1 FROM agent_activity
+			WHERE session_id = $1
+			  AND tool_name = 'run'
+			  AND success = true
+			  AND (input_summary LIKE '%curl%' OR input_summary LIKE '%wget%')
+			ORDER BY ts DESC LIMIT 10
+		) sub`, sessionID).Scan(&curlCount)
+	return curlCount > 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).
@@ -1913,7 +2117,7 @@ func (s *store) resolveArgEntityID(ctx context.Context, args map[string]any) uui
 // The (nullable) session_id column carries the conversation id. args is the
 // tool call's own arguments, used to best-effort tag the row with the
 // entity it acted on (see resolveArgEntityID).
-func (s *store) logActivity(ctx context.Context, agentID uuid.UUID, sessionID, toolName string, args map[string]any, inputSummary, outputSummary string, durationMs int, success bool, correlationID string) {
+func (s *store) logActivity(ctx context.Context, agentID uuid.UUID, sessionID, toolName string, args map[string]any, inputSummary, outputSummary string, durationMs int, success bool, correlationID string, tokenCount int) {
 	if s == nil || agentID == uuid.Nil {
 		return
 	}
@@ -1925,8 +2129,8 @@ func (s *store) logActivity(ctx context.Context, agentID uuid.UUID, sessionID, t
 	s.pool.Exec(ctx, `
 		INSERT INTO agent_activity
 		    (agent_id, session_id, activity_type, tool_name, entity_id, input_summary, output_summary,
-		     duration_ms, success, correlation_id)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
+		     duration_ms, success, correlation_id, token_count)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
 		agentID, sessionID, "tool_call", toolName, entityIDArg, inputSummary, outputSummary,
-		durationMs, success, correlationID)
+		durationMs, success, correlationID, tokenCount)
 }