oikos/cmd/nomos/agent.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/openai/openai-go"
	"github.com/openai/openai-go/option"
	"github.com/openai/openai-go/shared"
)

// maxIterations bounds one chat turn's tool-calling loop. Provisioning a
// service is a long chain (research → plan → request_execution → status), so
// 15 was too tight and turns died with "max iterations reached" mid-deploy.
const maxIterations = 25
const maxLLMRetries = 1

var refusalDenylist = []string{
	"我没有相关信息",
	"您可以尝试问我其它问题",
	"我无法",
	"抱歉，我无法",
	"关于这个问题，我没有",
}

type agent struct {
	client     *mcpClient
	system     string
	provider   *openai.Client
	model      string
	store      *store
	agentID    uuid.UUID
	reqOpts    []option.RequestOption
	apiBase    string // oikos HTTP API base, derived from NOMOS_MCP_URL, for chat-assent approvals
	httpClient *http.Client
}

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 == "" {
		// v4-pro over v4-flash: the flash tier over-narrates, occasionally
		// emits canned refusals, and is unreliable at multi-step tool use —
		// exactly the agentic provisioning path the operator needs to work.
		model = "deepseek/deepseek-v4-pro"
	}

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

	// Derive the oikos HTTP API base from the MCP URL (e.g.
	// "http://api:8090/mcp?session_id=..." -> "http://api:8090"). Used for
	// chat-assent approvals, which call the same decision endpoint the UI's
	// Approve button calls.
	mcpURL := os.Getenv("NOMOS_MCP_URL")
	apiBase := ""
	if idx := strings.Index(mcpURL, "/mcp"); idx > 0 {
		apiBase = mcpURL[:idx]
	}

	return &agent{
		client:     mcpClient,
		system:     system,
		provider:   &provider,
		model:      model,
		store:      st,
		agentID:    agentID,
		reqOpts:    reqOpts,
		apiBase:    apiBase,
		httpClient: &http.Client{Timeout: 15 * time.Second},
	}, 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.`
}

// assentWindowDuration is how long after an operator approves a plan that
// config_mutation commands auto-run without re-approval. The operator
// approved the plan; the agent should execute it end-to-end without
// stopping every step to re-ask. Destructive actions still always need
// explicit typed confirmation regardless of the window.
const assentWindowDuration = 30 * time.Minute

// openAssentWindow records an active assent window in autonomy_settings so
// the MCP run tool (separate process) can check it before requiring approval
// for config_mutation commands. Key is scoped to this agent's UUID.
func (a *agent) openAssentWindow(ctx context.Context) {
	if a.store == nil || a.store.pool == nil || a.agentID == uuid.Nil {
		return
	}
	key := "assent_window.agent:" + a.agentID.String()
	expires := time.Now().Add(assentWindowDuration).UTC().Format(time.RFC3339)
	_, err := a.store.pool.Exec(ctx,
		`INSERT INTO autonomy_settings (key, value) VALUES ($1, $2)
		 ON CONFLICT (key) DO UPDATE SET value = $2`, key, expires)
	if err != nil {
		slog.Warn("nomos: openAssentWindow", "error", err)
	} else {
		slog.Info("nomos: assent window opened", "agent", a.agentID, "expires", expires)
	}
}

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
	}

	system := a.system
	if snapshot := a.fleetSnapshot(); snapshot != "" {
		system += "\n\n" + snapshot
	}
	messages := []openai.ChatCompletionMessageParamUnion{openai.SystemMessage(system)}
	history, _ := a.store.getMessages(ctx, sessionID)
	var lastAssistantCalls []persistedCall
	for _, m := range history {
		text := extractText(m.Content)
		switch m.Role {
		case "user":
			messages = append(messages, openai.UserMessage(text))
		case "assistant":
			if calls := extractToolCalls(m.Content); len(calls) > 0 {
				messages = append(messages, assistantToolCallMessage(calls))
				for _, c := range calls {
					messages = append(messages, openai.ToolMessage(c.resultText(), c.id))
				}
				lastAssistantCalls = calls
			}
			if text != "" {
				messages = append(messages, openai.AssistantMessage(text))
			}
		}
	}
	if len(history) == 0 {
		messages = append(messages, openai.UserMessage(message))
	}

	// Chat-assent approval: if the immediately-preceding assistant turn
	// proposed gated action(s) and the operator's new message reads as
	// authorization ("go ahead", "yes", ...), grant them now — this is the
	// primary approval path; the Approve button in the UI is a fallback for
	// when the operator wants to click instead of type. Destructive-risk
	// actions are never granted by loose assent — they need the stricter
	// isTypedConfirmation ("I confirm ...", per SOUL.md's guidance for what
	// to ask the operator to type).
	pending := extractPendingApprovals(lastAssistantCalls)
	assent := isAssent(message)
	typedConfirm := isTypedConfirmation(message)
	if len(pending) > 0 && (assent || typedConfirm) {
		var granted, blocked []string
		for _, p := range pending {
			if p.destructive && !typedConfirm {
				blocked = append(blocked, p.execID)
				continue
			}
			if !p.destructive && !assent {
				continue // typed-confirm alone doesn't grant a non-destructive item without also reading as assent
			}
			ok, status, aerr := a.approveExecution(ctx, p.execID)
			if aerr != nil {
				slog.Error("nomos: chat-assent approve", "execution", p.execID, "error", aerr)
				continue
			}
			if ok {
				granted = append(granted, p.execID)
				slog.Info("nomos: chat-assent granted", "execution", p.execID, "status", status, "session", sessionID)
				emit(agentEvent{Type: "tool_use", Data: map[string]any{"name": "chat_assent", "args": map[string]any{"execution_id": p.execID}, "id": "assent-" + p.execID}, SessionID: sessionID})
				emit(agentEvent{Type: "tool_result", Data: map[string]any{"name": "chat_assent", "result": fmt.Sprintf("Approved via chat assent (%q). Status: %s.", message, status), "id": "assent-" + p.execID}, SessionID: sessionID})
			}
		}
		if len(granted) > 0 {
			a.openAssentWindow(ctx)
			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, ", "))
			messages = append(messages, openai.SystemMessage(note))
		}
		if len(blocked) > 0 {
			note := fmt.Sprintf("[System: execution(s) %s are classified DESTRUCTIVE and were NOT approved by loose assent — you must ask the operator for an explicit typed confirmation before they can run.]", strings.Join(blocked, ", "))
			messages = append(messages, openai.SystemMessage(note))
		}
	} else if assent && len(lastAssistantCalls) == 0 {
		// 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
		// 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.]"
		messages = append(messages, openai.SystemMessage(note))
		a.openAssentWindow(ctx)
	}

	for i := 0; i < maxIterations; i++ {
		params := openai.ChatCompletionNewParams{
			Model:    openai.ChatModel(a.model),
			Messages: messages,
			Tools:    tools,
		}

		var msg openai.ChatCompletionMessage
		var acc openai.ChatCompletionAccumulator

		for attempt := 0; attempt <= maxLLMRetries; attempt++ {
			acc = openai.ChatCompletionAccumulator{}
			stream := a.provider.Chat.Completions.NewStreaming(ctx, params, a.reqOpts...)
			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 {
				if attempt < maxLLMRetries {
					slog.Warn("nomos: llm stream error, retrying", "error", err, "attempt", attempt+1, "session", sessionID)
					continue
				}
				emit(agentEvent{Type: "error", Data: fmt.Sprintf("llm: %v", err), SessionID: sessionID})
				return
			}
			if len(acc.Choices) == 0 {
				if attempt < maxLLMRetries {
					slog.Warn("nomos: no choices in response, retrying", "attempt", attempt+1, "session", sessionID)
					continue
				}
				emit(agentEvent{Type: "error", Data: "no choices in response", SessionID: sessionID})
				return
			}

			msg = acc.Choices[0].Message

			if len(msg.ToolCalls) == 0 {
				if isRefusalOrEmpty(msg.Content) {
					if attempt < maxLLMRetries {
						slog.Warn("nomos: empty or refusal response, retrying",
							"session", sessionID, "iter", i+1, "attempt", attempt+1,
							"content_len", len(msg.Content))
						continue
					}
					emit(agentEvent{Type: "error", Data: "Nomos returned an empty or unusable response — please retry.", SessionID: sessionID})
					return
				}
			}
			break
		}

		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
}

// persistedCall is one merged tool_use+tool_result pair from a persisted
// assistant message's tool_calls array. The store keeps them as two entries
// sharing the same id (mirroring the SSE event pair); replay needs one
// entry per id to build a valid tool-calling assistant message.
type persistedCall struct {
	id     string
	name   string
	args   json.RawMessage
	result json.RawMessage
	errMsg string
}

func (c persistedCall) resultText() string {
	if c.errMsg != "" {
		return c.errMsg
	}
	if len(c.result) > 0 {
		return string(c.result)
	}
	return "null"
}

// extractToolCalls parses and merges a persisted message's tool_calls array,
// preserving first-seen order across ids.
func extractToolCalls(content json.RawMessage) []persistedCall {
	var m struct {
		ToolCalls []struct {
			ID     string          `json:"id"`
			Type   string          `json:"type"`
			Name   string          `json:"name"`
			Args   json.RawMessage `json:"args"`
			Result json.RawMessage `json:"result"`
			Error  string          `json:"error"`
		} `json:"tool_calls"`
	}
	if err := json.Unmarshal(content, &m); err != nil || len(m.ToolCalls) == 0 {
		return nil
	}

	byID := make(map[string]*persistedCall, len(m.ToolCalls))
	var order []string
	for _, tc := range m.ToolCalls {
		if tc.ID == "" {
			continue
		}
		pc, ok := byID[tc.ID]
		if !ok {
			pc = &persistedCall{id: tc.ID}
			byID[tc.ID] = pc
			order = append(order, tc.ID)
		}
		if tc.Name != "" {
			pc.name = tc.Name
		}
		if len(tc.Args) > 0 && string(tc.Args) != "null" {
			pc.args = tc.Args
		}
		if tc.Type == "tool_result" {
			pc.errMsg = tc.Error
			pc.result = tc.Result
		}
	}

	calls := make([]persistedCall, 0, len(order))
	for _, id := range order {
		calls = append(calls, *byID[id])
	}
	return calls
}

// assistantToolCallMessage builds the tool-calling assistant message that
// must precede the tool-role results being replayed.
func assistantToolCallMessage(calls []persistedCall) openai.ChatCompletionMessageParamUnion {
	toolCalls := make([]openai.ChatCompletionMessageToolCallParam, 0, len(calls))
	for _, c := range calls {
		args := string(c.args)
		if args == "" {
			args = "{}"
		}
		toolCalls = append(toolCalls, openai.ChatCompletionMessageToolCallParam{
			ID: c.id,
			Function: openai.ChatCompletionMessageToolCallFunctionParam{
				Name:      c.name,
				Arguments: args,
			},
		})
	}
	return openai.ChatCompletionMessageParamUnion{
		OfAssistant: &openai.ChatCompletionAssistantMessageParam{ToolCalls: toolCalls},
	}
}

// fleetSnapshot returns a compact, current-as-of-now fleet health line for
// the system prompt so the agent starts each turn already oriented instead
// of spending its first iteration rediscovering topology it already has
// tools to query. Best-effort: an empty string on any failure just means no
// snapshot, not an error for the turn.
func (a *agent) fleetSnapshot() string {
	result, err := a.client.callTool("get_health_summary", map[string]any{})
	if err != nil {
		return ""
	}
	rows, ok := result.([]any)
	if !ok {
		return ""
	}

	counts := map[string]int{}
	var attention []string
	for _, r := range rows {
		row, ok := r.(map[string]any)
		if !ok {
			continue
		}
		health, _ := row["health"].(string)
		counts[health]++
		if health != "healthy" && health != "" {
			if slug, ok := row["slug"].(string); ok && len(attention) < 10 {
				attention = append(attention, fmt.Sprintf("%s(%s)", slug, health))
			}
		}
	}
	if len(counts) == 0 {
		return ""
	}

	summary := fmt.Sprintf("Current fleet snapshot (as of now): healthy=%d degraded=%d down=%d stale=%d unknown=%d.",
		counts["healthy"], counts["degraded"], counts["down"], counts["stale"], counts["unknown"])
	if len(attention) > 0 {
		summary += " Needs attention: " + fmt.Sprintf("%v", attention) + "."
	}
	return summary
}

// isRefusalOrEmpty returns true when the LLM response is blank or looks like a
// canned non-English refusal to an English-language conversation. Flash-tier
// models occasionally emit Chinese boilerplate deflection instead of a real
// answer; this catches it before it reaches the UI.
func isRefusalOrEmpty(text string) bool {
	if strings.TrimSpace(text) == "" {
		return true
	}
	ascii, nonASCII := 0, 0
	for _, r := range text {
		if r <= 127 {
			ascii++
		} else {
			nonASCII++
		}
	}
	if nonASCII > ascii {
		return true
	}
	for _, pattern := range refusalDenylist {
		if strings.Contains(text, pattern) {
			return true
		}
	}
	return false
}

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
}