nomos+web: streaming, provider routing, event gap-fill, embedded UI; fix approval FK & session context

Agent (cmd/nomos):
- Stream LLM tokens via NewStreaming; emit text_delta then final text.
- OpenRouter provider routing: data_collection=deny (ZDR) + require_parameters;
  NOMOS_PROVIDER_SORT opt-in; Exacto via model suffix.
- Multi-turn: reload session history into context; UI passes session id.
- Fix agent_activity logging (agent_id/session_id) and mcpClient data race.

Events (live control-room feed):
- approval.created (mcp), approval.decided (api), execution.completed/failed
  (approved-action path), signal.raised/resolved + health.changed (scheduler,
  transition-gated).

Fixes:
- createApproval FK violation (reuse execution entity) — the agent's only
  write path; log the previously-swallowed errors.

Web UI:
- Embed web/dist via //go:embed (single binary); Dockerfile builds SPA into
  the Go stage; committed .gitkeep placeholder keeps backend-only builds green.
- Caddy: Authentik-gated /agent/* -> nomos so the UI reaches the agent
  same-origin in production.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

cmd/nomos/agent.go

@@ -0,0 +1,295 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"os"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/openai/openai-go"
+	"github.com/openai/openai-go/option"
+	"github.com/openai/openai-go/shared"
+)
+
+const maxIterations = 15
+
+type agent struct {
+	client   *mcpClient
+	system   string
+	provider *openai.Client
+	model    string
+	store    *store
+	agentID  uuid.UUID
+	reqOpts  []option.RequestOption
+}
+
+func newAgent(ctx context.Context, mcpClient *mcpClient, st *store, agentSlug string) (*agent, error) {
+	system := loadSoul()
+	apiKey := os.Getenv("OPENROUTER_API_KEY")
+	model := os.Getenv("NOMOS_MODEL")
+	if model == "" {
+		model = "deepseek/deepseek-v4-flash"
+	}
+
+	provider := openai.NewClient(
+		option.WithBaseURL("https://openrouter.ai/api/v1"),
+		option.WithAPIKey(apiKey),
+	)
+
+	agentID := st.resolveAgentID(ctx, agentSlug)
+	if agentID == uuid.Nil {
+		slog.Warn("nomos: agent entity not found; tool-call activity will not be logged", "slug", agentSlug)
+	}
+
+	// OpenRouter provider routing. data_collection=deny pins to zero-data-
+	// retention providers (privacy: conversations + tool results transit
+	// OpenRouter); require_parameters ensures the routed provider actually
+	// supports tool calling. NOMOS_PROVIDER_SORT (price|throughput|latency)
+	// and Exacto tool-accuracy routing are opt-in — the latter via a model
+	// suffix in NOMOS_MODEL (e.g. "deepseek/deepseek-v4-flash:exacto"), so an
+	// unsupported value never silently breaks the confirmed routing below.
+	providerRouting := map[string]any{
+		"data_collection":    "deny",
+		"require_parameters": true,
+	}
+	if sort := os.Getenv("NOMOS_PROVIDER_SORT"); sort != "" {
+		providerRouting["sort"] = sort
+	}
+	reqOpts := []option.RequestOption{option.WithJSONSet("provider", providerRouting)}
+
+	return &agent{
+		client:   mcpClient,
+		system:   system,
+		provider: &provider,
+		model:    model,
+		store:    st,
+		agentID:  agentID,
+		reqOpts:  reqOpts,
+	}, nil
+}
+
+func loadSoul() string {
+	paths := []string{"/app/nomos/SOUL.md", "nomos/SOUL.md"}
+	for _, p := range paths {
+		if data, err := os.ReadFile(p); err == nil {
+			return string(data)
+		}
+	}
+	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.`
+}
+
+type toolDef struct {
+	Name        string         `json:"name"`
+	Description string         `json:"description"`
+	InputSchema map[string]any `json:"inputSchema"`
+}
+
+type agentEvent struct {
+	Type       string `json:"type"`
+	Data       any    `json:"data,omitempty"`
+	SessionID  string `json:"session_id,omitempty"`
+	Iteration  int    `json:"iteration,omitempty"`
+}
+
+func (a *agent) chat(ctx context.Context, sessionID, message string, emit func(agentEvent)) {
+	correlationID := uuid.New().String()
+
+	tools, err := a.buildTools()
+	if err != nil {
+		emit(agentEvent{Type: "error", Data: fmt.Sprintf("build tools: %v", err), SessionID: sessionID})
+		return
+	}
+
+	// Rebuild conversation context from persisted history so sessions are
+	// multi-turn. The current user turn is saved by the HTTP handler before
+	// this runs, so it is already included in the history for real sessions.
+	// Intermediate tool_use/tool_result pairs are not replayed (their ids
+	// must match exactly or the API rejects them); prior final answers carry
+	// the salient context. Ephemeral sessions (no store) fall back to the
+	// single incoming message.
+	messages := []openai.ChatCompletionMessageParamUnion{openai.SystemMessage(a.system)}
+	history, _ := a.store.getMessages(ctx, sessionID)
+	for _, m := range history {
+		text := extractText(m.Content)
+		switch m.Role {
+		case "user":
+			messages = append(messages, openai.UserMessage(text))
+		case "assistant":
+			if text != "" {
+				messages = append(messages, openai.AssistantMessage(text))
+			}
+		}
+	}
+	if len(history) == 0 {
+		messages = append(messages, openai.UserMessage(message))
+	}
+
+	for i := 0; i < maxIterations; i++ {
+		params := openai.ChatCompletionNewParams{
+			Model:    openai.ChatModel(a.model),
+			Messages: messages,
+			Tools:    tools,
+		}
+
+		// Stream the completion, emitting token deltas as they arrive. The
+		// accumulator reassembles the full message (content + tool calls) for
+		// the loop's control flow.
+		stream := a.provider.Chat.Completions.NewStreaming(ctx, params, a.reqOpts...)
+		acc := openai.ChatCompletionAccumulator{}
+		for stream.Next() {
+			chunk := stream.Current()
+			acc.AddChunk(chunk)
+			if len(chunk.Choices) > 0 {
+				if delta := chunk.Choices[0].Delta.Content; delta != "" {
+					emit(agentEvent{Type: "text_delta", Data: delta, SessionID: sessionID, Iteration: i + 1})
+				}
+			}
+		}
+		if err := stream.Err(); err != nil {
+			emit(agentEvent{Type: "error", Data: fmt.Sprintf("llm: %v", err), SessionID: sessionID})
+			return
+		}
+		if len(acc.Choices) == 0 {
+			emit(agentEvent{Type: "error", Data: "no choices in response", SessionID: sessionID})
+			return
+		}
+
+		msg := acc.Choices[0].Message
+
+		if len(msg.ToolCalls) == 0 {
+			emit(agentEvent{Type: "text", Data: msg.Content, SessionID: sessionID})
+			emit(agentEvent{Type: "done", Data: map[string]any{
+				"session_id":     sessionID,
+				"usage":          acc.Usage,
+				"correlation_id": correlationID,
+				"iterations":     i + 1,
+			}, SessionID: sessionID})
+			return
+		}
+
+		slog.Info("nomos: tool calls", "count", len(msg.ToolCalls), "iter", i+1, "correlation", correlationID)
+
+		messages = append(messages, msg.ToParam())
+
+		for _, tc := range msg.ToolCalls {
+			var args map[string]any
+			if err := json.Unmarshal([]byte(tc.Function.Arguments), &args); err != nil {
+				args = map[string]any{}
+			}
+
+			emit(agentEvent{
+				Type:      "tool_use",
+				Data:      map[string]any{"name": tc.Function.Name, "args": args, "id": tc.ID},
+				SessionID: sessionID,
+				Iteration: i + 1,
+			})
+
+			start := time.Now()
+			result, callErr := a.client.callTool(tc.Function.Name, args)
+			elapsed := int(time.Since(start).Milliseconds())
+
+			inputJSON, _ := json.Marshal(args)
+			inputStr := string(inputJSON)
+
+			if callErr != nil {
+				a.store.logActivity(ctx, a.agentID, sessionID, tc.Function.Name, inputStr, callErr.Error(), elapsed, false, correlationID)
+
+				emit(agentEvent{
+					Type:      "tool_result",
+					Data:      map[string]any{"name": tc.Function.Name, "error": callErr.Error(), "id": tc.ID},
+					SessionID: sessionID,
+					Iteration: i + 1,
+				})
+				messages = append(messages, openai.ToolMessage(callErr.Error(), tc.ID))
+				slog.Error("nomos: tool error", "tool", tc.Function.Name, "error", callErr, "ms", elapsed)
+				continue
+			}
+
+			resultJSON, _ := json.Marshal(result)
+			a.store.logActivity(ctx, a.agentID, sessionID, tc.Function.Name, inputStr, string(resultJSON), elapsed, true, correlationID)
+
+			emit(agentEvent{
+				Type:      "tool_result",
+				Data:      map[string]any{"name": tc.Function.Name, "result": result, "id": tc.ID},
+				SessionID: sessionID,
+				Iteration: i + 1,
+			})
+			messages = append(messages, openai.ToolMessage(string(resultJSON), tc.ID))
+			slog.Info("nomos: tool success", "tool", tc.Function.Name, "ms", elapsed)
+		}
+	}
+
+	emit(agentEvent{Type: "text", Data: "Agent loop: max iterations reached without final answer.", SessionID: sessionID})
+	emit(agentEvent{Type: "done", Data: map[string]any{
+		"session_id":     sessionID,
+		"correlation_id": correlationID,
+		"iterations":     maxIterations,
+	}, SessionID: sessionID})
+}
+
+// extractText pulls the "text" field from a persisted message's JSONB content.
+func extractText(content json.RawMessage) string {
+	var m struct {
+		Text string `json:"text"`
+	}
+	if err := json.Unmarshal(content, &m); err != nil {
+		return ""
+	}
+	return m.Text
+}
+
+func (a *agent) buildTools() ([]openai.ChatCompletionToolParam, error) {
+	defs, err := a.client.listToolsFull()
+	if err != nil {
+		return nil, err
+	}
+
+	var tools []openai.ChatCompletionToolParam
+	for _, d := range defs {
+		params := shared.FunctionParameters(d.InputSchema)
+		if params == nil {
+			params = shared.FunctionParameters{"type": "object", "properties": map[string]any{}}
+		}
+
+		tools = append(tools, openai.ChatCompletionToolParam{
+			Type: "function",
+			Function: shared.FunctionDefinitionParam{
+				Name:        d.Name,
+				Description: openai.String(d.Description),
+				Parameters:  params,
+			},
+		})
+	}
+	return tools, nil
+}
+
+func (c *mcpClient) listToolsFull() ([]toolDef, error) {
+	resp, err := c.doRequest("tools/list", map[string]any{})
+	if err != nil {
+		return nil, err
+	}
+	var tr struct {
+		Tools []struct {
+			Name        string         `json:"name"`
+			Description string         `json:"description"`
+			InputSchema map[string]any `json:"inputSchema"`
+		} `json:"tools"`
+	}
+	if err := json.Unmarshal(resp.Result, &tr); err != nil {
+		return nil, err
+	}
+	out := make([]toolDef, len(tr.Tools))
+	for i, t := range tr.Tools {
+		out[i] = toolDef{
+			Name:        t.Name,
+			Description: t.Description,
+			InputSchema: t.InputSchema,
+		}
+	}
+	return out, nil
+}

cmd/nomos/main.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 )
@@ -20,7 +21,6 @@ func main() {
 		fmt.Fprintln(os.Stderr, "usage: nomos serve")
 		os.Exit(1)
 	}
-
 	mcpURL := os.Getenv("NOMOS_MCP_URL")
 	if mcpURL == "" {
 		mcpURL = "http://localhost:8090/mcp"
@@ -31,6 +31,11 @@ func main() {
 		agentSlug = "agent:nomos"
 	}
 
+	databaseURL := os.Getenv("DATABASE_URL")
+	if databaseURL == "" {
+		databaseURL = os.Getenv("OIKOS_DATABASE_URL")
+	}
+
 	switch os.Args[1] {
 	case "serve":
 		ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
@@ -42,6 +47,21 @@ func main() {
 			os.Exit(1)
 		}
 
+		st, err := newStore(ctx, databaseURL)
+		if err != nil {
+			slog.Error("nomos: db connect", "error", err)
+			os.Exit(1)
+		}
+		if st != nil {
+			defer st.close()
+		}
+
+		nAgent, err := newAgent(ctx, client, st, agentSlug)
+		if err != nil {
+			slog.Error("nomos: agent init", "error", err)
+			os.Exit(1)
+		}
+
 		mux := http.NewServeMux()
 		mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(200)
@@ -50,6 +70,15 @@ func main() {
 		mux.HandleFunc("/query", func(w http.ResponseWriter, r *http.Request) {
 			handleQuery(w, r, client, agentSlug, mcpURL)
 		})
+		mux.HandleFunc("/chat", func(w http.ResponseWriter, r *http.Request) {
+			handleChat(w, r, nAgent, st)
+		})
+		mux.HandleFunc("/sessions", func(w http.ResponseWriter, r *http.Request) {
+			handleSessionsList(w, r, st)
+		})
+		mux.HandleFunc("/sessions/", func(w http.ResponseWriter, r *http.Request) {
+			handleSessionDetail(w, r, st)
+		})
 
 		addr := os.Getenv("NOMOS_LISTEN")
 		if addr == "" {
@@ -58,7 +87,7 @@ func main() {
 
 		srv := &http.Server{Addr: addr, Handler: mux}
 		go func() {
-			slog.Info("nomos: gateway listening", "addr", addr, "mcp", mcpURL)
+			slog.Info("nomos: gateway listening", "addr", addr, "mcp", mcpURL, "db", databaseURL != "")
 			if err := srv.ListenAndServe(); err != http.ErrServerClosed {
 				slog.Error("nomos: serve", "error", err)
 			}
@@ -75,7 +104,130 @@ func main() {
 	}
 }
 
-// handleQuery maps structured queries to MCP tool calls.
+func sseEvent(w http.ResponseWriter, flusher http.Flusher, event agentEvent) {
+	data, _ := json.Marshal(event)
+	fmt.Fprintf(w, "data: %s\n\n", data)
+	flusher.Flush()
+}
+
+func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", 405)
+		return
+	}
+
+	var req struct {
+		SessionID string `json:"session_id"`
+		Message   string `json:"message"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, "bad request: "+err.Error(), 400)
+		return
+	}
+	if req.Message == "" {
+		http.Error(w, "message is required", 400)
+		return
+	}
+
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming not supported", 500)
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.WriteHeader(200)
+
+	ctx := r.Context()
+	sessionID := req.SessionID
+
+	if sessionID == "" {
+		title := truncate(req.Message, 80)
+		sess, err := st.createSession(ctx, title)
+		if err != nil {
+			slog.Error("nomos: create session", "error", err)
+			sessionID = "ephemeral"
+		} else {
+			sessionID = sess.ID
+		}
+	} else {
+		st.touchSession(ctx, sessionID)
+	}
+
+	slog.Info("nomos: chat", "session", sessionID, "message", truncate(req.Message, 100))
+
+	userMsg, _ := json.Marshal(map[string]any{"role": "user", "text": req.Message})
+	st.saveMessage(ctx, sessionID, "user", userMsg)
+
+	sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})
+
+	toolCalls := []map[string]any{}
+	var finalText string
+
+	a.chat(ctx, sessionID, req.Message, func(ev agentEvent) {
+		if ev.Type == "tool_use" || ev.Type == "tool_result" {
+			if m, ok := ev.Data.(map[string]any); ok {
+				m["type"] = ev.Type
+				toolCalls = append(toolCalls, m)
+			}
+		}
+		if ev.Type == "text" {
+			finalText, _ = ev.Data.(string)
+		}
+		sseEvent(w, flusher, ev)
+	})
+
+	assistantMsg, _ := json.Marshal(map[string]any{
+		"role":       "assistant",
+		"text":       finalText,
+		"tool_calls": toolCalls,
+	})
+	st.saveMessage(ctx, sessionID, "assistant", assistantMsg)
+}
+
+func handleSessionsList(w http.ResponseWriter, r *http.Request, st *store) {
+	if st == nil {
+		w.Header().Set("Content-Type", "application/json")
+		json.NewEncoder(w).Encode(map[string]any{"sessions": []any{}})
+		return
+	}
+
+	if r.Method == http.MethodOptions {
+		return
+	}
+
+	sessions, err := st.listSessions(r.Context())
+	if err != nil {
+		http.Error(w, err.Error(), 500)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"sessions": sessions})
+}
+
+func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store) {
+	if st == nil {
+		http.Error(w, "not found", 404)
+		return
+	}
+
+	id := strings.TrimPrefix(r.URL.Path, "/sessions/")
+	if id == "" {
+		http.Error(w, "session id required", 400)
+		return
+	}
+
+	messages, err := st.getMessages(r.Context(), id)
+	if err != nil {
+		http.Error(w, err.Error(), 500)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(map[string]any{"session_id": id, "messages": messages})
+}
+
 func handleQuery(w http.ResponseWriter, r *http.Request, client *mcpClient, agentSlug, mcpURL string) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "method not allowed", 405)
@@ -83,8 +235,8 @@ func handleQuery(w http.ResponseWriter, r *http.Request, client *mcpClient, agen
 	}
 
 	var req struct {
-		Query string `json:"query"`
-		Tool  string `json:"tool"`
+		Query string         `json:"query"`
+		Tool  string         `json:"tool"`
 		Args  map[string]any `json:"args"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
@@ -93,105 +245,79 @@ func handleQuery(w http.ResponseWriter, r *http.Request, client *mcpClient, agen
 	}
 
 	start := time.Now()
-	var result any
-	var err error
 
-	// Direct tool call (structured)
 	if req.Tool != "" {
-		result, err = client.callTool(req.Tool, req.Args)
-	} else {
-		// Natural-language-ish query routing
-		q := strings.ToLower(req.Query)
-		result, err = routeQuery(client, q, agentSlug)
-	}
-
-	duration := time.Since(start).Milliseconds()
-
-	if err != nil {
-		slog.Error("nomos: query failed", "query", req.Query, "error", err)
+		result, err := client.callTool(req.Tool, req.Args)
+		duration := time.Since(start).Milliseconds()
+		if err != nil {
+			slog.Error("nomos: query failed", "tool", req.Tool, "error", err)
+			w.Header().Set("Content-Type", "application/json")
+			json.NewEncoder(w).Encode(map[string]any{
+				"error":      err.Error(),
+				"elapsed_ms": duration,
+				"agent_slug": agentSlug,
+			})
+			return
+		}
 		w.Header().Set("Content-Type", "application/json")
 		json.NewEncoder(w).Encode(map[string]any{
-			"error":       err.Error(),
-			"elapsed_ms":  duration,
-			"agent_slug":  agentSlug,
+			"result":     result,
+			"elapsed_ms": duration,
+			"agent_slug": agentSlug,
+			"mcp_url":    mcpURL,
 		})
 		return
 	}
 
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]any{
-		"result":      result,
-		"elapsed_ms":  duration,
-		"agent_slug":  agentSlug,
-		"mcp_url":     mcpURL,
-	})
+	if req.Query != "" {
+		if strings.Contains(strings.ToLower(req.Query), "what can you do") ||
+			strings.Contains(strings.ToLower(req.Query), "help") {
+
+			tools, err := client.listTools()
+			duration := time.Since(start).Milliseconds()
+			if err != nil {
+				w.Header().Set("Content-Type", "application/json")
+				json.NewEncoder(w).Encode(map[string]any{
+					"error":      err.Error(),
+					"elapsed_ms": duration,
+				})
+				return
+			}
+			w.Header().Set("Content-Type", "application/json")
+			json.NewEncoder(w).Encode(map[string]any{
+				"message":    "natural language queries belong to /chat. Use structured /query with 'tool' for direct MCP calls.",
+				"tools":      tools,
+				"elapsed_ms": duration,
+			})
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		json.NewEncoder(w).Encode(map[string]any{
+			"message":    "natural language queries belong to /chat. Use structured /query with 'tool' for direct MCP calls.",
+			"elapsed_ms": time.Since(start).Milliseconds(),
+		})
+		return
+	}
+
+	http.Error(w, "either 'tool' or 'query' required", 400)
 }
 
-// routeQuery maps natural-language-style queries to MCP tool calls.
-func routeQuery(client *mcpClient, query, agentSlug string) (any, error) {
-	switch {
-	case strings.Contains(query, "depends on") || strings.Contains(query, "depend on"):
-		entity := extractEntity(query)
-		if entity == "" {
-			return nil, fmt.Errorf("no entity found in query: %s", query)
-		}
-		return client.callTool("get_blast_radius", map[string]any{
-			"entity_id": entity,
-		})
-
-	case strings.Contains(query, "what is") || strings.Contains(query, "describe"):
-		entity := extractEntity(query)
-		if entity == "" {
-			entity = query
-		}
-		return client.callTool("get_entity", map[string]any{
-			"slug_or_id": entity,
-		})
-
-	case strings.Contains(query, "health") || strings.Contains(query, "status"):
-		return client.callTool("get_health_summary", map[string]any{})
-
-	case strings.Contains(query, "restart") || strings.Contains(query, "reload"):
-		entity := extractEntity(query)
-		if entity == "" {
-			return nil, fmt.Errorf("no entity found in query: %s", query)
-		}
-		return client.callTool("request_execution", map[string]any{
-			"target": entity,
-			"action": "restart",
-		})
-
-	case strings.Contains(query, "what can you do") || strings.Contains(query, "help"):
-		return client.callTool("tools/list", nil)
-
-	default:
-		return client.callTool("get_health_summary", map[string]any{})
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
 	}
-}
-
-// extractEntity guesses an entity slug from a query.
-func extractEntity(query string) string {
-	for _, slug := range []string{"authentik", "caddy", "vaultwarden", "gitea", "immich"} {
-		if strings.Contains(query, slug) {
-			return "service:" + slug
-		}
-	}
-	if strings.Contains(query, "mac-mini") {
-		return "host:mac-mini"
-	}
-	if strings.Contains(query, "hubris") {
-		return "host:hubris"
-	}
-	return ""
+	return s[:n] + "..."
 }
 
 // ─── MCP Streamable HTTP client ────────────────────────────────────────
 
 type mcpClient struct {
-	baseURL    string
-	sessionID  string
-	http       *http.Client
-	nextID     int
+	baseURL   string
+	sessionID string
+	http      *http.Client
+	nextID    int
+	mu        sync.Mutex // MCP is one stateful session; serialize concurrent calls
 }
 
 func newMCPClient(baseURL string) (*mcpClient, error) {
@@ -200,7 +326,6 @@ func newMCPClient(baseURL string) (*mcpClient, error) {
 		http:    &http.Client{Timeout: 30 * time.Second},
 	}
 
-	// Initialize session
 	resp, err := c.doRequest("initialize", map[string]any{
 		"protocolVersion": "2024-11-05",
 		"capabilities":    map[string]any{},
@@ -214,7 +339,6 @@ func newMCPClient(baseURL string) (*mcpClient, error) {
 	}
 	c.sessionID = resp.sessionID
 
-	// Send initialized notification
 	c.doRequest("notifications/initialized", map[string]any{})
 
 	slog.Info("nomos: mcp connected", "session", c.sessionID[:16]+"...")
@@ -228,6 +352,8 @@ type mcpJSONRPCResponse struct {
 }
 
 func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	c.nextID++
 	body, _ := json.Marshal(map[string]any{
 		"jsonrpc": "2.0",
@@ -255,7 +381,6 @@ func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPC
 	result := &mcpJSONRPCResponse{}
 	result.sessionID = resp.Header.Get("Mcp-Session-Id")
 
-	// Parse SSE stream: "event: message\ndata: <json>\n\n"
 	scanner := bufio.NewScanner(resp.Body)
 	for scanner.Scan() {
 		line := scanner.Text()
@@ -287,7 +412,6 @@ func (c *mcpClient) callTool(name string, args map[string]any) (any, error) {
 		return nil, err
 	}
 
-	// Parse MCP content: { "content": [{ "type": "text", "text": "..." }] }
 	var toolResult struct {
 		Content []struct {
 			Type string `json:"type"`
@@ -301,7 +425,6 @@ func (c *mcpClient) callTool(name string, args map[string]any) (any, error) {
 	var texts []string
 	for _, c := range toolResult.Content {
 		if c.Type == "text" {
-			// Try to parse as JSON for structured display
 			var parsed any
 			if json.Unmarshal([]byte(c.Text), &parsed) == nil {
 				return parsed, nil
@@ -337,5 +460,4 @@ func (c *mcpClient) listTools() ([]string, error) {
 }
 
 func (c *mcpClient) close() {
-	// MCP sessions are ephemeral; no explicit close needed
 }

cmd/nomos/store.go

@@ -0,0 +1,156 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type store struct {
+	pool *pgxpool.Pool
+}
+
+func newStore(ctx context.Context, databaseURL string) (*store, error) {
+	if databaseURL == "" {
+		return nil, nil
+	}
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("connect db: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("ping db: %w", err)
+	}
+	return &store{pool: pool}, nil
+}
+
+func (s *store) close() {
+	if s.pool != nil {
+		s.pool.Close()
+	}
+}
+
+type session struct {
+	ID           string    `json:"id"`
+	Title        string    `json:"title"`
+	Actor        string    `json:"actor"`
+	CreatedAt    time.Time `json:"created_at"`
+	LastActiveAt time.Time `json:"last_active_at"`
+}
+
+type message struct {
+	ID        string          `json:"id"`
+	SessionID string          `json:"session_id"`
+	Role      string          `json:"role"`
+	Content   json.RawMessage `json:"content"`
+	CreatedAt time.Time       `json:"created_at"`
+}
+
+func (s *store) createSession(ctx context.Context, title string) (*session, error) {
+	if s == nil {
+		return &session{ID: "ephemeral", Title: title, Actor: "agent:nomos"}, nil
+	}
+	var id string
+	err := s.pool.QueryRow(ctx,
+		`INSERT INTO agent_sessions (title, actor) VALUES ($1, 'agent:nomos') RETURNING id`,
+		title).Scan(&id)
+	if err != nil {
+		return nil, err
+	}
+	return &session{ID: id, Title: title, Actor: "agent:nomos", CreatedAt: time.Now(), LastActiveAt: time.Now()}, nil
+}
+
+func (s *store) saveMessage(ctx context.Context, sessionID, role string, content json.RawMessage) error {
+	if s == nil {
+		return nil
+	}
+	_, err := s.pool.Exec(ctx,
+		`INSERT INTO agent_messages (session_id, role, content) VALUES ($1, $2, $3)`,
+		sessionID, role, content)
+	return err
+}
+
+func (s *store) touchSession(ctx context.Context, id string) {
+	if s != nil {
+		s.pool.Exec(ctx, `UPDATE agent_sessions SET last_active_at=now() WHERE id=$1`, id)
+	}
+}
+
+func (s *store) listSessions(ctx context.Context) ([]session, error) {
+	if s == nil {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx,
+		`SELECT id, title, actor, created_at, last_active_at FROM agent_sessions ORDER BY last_active_at DESC LIMIT 50`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []session
+	for rows.Next() {
+		var sess session
+		if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.CreatedAt, &sess.LastActiveAt); err != nil {
+			return nil, err
+		}
+		out = append(out, sess)
+	}
+	return out, rows.Err()
+}
+
+func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, error) {
+	if s == nil {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx,
+		`SELECT id, session_id, role, content, created_at FROM agent_messages WHERE session_id=$1 ORDER BY created_at ASC`,
+		sessionID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []message
+	for rows.Next() {
+		var m message
+		if err := rows.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &m.CreatedAt); err != nil {
+			return nil, err
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// resolveAgentID looks up the UUID of the agent entity (e.g. "agent:nomos").
+// Returns uuid.Nil if the store is absent or the slug is unknown.
+func (s *store) resolveAgentID(ctx context.Context, slug string) uuid.UUID {
+	if s == nil {
+		return uuid.Nil
+	}
+	var id uuid.UUID
+	if err := s.pool.QueryRow(ctx, `SELECT id FROM entities WHERE slug = $1`, slug).Scan(&id); err != nil {
+		return uuid.Nil
+	}
+	return id
+}
+
+// logActivity records a tool call. agent_id is the agent entity UUID and is
+// NOT NULL in the schema, so we skip logging when it can't be resolved.
+// The (nullable) session_id column carries the conversation id.
+func (s *store) logActivity(ctx context.Context, agentID uuid.UUID, sessionID, toolName, inputSummary, outputSummary string, durationMs int, success bool, correlationID string) {
+	if s == nil || agentID == uuid.Nil {
+		return
+	}
+	s.pool.Exec(ctx, `
+		INSERT INTO agent_activity
+		    (agent_id, session_id, activity_type, tool_name, input_summary, output_summary,
+		     duration_ms, success, correlation_id)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
+		agentID, sessionID, "tool_call", toolName, inputSummary, outputSummary,
+		durationMs, success, correlationID)
+}