oikos/cmd/nomos/main.go

package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/dtoro/oikos/internal/safego"
	"github.com/google/uuid"
)

func main() {
	if len(os.Args) < 2 {
		fmt.Fprintln(os.Stderr, "usage: nomos serve")
		os.Exit(1)
	}
	mcpURL := os.Getenv("NOMOS_MCP_URL")
	if mcpURL == "" {
		mcpURL = "http://localhost:8090/mcp"
	}
	// api's combinedAuth requires a bearer token on every request (no
	// dev-open bypass — plans/2026-07-12-wails-desktop-app.md 0.4); this is
	// the same shared secret api validates against (OIKOS_MCP_BEARER_TOKEN).
	mcpToken := os.Getenv("OIKOS_MCP_BEARER_TOKEN")

	agentSlug := os.Getenv("NOMOS_AGENT_SLUG")
	if agentSlug == "" {
		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)
		defer cancel()

		// One MCP client PER SESSION, not one shared client for the whole
		// process — see mcpClientPool's doc comment. A dedicated client is
		// created lazily on each session's first tool call.
		clientPool := newMCPClientPool(mcpURL, mcpToken)
		// Prove connectivity at startup the same way the old single-client
		// constructor did, so a misconfigured/unreachable MCP endpoint still
		// fails fast on boot instead of only on the first real chat. Doesn't
		// reuse the pool (nothing to key it by yet) — just a throwaway probe.
		if probe, err := newMCPClient(mcpURL, mcpToken); err != nil {
			slog.Error("nomos: mcp connect", "url", mcpURL, "error", err)
			os.Exit(1)
		} else {
			probe.close()
		}

		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, clientPool, st, agentSlug)
		if err != nil {
			slog.Error("nomos: agent init", "error", err)
			os.Exit(1)
		}

		// Event-driven auto-continuation: feed finished async executions back
		// into the agent so an approved plan runs to completion (and recovers
		// from failures) without the operator ticking it forward each step.
		safego.Go("nomos:continuation-worker", func() { nAgent.runContinuationWorker(ctx) })

		// Idle sweep for stalled goal-bearing tasks (fix 2+3 of
		// plans/2026-07-11-task-completion-safety-net.md) — a coarser,
		// slower-ticking counterpart to the continuation worker above.
		safego.Go("nomos:idle-sweep-worker", func() { nAgent.runIdleSweepWorker(ctx) })

		safego.Go("nomos:mcp-pool-sweeper", func() {
			ticker := time.NewTicker(5 * time.Minute)
			defer ticker.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					clientPool.sweep()
				}
			}
		})

		// Stale execution sweep: cancels non-terminal executions older than
		// 10 minutes (orphaned by MCP timeouts — see cleanupStaleExecutions).
		safego.Go("nomos:stale-execution-sweeper", func() {
			ticker := time.NewTicker(5 * time.Minute)
			defer ticker.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					st.cleanupStaleExecutions(ctx, 10*time.Minute)
				}
			}
		})

		mux := http.NewServeMux()
		mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
			w.WriteHeader(200)
			w.Write([]byte("ok"))
		})
		mux.HandleFunc("/query", func(w http.ResponseWriter, r *http.Request) {
			handleQuery(w, r, clientPool, 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, nAgent)
		})

		addr := os.Getenv("NOMOS_LISTEN")
		if addr == "" {
			addr = ":8092"
		}

		srv := &http.Server{Addr: addr, Handler: mux}
		safego.Go("nomos:http-server", func() {
			slog.Info("nomos: gateway listening", "addr", addr, "mcp", mcpURL, "db", databaseURL != "")
			if err := srv.ListenAndServe(); err != http.ErrServerClosed {
				slog.Error("nomos: serve", "error", err)
			}
		})

		<-ctx.Done()
		slog.Info("nomos: shutting down")
		srv.Shutdown(context.Background())
		clientPool.closeAll()

	default:
		fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1])
		os.Exit(1)
	}
}

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 == "" && 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() {
			base := "[System: the operator's connection was re-established. The task may have progressed in the background.]"
			note := st.enrichResumeNote(context.Background(), req.SessionID, base)
			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)
		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

	// pctx (persistence context) is deliberately context.Background(), not
	// ctx/r.Context(), for every DB write in this handler — ctx cancels the
	// instant the client disconnects (Stop button, tab close, network blip),
	// and a write made with an already-cancelled context fails. Before this
	// fix, the assistant message was only ever saved ONCE, at the very end,
	// using ctx — so a disconnect mid-turn silently lost the ENTIRE turn's
	// tool-call history from the persisted transcript, even though real work
	// (executions launched, knowledge written) had already happened
	// server-side. The agent's own work (a.chat below) still correctly stops
	// when ctx cancels — this only changes what happens to persistence.
	pctx := context.Background()

	if sessionID == "" {
		title := truncate(req.Message, 80)
		sess, err := st.createSession(pctx, title)
		if err != nil {
			slog.Error("nomos: create session", "error", err)
			sessionID = "ephemeral"
		} else {
			sessionID = sess.ID
		}
	} else {
		// P2 iteration: if the operator sends a follow-up on a session
		// that already reached a terminal state (done/failed), reopen it
		// so a new sub-task can be framed (set_goal → propose_plan →
		// execute). reopenSession marks the prior plan's steps as
		// `replaced` (proposePlan ignores those) and clears outcome/
		// summary. Without this, propose_plan refuses the follow-up with
		// errPlanInFlight because the prior steps are all `done`. If the
		// session is still active, reopen is a no-op — the follow-up is
		// just a continuation of in-flight work.
		st.reopenSession(pctx, sessionID)
		st.touchSession(pctx, 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(pctx, sessionID, "user", userMsg)

	// If this task has a pending operator question, the incoming message IS the
	// answer — close it so the panel clears. No separate resume needed: this
	// chat turn is the resume, and the agent sees the question + answer in its
	// replayed history.
	if qid := st.openQuestionID(pctx, sessionID); qid != "" {
		st.answerQuestion(pctx, sessionID, qid, req.Message)
	}

	sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})

	toolCalls := []map[string]any{}
	// P3: accumulate per-iteration reasoning instead of overwriting with
	// the final `text` event. The agent loop emits a `text` event for each
	// LLM iteration that produced text (intermediate reasoning before tool
	// calls + the final answer). Without accumulation, only the last `text`
	// survives in the persisted row — a reload shows the final summary but
	// not the thinking that led to each tool call.
	var textParts []string
	var finalText string

	// Incremental persistence, mirroring resumeSession's existing
	// placeholder+update pattern (continue.go): insert a placeholder now,
	// update the SAME row after every tool call, so whatever happened before
	// an abort is never lost — only what hadn't happened yet is.
	placeholder, _ := json.Marshal(map[string]any{"role": "assistant", "text": ""})
	msgID, err := st.insertMessageReturningID(pctx, sessionID, "assistant", placeholder)
	if err != nil {
		slog.Error("nomos: chat placeholder insert failed", "session", sessionID, "error", err)
	}
	persist := func() {
		if msgID == uuid.Nil {
			return
		}
		body, _ := json.Marshal(map[string]any{
			"role":       "assistant",
			"text":       finalText,
			"tool_calls": toolCalls,
		})
		st.updateMessage(pctx, msgID, body)
	}

	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
				// One entry per tool call: tool_use creates it, tool_result
				// merges the result into the same entry (matched by id).
				// Before this fix, both events appended separate entries,
				// doubling every tool call in the persisted transcript
				// (confirmed pre-existing in d9cdcee1, v0.3.x era).
				id, _ := m["id"].(string)
				if id != "" && ev.Type == "tool_result" {
					for _, existing := range toolCalls {
						if eID, _ := existing["id"].(string); eID == id {
							for k, v := range m {
								existing[k] = v
							}
							break
						}
					}
				} else {
					toolCalls = append(toolCalls, m)
				}
			}
			persist() // live: survives even if the client disconnects right after
		}
		if ev.Type == "text" {
			// P3: accumulate. Each `text` event is one iteration's reasoning
			// (or the final answer). Join with newlines so the persisted row
			// reads as the full transcript of what the agent said, not just
			// the last thing.
			if t, ok := ev.Data.(string); ok && t != "" {
				textParts = append(textParts, t)
				finalText = strings.Join(textParts, "\n\n")
				persist()
			}
		}
		sseEvent(w, flusher, ev)
	})

	// B.6: if the turn ended with no text and no tool calls (the model
	// empty-response'd and all retries failed), delete the placeholder row
	// instead of persisting an empty bubble. The error event was already
	// streamed to the frontend via the 'done with error=true' event, so the
	// operator sees the error inline — an empty assistant bubble in the
	// transcript adds nothing and looks like the agent is broken.
	if finalText == "" && len(toolCalls) == 0 && msgID != uuid.Nil {
		st.deleteMessage(pctx, msgID)
	} else {
		persist() // final state — same row, updated one last time with the concluding text
	}

	// Generate a meaningful title from the assistant's first answer
	// instead of reusing the raw user message for every session.
	if finalText != "" && sessionID != "ephemeral" {
		title := truncate(finalText, 80)
		if title != "" {
			st.updateSessionTitle(pctx, sessionID, title)
		}
	}
}

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, a *agent) {
	if st == nil {
		http.Error(w, "not found", 404)
		return
	}

	rest := strings.TrimPrefix(r.URL.Path, "/sessions/")
	parts := strings.Split(rest, "/")
	id := parts[0]
	if id == "" {
		http.Error(w, "session id required", 400)
		return
	}

	// POST /sessions/{id}/questions/{qid}/answer — the operator answers a
	// pinned question from the context panel; resume the agent with the answer.
	if len(parts) == 4 && parts[1] == "questions" && parts[3] == "answer" {
		if r.Method != http.MethodPost {
			http.Error(w, "method not allowed", 405)
			return
		}
		handleAnswerQuestion(w, r, st, a, id, parts[2])
		return
	}

	// POST /sessions/{id}/resume — the operator asks the agent to continue.
	if len(parts) == 2 && parts[1] == "resume" && r.Method == http.MethodPost {
		base := "[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.]"
		note := st.enrichResumeNote(context.Background(), id, base)
		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.
	if len(parts) == 2 && r.Method == http.MethodGet {
		switch parts[1] {
		case "plan":
			steps, err := st.getPlanSteps(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{"steps": steps})
			return
		case "questions":
			questions, err := st.getQuestions(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{"questions": questions})
			return
		}
	}

	switch r.Method {
	case http.MethodDelete:
		if err := st.deleteSession(r.Context(), id); err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		w.WriteHeader(204)

	case http.MethodGet:
		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})

	default:
		http.Error(w, "method not allowed", 405)
	}
}

