The last backend piece: when the agent hits a decision only the operator can
make, it surfaces a structured question instead of guessing or stalling.
- ask_operator(prompt, why?, options?, context_entities?): nomos-local tool
that records a session_questions row, moves the task to awaiting_input, emits
question.raised, and ENDS the turn (the agent loop returns after it, so the
agent can't barrel past its own question). The prompt becomes the assistant's
visible message so the question also shows inline in the transcript.
- Two resume paths, both close the question + emit question.answered + return
the task to executing:
- Panel: POST /sessions/{id}/questions/{qid}/answer → resumes the agent in the
background with the answer injected (reusing the continuation machinery,
refactored continueSession → resumeSession). Returns 202; the reply lands via
message polling.
- Chat reply: the next chat message on a task with an open question IS the
answer — auto-closed in handleChat; the turn itself is the resume.
Verified end-to-end: forcing a decision paused the task at awaiting_input with
the structured question (prompt/why/options/entities); a panel answer resumed
the agent (it acknowledged host:strong and continued); a plain chat reply
auto-closed a second question. Cleanup + tests green.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
700 lines
25 KiB
Go
700 lines
25 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 → request_execution → 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 = 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)) {
|
|
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()
|
|
|
|
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})
|
|
|
|
// 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)
|
|
slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
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. 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(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)
|
|
}
|
|
|
|
// 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
|
|
}
|
|
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()
|
|
// Session-scoped task tools are handled in-process; everything else
|
|
// is forwarded to the shared MCP server.
|
|
var result any
|
|
var callErr error
|
|
if localRes, handled := a.handleTaskTool(ctx, sessionID, tc.Function.Name, args); handled {
|
|
result = localRes
|
|
} else {
|
|
result, callErr = a.client.callTool(tc.Function.Name, args)
|
|
}
|
|
elapsed := int(time.Since(start).Milliseconds())
|
|
|
|
inputJSON, _ := json.Marshal(args)
|
|
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)
|
|
|
|
// 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() 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
|
|
}
|
|
// 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
|
|
}
|
|
|
|
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
|
|
}
|