feat(tasks): phase 3 — close the knowledge loop (complete_task + retrieval)

Adds the compounding knowledge loop the task model is built around: - complete_task(outcome, summary): a nomos-LOCAL, session-scoped tool (the shared MCP server has no session id). Introduces the local-tool mechanism — buildTools appends task tools, the agent loop routes them to handleTaskTool instead of the MCP client. Sets the task's terminal status/outcome/summary, mirrors it onto the task entity, and emits task.status. - Knowledge → task linkage: after a successful upsert_knowledge in a task, nomos links the note to the task entity (documents) and emits knowledge.recorded, so the task's outcome view shows what it learned. The note's about-link to the involved entity (written by upsert_knowledge) is the retrieval path future tasks use. - SOUL: every chat is a task loop — retrieve prior knowledge FIRST (get_entity_knowledge on the target), plan, execute, record learnings, then complete_task. Scales down for trivial read-only tasks. - deleteSession now cleans up the task entity, its relationships, and its task-scoped events (was orphaning them); the knowledge doc itself and its about-links survive, as knowledge should outlive the task. Verified end-to-end: a task recorded a note and completed; task.status + knowledge.recorded hit the SSE stream; status=done/outcome=success persisted; the note linked to both lxc:caddy (retrieval) and the task; a future get_entity_knowledge(lxc:caddy) surfaces it; delete cleaned edges+events (0/0/0) while the knowledge survived. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-11 12:43:29 +02:00
parent 3dba2e550a
commit 532310bb4b
4 changed files with 214 additions and 8 deletions
--- a/cmd/nomos/agent.go
+++ b/cmd/nomos/agent.go
@@ -360,7 +360,15 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
 			})

 			start := time.Now()
-			result, callErr := a.client.callTool(tc.Function.Name, args)
+			// Session-scoped task tools are handled in-process; everything else
+			// is forwarded to the shared MCP server.
+			var result any
+			var callErr error
+			if localRes, handled := a.handleTaskTool(ctx, sessionID, tc.Function.Name, args); handled {
+				result = localRes
+			} else {
+				result, callErr = a.client.callTool(tc.Function.Name, args)
+			}
 			elapsed := int(time.Since(start).Milliseconds())

 			inputJSON, _ := json.Marshal(args)
@@ -397,6 +405,12 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
 			// results — so a bulk query doesn't drag the whole fleet in.
 			a.store.recordTouched(ctx, sessionID, tc.Function.Name, args)