// handleAnswerQuestion records the operator's answer to a pinned question and
// resumes the agent in the background with that answer injected. Returns 202 —
// the agent's response lands via the normal message-polling path, not this POST.
func handleAnswerQuestion(w http.ResponseWriter, r *http.Request, st *store, a *agent, sessionID, questionID string) {
	var req struct {
		Answer string `json:"answer"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil || strings.TrimSpace(req.Answer) == "" {
		http.Error(w, "answer is required", 400)
		return
	}
	prompt, _, _ := st.getQuestion(r.Context(), questionID)
	if err := st.answerQuestion(r.Context(), sessionID, questionID, req.Answer); err != nil {
		http.Error(w, err.Error(), 500)
		return
	}
	if a != nil {
		base := fmt.Sprintf("[System: the operator answered your question %q with: %q. "+
			"Continue the task from here — do not re-ask.]", prompt, req.Answer)
		note := st.enrichResumeNote(context.Background(), sessionID, base)
		safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), sessionID, note) })
	}
	w.WriteHeader(202)
}

func handleQuery(w http.ResponseWriter, r *http.Request, pool *mcpClientPool, agentSlug, mcpURL string) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", 405)
		return
	}

	var req struct {
		Query string         `json:"query"`
		Tool  string         `json:"tool"`
		Args  map[string]any `json:"args"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "bad request: "+err.Error(), 400)
		return
	}

	// The structured /query endpoint is stateless/session-less — "query" is a
	// fixed pool key (not a real session id) so repeated calls reuse one
	// dedicated connection instead of paying a fresh MCP handshake every time,
	// while still never sharing a connection with an actual chat task.
	client, err := pool.get("query")
	if err != nil {
		http.Error(w, "mcp unavailable: "+err.Error(), 502)
		return
	}

	start := time.Now()

	if req.Tool != "" {
		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{
			"result":     result,
			"elapsed_ms": duration,
			"agent_slug": agentSlug,
			"mcp_url":    mcpURL,
		})
		return
	}

	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)
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "..."
}

