diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index fde5502..b032b8d 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "os" + "strings" "time" "github.com/google/uuid" @@ -15,6 +16,15 @@ import ( ) const maxIterations = 15 +const maxLLMRetries = 1 + +var refusalDenylist = []string{ + "我没有相关信息", + "您可以尝试问我其它问题", + "我无法", + "抱歉，我无法", + "关于这个问题，我没有", +} type agent struct { client *mcpClient @@ -105,14 +115,6 @@ func (a *agent) chat(ctx context.Context, sessionID, message string, emit func(a return } - // Rebuild conversation context from persisted history so sessions are - // multi-turn. The current user turn is saved by the HTTP handler before - // this runs, so it is already included in the history for real sessions. - // Prior tool_use/tool_result pairs are replayed as a tool-calling - // assistant message followed by matching tool-role results, so the agent - // starts each turn already knowing what it already checked instead of - // re-querying the same tools from scratch. Ephemeral sessions (no store) - // fall back to the single incoming message. system := a.system if snapshot := a.fleetSnapshot(); snapshot != "" { system += "\n\n" + snapshot @@ -147,30 +149,54 @@ func (a *agent) chat(ctx context.Context, sessionID, message string, emit func(a Tools: tools, } - // Stream the completion, emitting token deltas as they arrive. The - // accumulator reassembles the full message (content + tool calls) for - // the loop's control flow. - stream := a.provider.Chat.Completions.NewStreaming(ctx, params, a.reqOpts...) - acc := openai.ChatCompletionAccumulator{} - for stream.Next() { - chunk := stream.Current() - acc.AddChunk(chunk) - if len(chunk.Choices) > 0 { - if delta := chunk.Choices[0].Delta.Content; delta != "" { - emit(agentEvent{Type: "text_delta", Data: delta, SessionID: sessionID, Iteration: i + 1}) + 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 { - emit(agentEvent{Type: "error", Data: fmt.Sprintf("llm: %v", err), SessionID: sessionID}) - return - } - if len(acc.Choices) == 0 { - emit(agentEvent{Type: "error", Data: "no choices in response", SessionID: sessionID}) - return - } + 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 + 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}) @@ -388,6 +414,33 @@ func (a *agent) fleetSnapshot() string { 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 { diff --git a/cmd/nomos/main.go b/cmd/nomos/main.go index eeec8b3..c4fa597 100644 --- a/cmd/nomos/main.go +++ b/cmd/nomos/main.go @@ -186,6 +186,15 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) { "tool_calls": toolCalls, }) st.saveMessage(ctx, sessionID, "assistant", assistantMsg) + + // 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(ctx, sessionID, title) + } + } } func handleSessionsList(w http.ResponseWriter, r *http.Request, st *store) { @@ -220,13 +229,26 @@ func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store) { return } - messages, err := st.getMessages(r.Context(), id) - if err != nil { - http.Error(w, err.Error(), 500) - 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) } - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"session_id": id, "messages": messages}) } func handleQuery(w http.ResponseWriter, r *http.Request, client *mcpClient, agentSlug, mcpURL string) { diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index f0d04a6..1661b2f 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -10,6 +10,8 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) +const maxToolResultSize = 4096 + type store struct { pool *pgxpool.Pool } @@ -71,10 +73,45 @@ func (s *store) saveMessage(ctx context.Context, sessionID, role string, content } _, err := s.pool.Exec(ctx, `INSERT INTO agent_messages (session_id, role, content) VALUES ($1, $2, $3)`, - sessionID, role, content) + sessionID, role, truncateToolResults(content)) return err } +func truncateToolResults(content json.RawMessage) json.RawMessage { + var m map[string]any + if err := json.Unmarshal(content, &m); err != nil { + return content + } + toolCalls, ok := m["tool_calls"].([]any) + if !ok || len(toolCalls) == 0 { + return content + } + changed := false + for i, raw := range toolCalls { + tc, ok := raw.(map[string]any) + if !ok { + continue + } + if result, ok := tc["result"]; ok { + resultJSON, _ := json.Marshal(result) + if len(resultJSON) > maxToolResultSize { + tc["result"] = string(resultJSON[:maxToolResultSize]) + fmt.Sprintf("...truncated (%d bytes total)", len(resultJSON)) + toolCalls[i] = tc + changed = true + } + } + } + if !changed { + return content + } + m["tool_calls"] = toolCalls + out, err := json.Marshal(m) + if err != nil { + return content + } + return out +} + func (s *store) touchSession(ctx context.Context, id string) { if s != nil { s.pool.Exec(ctx, `UPDATE agent_sessions SET last_active_at=now() WHERE id=$1`, id) @@ -126,6 +163,26 @@ func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, e return out, rows.Err() } +func (s *store) deleteSession(ctx context.Context, id string) error { + if s == nil { + return nil + } + _, err := s.pool.Exec(ctx, `DELETE FROM agent_messages WHERE session_id = $1`, id) + if err != nil { + return err + } + _, err = s.pool.Exec(ctx, `DELETE FROM agent_sessions WHERE id = $1`, id) + return err +} + +func (s *store) updateSessionTitle(ctx context.Context, id, title string) error { + if s == nil { + return nil + } + _, err := s.pool.Exec(ctx, `UPDATE agent_sessions SET title = $1 WHERE id = $2`, title, id) + return err +} + // resolveAgentID looks up the UUID of the agent entity (e.g. "agent:nomos"). // Returns uuid.Nil if the store is absent or the slug is unknown. func (s *store) resolveAgentID(ctx context.Context, slug string) uuid.UUID { diff --git a/internal/mcp/server.go b/internal/mcp/server.go index fa32508..8b0cc35 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -698,6 +698,8 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { COALESCE(st.last_check_at::text, '') AS last_check FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id + WHERE e.state IS NOT NULL + OR st.health IS NOT NULL ORDER BY st.health, e.slug LIMIT 200 `), nil diff --git a/nomos/SOUL.md b/nomos/SOUL.md index 8973f44..b31b96f 100644 --- a/nomos/SOUL.md +++ b/nomos/SOUL.md @@ -23,14 +23,30 @@ the actuator (a separate container with restricted SSH key) picks up. ## Key MCP tools -- `get_entity`, `list_entities` — resolve slugs to state +- `list_lxcs` — all LXC containers with host, IP, health (use for fleet-wide questions) +- `get_lxc_state` — per-container `pct status` (use only for a specific named container) +- `get_state_snapshot` — fleet health, disk, drift at a glance +- `get_health_summary` — fleet health counts +- `query_metrics` — time-series metrics (prefer over per-entity `get_trend` for fleet-wide) +- `list_entities` — resolve slugs to state (pass `type` filter when possible) +- `get_entity` — single-entity detail - `get_blast_radius` — understand impact before requesting action -- `get_health_summary` — fleet status at a glance - `get_signal_history` — open alerts -- `get_trend` — metric trends for decisions +- `get_trend` — metric trends for a specific entity (single-entity only) - `request_execution` — the ONLY mutation path - `get_agent_activity` — your own behavior log +### Tool selection rules + +- **Fleet-wide questions** (e.g. "which hosts are saturated?", "what needs updating?"): + prefer bulk tools: `list_lxcs`, `get_health_summary`, `get_state_snapshot`, + `query_metrics`. Only fall back to per-entity tools (`get_lxc_state`, `tail_log`, + `get_trend`) for a specific named entity the user asked about. +- **One call > many calls**: each `get_lxc_state` is a live SSH round-trip. + `list_lxcs` answers the same question in one call. Use it. +- When a bulk tool's summary isn't enough for a specific entity, call the + per-entity tool for that one entity — not for every entity in the fleet. + ## Policy awareness Before calling `request_execution`: diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 924b44f..204440f 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -31,6 +31,11 @@ export async function fetchMessages(sessionId: string): Promise { return data.messages ?? [] } +export async function deleteSession(sessionId: string): Promise { + const res = await fetch(`${BASE}/sessions/${sessionId}`, { method: 'DELETE' }) + return res.ok +} + export interface ChatEvent { type: string data: any diff --git a/web/src/lib/components/SessionRail.svelte b/web/src/lib/components/SessionRail.svelte index 6ce7dc4..b3faac5 100644 --- a/web/src/lib/components/SessionRail.svelte +++ b/web/src/lib/components/SessionRail.svelte @@ -1,20 +1,39 @@