+			// When the agent records knowledge, link that note to this task so
+			// the task's outcome view shows what it learned (and pulse it live).
+			if tc.Function.Name == "upsert_knowledge" {
+				a.store.linkKnowledgeToTask(ctx, sessionID, string(resultJSON))
+			}
+
 			emit(agentEvent{
 				Type:      "tool_result",
 				Data:      map[string]any{"name": tc.Function.Name, "result": result, "id": tc.ID},
@@ -619,6 +633,9 @@ func (a *agent) buildTools() ([]openai.ChatCompletionToolParam, error) {
 	if err != nil {
 		return nil, err
 	}
+	// Append nomos-local, session-scoped task tools (complete_task, …) to the
+	// MCP tool list. They're routed to handleTaskTool, not the MCP client.
+	defs = append(defs, taskToolDefs()...)

 	var tools []openai.ChatCompletionToolParam
 	for _, d := range defs {
--- a/cmd/nomos/store.go
+++ b/cmd/nomos/store.go
@@ -242,12 +242,101 @@ func (s *store) deleteSession(ctx context.Context, id string) error {
 	if s == nil {
 		return nil
 	}
-	_, err := s.pool.Exec(ctx, `DELETE FROM agent_messages WHERE session_id = $1`, id)
-	if err != nil {
+	// Resolve the task entity so we can clean up its graph edges and events too
+	// — otherwise deleting a session orphans its task:<id> entity, its involves/
+	// documents relationships, and its task-scoped events.
+	var entID uuid.UUID
+	s.pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, id).Scan(&entID)
+
+	if _, err := s.pool.Exec(ctx, `DELETE FROM agent_messages WHERE session_id = $1`, id); err != nil {
 		return err
 	}
-	_, err = s.pool.Exec(ctx, `DELETE FROM agent_sessions WHERE id = $1`, id)
-	return err
+	// task.status / entity.touched / knowledge.recorded are all correlated by
+	// session id.
+	s.pool.Exec(ctx, `DELETE FROM events WHERE correlation_id = $1`, id)
+	if _, err := s.pool.Exec(ctx, `DELETE FROM agent_sessions WHERE id = $1`, id); err != nil {
+		return err
+	}
+	if entID != uuid.Nil {
+		// relationships FK is ON DELETE RESTRICT, so drop the task's edges first.
+		s.pool.Exec(ctx, `DELETE FROM relationships WHERE source_id = $1 OR target_id = $1`, entID)
+		s.pool.Exec(ctx, `DELETE FROM entities WHERE id = $1`, entID)
+	}
+	return nil
+}
+
+// completeTask sets a task's terminal state, outcome, and one-line summary,
+// mirrors the outcome onto the task entity's attributes (so the board/graph
+// show it), and publishes task.status for the live context panel. outcome is
+// success|failure|partial; status is derived (failure → failed, else done).
+func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary string) error {
+	if s == nil || sessionID == "" || sessionID == "ephemeral" {
+		return nil
+	}
+	status := "done"
+	if outcome == "failure" {
+		status = "failed"
+	}
+	if _, err := s.pool.Exec(ctx, `
+		UPDATE agent_sessions SET status = $2, outcome = $3, summary = $4, last_active_at = now()
+		WHERE id = $1`, sessionID, status, outcome, summary); err != nil {
+		return err
+	}
+	var entID uuid.UUID
+	s.pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, sessionID).Scan(&entID)
+	var entPtr *uuid.UUID
+	if entID != uuid.Nil {
+		attrs, _ := json.Marshal(map[string]any{"outcome": outcome, "status": status, "summary": summary})
+		s.pool.Exec(ctx, `UPDATE entities SET attributes = attributes || $2::jsonb, updated_at = now() WHERE id = $1`,
+			entID, string(attrs))
+		entPtr = &entID
+	}
+	severity := "info"
+	if outcome == "failure" {
+		severity = "warning"
+	}
+	_ = observability.Event(ctx, sqlcgen.New(s.pool), "task.status", entPtr, severity, "nomos", sessionID,
+		map[string]any{"status": status, "outcome": outcome, "summary": summary})
+	return nil
+}
+
+// knowledgeSlugRe matches a nomos knowledge doc slug (<kind>:nomos/<title>) as
+// printed in upsert_knowledge's result text.
+var knowledgeSlugRe = regexp.MustCompile(`[a-z]+:nomos/[a-z0-9-]+`)
+
+// linkKnowledgeToTask runs after a successful upsert_knowledge call within a
+// task: it links the created knowledge doc to the task entity (documents) so
+// get_relations(task) surfaces what the task learned, and publishes
+// knowledge.recorded for the live panel. Best-effort. The doc is ALSO linked to
+// the entity it's "about" by upsert_knowledge itself — that about-link is the
+// retrieval path future tasks use (get_entity_knowledge); this task-link is for
+// the task's own outcome/knowledge view.
+func (s *store) linkKnowledgeToTask(ctx context.Context, sessionID, resultText string) {
+	if s == nil || sessionID == "" || sessionID == "ephemeral" {
+		return
+	}
+	slug := knowledgeSlugRe.FindString(resultText)
+	if slug == "" {
+		return
+	}
+	var taskID, docID uuid.UUID
+	if err := s.pool.QueryRow(ctx,
+		`SELECT entity_id FROM agent_sessions WHERE id = $1`, sessionID).Scan(&taskID); err != nil || taskID == uuid.Nil {
+		return
+	}
+	if err := s.pool.QueryRow(ctx,
+		`SELECT id FROM entities WHERE slug = $1`, slug).Scan(&docID); err != nil {
+		return
+	}
+	s.pool.Exec(ctx, `
+		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
+		SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now()
+		WHERE NOT EXISTS (
+			SELECT 1 FROM relationships
+			WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`,
+		docID, taskID)
+	_ = observability.Event(ctx, sqlcgen.New(s.pool), "knowledge.recorded", &docID, "info", "nomos", sessionID,
+		map[string]any{"slug": slug})
 }

 func (s *store) updateSessionTitle(ctx context.Context, id, title string) error {
--- a/cmd/nomos/tasks.go
+++ b/cmd/nomos/tasks.go
@@ -0,0 +1,72 @@
+package main
+
+import (
+	"context"
+	"fmt"
+)
+
+// Task tools are nomos-LOCAL, not MCP tools. They are session-scoped, and the
+// shared MCP server (api:8090/mcp) has no session id — so these are handled
+// in-process by nomos, which knows the session/task and holds the store.
+// buildTools appends these to the model's tool list; the agent loop routes a
+// call whose name isTaskTool to handleTaskTool instead of the MCP client.
+//
+// Phase 3 ships complete_task; set_goal / propose_plan / update_plan_step /
+// ask_operator land in later phases through the same mechanism.
+
+func taskToolDefs() []toolDef {
+	return []toolDef{
+		{
+			Name: "complete_task",
+			Description: "Mark the current task finished. Call this once the goal is " +
+				"verified done — or when you've genuinely failed or only partially " +
+				"succeeded. Sets the task's outcome and a one-line summary shown on the " +
+				"task board. Record what you learned with upsert_knowledge BEFORE " +
+				"completing, so future tasks on the same entities benefit.",
+			InputSchema: map[string]any{
+				"type": "object",
+				"properties": map[string]any{
+					"outcome": map[string]any{
+						"type":        "string",
+						"enum":        []string{"success", "failure", "partial"},
+						"description": "Did the task achieve its goal?",
+					},
+					"summary": map[string]any{
+						"type":        "string",
+						"description": "One line describing the result (shown on the task card).",
+					},
+				},
+				"required": []string{"outcome", "summary"},
+			},
+		},
+	}
+}
+
+func isTaskTool(name string) bool {
+	switch name {
+	case "complete_task":
+		return true
+	default:
+		return false
+	}
+}
+
+// handleTaskTool executes a nomos-local task tool. Returns (result, true) if it
+// handled the call, or (nil, false) if name is not a local task tool (so the
+// caller forwards it to the MCP client).
+func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args map[string]any) (any, bool) {
+	switch name {
+	case "complete_task":
+		outcome, _ := args["outcome"].(string)
+		summary, _ := args["summary"].(string)
+		if outcome == "" {
+			outcome = "success"
+		}
+		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
+	default:
+		return nil, false
+	}
+}