// ─── MCP Streamable HTTP client ────────────────────────────────────────

type mcpClient struct {
	baseURL   string
	token     string // OIKOS_MCP_BEARER_TOKEN — api's combinedAuth requires it on every request (no dev-open bypass)
	sessionID string
	http      *http.Client
	nextID    int
	mu        sync.Mutex // this client is one stateful MCP session; serialize ITS OWN calls

	// toolsCache holds the last tools/list result. The tool list is static
	// for the lifetime of one MCP connection — it only changes when the api
	// process (re)registers tools, i.e. on a restart, which this client
	// already detects and reacts to via reconnectLocked. Without this,
	// buildTools (called at the start of EVERY chat turn, including every
	// auto-continuation resume) paid a full tools/list round-trip every
	// single time for a list that's almost always identical to the last one.
	// Guarded separately from mu (not reused) so a cache check never
	// contends with an in-flight doRequest call for a different method.
	toolsMu    sync.Mutex
	toolsCache []toolDef
}

func newMCPClient(baseURL, token string) (*mcpClient, error) {
	c := &mcpClient{
		baseURL: baseURL,
		token:   token,
		http:    &http.Client{Timeout: 30 * time.Second},
	}

	resp, err := c.doRequest("initialize", map[string]any{
		"protocolVersion": "2024-11-05",
		"capabilities":    map[string]any{},
		"clientInfo":      map[string]any{"name": "nomos", "version": "2.0"},
	})
	if err != nil {
		return nil, fmt.Errorf("initialize: %w", err)
	}
	if resp.sessionID == "" {
		return nil, fmt.Errorf("no session ID in initialize response")
	}
	c.sessionID = resp.sessionID

	c.doRequest("notifications/initialized", map[string]any{})

	slog.Info("nomos: mcp connected", "session", c.sessionID[:16]+"...")
	return c, nil
}

