dtoro
3c3b12df5e
The assent system note said 'Do not re-request or call run again for these' — the LLM interpreted this as 'don't call run at all' and produced empty responses (finish_reason=stop, content_len=0) until retries were exhausted, leaving the session stuck in 'executing'. Fix: rewrite both assent notes (pending-approval path and pure-plan-approval path) to be directive about WHAT TO DO NEXT: call update_plan_step(running) then run for each remaining step. The 'don't re-request' guidance is now scoped to 'THOSE SPECIFIC' executions, not all run calls. Also bump maxLLMRetries from 2 to 3 — the empty-response flake on complex multi-turn flows benefits from one more retry.
845 lines
32 KiB
Go
845 lines
32 KiB
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 → run → per-step
|
|
// install/verify run calls), so this must be generous; a full deploy with the
|
|
// decomposed pct_create flow can legitimately need many steps. On exhaustion
|
|
// the loop now produces a real summary (finalSummary) rather than a dead end.
|
|
const maxIterations = 40
|
|
const maxLLMRetries = 3
|
|
|
|
// historyWindowSize bounds how many of a session's most recent persisted
|
|
// messages are replayed into the LLM's context on each turn — see
|
|
// store.go's getRecentMessages for why this exists (fix A2 of
|
|
// plans/2026-07-11-nomos-agent-code-review.md: unbounded history replay was
|
|
// a real, observed-in-production cost/latency/eventual-context-limit risk).
|
|
// 30 is a fixed-window choice, not token-budget-aware: simplest option that
|
|
// still keeps roughly the current task's working context, at the cost of
|
|
// occasionally dropping something a very long task still needed — the
|
|
// system note injected when truncation happens tells the model to check
|
|
// upsert_knowledge/search_knowledge rather than assume something didn't
|
|
// happen. A token-aware trim or LLM-summarize-on-drop are documented
|
|
// stretch options if a fixed window proves insufficient in practice.
|
|
const historyWindowSize = 30
|
|
|
|
var refusalDenylist = []string{
|
|
"我没有相关信息",
|
|
"您可以尝试问我其它问题",
|
|
"我无法",
|
|
"抱歉，我无法",
|
|
"关于这个问题，我没有",
|
|
}
|
|
|
|
type agent struct {
|
|
clients *mcpClientPool // one MCP client PER SESSION, not shared — see mcpClientPool's doc comment
|
|
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
|
|
apiToken string // OIKOS_MCP_BEARER_TOKEN — api's combinedAuth requires it (no dev-open bypass)
|
|
httpClient *http.Client
|
|
}
|
|
|
|
func newAgent(ctx context.Context, clients *mcpClientPool, 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{
|
|
clients: clients,
|
|
system: system,
|
|
provider: &provider,
|
|
model: model,
|
|
store: st,
|
|
agentID: agentID,
|
|
reqOpts: reqOpts,
|
|
apiBase: apiBase,
|
|
apiToken: os.Getenv("OIKOS_MCP_BEARER_TOKEN"),
|
|
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 run. 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 AND this
|
|
// session/task — see store.go's assentWindowActive for why: without the
|
|
// session dimension, approving one task's plan would silently auto-run
|
|
// unapproved actions in any other concurrently-running task.
|
|
func (a *agent) openAssentWindow(ctx context.Context, sessionID string) {
|
|
if a.store == nil || a.store.pool == nil || a.agentID == uuid.Nil || sessionID == "" {
|
|
return
|
|
}
|
|
key := assentWindowKey(a.agentID, sessionID)
|
|
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, "session", sessionID, "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)) {
|
|
a.chatWith(ctx, sessionID, message, "", emit)
|
|
}
|
|
|
|
// chatWith is chat() with an optional system-injected note appended after the
|
|
// replayed history. The auto-continuation worker uses it to resume a session
|
|
// with a finished execution's result ("execution X completed: … — continue the
|
|
// plan") without persisting a fake user turn. message is normally the new user
|
|
// message; for a worker continuation it is empty and systemInject carries the
|
|
// note.
|
|
func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject string, emit func(agentEvent)) {
|
|
correlationID := uuid.New().String()
|
|
|
|
// emitError emits an error event followed by a done event. The done
|
|
// event is CRITICAL on every terminal path: the frontend's
|
|
// onComplete handler (chat.ts) treats a missing `done` as a severed
|
|
// network connection and triggers an auto-reconnect → resumeSession.
|
|
// Before this fix, a model empty-response (the most common case here)
|
|
// returned without `done`, was misclassified as a network drop, and
|
|
// the reconnect logic re-invoked the agent with a generic "report
|
|
// your state" note — which caused the agent to re-propose the plan
|
|
// and duplicate it in the sidebar (operator-reported 2026-07-14).
|
|
// Every error return below must go through emitError so the frontend
|
|
// shows the error inline instead of silently reconnecting.
|
|
emitError := func(data string) {
|
|
emit(agentEvent{Type: "error", Data: data, SessionID: sessionID})
|
|
emit(agentEvent{Type: "done", Data: map[string]any{
|
|
"session_id": sessionID,
|
|
"correlation_id": correlationID,
|
|
"iterations": 0,
|
|
"error": true,
|
|
}, SessionID: sessionID})
|
|
}
|
|
|
|
tools, err := a.buildTools(sessionID)
|
|
if err != nil {
|
|
emitError(fmt.Sprintf("build tools: %v", err))
|
|
return
|
|
}
|
|
|
|
system := a.system
|
|
if snapshot := a.fleetSnapshot(sessionID); snapshot != "" {
|
|
system += "\n\n" + snapshot
|
|
}
|
|
messages := []openai.ChatCompletionMessageParamUnion{openai.SystemMessage(system)}
|
|
history, truncatedHistory, _ := a.store.getRecentMessages(ctx, sessionID, historyWindowSize)
|
|
if truncatedHistory {
|
|
// Tell the model explicitly rather than silently dropping older
|
|
// turns — otherwise it might assume something wasn't done just
|
|
// because it doesn't see the turn that did it.
|
|
messages = append(messages, openai.SystemMessage(fmt.Sprintf(
|
|
"[System: this task has been running long enough that only the most recent %d turns of its history are included above your context — earlier turns happened but aren't shown. If you need to know what was already tried or found, check search_knowledge/get_entity_knowledge (if you recorded it) rather than assuming it didn't happen.]",
|
|
historyWindowSize)))
|
|
}
|
|
// sawSetGoal / sawCompleteTask track whether this session has EVER framed
|
|
// itself as a structured task (set_goal) or already reached a terminal
|
|
// state (complete_task) — across both replayed history and this turn's
|
|
// own tool calls (updated again below as they happen live). Used by the
|
|
// end-of-turn safety net (plans/2026-07-11-task-completion-safety-net.md,
|
|
// fix 1): most sessions are a single trivial Q&A exchange that answers in
|
|
// text and never calls either tool, leaving agent_sessions.status stuck
|
|
// at its creation-time default forever. If a session never framed itself
|
|
// as a task, its first plain-text turn-end IS the task ending.
|
|
var sawSetGoal, sawCompleteTask bool
|
|
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))
|
|
switch c.name {
|
|
case "set_goal":
|
|
sawSetGoal = true
|
|
case "complete_task":
|
|
sawCompleteTask = true
|
|
}
|
|
}
|
|
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})
|
|
|
|
// An explicit typed confirmation for a destructive action
|
|
// opens a short, target-scoped window so the rest of a
|
|
// destructive recovery sequence on the SAME target (e.g.
|
|
// stop -> destroy) doesn't need a second typed confirmation.
|
|
if p.destructive && typedConfirm {
|
|
if execUUID, perr := uuid.Parse(p.execID); perr == nil {
|
|
if target := a.store.executionTarget(ctx, execUUID); target != "" {
|
|
a.store.openDestructiveWindow(ctx, a.agentID, target, sessionID)
|
|
slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target, "session", sessionID)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if len(granted) > 0 {
|
|
a.openAssentWindow(ctx, sessionID)
|
|
note := fmt.Sprintf("[System: the operator 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. The approved execution(s) are already dispatched — do not re-request THOSE SPECIFIC ones. But you MUST continue executing the REMAINING plan steps: call update_plan_step(seq=N, status=\"running\") then run(...) for each unstarted step. Do not stop and wait for 'continue'. 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. Once they do confirm, further destructive steps on that SAME target (e.g. finishing a stop-then-destroy sequence) will auto-run for 15 minutes without asking again — but a different target always needs its own confirmation.]", strings.Join(blocked, ", "))
|
|
messages = append(messages, openai.SystemMessage(note))
|
|
}
|
|
} else if assent && len(pending) == 0 {
|
|
// The operator said "proceed"/"go ahead"/"yes" but there are no
|
|
// pending approvals from the preceding turn — meaning the agent
|
|
// proposed a plan (via propose_plan, possibly with pre-plan research
|
|
// tool calls) and asked "shall I?" without calling run yet. Inject
|
|
// a system note telling the agent the operator approved — go execute
|
|
// the plan now. The old check (len(lastAssistantCalls) == 0) was too
|
|
// restrictive: it only fired when the assistant had ZERO tool calls,
|
|
// but propose_plan + research tools are tool calls. The right check
|
|
// is "no pending APPROVALS" (len(pending) == 0), not "no tool calls."
|
|
note := "[System: The operator approved your proposed plan. Execute it now — for each step, call update_plan_step(seq=N, status=\"running\") then run(...) for that step's target, then update_plan_step(seq=N, status=\"done\"). The assent window is active: config_mutation commands will auto-run. Do not re-describe the plan or ask for confirmation again. Do not wait for 'continue' — execute all steps in this turn.]"
|
|
messages = append(messages, openai.SystemMessage(note))
|
|
a.openAssentWindow(ctx, sessionID)
|
|
}
|
|
|
|
// Worker continuation: append the finished-execution note so the model
|
|
// sees the result and decides the next step (proceed / recover / done).
|
|
if systemInject != "" {
|
|
messages = append(messages, openai.SystemMessage(systemInject))
|
|
}
|
|
|
|
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
|
|
}
|
|
emitError(fmt.Sprintf("llm: %v", err))
|
|
return
|
|
}
|
|
if len(acc.Choices) == 0 {
|
|
if attempt < maxLLMRetries {
|
|
slog.Warn("nomos: no choices in response, retrying", "attempt", attempt+1, "session", sessionID)
|
|
continue
|
|
}
|
|
emitError("no choices in response (the model returned zero completions — likely a provider or rate-limit issue)")
|
|
return
|
|
}
|
|
|
|
msg = acc.Choices[0].Message
|
|
finishReason := acc.Choices[0].FinishReason
|
|
|
|
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), "finish_reason", finishReason)
|
|
continue
|
|
}
|
|
// B.4: surface the real error context (finish_reason +
|
|
// refusal text) instead of a generic "empty response" —
|
|
// the operator can tell "content_filter — rephrase" from
|
|
// "length — token limit hit" from "stop — model no-op'd".
|
|
detail := "empty response"
|
|
if msg.Refusal != "" {
|
|
detail = fmt.Sprintf("refusal: %s", msg.Refusal)
|
|
} else if finishReason != "" && finishReason != "stop" {
|
|
detail = fmt.Sprintf("finish_reason=%s", finishReason)
|
|
}
|
|
emitError(fmt.Sprintf("Nomos returned an empty or unusable response (%s). Retry or rephrase.", detail))
|
|
return
|
|
}
|
|
}
|
|
break
|
|
}
|
|
|
|
if len(msg.ToolCalls) == 0 {
|
|
emit(agentEvent{Type: "text", Data: msg.Content, SessionID: sessionID})
|
|
if !sawSetGoal && !sawCompleteTask {
|
|
a.autoCompleteTrivialTask(ctx, sessionID, msg.Content)
|
|
}
|
|
emit(agentEvent{Type: "done", Data: map[string]any{
|
|
"session_id": sessionID,
|
|
"usage": acc.Usage,
|
|
"correlation_id": correlationID,
|
|
"iterations": i + 1,
|
|
}, SessionID: sessionID})
|
|
return
|
|
}
|
|
|
|
// P3: persist intermediate reasoning. When the model produces text
|
|
// AND tool calls in the same iteration, the text is its reasoning
|
|
// before the tool calls — the operator saw it live via text_delta,
|
|
// but without emitting it as a `text` event here, the persist layer
|
|
// (main.go/continue.go) never captures it and a reload shows only
|
|
// the final summary + a flat tool-call list, not the thinking that
|
|
// led to each step. Emitting it lets the persist layer accumulate
|
|
// per-iteration reasoning into the row's text field.
|
|
if strings.TrimSpace(msg.Content) != "" {
|
|
emit(agentEvent{Type: "text", Data: msg.Content, SessionID: sessionID})
|
|
}
|
|
|
|
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{}
|
|
}
|
|
|
|
switch tc.Function.Name {
|
|
case "set_goal":
|
|
sawSetGoal = true
|
|
case "complete_task":
|
|
sawCompleteTask = true
|
|
}
|
|
|
|
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()
|
|
// 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 {
|
|
// _session_id rides along on the wire call only — never in
|
|
// `args` (which is what gets emitted/logged/persisted as the
|
|
// model's own tool call) — so the MCP-side assent/destructive
|
|
// window checks can scope to THIS task instead of bleeding
|
|
// across every concurrently-running one sharing this agent
|
|
// identity. Not part of any tool's declared InputSchema, so
|
|
// the model never sees or supplies it.
|
|
wireArgs := make(map[string]any, len(args)+1)
|
|
for k, v := range args {
|
|
wireArgs[k] = v
|
|
}
|
|
wireArgs["_session_id"] = sessionID
|
|
var client *mcpClient
|
|
client, callErr = a.clients.get(sessionID)
|
|
if callErr == nil {
|
|
result, callErr = client.callTool(tc.Function.Name, wireArgs)
|
|
}
|
|
}
|
|
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, args, 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, args, inputStr, string(resultJSON), elapsed, true, correlationID)
|
|
|
|
// Link any execution this tool queued/started back to this
|
|
// session, so the auto-continuation worker can feed its result
|
|
// back here when it finishes (see cmd/nomos/continue.go). Async
|
|
// executions (pct_create, apt_upgrade) are the ones that matter —
|
|
// their result lands after this turn ends.
|
|
for _, execID := range extractExecutionIDs(string(resultJSON)) {
|
|
a.store.linkExecution(ctx, execID, sessionID)
|
|
}
|
|
|
|
// Record which entities this task touched (task —involves→ entity)
|
|
// and pulse them on the live context panel. Args only — never
|
|
// 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},
|
|
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)
|
|
|
|
// ask_operator pauses the task: the agent has posed a decision only
|
|
// the operator can make. End the turn here so it doesn't barrel past
|
|
// its own question — the answer (panel or chat reply) resumes it.
|
|
// The prompt becomes the assistant's visible message so the question
|
|
// also shows inline in the transcript.
|
|
if tc.Function.Name == "ask_operator" {
|
|
prompt, _ := args["prompt"].(string)
|
|
emit(agentEvent{Type: "text", Data: prompt, SessionID: sessionID})
|
|
emit(agentEvent{Type: "done", Data: map[string]any{
|
|
"session_id": sessionID,
|
|
"correlation_id": correlationID,
|
|
"iteration": i + 1,
|
|
}, SessionID: sessionID})
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Hitting the step limit used to end the turn with a bare "max iterations
|
|
// reached without final answer" — a dead end that made the operator ask
|
|
// "status?" to find out what actually happened after a long working turn.
|
|
// Instead, spend one final call asking the model to summarize what it did
|
|
// and the current state, so the turn always ends with a real report.
|
|
messages = append(messages, openai.SystemMessage("[System: you've reached the step limit for this turn. STOP calling tools now and write a concise status report: what you accomplished, the current state of the goal, anything that failed, and what remains. This is what the operator sees.]"))
|
|
summary := a.finalSummary(ctx, messages)
|
|
if summary == "" {
|
|
summary = "I hit this turn's step limit while working. I've done a lot but couldn't wrap up cleanly — ask me for a status update and I'll summarize the current state."
|
|
}
|
|
emit(agentEvent{Type: "text", Data: summary, SessionID: sessionID})
|
|
emit(agentEvent{Type: "done", Data: map[string]any{
|
|
"session_id": sessionID,
|
|
"correlation_id": correlationID,
|
|
"iterations": maxIterations,
|
|
}, SessionID: sessionID})
|
|
}
|
|
|
|
// finalSummary makes one non-tool LLM call to turn an exhausted tool-loop into
|
|
// a real status report instead of a dead-end message. Best-effort: empty on
|
|
// any error, and the caller has a fallback.
|
|
func (a *agent) finalSummary(ctx context.Context, messages []openai.ChatCompletionMessageParamUnion) string {
|
|
params := openai.ChatCompletionNewParams{
|
|
Model: openai.ChatModel(a.model),
|
|
Messages: messages,
|
|
// No Tools: force a text answer.
|
|
}
|
|
resp, err := a.provider.Chat.Completions.New(ctx, params, a.reqOpts...)
|
|
if err != nil || len(resp.Choices) == 0 {
|
|
return ""
|
|
}
|
|
return resp.Choices[0].Message.Content
|
|
}
|
|
|
|
// 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(sessionID string) string {
|
|
client, err := a.clients.get(sessionID)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
result, err := 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(sessionID string) ([]openai.ChatCompletionToolParam, error) {
|
|
client, err := a.clients.get(sessionID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defs, err := client.listToolsFull()
|
|
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 {
|
|
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
|
|
}
|
|
|
|
// listToolsFull returns the MCP server's tool list, cached on this client
|
|
// after the first call (see mcpClient.toolsCache). Fix F1 of
|
|
// plans/2026-07-11-nomos-agent-code-review.md: buildTools calls this at the
|
|
// start of every chat turn, including every auto-continuation resume — the
|
|
// tool list is static for the lifetime of one MCP connection, so re-fetching
|
|
// it every single time was avoidable network+parsing work on the hot path.
|
|
// Cache invalidates on reconnectLocked (an api restart may change what's
|
|
// registered).
|
|
func (c *mcpClient) listToolsFull() ([]toolDef, error) {
|
|
c.toolsMu.Lock()
|
|
if c.toolsCache != nil {
|
|
cached := c.toolsCache
|
|
c.toolsMu.Unlock()
|
|
return cached, nil
|
|
}
|
|
c.toolsMu.Unlock()
|
|
|
|
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,
|
|
}
|
|
}
|
|
|
|
c.toolsMu.Lock()
|
|
c.toolsCache = out
|
|
c.toolsMu.Unlock()
|
|
return out, nil
|
|
}
|