dtoro
604b608fa8
search_knowledge and get_entity_knowledge only ever returned a ts_headline
snippet/short headline — enough to find a note, not enough to act on it.
Add get_knowledge_content(slug), mirroring the web UI's
/api/v1/knowledge/content/{id}, so the agent can read a document/
investigation/runbook's full markdown body once it knows which one it
needs. upsert_knowledge already covered the write side. Cross-referenced
all three tool descriptions so the agent discovers the full-read path.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1739 lines
79 KiB
Go
1739 lines
79 KiB
Go
// Package mcp implements the Oikos MCP interface (plan R3-10).
|
|
// Uses the official MCP Go SDK with Streamable HTTP transport.
|
|
package mcp
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"html"
|
|
"io"
|
|
"log/slog"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
|
"github.com/dtoro/oikos/internal/observability"
|
|
"github.com/dtoro/oikos/internal/policy"
|
|
"github.com/dtoro/oikos/internal/safego"
|
|
"github.com/google/jsonschema-go/jsonschema"
|
|
"github.com/google/uuid"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
// prop is one input-schema property (name → type + description).
|
|
type prop struct {
|
|
name, typ, desc string
|
|
}
|
|
|
|
// objSchema builds an "object" JSON Schema from a list of properties. The
|
|
// MCP SDK requires every tool to declare an object input schema so tools
|
|
// are self-describing to the agent; a nil schema panics at registration.
|
|
func objSchema(props ...prop) *jsonschema.Schema {
|
|
s := &jsonschema.Schema{Type: "object", Properties: map[string]*jsonschema.Schema{}}
|
|
for _, p := range props {
|
|
s.Properties[p.name] = &jsonschema.Schema{Type: p.typ, Description: p.desc}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// NewHandler creates an http.Handler that serves the Oikos MCP server.
|
|
// agentID is the Nomos agent entity UUID; tool calls are logged to agent_activity.
|
|
func NewHandler(pool *db.Pool, token string, agentID uuid.UUID) http.Handler {
|
|
s := newServer(pool, agentID)
|
|
handler := mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server {
|
|
if token != "" {
|
|
if r.Header.Get("Authorization") != "Bearer "+token {
|
|
return nil
|
|
}
|
|
}
|
|
return s
|
|
}, nil)
|
|
return handler
|
|
}
|
|
|
|
// toolHandler is the function signature registered via AddTool.
|
|
type toolHandler = mcp.ToolHandler
|
|
|
|
func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
|
|
s := mcp.NewServer(&mcp.Implementation{Name: "oikos", Version: "dev"}, &mcp.ServerOptions{
|
|
Logger: slog.Default(),
|
|
})
|
|
|
|
register := func(tool *mcp.Tool, handler toolHandler) {
|
|
s.AddTool(tool, withActivityLogging(pool, agentID, tool.Name, handler))
|
|
}
|
|
|
|
register(&mcp.Tool{Name: "get_entity", Description: "Get an entity by slug or UUID",
|
|
InputSchema: objSchema(prop{"slug_or_id", "string", "Entity slug (e.g. host:hubris) or UUID"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
idOrSlug, _ := args["slug_or_id"].(string)
|
|
return queryEntity(ctx, pool, idOrSlug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "list_entities", Description: "List entities filtered by type, state, or search",
|
|
InputSchema: objSchema(
|
|
prop{"type", "string", "Filter by entity type"},
|
|
prop{"state", "string", "Filter by lifecycle state"},
|
|
prop{"q", "string", "Substring match on slug or name"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state, e.version, e.created_at, e.updated_at
|
|
FROM entities e
|
|
WHERE ($1::text IS NULL OR e.type = $1)
|
|
AND ($2::text IS NULL OR e.state = $2)
|
|
AND ($3::text IS NULL OR e.slug ILIKE '%'||$3||'%' OR e.name ILIKE '%'||$3||'%')
|
|
ORDER BY e.slug LIMIT $4`,
|
|
nStr(args["type"]), nStr(args["state"]), nStr(args["q"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_relations", Description: "Get relationships for an entity",
|
|
InputSchema: objSchema(prop{"entity_id", "string", "Entity slug"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT r.type, src.slug AS source, tgt.slug AS target
|
|
FROM relationships r
|
|
JOIN entities src ON src.id = r.source_id
|
|
JOIN entities tgt ON tgt.id = r.target_id
|
|
WHERE (src.slug = $1 OR tgt.slug = $1) AND r.valid_to IS NULL
|
|
ORDER BY r.type`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_blast_radius", Description: "Find entities affected if this entity goes down",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"depth", "integer", "Traversal depth (default 3)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
depth := int(getFloat(args, "depth", 3))
|
|
return queryRows(ctx, pool,
|
|
"SELECT e.slug, CAST(b.depth AS int) FROM blast_radius((SELECT id FROM entities WHERE slug = $1), $2) b JOIN entities e ON e.id = b.entity_id",
|
|
slug, depth), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_health_summary", Description: "Current fleet health summary",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, st.health, st.last_check_at
|
|
FROM entity_status st JOIN entities e ON e.id = st.entity_id
|
|
WHERE e.type <> 'check'
|
|
ORDER BY e.slug`), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_audit_trail", Description: "Query the audit log",
|
|
InputSchema: objSchema(prop{"entity_id", "string", "Filter by affected entity UUID"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT id, ts, actor_type, action, entity_id::text, method, path, correlation_id
|
|
FROM audit_log
|
|
WHERE ($1::text IS NULL OR entity_id::text = $1)
|
|
ORDER BY ts DESC LIMIT 50`, nStr(args["entity_id"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation (PostgreSQL FTS with ts_rank ranking). Returns a short snippet per hit, not the full note — call get_knowledge_content with the returned slug to read the whole thing.",
|
|
InputSchema: objSchema(prop{"query", "string", "Search terms"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
q := nStr(args["query"])
|
|
return queryRows(ctx, pool, `
|
|
SELECT ke.title, e.slug,
|
|
ts_rank(ke.search, plainto_tsquery('english', $1)) AS rank,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', $1),
|
|
'MaxWords=40, MinWords=15, ShortWord=3, MaxFragments=3,
|
|
FragmentDelimiter=" ... "') AS snippet,
|
|
ke.source, ke.tags
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE ke.search @@ plainto_tsquery('english', $1)
|
|
ORDER BY rank DESC
|
|
LIMIT 20`, q), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_entity_knowledge", Description: "All documents, investigations, and runbooks linked to an entity. Returns a headline per note, not the full text — call get_knowledge_content with the returned slug to read the whole thing.",
|
|
InputSchema: objSchema(prop{"entity_slug", "string", "Entity slug (e.g. lxc:jellyfin, service:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entities target ON target.id = r.target_id
|
|
WHERE target.slug = $1
|
|
AND r.valid_to IS NULL
|
|
AND r.type IN ('documents', 'about')
|
|
UNION
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entity_types target_type ON target_type.name = (SELECT type FROM entities WHERE slug = $1)
|
|
JOIN entities ent ON ent.type = target_type.name AND ent.slug = $1
|
|
WHERE r.valid_to IS NULL
|
|
AND r.type = 'procedure-for'
|
|
ORDER BY 1`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_knowledge_content", Description: "Full markdown body of one document/investigation/runbook, by its own entity slug. search_knowledge and get_entity_knowledge only return short snippets/headlines — once you know which note you need (from either of those, or because you already know its slug), call this to read the whole thing before acting on it.",
|
|
InputSchema: objSchema(prop{"slug", "string", "The knowledge entity's own slug (e.g. document:containers/101-jellyfin, runbook:client-enrollment) — not the slug of an entity it's about."}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["slug"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT ke.title, e.slug, e.type AS kind, ke.content, ke.source, ke.tags, ke.updated_at::text
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE e.slug = $1`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "upsert_knowledge", Description: "Write back what you learned so future sessions (and future you) benefit — this is how the system gets smarter over time. Use it AFTER solving a non-obvious problem, deploying a service, or discovering a gotcha: record the finding, the fix, and any caveats. Re-calling with the same title updates the existing note instead of duplicating. This is the ONLY way to persist knowledge; a chat message alone is forgotten. search_knowledge/get_entity_knowledge find it, get_knowledge_content reads the full body back.",
|
|
InputSchema: objSchema(
|
|
prop{"title", "string", "Short, specific, searchable title (e.g. 'Dragonfly memlock rlimit in unprivileged LXCs', not 'notes')."},
|
|
prop{"content", "string", "The knowledge itself, in markdown. Be concrete: symptom, root cause, the exact fix/commands, and any caveats. Written for someone hitting this fresh."},
|
|
prop{"about", "string", "Optional entity slug this knowledge concerns (e.g. lxc:typetype, host:strong) — links the note to that entity so get_entity_knowledge surfaces it."},
|
|
prop{"tags", "string", "Optional comma-separated tags (e.g. 'docker,networking,gotcha')."},
|
|
prop{"kind", "string", "One of: investigation (a finding/incident analysis — default), document (reference), runbook (a repeatable procedure)."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return upsertKnowledge(ctx, pool, args)
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "update_entity_attributes", Description: "Merge new/changed attributes into an entity — the OTHER half of avoiding knowledge-base drift (upsert_knowledge records what you learned; this keeps the entity's own facts current). Use it when you discover something concrete about an entity's actual state that the graph doesn't reflect yet: a new IP, a version number, a config value, a discovered port — anything a FUTURE task would otherwise have to rediscover from scratch. Does NOT require approval (this updates the knowledge graph, not the live infrastructure). Merges shallowly — existing keys not mentioned are kept; keys you pass overwrite.",
|
|
InputSchema: objSchema(
|
|
prop{"slug", "string", "Entity slug to update (e.g. lxc:typetype, host:strong)."},
|
|
prop{"attributes", "string", "JSON object string of attributes to merge in, e.g. {\"lan_ip\":\"192.168.8.50\",\"os\":\"debian-12\"}."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["slug"].(string)
|
|
attrsStr, _ := args["attributes"].(string)
|
|
if slug == "" || attrsStr == "" {
|
|
return textResult("error: slug and attributes are required"), nil
|
|
}
|
|
var attrs map[string]any
|
|
if err := json.Unmarshal([]byte(attrsStr), &attrs); err != nil {
|
|
return textResult(fmt.Sprintf("error: attributes is not valid JSON: %v", err)), nil
|
|
}
|
|
attrsJSON, _ := json.Marshal(attrs)
|
|
ct, err := pool.Exec(ctx, `
|
|
UPDATE entities SET attributes = attributes || $2::jsonb, updated_at = now()
|
|
WHERE slug = $1`, slug, string(attrsJSON))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error updating %s: %v", slug, err)), nil
|
|
}
|
|
if ct.RowsAffected() == 0 {
|
|
return textResult(fmt.Sprintf("error: entity %q not found", slug)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Updated %s with %d attribute(s).", slug, len(attrs))), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "create_relationship", Description: "Record a relationship you discovered between two entities — the graph-structure half of keeping the knowledge base current (alongside update_entity_attributes and upsert_knowledge). Use it when you learn that one entity depends on, hosts, routes to, etc. another, and that edge isn't in the graph yet. type must be an existing relationship type (see get_relations output on similar entities for examples: hosts, provides, depends-on, configured-by, about, documents, ...). Idempotent — re-calling the same source/target/type is a no-op. Does NOT require approval.",
|
|
InputSchema: objSchema(
|
|
prop{"source", "string", "Source entity slug."},
|
|
prop{"target", "string", "Target entity slug."},
|
|
prop{"type", "string", "Relationship type name (must already exist in the ontology)."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
source, _ := args["source"].(string)
|
|
target, _ := args["target"].(string)
|
|
relType, _ := args["type"].(string)
|
|
if source == "" || target == "" || relType == "" {
|
|
return textResult("error: source, target, and type are required"), nil
|
|
}
|
|
var sourceID, targetID uuid.UUID
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", source).Scan(&sourceID); err != nil {
|
|
return textResult(fmt.Sprintf("error: source entity %q not found", source)), nil
|
|
}
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", target).Scan(&targetID); err != nil {
|
|
return textResult(fmt.Sprintf("error: target entity %q not found", target)), nil
|
|
}
|
|
_, err := pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT $1, $2, $3, '{"by":"nomos"}'::jsonb, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL)`,
|
|
sourceID, targetID, relType)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error creating relationship: %v (is %q a valid relationship type?)", err, relType)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Recorded: %s —%s→ %s", source, relType, target)), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "query_metrics", Description: "Query time-series metrics",
|
|
InputSchema: objSchema(prop{"hours", "integer", "Look-back window in hours (default 24)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hours := int(getFloat(args, "hours", 24))
|
|
return queryRows(ctx, pool, `
|
|
SELECT time_bucket('1 hour', ts) AS bucket,
|
|
entity_id::text, metric,
|
|
ROUND(avg(value)::numeric, 2) AS avg,
|
|
ROUND(min(value)::numeric, 2) AS min,
|
|
ROUND(max(value)::numeric, 2) AS max
|
|
FROM metric_samples
|
|
WHERE ts > now() - make_interval(hours => $1)
|
|
GROUP BY bucket, entity_id, metric
|
|
ORDER BY bucket DESC LIMIT 100`, hours), nil
|
|
})
|
|
|
|
// ─── Phase 4: new tools ──────────────────────────────────────────
|
|
|
|
register(&mcp.Tool{Name: "get_signal_history", Description: "Query open and recent signals",
|
|
InputSchema: objSchema(
|
|
prop{"entity_slug", "string", "Filter by target entity slug"},
|
|
prop{"state", "string", "Filter by signal state (raised, resolved)"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT s.entity_id::text, s.kind, s.severity, s.state,
|
|
s.occurrence_count, e.slug AS target_slug,
|
|
s.first_seen_at, s.last_seen_at
|
|
FROM signals s
|
|
LEFT JOIN entities e ON e.id = s.target_entity_id
|
|
WHERE ($1::text IS NULL OR e.slug = $1)
|
|
AND ($2::text IS NULL OR s.state = $2)
|
|
ORDER BY s.last_seen_at DESC LIMIT $3`,
|
|
nStr(args["entity_slug"]), nStr(args["state"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_patterns", Description: "List learned action patterns",
|
|
InputSchema: objSchema(
|
|
prop{"status", "string", "Filter by status (hypothesized, validated, active)"},
|
|
prop{"entity_type", "string", "Filter by applies_type"},
|
|
prop{"action", "string", "Filter by action"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT p.entity_id::text, p.applies_type, p.action, p.pattern,
|
|
p.confidence, p.evidence_count, p.success_count, p.failure_count,
|
|
p.status, p.quarantined, p.version, p.last_validated_at
|
|
FROM patterns p
|
|
WHERE ($1::text IS NULL OR p.status = $1)
|
|
AND ($2::text IS NULL OR p.applies_type = $2)
|
|
AND ($3::text IS NULL OR p.action = $3)
|
|
ORDER BY p.applies_type, p.action`,
|
|
nStr(args["status"]), nStr(args["entity_type"]), nStr(args["action"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_skills", Description: "List available automation skills",
|
|
InputSchema: objSchema(
|
|
prop{"status", "string", "Filter by status (active, inactive, deprecated)"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT s.entity_id::text, s.version, s.name, LEFT(s.procedure::text, 300) AS procedure_preview,
|
|
s.applies_type, s.action, s.status, s.success_rate,
|
|
s.changed_by::text, s.change_reason, s.last_used_at
|
|
FROM skills s
|
|
WHERE ($1::text IS NULL OR s.status = $1)
|
|
ORDER BY s.name, s.version DESC`,
|
|
nStr(args["status"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "request_execution", Description: "Request a gated execution (agent-only mutation path). Supported actions: restart, systemctl, pct_exec, apt_upgrade, pct_create. pct_create provisions a NEW LXC and installs its service in one approved step — you do NOT need follow-up pct_exec calls for package installs.",
|
|
InputSchema: objSchema(
|
|
prop{"target", "string", "Target entity slug. For pct_create this MUST be the Proxmox HOST that will run the container (e.g. host:strong) — NOT the new LXC's name. For restart/systemctl/apt_upgrade/pct_exec use the target service/LXC slug (e.g. lxc:caddy)."},
|
|
prop{"action", "string", "Action: restart, systemctl, pct_exec, apt_upgrade, pct_create"},
|
|
prop{"params", "string", "For systemctl: 'enable|disable|reload'. For pct_exec: the shell command. For apt_upgrade: 'audit|upgrade'. For pct_create: a JSON object string with keys: vmid (int, required, unused id), hostname (string, required), cores (int), memory (MB int), disk_gb (int), ip (CIDR e.g. 192.168.8.50/24, or omit/\"dhcp\" — DHCP is the safe default, see below), gw (gateway ip, static only), bridge (e.g. vmbr0/vmbr1 — WHICH BRIDGE REACHES WHICH SUBNET IS DIFFERENT PER HOST, never assume vmbr0; see below), storage (default local-lvm), template (optional — omit to auto-pick newest debian on the host), privileged (bool), nesting (bool), mounts ([]string of 'src,mp=/dst'). Example: {\"vmid\":150,\"hostname\":\"typetype\",\"cores\":2,\"memory\":2048,\"disk_gb\":16,\"ip\":\"192.168.8.50/24\",\"gw\":\"192.168.8.2\",\"bridge\":\"vmbr1\",\"nesting\":true}. pct_create is ATOMIC — it ONLY creates and starts the container (no services/post_install params anymore). Once it completes you will be automatically re-invoked with the result; install packages and run setup by issuing your OWN `run` calls against the new lxc:<hostname> target, one step at a time — you'll see each step's real output and can fix exactly the one that fails, instead of one opaque multi-minute install that either fully works or fully doesn't. STATIC IP RULE: before setting ip/gw/bridge to anything other than DHCP, use list_entities/get_entity_knowledge to find an EXISTING lxc on the SAME host whose IP is in the same /28 block, and copy its exact gw+bridge — do not invent a gateway. If no such neighbor exists, prefer ip:\"dhcp\" (proven to work, gets a real routable address) over guessing; a wrong bridge/gateway pair fails a fast pre-flight ping check now (seconds, not minutes) but is still a wasted turn — better to not guess at all."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
targetSlug, _ := args["target"].(string)
|
|
action, _ := args["action"].(string)
|
|
params, _ := args["params"].(string)
|
|
sessionID, _ := args["_session_id"].(string)
|
|
if targetSlug == "" || action == "" {
|
|
return textResult("error: target and action required"), nil
|
|
}
|
|
|
|
var targetID uuid.UUID
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
|
|
return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
|
|
}
|
|
|
|
// restart, pct_exec, and systemctl (outside enable/disable) route
|
|
// through the same classify→gate path as `run` instead of executing
|
|
// immediately over SSH with a hardcoded risk_class='reversible_low'
|
|
// that was never actually checked against anything. Found live
|
|
// 2026-07-10: a chat request to "restart caddy" — the fleet's
|
|
// reverse proxy — executed instantly with zero approval, because
|
|
// this action bypassed the classifier entirely. classifyAndGate
|
|
// applies the same read-only/config-mutation/destructive
|
|
// classification and approval flow the `run` tool already uses.
|
|
if action == "restart" || action == "pct_exec" || (action == "systemctl" && params != "enable" && params != "disable") {
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
var cmd, purpose string
|
|
switch action {
|
|
case "restart":
|
|
cmd = fmt.Sprintf("systemctl restart %s; sleep 1; systemctl is-active %s", svc, svc)
|
|
purpose = "restart " + svc
|
|
case "pct_exec":
|
|
cmd = params
|
|
purpose = "pct_exec (legacy) on " + targetSlug
|
|
case "systemctl":
|
|
cmd = fmt.Sprintf("systemctl %s %s; sleep 1; systemctl is-active %s", params, svc, svc)
|
|
purpose = "systemctl " + params + " " + svc
|
|
}
|
|
return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, cmd, purpose, "", sessionID), nil
|
|
}
|
|
|
|
// Deduplicate: if a pending execution already exists for the same
|
|
// target+action, return the existing one instead of creating a
|
|
// duplicate. Prevents the LLM from re-requesting the same gated
|
|
// action in a tool-calling loop. Only blocks when a pending
|
|
// execution exists; completed/failed ones don't block.
|
|
if action == "systemctl" || action == "apt_upgrade" || action == "pct_create" {
|
|
execNamePrefix := action + " on " + targetSlug
|
|
var existingID string
|
|
err := pool.QueryRow(ctx, `
|
|
SELECT e.id::text FROM entities e
|
|
JOIN executions ex ON ex.entity_id = e.id
|
|
WHERE e.type = 'execution' AND e.name LIKE $1 AND ex.status = 'pending_approval'
|
|
ORDER BY e.created_at DESC LIMIT 1`, execNamePrefix+"%").Scan(&existingID)
|
|
if err == nil && existingID != "" {
|
|
return textResult(fmt.Sprintf("%s on %s is already queued for approval — execution %s. Wait for operator approval. Do not re-request.",
|
|
action, targetSlug, existingID)), nil
|
|
}
|
|
}
|
|
|
|
id, _ := uuid.NewV7()
|
|
correlationID := uuid.New().String()
|
|
|
|
// Full UUID, not a truncated prefix: UUIDv7's leading bytes encode a
|
|
// millisecond timestamp, so an 8-char prefix collides for real under
|
|
// back-to-back requests (observed live: two `run` calls seconds
|
|
// apart hit entities_slug_key). The full string is guaranteed unique.
|
|
execName := action + " on " + targetSlug + " (" + id.String() + ")"
|
|
execSlug := "exec:" + targetSlug + ":" + id.String()
|
|
_, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
|
|
id, execSlug, execName)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: failed to create execution: %v", err)), nil
|
|
}
|
|
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, 'reversible_low', 'running', $4, $5) ON CONFLICT DO NOTHING`,
|
|
id, targetID, action+":"+params, correlationID, agentID)
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
|
|
id, targetID)
|
|
if sessionID != "" {
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
|
|
FROM entities t WHERE t.slug = $2
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
|
|
id, "task:"+sessionID)
|
|
}
|
|
|
|
// Execute reversible actions immediately. restart/pct_exec/systemctl
|
|
// (outside enable/disable) never reach here — they're routed through
|
|
// classifyAndGate above, before this dedup+insert block.
|
|
switch action {
|
|
case "systemctl":
|
|
// Only enable/disable reach this case now.
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "systemctl", svc+":"+params, "config_mutation")
|
|
return textResult(fmt.Sprintf("systemctl %s on %s requires approval — execution %s queued", params, svc, id)), nil
|
|
|
|
case "apt_upgrade":
|
|
if params == "audit" {
|
|
host, user, err := resolveHost(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, "apt update -qq 2>&1 >/dev/null; apt list --upgradable 2>/dev/null | tail -n +2 | wc -l; apt list --upgradable 2>/dev/null | tail -n +2 | head -20")
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("apt audit error: %v", err)), nil
|
|
}
|
|
return textResult("apt audit:\n" + out), nil
|
|
}
|
|
// During an active assent window, auto-approve.
|
|
if assentWindowActive(ctx, pool, agentID, sessionID) {
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
|
|
// Do NOT pre-flip approvals/executions status here (that was
|
|
// the previous, broken "autoApprove" helper). DecideApproval
|
|
// (invoked below) is the ONE place that transitions
|
|
// pending_approval -> approved and dispatches the real SSH
|
|
// work — it specifically looks for status='pending_approval'
|
|
// to find what to run. Pre-flipping the status past that
|
|
// state meant DecideApproval's own lookup found nothing,
|
|
// silently no-opped, and the execution sat at 'approved'
|
|
// forever with nothing actually running. Found live: every
|
|
// assent-window auto-approved pct_create/apt_upgrade has
|
|
// never actually executed, via this exact bug. Calling
|
|
// executeApprovedViaAPI directly against the untouched
|
|
// pending_approval row makes this identical to the manual
|
|
// Approve-button path, just without a human click.
|
|
//
|
|
// context.Background(), NOT ctx: ctx is scoped to this MCP
|
|
// tool call, cancelled the instant the chat turn's HTTP
|
|
// response completes (every normal turn) — a goroutine
|
|
// meant to outlive the request must not inherit its context.
|
|
safego.Go("mcp:executeApprovedViaAPI:apt_upgrade", func() {
|
|
executeApprovedViaAPI(context.Background(), id, targetSlug, "apt_upgrade:"+params)
|
|
})
|
|
slog.Info("mcp: apt_upgrade auto-approved via assent window", "execution_id", id)
|
|
return textResult(fmt.Sprintf("apt_upgrade on %s auto-approved via assent window — execution %s running.", targetSlug, id)), nil
|
|
}
|
|
// upgrade requires approval — queue
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
|
|
return textResult(fmt.Sprintf("apt_upgrade on %s requires approval — execution %s queued", targetSlug, id)), nil
|
|
|
|
case "pct_create":
|
|
// During an active assent window, auto-approve and execute
|
|
// instead of queuing — the operator already approved the plan.
|
|
if assentWindowActive(ctx, pool, agentID, sessionID) {
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
|
|
// See the apt_upgrade case above for why there's no
|
|
// pre-flip-status "autoApprove" step here anymore, and why
|
|
// this uses context.Background().
|
|
safego.Go("mcp:executeApprovedViaAPI:pct_create", func() {
|
|
executeApprovedViaAPI(context.Background(), id, targetSlug, "pct_create:"+params)
|
|
})
|
|
slog.Info("mcp: pct_create auto-approved via assent window", "execution_id", id)
|
|
return textResult(fmt.Sprintf("pct_create on %s auto-approved via assent window — execution %s running. The LXC is being provisioned now.", targetSlug, id)), nil
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
|
|
return textResult(fmt.Sprintf("pct_create on %s requires approval — execution %s queued. The LXC will be provisioned once approved.", targetSlug, id)), nil
|
|
|
|
default:
|
|
return textResult(fmt.Sprintf("unknown action: %s. Supported: restart, systemctl, pct_exec, apt_upgrade, pct_create", action)), nil
|
|
}
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "run", Description: "Run ANY shell command against any host or LXC. This is the general execution primitive — prefer it over asking the operator to run something manually, and don't wait for a matching fixed action to exist. Every command is automatically risk-classified: read-only inspection (cat, systemctl status, docker ps, journalctl, df, git status, ...) runs immediately; anything that changes state requires operator approval (granted by the operator replying \"go ahead\"/\"yes\" in chat, or via the Approve button); commands matching a destructive pattern (rm -rf, dd, mkfs, pct/qm destroy, DROP TABLE, reboot, piping curl into a shell, ...) always require approval regardless of what you declare. You cannot talk your way past the destructive check by declaring a lower risk.",
|
|
InputSchema: objSchema(
|
|
prop{"target", "string", "Target entity slug: host:<slug> (e.g. host:strong) or lxc:<slug> (e.g. lxc:caddy). LXC commands run via pct exec on its Proxmox host automatically."},
|
|
prop{"command", "string", "The shell command to run. Can be a full script (multi-line, &&-chained). Runs as root."},
|
|
prop{"purpose", "string", "One sentence: why you're running this. Shown to the operator alongside the approval — be specific, this is what they're approving."},
|
|
prop{"declared_risk", "string", "Optional self-assessment: read_only, reversible_low, config_mutation, or destructive. This can only ESCALATE the automatic classification, never lower it — declaring a mutating command as read_only has no effect."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
targetSlug, _ := args["target"].(string)
|
|
command, _ := args["command"].(string)
|
|
purpose, _ := args["purpose"].(string)
|
|
declaredRisk, _ := args["declared_risk"].(string)
|
|
sessionID, _ := args["_session_id"].(string)
|
|
if targetSlug == "" || command == "" {
|
|
return textResult("error: target and command are required"), nil
|
|
}
|
|
|
|
var targetID uuid.UUID
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
|
|
return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
|
|
}
|
|
|
|
return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, command, purpose, declaredRisk, sessionID), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "http_get", Description: "Fetch a public web page or raw file (e.g. a GitHub README/raw URL) and return sanitized text. Use this to research how to deploy a service before provisioning. HTTP/HTTPS only; body is truncated to ~16KB.",
|
|
InputSchema: objSchema(
|
|
prop{"url", "string", "Absolute http(s) URL to fetch"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
rawURL, _ := args["url"].(string)
|
|
return httpGet(ctx, rawURL), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_execution_status", Description: "Check the status of a requested execution",
|
|
InputSchema: objSchema(
|
|
prop{"execution_id", "string", "Execution UUID (from request_execution output)"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
execID, _ := args["execution_id"].(string)
|
|
if execID == "" {
|
|
return textResult("execution_id required"), nil
|
|
}
|
|
eid, err := uuid.Parse(execID)
|
|
if err != nil {
|
|
// Try finding by exec slug prefix
|
|
var found uuid.UUID
|
|
err2 := pool.QueryRow(ctx, "SELECT entity_id FROM executions WHERE entity_id::text LIKE $1 LIMIT 1", execID+"%").Scan(&found)
|
|
if err2 != nil {
|
|
return textResult(fmt.Sprintf("execution not found: %s", execID)), nil
|
|
}
|
|
eid = found
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.entity_id::text, e.action, e.risk_class, e.status,
|
|
e.result::text, e.duration_ms, e.started_at::text,
|
|
e.completed_at::text, e.correlation_id
|
|
FROM executions e
|
|
WHERE e.entity_id = $1`, eid), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_trend", Description: "Get metric trends for an entity",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"days", "integer", "Look-back window in days (default 7)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
days := int(getFloat(args, "days", 7))
|
|
return queryRows(ctx, pool, `
|
|
SELECT metric,
|
|
ROUND(avg(value)::numeric, 2) AS avg_val,
|
|
ROUND(stddev(value)::numeric, 2) AS std_val,
|
|
count(*) AS sample_count,
|
|
ROUND(regr_slope(value, EXTRACT(EPOCH FROM ts)::numeric)::numeric, 4) AS slope
|
|
FROM metric_samples ms
|
|
JOIN entities e ON e.id = ms.entity_id
|
|
WHERE e.slug = $1 AND ts >= now() - make_interval(days => $2)
|
|
GROUP BY metric
|
|
ORDER BY metric`, slug, days), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_event_timeline", Description: "Get recent events",
|
|
InputSchema: objSchema(
|
|
prop{"severity", "string", "Filter by severity (info, warn, error)"},
|
|
prop{"entity_slug", "string", "Filter by entity slug"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT ev.ts, ev.type, ev.severity, ev.source, e.slug AS entity_slug,
|
|
ev.data::text AS message, ev.correlation_id
|
|
FROM events ev
|
|
LEFT JOIN entities e ON e.id = ev.entity_id
|
|
WHERE ($1::text IS NULL OR ev.severity = $1)
|
|
AND ($2::text IS NULL OR e.slug = $2)
|
|
ORDER BY ev.ts DESC LIMIT $3`,
|
|
nStr(args["severity"]), nStr(args["entity_slug"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_agent_activity", Description: "Agent self-inspection: query agent activity log",
|
|
InputSchema: objSchema(
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT id, ts, agent_id::text, session_id, activity_type, tool_name,
|
|
entity_id::text, left(input_summary, 200) AS input_summary,
|
|
left(output_summary, 200) AS output_summary,
|
|
duration_ms, token_count, success, correlation_id
|
|
FROM agent_activity
|
|
WHERE agent_id = $1
|
|
ORDER BY ts DESC LIMIT $2`, agentID, limit), nil
|
|
})
|
|
|
|
// ─── Phase 5: operational MCP tools ──────────────────────────────
|
|
|
|
register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, and state",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id,
|
|
e.attributes->>'lan_ip' AS lan_ip,
|
|
st.health, st.last_check_at
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.type = 'lxc'
|
|
ORDER BY (e.attributes->>'pve_id')::int`), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "ping_service", Description: "Check if a service is reachable via HTTP",
|
|
InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
rows, err := pool.Query(ctx, `
|
|
SELECT st.health, st.last_check_at, e.attributes->>'url' AS url
|
|
FROM entity_status st
|
|
JOIN entities e ON e.id = st.entity_id
|
|
WHERE e.slug = $1`, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("query error: %v", err)), nil
|
|
}
|
|
defer rows.Close()
|
|
if !rows.Next() {
|
|
return textResult(fmt.Sprintf("service not found: %s", slug)), nil
|
|
}
|
|
var health, lastCheck, url string
|
|
rows.Scan(&health, &lastCheck, &url)
|
|
if url == "" {
|
|
url = "(no URL in entity attributes)"
|
|
}
|
|
return textResult(fmt.Sprintf("health=%s last_check=%s url=%s", health, lastCheck, url)), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "tail_log", Description: "Get recent log lines from a service via journalctl",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"},
|
|
prop{"lines", "integer", "Number of lines (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
n := int(getFloat(args, "lines", 50))
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(slug, "lxc:")
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("journalctl -u %s -n %d --no-pager 2>&1 || true", svc, n))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_service_status", Description: "Check systemd service status on a host",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(slug, "lxc:")
|
|
out, err := sshExec(ctx, host, user,
|
|
fmt.Sprintf("systemctl is-active %s; systemctl is-enabled %s; systemctl show %s -p ActiveEnterTimestamp -p SubState 2>&1 || true", svc, svc, svc))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_lxc_state", Description: "Get LXC container resource state from Proxmox host",
|
|
InputSchema: objSchema(
|
|
prop{"lxc_slug", "string", "LXC entity slug (e.g. lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["lxc_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("lxc_slug is required"), nil
|
|
}
|
|
var pveID string
|
|
err := pool.QueryRow(ctx, "SELECT attributes->>'pve_id' FROM entities WHERE slug = $1", slug).Scan(&pveID)
|
|
if err != nil || pveID == "" {
|
|
return textResult(fmt.Sprintf("LXC not found or missing pve_id: %s", slug)), nil
|
|
}
|
|
// Resolve the Proxmox host — find the host that runs this LXC
|
|
var hostID uuid.UUID
|
|
err = pool.QueryRow(ctx, `
|
|
SELECT t.id FROM entities t
|
|
JOIN relationships r ON r.source_id = t.id
|
|
JOIN entities s ON s.id = r.target_id
|
|
WHERE s.slug = $1 AND r.type = 'hosts' AND r.valid_to IS NULL
|
|
LIMIT 1`, slug).Scan(&hostID)
|
|
if err != nil {
|
|
// Fallback: use the inventory host attribute if no relationship
|
|
var hostSlug string
|
|
err = pool.QueryRow(ctx, "SELECT attributes->>'host' FROM entities WHERE slug = $1", slug).Scan(&hostSlug)
|
|
if err != nil || hostSlug == "" {
|
|
return textResult(fmt.Sprintf("cannot resolve Proxmox host for %s", slug)), nil
|
|
}
|
|
var host, user string
|
|
host, user, err = resolveHost(ctx, pool, "host:"+hostSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
out, err2 := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
|
|
if err2 != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err2)), nil
|
|
}
|
|
return textResult(out), nil
|
|
}
|
|
var hostSlug string
|
|
pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", hostID).Scan(&hostSlug)
|
|
host, user, err := resolveHost(ctx, pool, hostSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
// ─── Client introspection tools (plan: client-lifecycle Phase 3) ──
|
|
|
|
register(&mcp.Tool{Name: "whoami", Description: "Get the current entity record, peers, and health for a host",
|
|
InputSchema: objSchema(prop{"hostname", "string", "Hostname of the calling machine"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hostname, _ := args["hostname"].(string)
|
|
if hostname == "" {
|
|
return textResult("error: hostname required"), nil
|
|
}
|
|
slug := "ws:" + hostname
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.attributes->>'mesh_ip' AS mesh_ip,
|
|
e.attributes->>'age_pubkey' AS age_pubkey,
|
|
e.enrolled_at
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1
|
|
ORDER BY e.slug`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "explain", Description: "Compact context card for a service: type, state, health, relations, risk",
|
|
InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug (e.g. service:jellyfin, lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("error: service_slug required"), nil
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.version, e.updated_at,
|
|
COALESCE(e.attributes::text, '{}') AS attrs
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "preflight", Description: "Risk classification for an action on a service",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Entity slug"},
|
|
prop{"action", "string", "Planned action (restart, deploy, destroy, etc.)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
action, _ := args["action"].(string)
|
|
if slug == "" || action == "" {
|
|
return textResult("error: service_slug and action required"), nil
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
CASE
|
|
WHEN $2 IN ('restart', 'logs', 'status') THEN 'reversible_low'
|
|
WHEN $2 IN ('deploy', 'upgrade', 'configure') THEN 'config_mutation'
|
|
WHEN $2 IN ('destroy', 'wipe', 'revoke') THEN 'destructive'
|
|
ELSE 'read_only'
|
|
END AS risk_class,
|
|
CASE
|
|
WHEN $2 IN ('read_only','reversible_low') THEN 'auto-act'
|
|
WHEN $2 = 'config_mutation' THEN 'operator-approval'
|
|
ELSE 'operator-approval+confirmation'
|
|
END AS approval
|
|
FROM entities e WHERE e.slug = $1`, slug, action), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_change_history", Description: "Last N change-ledger entries for an entity",
|
|
InputSchema: objSchema(
|
|
prop{"entity_slug", "string", "Entity slug"},
|
|
prop{"limit", "integer", "Max entries (default 20)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
limit := int(getFloat(args, "limit", 20))
|
|
return queryRows(ctx, pool, `
|
|
SELECT al.ts AS timestamp, al.actor_type, al.actor_id::text AS actor_label,
|
|
al.action, al.method, al.path,
|
|
al.detail::text AS details
|
|
FROM audit_log al
|
|
JOIN entities e ON e.id = al.entity_id
|
|
WHERE e.slug = $1
|
|
ORDER BY al.ts DESC
|
|
LIMIT $2`, slug, limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_state_snapshot", Description: "Last scheduler Observe-pass: fleet health, disk, drift count",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
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
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "list_my_secrets", Description: "List secrets accessible to this client by public key",
|
|
InputSchema: objSchema(prop{"caller_pubkey", "string", "Age public key of the caller (optional)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
pubkey, _ := args["caller_pubkey"].(string)
|
|
// Match entities where age_pubkey attribute contains the caller's key.
|
|
query := `
|
|
SELECT e.slug, e.type, e.name,
|
|
e.attributes->>'age_pubkey' AS age_pubkey
|
|
FROM entities e
|
|
WHERE e.attributes->>'age_pubkey' IS NOT NULL`
|
|
var dbArgs []any
|
|
if pubkey != "" {
|
|
query += ` AND e.attributes->>'age_pubkey' = $1`
|
|
dbArgs = append(dbArgs, pubkey)
|
|
}
|
|
query += ` ORDER BY e.slug LIMIT 100`
|
|
return queryRows(ctx, pool, query, dbArgs...), nil
|
|
})
|
|
|
|
return s
|
|
}
|
|
|
|
// withActivityLogging wraps a tool handler to record agent_activity rows.
|
|
func withActivityLogging(pool *db.Pool, agentID uuid.UUID, toolName string, next mcp.ToolHandler) mcp.ToolHandler {
|
|
if agentID == uuid.Nil {
|
|
return next
|
|
}
|
|
return func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
start := time.Now()
|
|
result, err := next(ctx, req)
|
|
duration := int(time.Since(start).Milliseconds())
|
|
|
|
// Build input summary (first 500 chars of args)
|
|
inputSummary := ""
|
|
if req != nil && len(req.Params.Arguments) > 0 {
|
|
inputSummary = string(req.Params.Arguments)
|
|
}
|
|
if len(inputSummary) > 500 {
|
|
inputSummary = inputSummary[:500]
|
|
}
|
|
|
|
// Build output summary
|
|
outputSummary := ""
|
|
success := err == nil
|
|
if result != nil {
|
|
for _, c := range result.Content {
|
|
if tc, ok := c.(*mcp.TextContent); ok {
|
|
outputSummary = tc.Text
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
outputSummary = err.Error()
|
|
success = false
|
|
}
|
|
if len(outputSummary) > 500 {
|
|
outputSummary = outputSummary[:500]
|
|
}
|
|
|
|
correlationID := uuid.New().String()
|
|
|
|
entityID := resolveArgEntityID(ctx, pool, argsMap(req))
|
|
var entityIDArg any
|
|
if entityID != uuid.Nil {
|
|
entityIDArg = entityID
|
|
}
|
|
|
|
_, logErr := pool.Exec(ctx, `
|
|
INSERT INTO agent_activity
|
|
(agent_id, activity_type, tool_name, entity_id, input_summary, output_summary,
|
|
duration_ms, success, correlation_id)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
|
|
agentID, "tool_call", toolName, entityIDArg, inputSummary, outputSummary,
|
|
duration, success, correlationID)
|
|
if logErr != nil {
|
|
slog.Warn("mcp: log agent_activity", "error", logErr)
|
|
}
|
|
|
|
return result, err
|
|
}
|
|
}
|
|
|
|
// entityArgKeys lists tool-argument keys, in priority order, that commonly
|
|
// carry the target entity's slug or UUID. Tool input schemas aren't
|
|
// consistent about naming this (target, entity_slug, slug, service_slug,
|
|
// lxc_slug, entity_id all appear across server.go's tool registrations), so
|
|
// this is a best-effort lookup used to tag agent_activity rows with the
|
|
// entity a tool call acted on.
|
|
var entityArgKeys = []string{
|
|
"target", "entity_slug", "slug", "slug_or_id",
|
|
"service_slug", "lxc_slug", "entity_id", "about",
|
|
}
|
|
|
|
// resolveArgEntityID best-effort resolves the entity a tool call acted on
|
|
// from its arguments, trying entityArgKeys in order. Returns uuid.Nil if no
|
|
// key is present or none resolves to a known entity.
|
|
func resolveArgEntityID(ctx context.Context, pool *db.Pool, args map[string]any) uuid.UUID {
|
|
for _, key := range entityArgKeys {
|
|
v, _ := args[key].(string)
|
|
if v == "" {
|
|
continue
|
|
}
|
|
if u, err := uuid.Parse(v); err == nil {
|
|
return u
|
|
}
|
|
var id uuid.UUID
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", v).Scan(&id); err == nil {
|
|
return id
|
|
}
|
|
}
|
|
return uuid.Nil
|
|
}
|
|
|
|
// ─── Helpers ──────────────────────────────────────────────────────────
|
|
|
|
func argsMap(req *mcp.CallToolRequest) map[string]any {
|
|
if req == nil || len(req.Params.Arguments) == 0 {
|
|
return nil
|
|
}
|
|
var m map[string]any
|
|
json.Unmarshal(req.Params.Arguments, &m)
|
|
return m
|
|
}
|
|
|
|
func getFloat(m map[string]any, key string, def float64) float64 {
|
|
if m == nil {
|
|
return def
|
|
}
|
|
switch v := m[key].(type) {
|
|
case float64:
|
|
return v
|
|
case int:
|
|
return float64(v)
|
|
case json.Number:
|
|
f, err := v.Float64()
|
|
if err != nil {
|
|
return def
|
|
}
|
|
return f
|
|
}
|
|
return def
|
|
}
|
|
|
|
func nStr(v any) any {
|
|
if v == nil {
|
|
return nil
|
|
}
|
|
s, _ := v.(string)
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|
|
|
|
func textResult(s string) *mcp.CallToolResult {
|
|
return &mcp.CallToolResult{
|
|
Content: []mcp.Content{&mcp.TextContent{Text: s}},
|
|
}
|
|
}
|
|
|
|
// jsonOut builds a valid {"output": "..."} JSON payload for an execution's
|
|
// result column. Command output contains quotes/backslashes/control chars, so
|
|
// it must be JSON-marshaled — a hand-built string fails the ::jsonb cast and
|
|
// silently drops the status update, leaving the execution stuck.
|
|
func jsonOut(out string) []byte {
|
|
b, _ := json.Marshal(map[string]any{"output": out})
|
|
return b
|
|
}
|
|
|
|
// jsonErr builds a valid {"error": "..."} JSON payload for an execution's
|
|
// result column — same rationale as jsonOut, for the failure path.
|
|
func jsonErr(format string, args ...any) []byte {
|
|
b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)})
|
|
return b
|
|
}
|
|
|
|
func queryEntity(ctx context.Context, pool *db.Pool, idOrSlug string) *mcp.CallToolResult {
|
|
var id uuid.UUID
|
|
if u, err := uuid.Parse(idOrSlug); err == nil {
|
|
id = u
|
|
} else {
|
|
pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&id)
|
|
}
|
|
if id == uuid.Nil {
|
|
return textResult(fmt.Sprintf("entity not found: %s", idOrSlug))
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT slug, type, name, state, attributes,
|
|
maintenance_until::text, version, created_at, updated_at
|
|
FROM entities WHERE id = $1`, id)
|
|
}
|
|
|
|
func queryRows(ctx context.Context, pool *db.Pool, query string, args ...any) *mcp.CallToolResult {
|
|
rows, err := pool.Query(ctx, query, args...)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
defer rows.Close()
|
|
|
|
cols := rows.FieldDescriptions()
|
|
var items []map[string]any
|
|
items = make([]map[string]any, 0)
|
|
for rows.Next() {
|
|
vals, err := rows.Values()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
m := make(map[string]any)
|
|
for i, col := range cols {
|
|
m[string(col.Name)] = fmt.Sprintf("%v", vals[i])
|
|
}
|
|
items = append(items, m)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
|
|
data, _ := json.MarshalIndent(items, "", " ")
|
|
return textResult(string(data))
|
|
}
|
|
|
|
// ─── SSH helpers ─────────────────────────────────────────────────────────
|
|
|
|
var (
|
|
sshUser string
|
|
sshKey []byte
|
|
sshPool = make(map[string]*ssh.Client)
|
|
sshPoolMu sync.Mutex
|
|
)
|
|
|
|
func initSSH() {
|
|
if sshUser == "" {
|
|
sshUser = os.Getenv("OIKOS_SSH_USER")
|
|
if sshUser == "" {
|
|
sshUser = "root"
|
|
}
|
|
}
|
|
keyPath := os.Getenv("OIKOS_SSH_KEY_PATH")
|
|
if keyPath == "" {
|
|
keyPath = "/etc/oikos/ssh_key"
|
|
}
|
|
if len(sshKey) == 0 {
|
|
var err error
|
|
sshKey, err = os.ReadFile(keyPath)
|
|
if err != nil {
|
|
slog.Warn("mcp ssh: cannot read key", "path", keyPath, "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// sshExecTimeout bounds how long a single remote command may run — see the
|
|
// matching constant/comment in httpapi/phase3.go. Without it, a hung remote
|
|
// command (piped install script stuck retrying DNS, etc.) blocks this
|
|
// goroutine forever with no way for the caller to ever get an answer.
|
|
const sshExecTimeout = 10 * time.Minute
|
|
|
|
func sshExec(ctx context.Context, host, user, command string) (string, error) {
|
|
initSSH()
|
|
if len(sshKey) == 0 {
|
|
return "", fmt.Errorf("no SSH key available")
|
|
}
|
|
if user == "" {
|
|
user = sshUser
|
|
}
|
|
|
|
addr := host + ":22"
|
|
signer, err := ssh.ParsePrivateKey(sshKey)
|
|
if err != nil {
|
|
return "", fmt.Errorf("parse key: %w", err)
|
|
}
|
|
|
|
cfg := &ssh.ClientConfig{
|
|
User: user,
|
|
Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
|
|
HostKeyCallback: ssh.InsecureIgnoreHostKey(),
|
|
Timeout: 10 * time.Second,
|
|
}
|
|
|
|
client, err := ssh.Dial("tcp", addr, cfg)
|
|
if err != nil {
|
|
return "", fmt.Errorf("dial %s: %w", host, err)
|
|
}
|
|
defer client.Close()
|
|
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
return "", fmt.Errorf("session: %w", err)
|
|
}
|
|
defer session.Close()
|
|
|
|
type result struct {
|
|
out []byte
|
|
err error
|
|
}
|
|
done := make(chan result, 1)
|
|
go func() {
|
|
// Recovers a panic in CombinedOutput (SSH library internals, rare but
|
|
// not impossible) and reports it as a failed command instead of
|
|
// crashing the whole api process — every gated action runs through
|
|
// this function, so an unrecovered panic here would take down every
|
|
// concurrently-running task's execution, not just this one. Without
|
|
// this, a panic would ALSO silently degrade to "wait out the full
|
|
// timeout" (done never receives, the select below falls through to
|
|
// its time.After case) rather than crashing outright — recovering
|
|
// and sending an immediate result is strictly better: the caller
|
|
// finds out now, not after sshExecTimeout.
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)}
|
|
}
|
|
}()
|
|
out, err := session.CombinedOutput(command)
|
|
done <- result{out, err}
|
|
}()
|
|
|
|
select {
|
|
case r := <-done:
|
|
text := strings.TrimSpace(string(r.out))
|
|
// A non-zero exit MUST surface as an error — matching the fix
|
|
// applied to httpapi's sshExec (this copy still had the original
|
|
// bug: only erroring when there was no output at all, so a command
|
|
// that failed but printed something was silently reported as
|
|
// success).
|
|
if r.err != nil {
|
|
if text != "" {
|
|
return text, fmt.Errorf("%w: %s", r.err, text)
|
|
}
|
|
return text, fmt.Errorf("exec: %w", r.err)
|
|
}
|
|
return text, nil
|
|
case <-time.After(sshExecTimeout):
|
|
session.Close()
|
|
client.Close()
|
|
return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
|
|
case <-ctx.Done():
|
|
session.Close()
|
|
client.Close()
|
|
return "", ctx.Err()
|
|
}
|
|
}
|
|
|
|
func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUser string, err error) {
|
|
var attrs string
|
|
err = pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE slug = $1", entitySlug).Scan(&attrs)
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("entity not found: %s", entitySlug)
|
|
}
|
|
|
|
var m map[string]interface{}
|
|
if err := json.Unmarshal([]byte(attrs), &m); err != nil {
|
|
return "", "", fmt.Errorf("parse attributes: %w", err)
|
|
}
|
|
|
|
if ip, ok := m["lan_ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
if mesh, ok := m["mesh"].(map[string]interface{}); ok {
|
|
for _, proto := range []string{"netbird", "tailscale"} {
|
|
if p, ok := mesh[proto].(map[string]interface{}); ok {
|
|
if ip, ok := p["ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return "", "", fmt.Errorf("no IP found for %s", entitySlug)
|
|
}
|
|
|
|
// htmlTagRe strips HTML tags for the naive text extraction in httpGet.
|
|
var htmlTagRe = regexp.MustCompile(`(?s)<(script|style)[^>]*>.*?</(script|style)>|<[^>]+>`)
|
|
|
|
// httpGet fetches a public URL and returns sanitized, size-capped text so the
|
|
// agent can read a service's README/site before provisioning. Guards: scheme
|
|
// allow-list, request timeout, 16KB body cap, and blocking of RFC1918/loopback
|
|
// hosts to avoid using the tool as an SSRF pivot into the private mesh.
|
|
func httpGet(ctx context.Context, rawURL string) *mcp.CallToolResult {
|
|
if rawURL == "" {
|
|
return textResult("error: url required")
|
|
}
|
|
u, err := url.Parse(strings.TrimSpace(rawURL))
|
|
if err != nil || (u.Scheme != "http" && u.Scheme != "https") {
|
|
return textResult("error: url must be an absolute http(s) URL")
|
|
}
|
|
if isPrivateHost(u.Hostname()) {
|
|
return textResult("error: refusing to fetch private/loopback address")
|
|
}
|
|
|
|
cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
defer cancel()
|
|
hreq, err := http.NewRequestWithContext(cctx, http.MethodGet, u.String(), nil)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
hreq.Header.Set("User-Agent", "oikos-nomos/1.0 (+homelab agent)")
|
|
hreq.Header.Set("Accept", "text/plain, text/html, application/json;q=0.9, */*;q=0.5")
|
|
|
|
client := &http.Client{Timeout: 20 * time.Second}
|
|
resp, err := client.Do(hreq)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: fetch failed: %v", err))
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
const cap = 256 * 1024 // read a bit extra pre-strip; final output capped below
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, cap))
|
|
ct := resp.Header.Get("Content-Type")
|
|
text := sanitizeBody(ct, string(body))
|
|
return textResult(fmt.Sprintf("GET %s → %d %s\n\n%s", u.String(), resp.StatusCode, ct, text))
|
|
}
|
|
|
|
// sanitizeBody strips scripts/styles/tags from HTML, unescapes entities,
|
|
// collapses whitespace, and caps the result to ~16KB of readable text.
|
|
func sanitizeBody(contentType, raw string) string {
|
|
text := raw
|
|
if strings.Contains(contentType, "html") {
|
|
text = htmlTagRe.ReplaceAllString(text, " ")
|
|
text = html.UnescapeString(text)
|
|
text = strings.Join(strings.Fields(text), " ")
|
|
}
|
|
if len(text) > 16*1024 {
|
|
text = text[:16*1024] + "\n…[truncated]"
|
|
}
|
|
return text
|
|
}
|
|
|
|
// isPrivateHost reports whether host is loopback, link-local, or RFC1918.
|
|
func isPrivateHost(host string) bool {
|
|
host = strings.ToLower(host)
|
|
if host == "localhost" || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") {
|
|
return true
|
|
}
|
|
ip := net.ParseIP(host)
|
|
if ip == nil {
|
|
return false // hostname; DNS may still resolve private — acceptable for a homelab tool
|
|
}
|
|
return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsUnspecified()
|
|
}
|
|
|
|
// resolveExecTarget resolves any target slug (host: or lxc:) to the SSH
|
|
// endpoint that will actually run the command, and a wrap function that turns
|
|
// a plain shell command into whatever must actually be sent over that SSH
|
|
// connection: identity for a host, `pct exec <pve_id> -- ...` for an LXC.
|
|
//
|
|
// The lxc.attributes.host value is stored WITHOUT a "host:" prefix (e.g.
|
|
// "strong", not "host:strong") — see pct_create's entity registration. The
|
|
// pre-existing pct_exec handler queried resolveHost with that bare value
|
|
// directly, which can never match a "host:*" slug and always fails; this
|
|
// prefixes it correctly.
|
|
func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(cmd string) string, err error) {
|
|
if strings.HasPrefix(targetSlug, "host:") {
|
|
host, user, err = resolveHost(ctx, pool, targetSlug)
|
|
return host, user, func(cmd string) string { return cmd }, err
|
|
}
|
|
if strings.HasPrefix(targetSlug, "lxc:") {
|
|
var pveID, hostAttr string
|
|
// COALESCE the host column: many older LXC entities (seeded from
|
|
// inventory, not provisioned by pct_create) have pve_id but no host
|
|
// attribute at all. Scanning a SQL NULL into a plain string errors
|
|
// the whole row, wrongly reporting "missing pve_id" even when it was
|
|
// present — COALESCE avoids the NULL, "" is handled below.
|
|
if qerr := pool.QueryRow(ctx, "SELECT attributes->>'pve_id', COALESCE(attributes->>'host', '') FROM entities WHERE slug = $1", targetSlug).Scan(&pveID, &hostAttr); qerr != nil || pveID == "" {
|
|
return "", "", nil, fmt.Errorf("LXC not found or missing pve_id: %s", targetSlug)
|
|
}
|
|
hostSlug := hostAttr
|
|
if hostSlug == "" {
|
|
hostSlug = "hubris" // documented default Proxmox host when unset
|
|
}
|
|
if !strings.HasPrefix(hostSlug, "host:") {
|
|
hostSlug = "host:" + hostSlug
|
|
}
|
|
host, user, err = resolveHost(ctx, pool, hostSlug)
|
|
id := pveID
|
|
return host, user, func(cmd string) string {
|
|
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
|
|
return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", id, b64)
|
|
}, err
|
|
}
|
|
return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug> or lxc:<slug>", targetSlug)
|
|
}
|
|
|
|
// classifyAndGate is the shared classify→execute-or-queue path for every
|
|
// mutating command, used by both the general `run` tool and
|
|
// request_execution's restart/systemctl/pct_exec actions. Those legacy
|
|
// actions used to execute immediately over SSH with a hardcoded
|
|
// risk_class='reversible_low' that was never actually evaluated against the
|
|
// command — found live 2026-07-10 when a chat request to restart caddy (the
|
|
// fleet's reverse proxy) executed instantly with no approval at all. Routing
|
|
// every mutating path through the same classifier + approval-queue logic
|
|
// closes that gap without special-casing each caller.
|
|
func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.UUID, targetSlug, command, purpose, declaredRisk, sessionID string) *mcp.CallToolResult {
|
|
riskClass := policy.ClassifyCommand(command, declaredRisk)
|
|
runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose})
|
|
actionCol := "run:" + string(runParams)
|
|
|
|
// Dedup: an identical pending command (same target, command, and
|
|
// purpose) blocks a re-request — stops a tool-calling loop from queuing
|
|
// the same approval repeatedly.
|
|
var existingID string
|
|
derr := pool.QueryRow(ctx, `
|
|
SELECT e.id::text FROM entities e
|
|
JOIN executions ex ON ex.entity_id = e.id
|
|
WHERE e.type = 'execution' AND ex.target_entity_id = $1
|
|
AND ex.action = $2 AND ex.status = 'pending_approval'
|
|
ORDER BY e.created_at DESC LIMIT 1`,
|
|
targetID, actionCol).Scan(&existingID)
|
|
if derr == nil && existingID != "" {
|
|
return textResult(fmt.Sprintf("An identical command is already queued for approval on %s — execution %s. Wait for the operator, don't re-request.", targetSlug, existingID))
|
|
}
|
|
|
|
id, _ := uuid.NewV7()
|
|
correlationID := uuid.New().String()
|
|
execName := "run on " + targetSlug + " (" + id.String() + ")"
|
|
execSlug := "exec:" + targetSlug + ":" + id.String()
|
|
if _, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
|
|
id, execSlug, execName); err != nil {
|
|
return textResult(fmt.Sprintf("error: failed to create execution: %v", err))
|
|
}
|
|
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, $4, 'running', $5, $6) ON CONFLICT DO NOTHING`,
|
|
id, targetID, actionCol, riskClass, correlationID, agentID)
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
|
|
id, targetID)
|
|
if sessionID != "" {
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
|
|
FROM entities t WHERE t.slug = $2
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
|
|
id, "task:"+sessionID)
|
|
}
|
|
|
|
if riskClass == policy.RiskReadOnly {
|
|
host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
|
|
if rerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
|
|
return textResult(fmt.Sprintf("resolve target: %v", rerr))
|
|
}
|
|
out, xerr := sshExec(ctx, host, user, wrap(command))
|
|
if xerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
|
|
return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
|
|
return textResult(fmt.Sprintf("run on %s (read_only, auto): %s", targetSlug, out))
|
|
}
|
|
|
|
// Assent window: if the operator recently approved a plan in this
|
|
// agent's chat session, config_mutation commands auto-run without
|
|
// re-approval. This is the "approve the plan, carry it out" path — the
|
|
// operator approved the overall direction; individual config steps
|
|
// within the window don't each need a separate yes. Destructive
|
|
// commands never auto-run, regardless of window.
|
|
if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID, sessionID) {
|
|
host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
|
|
if rerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
|
|
return textResult(fmt.Sprintf("resolve target: %v", rerr))
|
|
}
|
|
out, xerr := sshExec(ctx, host, user, wrap(command))
|
|
if xerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
|
|
return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
|
|
slog.Info("mcp: run auto-executed via assent window", "target", targetSlug, "execution_id", id)
|
|
return textResult(fmt.Sprintf("run on %s (config_mutation, auto via assent window): %s", targetSlug, out))
|
|
}
|
|
|
|
// Destructive window: a narrow, TARGET-scoped grant opened only after an
|
|
// operator's explicit typed confirmation ("I confirm") on this same
|
|
// target — never by loose assent. Exists for multi-step destructive
|
|
// recovery (e.g. a failed destroy needing stop, then destroy) so the
|
|
// operator isn't asked to re-type "I confirm" for every single command
|
|
// against the thing they just confirmed.
|
|
if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug, sessionID) {
|
|
host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
|
|
if rerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
|
|
return textResult(fmt.Sprintf("resolve target: %v", rerr))
|
|
}
|
|
out, xerr := sshExec(ctx, host, user, wrap(command))
|
|
if xerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
|
|
return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
|
|
slog.Info("mcp: run auto-executed via destructive window", "target", targetSlug, "execution_id", id)
|
|
return textResult(fmt.Sprintf("run on %s (destructive, auto via confirmed-target window): %s", targetSlug, out))
|
|
}
|
|
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class=$2 WHERE entity_id=$1`, id, riskClass)
|
|
createApproval(ctx, pool, id, targetID, "run", string(runParams), riskClass)
|
|
confirmNote := ""
|
|
if riskClass == policy.RiskDestructive {
|
|
confirmNote = " This is classified DESTRUCTIVE — flag that clearly to the operator; it needs explicit confirmation, not just a casual \"go ahead\"."
|
|
}
|
|
return textResult(fmt.Sprintf("run on %s requires approval (risk: %s) — execution %s queued.%s Present the command and purpose to the operator and wait; do not re-request.",
|
|
targetSlug, riskClass, id, confirmNote))
|
|
}
|
|
|
|
// autoApprove updates the approval + execution status in the DB to approved,
|
|
// mirroring what DecideApproval does. Returns true on success. This is used
|
|
// by the assent-window path to skip the operator-approval queue when the
|
|
// operator already approved the overall plan via chat assent.
|
|
// executeApprovedViaAPI calls the HTTP API's approval-decision endpoint to
|
|
// trigger the actual execution. The API server (phase3.executeApprovedAction)
|
|
// handles the real SSH work (pct create, apt upgrade, etc.) in a goroutine.
|
|
// We POST to the decision endpoint to reuse the exact same execution path
|
|
// as a manual Approve-button click, ensuring the audit trail is consistent.
|
|
func executeApprovedViaAPI(ctx context.Context, execID uuid.UUID, targetSlug, actionStr string) {
|
|
apiBase := os.Getenv("OIKOS_API_BASE")
|
|
if apiBase == "" {
|
|
apiBase = "http://api:8090"
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"decision": "approve"})
|
|
client := &http.Client{Timeout: 10 * time.Second}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
|
|
apiBase+"/api/v1/approvals/"+execID.String()+"/decision", bytes.NewReader(body))
|
|
if err != nil {
|
|
slog.Error("mcp: executeApprovedViaAPI request", "error", err)
|
|
return
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
slog.Error("mcp: executeApprovedViaAPI call", "error", err)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusOK {
|
|
// A non-200 here means the real SSH work was never dispatched — this
|
|
// is the call that actually triggers executeApprovedAction via
|
|
// DecideApproval. (A previous version of this comment claimed a
|
|
// non-200 was fine because a since-removed "autoApprove" step had
|
|
// already triggered execution via a raw DB update — it hadn't; that
|
|
// was the bug where auto-approved pct_create/apt_upgrade never
|
|
// actually ran. There is no other path that dispatches the work.)
|
|
slog.Error("mcp: executeApprovedViaAPI non-200 — execution was NOT dispatched", "status", resp.StatusCode, "execution", execID)
|
|
}
|
|
}
|
|
|
|
// assentWindowActive checks whether the operator has recently approved a plan
|
|
// in THIS TASK's chat session. The agent sets an
|
|
// assent_window.agent:<uuid>.session:<id> key in autonomy_settings with an
|
|
// expiry timestamp when chat-assent grants a pending execution. While
|
|
// active, config_mutation commands auto-run without re-approval — the
|
|
// operator approved the overall plan, not each step. Scoped by session, not
|
|
// just agent: with one agent:nomos entity serving every concurrent task, an
|
|
// agent-only key would let approving Task A's plan silently auto-run
|
|
// unapproved actions from a concurrently-running Task B. sessionID comes
|
|
// from the `_session_id` nomos injects into every tool call's wire args
|
|
// (never part of any tool's declared InputSchema, so the model never
|
|
// supplies or sees it) — see cmd/nomos/agent.go's tool dispatch loop.
|
|
func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, sessionID string) bool {
|
|
if agentID == uuid.Nil || sessionID == "" {
|
|
return false // fail closed: no session to scope to means no window
|
|
}
|
|
var expiresStr string
|
|
err := pool.QueryRow(ctx,
|
|
"SELECT value FROM autonomy_settings WHERE key = $1",
|
|
"assent_window.agent:"+agentID.String()+".session:"+sessionID).Scan(&expiresStr)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
expires, err := time.Parse(time.RFC3339, expiresStr)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return time.Now().UTC().Before(expires)
|
|
}
|
|
|
|
// destructiveWindowActive reports whether targetSlug has a live, explicitly-
|
|
// confirmed destructive grant for this agent WITHIN THIS SESSION/TASK. Key
|
|
// format ("destructive_window.agent:<id>.target:<slug>.session:<id>") must
|
|
// match cmd/nomos/store.go's openDestructiveWindow — both processes
|
|
// read/write the same autonomy_settings row. Scoped to one target AND one
|
|
// session so a typed confirmation for destroying container A in task X can
|
|
// never be read as authorizing anything against container A from a
|
|
// different, concurrently-running task Y.
|
|
func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug, sessionID string) bool {
|
|
if agentID == uuid.Nil || targetSlug == "" || sessionID == "" {
|
|
return false
|
|
}
|
|
var expiresStr string
|
|
err := pool.QueryRow(ctx,
|
|
"SELECT value FROM autonomy_settings WHERE key = $1",
|
|
"destructive_window.agent:"+agentID.String()+".target:"+targetSlug+".session:"+sessionID).Scan(&expiresStr)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
expires, err := time.Parse(time.RFC3339, expiresStr)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return time.Now().UTC().Before(expires)
|
|
}
|
|
|
|
// knowledgeSlugRe strips a title down to a slug segment.
|
|
var knowledgeSlugRe = regexp.MustCompile(`[^a-z0-9]+`)
|
|
|
|
func knowledgeSlug(kind, title string) string {
|
|
s := strings.ToLower(strings.TrimSpace(title))
|
|
s = knowledgeSlugRe.ReplaceAllString(s, "-")
|
|
s = strings.Trim(s, "-")
|
|
if s == "" {
|
|
s = "note"
|
|
}
|
|
if len(s) > 80 {
|
|
s = s[:80]
|
|
}
|
|
return kind + ":nomos/" + s
|
|
}
|
|
|
|
// upsertKnowledge is the agent's write-back path — the missing half of the
|
|
// knowledge loop (search_knowledge/get_entity_knowledge could only read).
|
|
// Without this, everything the agent learned lived only in an ephemeral chat
|
|
// message and was lost; the system could never actually "get better." A
|
|
// knowledge doc IS an entity (type document/investigation/runbook) with a row
|
|
// in knowledge_entities; re-titling the same thing updates in place rather
|
|
// than duplicating. Optionally linked to the entity it's about so
|
|
// get_entity_knowledge surfaces it there.
|
|
func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*mcp.CallToolResult, error) {
|
|
title, _ := args["title"].(string)
|
|
content, _ := args["content"].(string)
|
|
about, _ := args["about"].(string)
|
|
tagsRaw, _ := args["tags"].(string)
|
|
kind, _ := args["kind"].(string)
|
|
|
|
title = strings.TrimSpace(title)
|
|
content = strings.TrimSpace(content)
|
|
if title == "" || content == "" {
|
|
return textResult("error: title and content are required"), nil
|
|
}
|
|
switch kind {
|
|
case "document", "investigation", "runbook":
|
|
case "":
|
|
kind = "investigation"
|
|
default:
|
|
return textResult(fmt.Sprintf("error: kind must be document, investigation, or runbook (got %q)", kind)), nil
|
|
}
|
|
|
|
var tags []string
|
|
for _, t := range strings.Split(tagsRaw, ",") {
|
|
if t = strings.TrimSpace(t); t != "" {
|
|
tags = append(tags, t)
|
|
}
|
|
}
|
|
|
|
slug := knowledgeSlug(kind, title)
|
|
|
|
// Upsert the knowledge-doc entity, getting its id whether it already
|
|
// existed or we just created it.
|
|
docID, _ := uuid.NewV7()
|
|
err := pool.QueryRow(ctx, `
|
|
INSERT INTO entities (id, slug, type, name, attributes)
|
|
VALUES ($1, $2, $3, $4, '{}')
|
|
ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name, updated_at = now()
|
|
RETURNING id`, docID, slug, kind, title).Scan(&docID)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error creating knowledge entity: %v", err)), nil
|
|
}
|
|
|
|
// Upsert the knowledge content (search column is generated, don't set it).
|
|
_, err = pool.Exec(ctx, `
|
|
INSERT INTO knowledge_entities (entity_id, title, content, source, tags, updated_at)
|
|
VALUES ($1, $2, $3, 'nomos-agent', $4, now())
|
|
ON CONFLICT (entity_id) DO UPDATE
|
|
SET title = EXCLUDED.title, content = EXCLUDED.content,
|
|
tags = EXCLUDED.tags, updated_at = now()`,
|
|
docID, title, content, tags)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error writing knowledge: %v", err)), nil
|
|
}
|
|
|
|
// Link it to the entity it's about, if given and not already linked.
|
|
linked := ""
|
|
if about = strings.TrimSpace(about); about != "" {
|
|
var targetID uuid.UUID
|
|
if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", about).Scan(&targetID); qerr == nil {
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`,
|
|
docID, targetID)
|
|
linked = " and linked to " + about
|
|
} else {
|
|
linked = fmt.Sprintf(" (note: entity %q not found, saved unlinked)", about)
|
|
}
|
|
}
|
|
|
|
_ = observability.Event(ctx, sqlcgen.New(pool), "knowledge.upserted", &docID, "info", "mcp", "",
|
|
map[string]any{"slug": slug, "title": title, "kind": kind})
|
|
|
|
return textResult(fmt.Sprintf("Saved knowledge %q as %s%s. It's now searchable via search_knowledge and will surface in future sessions.", title, slug, linked)), nil
|
|
}
|
|
|
|
func createApproval(ctx context.Context, pool *db.Pool, execID, targetID uuid.UUID, action, params, riskClass string) {
|
|
p := map[string]any{"action": action, "params": params, "execution_id": execID.String()}
|
|
payload, _ := json.Marshal(p)
|
|
// approvals.entity_id is PK + FK to entities(id). Reuse the execution's
|
|
// entity (already inserted by request_execution) so the FK is satisfied —
|
|
// a fresh UUID here had no matching entities row, so the INSERT silently
|
|
// failed, orphaning the execution and never alerting the operator. One
|
|
// execution maps to at most one approval, so the 1:1 identity holds.
|
|
if _, err := pool.Exec(ctx, `
|
|
INSERT INTO approvals (entity_id, subject_entity_id, action, risk_class,
|
|
kind, payload, status, expires_at, created_at)
|
|
VALUES ($1, $2, $3, $4, 'execution', $5::jsonb, 'pending',
|
|
now() + interval '1 hour', now())`,
|
|
execID, targetID, action, riskClass, string(payload)); err != nil {
|
|
slog.Error("createApproval: insert approval", "error", err, "execution", execID)
|
|
return
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE executions SET approval_id = $1 WHERE entity_id = $1`, execID); err != nil {
|
|
slog.Error("createApproval: link approval to execution", "error", err, "execution", execID)
|
|
}
|
|
|
|
// Emit for SSE fan-out — the operator-facing moment: an agent-requested
|
|
// gated action is now awaiting a decision.
|
|
_ = observability.Event(ctx, sqlcgen.New(pool), "approval.created", &execID, "warning", "mcp", "",
|
|
map[string]any{"action": action, "params": params, "risk_class": riskClass})
|
|
} |