type mcpJSONRPCResponse struct {
	sessionID string
	Result    json.RawMessage `json:"result"`
	Error     json.RawMessage `json:"error"`
}

// errStaleSession signals that the MCP server rejected our session id (e.g.
// after an api/MCP restart), so the client should re-initialize and retry.
var errStaleSession = fmt.Errorf("mcp session stale")

// doRequest serializes MCP calls and transparently re-initializes the session
// if the server has forgotten it (common after an api redeploy), retrying the
// original call once. Without this, an api restart permanently breaks nomos
// until it is itself restarted.
func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	resp, err := c.send(method, params)
	if err != nil && method != "initialize" && errors.Is(err, errStaleSession) {
		slog.Warn("nomos: mcp session stale, reconnecting")
		if rerr := c.reconnectLocked(); rerr != nil {
			return nil, fmt.Errorf("mcp reconnect: %w (original: %v)", rerr, err)
		}
		return c.send(method, params)
	}
	return resp, err
}

// reconnectLocked re-initializes the MCP session. The caller must hold c.mu.
func (c *mcpClient) reconnectLocked() error {
	c.sessionID = ""
	// A reconnect means the api process was restarted (or forgot us) — its
	// tool registration may have changed, so the cached list is no longer
	// trustworthy.
	c.toolsMu.Lock()
	c.toolsCache = nil
	c.toolsMu.Unlock()
	resp, err := c.send("initialize", map[string]any{
		"protocolVersion": "2024-11-05",
		"capabilities":    map[string]any{},
		"clientInfo":      map[string]any{"name": "nomos", "version": "2.0"},
	})
	if err != nil {
		return err
	}
	if resp.sessionID == "" {
		return fmt.Errorf("no session ID on re-initialize")
	}
	c.sessionID = resp.sessionID
	_, _ = c.send("notifications/initialized", map[string]any{})
	slog.Info("nomos: mcp reconnected", "session", c.sessionID[:16]+"...")
	return nil
}

// send performs one MCP round-trip. It does not lock; callers hold c.mu.
func (c *mcpClient) send(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
	c.nextID++
	body, _ := json.Marshal(map[string]any{
		"jsonrpc": "2.0",
		"method":  method,
		"params":  params,
		"id":      c.nextID,
	})

	req, err := http.NewRequest(http.MethodPost, c.baseURL, bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/json, text/event-stream")
	if c.sessionID != "" {
		req.Header.Set("Mcp-Session-Id", c.sessionID)
	}
	if c.token != "" {
		req.Header.Set("Authorization", "Bearer "+c.token)
	}

	resp, err := c.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// A rejected/unknown session comes back as 4xx (commonly 400/404).
	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest {
		return nil, errStaleSession
	}

	result := &mcpJSONRPCResponse{}
	result.sessionID = resp.Header.Get("Mcp-Session-Id")

	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
	gotData := false
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "data: ") {
			gotData = true
			data := line[6:]
			if err := json.Unmarshal([]byte(data), result); err != nil {
				return nil, fmt.Errorf("parse response: %w", err)
			}
		}
	}

	if result.Error != nil {
		return nil, fmt.Errorf("rpc error: %s", string(result.Error))
	}

	// Empty body with no result and a session set: the server likely dropped
	// our session. Notifications legitimately return no data, so exempt them.
	if !gotData && result.Result == nil && method != "notifications/initialized" {
		return nil, errStaleSession
	}

	if result.sessionID != "" {
		c.sessionID = result.sessionID
	}

	return result, nil
}

func (c *mcpClient) callTool(name string, args map[string]any) (any, error) {
	resp, err := c.doRequest("tools/call", map[string]any{
		"name":      name,
		"arguments": args,
	})
	if err != nil {
		return nil, err
	}

	var toolResult struct {
		Content []struct {
			Type string `json:"type"`
			Text string `json:"text"`
		} `json:"content"`
	}
	if err := json.Unmarshal(resp.Result, &toolResult); err != nil {
		return string(resp.Result), nil
	}

	var texts []string
	for _, c := range toolResult.Content {
		if c.Type == "text" {
			var parsed any
			if json.Unmarshal([]byte(c.Text), &parsed) == nil {
				return parsed, nil
			}
			texts = append(texts, c.Text)
		}
	}
	if len(texts) == 1 {
		return texts[0], nil
	}
	return texts, nil
}

func (c *mcpClient) listTools() ([]string, 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"`
		} `json:"tools"`
	}
	if err := json.Unmarshal(resp.Result, &tr); err != nil {
		return nil, err
	}
	var names []string
	for _, t := range tr.Tools {
		names = append(names, t.Name)
	}
	return names, nil
}

func (c *mcpClient) close() {
}

// ─── Per-session MCP client pool ────────────────────────────────────────
//
// A single shared mcpClient serializes EVERY tool call across EVERY
// concurrently-running task through one mutex (see mcpClient.mu) — `run`
// executes its SSH command synchronously inside that lock and is capped at
// up to 10 minutes, so one task mid-`run` stalled every other task's tool
// calls, even trivial reads, behind it. The MCP *server* has no per-
// connection state to protect (newServer in internal/mcp/server.go returns
// one shared *mcp.Server instance whose tool handlers close only over the DB
// pool, which is already safe for concurrent use) — the mutex existed purely
// because the *client* reused one stateful transport session, not because
// the server needed it. Giving each task's own session its own client
// removes the cross-task serialization entirely: a task's own tool calls
// stay sequential (which they already are — the agent loop calls tools one
// at a time within a turn), but no longer block anyone else's.
type mcpClientPool struct {
	baseURL string
	token   string
	mu      sync.Mutex
	clients map[string]*pooledMCPClient
}

type pooledMCPClient struct {
	client   *mcpClient
	lastUsed time.Time
}

func newMCPClientPool(baseURL, token string) *mcpClientPool {
	return &mcpClientPool{baseURL: baseURL, token: token, clients: make(map[string]*pooledMCPClient)}
}

// get returns the client for sessionID, creating and initializing one (a
// real MCP handshake) on first use. Session ids that don't identify a real
// persisted conversation ("" / "ephemeral", the no-DB-store path; "query",
// the structured /query endpoint) still get exactly one dedicated,
// reused client each via the same map — just keyed on a fixed string instead
// of a real session id — so that traffic doesn't pay a fresh handshake per
// request while still never sharing a connection with an actual task.
func (p *mcpClientPool) get(sessionID string) (*mcpClient, error) {
	key := sessionID
	if key == "" {
		key = "ephemeral"
	}

	p.mu.Lock()
	if pc, ok := p.clients[key]; ok {
		pc.lastUsed = time.Now()
		p.mu.Unlock()
		return pc.client, nil
	}
	p.mu.Unlock()

	// Initialize outside the lock — it's a network round-trip, and holding
	// the pool mutex for it would serialize unrelated sessions' first calls
	// behind each other, undermining the whole point of this pool.
	c, err := newMCPClient(p.baseURL, p.token)
	if err != nil {
		return nil, err
	}

	p.mu.Lock()
	// Another goroutine may have created one for the same key while we were
	// initializing (two of this session's tool calls racing on a cold
	// start); keep whichever won, close out the loser's connection (a no-op
	// today, but future-proof if mcpClient.close ever does real teardown).
	if existing, ok := p.clients[key]; ok {
		p.mu.Unlock()
		c.close()
		return existing.client, nil
	}
	p.clients[key] = &pooledMCPClient{client: c, lastUsed: time.Now()}
	p.mu.Unlock()
	return c, nil
}

// mcpClientIdleTimeout is how long an idle session's MCP client is kept
// before eviction — long enough to outlive a single slow `run` (capped at 10
// minutes server-side) plus normal think-time between a task's tool calls,
// short enough not to accumulate one abandoned connection per finished task
// forever.
const mcpClientIdleTimeout = 20 * time.Minute

// sweep evicts clients idle past mcpClientIdleTimeout. Call on a ticker.
func (p *mcpClientPool) sweep() {
	cutoff := time.Now().Add(-mcpClientIdleTimeout)
	p.mu.Lock()
	defer p.mu.Unlock()
	for key, pc := range p.clients {
		if pc.lastUsed.Before(cutoff) {
			pc.client.close()
			delete(p.clients, key)
		}
	}
}

func (p *mcpClientPool) closeAll() {
	p.mu.Lock()
	defer p.mu.Unlock()
	for key, pc := range p.clients {
		pc.client.close()
		delete(p.clients, key)
	}
}