0.22.0 — full MCP agent surface: 19 new tools, async run, session reliability

Stage 1 — Foundation: - Target validation: iptables + systemctl/docker target-type gates - Async run for long-running commands (sleep/wait/poll loops) - audit_log.session_id plumbing (SQL, sqlcgen, 18 call sites) Stage 2 — External agent observe (11 tools): - get_dashboard_summary, get_ontology, list_checks, list_executions - get_knowledge_revisions, get_knowledge_duplicates, get_knowledge_orphans - list_knowledge_tags, list_entity_sessions, find_entities_by - 3 resource templates: oikos://entity/{slug}, knowledge/{id}, execution/{id} Stage 3 — Nomos reliability: - complete_task(success) refused without verification (upgraded from warn) - sessionHasPlan excludes replaced steps (forces propose_plan after reopen) - Bash syntax validation in run() (rejects literal \n, flag-space typos) - Scope gate in SOUL.md (ask before pivoting to unrelated subsystem) Stage 4 — External agent act (9 mutation tools): - ack_signal, resolve_signal, mute_signal, cancel_execution - update_check, delete_knowledge, restore_knowledge - merge_knowledge, rename_knowledge_tag
2026-08-04 23:51:55 +02:00
parent 85f0bb67fa
commit 4e294b3630
20 changed files with 1264 additions and 41 deletions
--- a/internal/mcp/server.go
+++ b/internal/mcp/server.go
@@ -72,9 +72,121 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
 	for _, t := range allTools(pool, agentID) {
 		s.AddTool(t.tool, withActivityLogging(pool, agentID, t.tool.Name, t.handler))
 	}
+
+	// Resource templates: let MCP clients browse and attach entities,
+	// knowledge entries, and executions as conversation resources.
+	s.AddResourceTemplate(&mcp.ResourceTemplate{
+		URITemplate: "oikos://entity/{slug}",
+		Name:        "Entity",
+		Description: "Oikos entity by slug (e.g. host:hubris, lxc:jellyfin)",
+		MIMEType:    "application/json",
+	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
+		slug := matches["slug"]
+		var id uuid.UUID
+		if u, err := uuid.Parse(slug); err == nil {
+			id = u
+		} else {
+			pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&id)
+		}
+		if id == uuid.Nil {
+			return "", fmt.Errorf("entity not found: %s", slug)
+		}
+		result := queryEntity(ctx, pool, slug)
+		return result.Content[0].(*mcp.TextContent).Text, nil
+	}))
+
+	s.AddResourceTemplate(&mcp.ResourceTemplate{
+		URITemplate: "oikos://knowledge/{id}",
+		Name:        "Knowledge",
+		Description: "Knowledge entry by entity slug or UUID",
+		MIMEType:    "application/json",
+	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
+		idOrSlug := matches["id"]
+		var entityID uuid.UUID
+		if u, err := uuid.Parse(idOrSlug); err == nil {
+			entityID = u
+		} else {
+			pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&entityID)
+		}
+		if entityID == uuid.Nil {
+			return "", fmt.Errorf("knowledge not found: %s", idOrSlug)
+		}
+		result := queryRows(ctx, pool, `
+			SELECT ke.title, ke.content, ke.tags::text, e.slug, e.type AS kind,
+			       ke.updated_at::text
+			FROM knowledge_entities ke
+			JOIN entities e ON e.id = ke.entity_id
+			WHERE ke.entity_id = $1`, entityID)
+		return result.Content[0].(*mcp.TextContent).Text, nil
+	}))
+
+	s.AddResourceTemplate(&mcp.ResourceTemplate{
+		URITemplate: "oikos://execution/{id}",
+		Name:        "Execution",
+		Description: "Execution by UUID (returns status, result, timing)",
+		MIMEType:    "application/json",
+	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
+		result := queryRows(ctx, pool, `
+			SELECT e.entity_id, te.slug AS target, e.action, e.risk_class,
+			       e.status, e.result::text, e.duration_ms,
+			       e.started_at::text, e.completed_at::text
+			FROM executions e
+			JOIN entities te ON te.id = e.target_entity_id
+			WHERE e.entity_id = $1`, matches["id"])
+		return result.Content[0].(*mcp.TextContent).Text, nil
+	}))
+
 	return s
 }

+// resourceHandler adapts a simple func(ctx, params) → (string, error) into
+// an MCP ResourceHandler, reading the URI matched by a ResourceTemplate.
+func resourceHandler(pool *db.Pool, fn func(ctx context.Context, matches map[string]string) (string, error)) mcp.ResourceHandler {
+	return func(ctx context.Context, req *mcp.ReadResourceRequest) (*mcp.ReadResourceResult, error) {
+		uri := req.Params.URI
+		matches := matchURITemplate(uri)
+		if matches == nil {
+			return nil, mcp.ResourceNotFoundError(uri)
+		}
+
+		text, err := fn(ctx, matches)
+		if err != nil {
+			return nil, mcp.ResourceNotFoundError(uri)
+		}
+
+		result, err := json.MarshalIndent(json.RawMessage(text), "", "  ")
+		if err != nil {
+			result = []byte(text)
+		}
+
+		return &mcp.ReadResourceResult{
+			Contents: []*mcp.ResourceContents{{
+				URI:      uri,
+				MIMEType: "application/json",
+				Text:     string(result),
+			}},
+		}, nil
+	}
+}
+
+// matchURITemplate extracts parameters from a URI that matches one of the
+// oikos:// resource templates. Returns nil if the URI doesn't match.
+func matchURITemplate(uri string) map[string]string {
+	// oikos://entity/{slug}
+	if rest, ok := strings.CutPrefix(uri, "oikos://entity/"); ok && rest != "" {
+		return map[string]string{"slug": rest}
+	}
+	// oikos://knowledge/{id}
+	if rest, ok := strings.CutPrefix(uri, "oikos://knowledge/"); ok && rest != "" {
+		return map[string]string{"id": rest}
+	}
+	// oikos://execution/{id}
+	if rest, ok := strings.CutPrefix(uri, "oikos://execution/"); ok && rest != "" {
+		return map[string]string{"id": rest}
+	}
+	return nil
+}
+
 // 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 {
@@ -628,6 +740,90 @@ func autoRun(ctx context.Context, pool *db.Pool, id uuid.UUID, targetSlug, comma
 	return out, nil
 }

+// autoRunAsync starts a command in a goroutine, marking it running and returning
+// immediately. The caller gets an execution_id to poll with get_execution_status.
+// Used for commands containing sleep/wait/poll loops that would exceed the MCP
+// client timeout (120s) — the execution continues server-side.
+func autoRunAsync(ctx context.Context, pool *db.Pool, id uuid.UUID, targetSlug, command string) {
+	startedAt := time.Now()
+	if _, err := pool.Exec(ctx,
+		`UPDATE executions SET status='running', started_at=$2 WHERE entity_id=$1`,
+		id, startedAt); err != nil {
+		slog.Error("mcp: mark execution running (async)", "error", err, "execution_id", id)
+	}
+
+	host, user, wrap, err := resolveExecTarget(ctx, pool, targetSlug)
+	if err != nil {
+		pool.Exec(ctx,
+			`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
+			id, jsonErr("%s", err.Error()), int(time.Since(startedAt).Milliseconds()))
+		slog.Error("mcp: async run resolve target", "error", err, "execution_id", id, "target", targetSlug)
+		return
+	}
+
+	var correlationID string
+	if qerr := pool.QueryRow(ctx,
+		`SELECT correlation_id FROM executions WHERE entity_id = $1`, id).Scan(&correlationID); qerr != nil {
+		correlationID = ""
+	}
+
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("mcp: async run panic", "panic", r, "execution_id", id)
+				pool.Exec(context.Background(),
+					`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
+					id, jsonErr("panic: %v", r), int(time.Since(startedAt).Milliseconds()))
+			}
+		}()
+
+		sink, flush := execlog.New(context.Background(), pool, id, correlationID)
+		out, execErr := sshExecStream(context.Background(), host, user, wrap(command), sink)
+		flush()
+		if execErr != nil {
+			pool.Exec(context.Background(),
+				`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
+				id, jsonErr("%s: %s", execErr.Error(), out), int(time.Since(startedAt).Milliseconds()))
+			slog.Error("mcp: async run failed", "error", execErr, "execution_id", id, "output", out)
+		} else {
+			pool.Exec(context.Background(),
+				`UPDATE executions SET status='completed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
+				id, jsonOut(out), int(time.Since(startedAt).Milliseconds()))
+		}
+	}()
+}
+
+// isLongRunningCommand detects shell commands containing sleep, wait, or poll
+// loops that indicate the command will exceed the MCP client timeout (120s).
+// These commands should use autoRunAsync to avoid the client timing out while
+// the command continues server-side.
+func isLongRunningCommand(cmd string) bool {
+	cmd = strings.TrimSpace(cmd)
+
+	// sleep with duration — `sleep 30`, `sleep 1m`, etc.
+	if sleepRe.MatchString(cmd) {
+		return true
+	}
+
+	// while/shell poll loops with sleep: `while ...; do ... sleep; done`
+	if pollRe.MatchString(cmd) {
+		return true
+	}
+
+	// standalone wait command
+	if waitRe.MatchString(cmd) {
+		return true
+	}
+
+	return false
+}
+
+var (
+	sleepRe = regexp.MustCompile(`\bsleep\s+\d`)
+	pollRe  = regexp.MustCompile(`\bwhile\b.*\bsleep\b`)
+	waitRe  = regexp.MustCompile(`\bwait\s+\d|[&;]\s*wait\b`)
+)
+
 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})
@@ -651,6 +847,14 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 	// not be dispatched against lxc:/vm: targets — those aren't Proxmox hosts
 	// and don't have these tools. Caught live 2026-08-04: the agent ran
 	// `qm stop 100` against lxc:dns, wasting a turn.
+
+	// Command syntax validation: catch LLM-generated bash bugs before they
+	// hit the shell. The model sometimes inserts literal \n between commands
+	// or puts spaces inside flags — these always fail, so reject early.
+	if syntaxErr := validateCommandSyntax(command); syntaxErr != "" {
+		return textResult(syntaxErr)
+	}
+
 	if cmdPrefix, hostOnly := hostOnlyCommand(command); hostOnly && !strings.HasPrefix(targetSlug, "host:") {
 		hostSuggestion := resolveProxmoxHostSlug(ctx, pool, targetSlug, "")
 		if hostSuggestion == "" {
@@ -660,6 +864,14 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 			cmdPrefix, targetSlug, cmdPrefix, hostSuggestion))
 	}

+	// systemctl and docker work on hosts and LXCs, but not VMs.
+	if cmdPrefix, hostLxc := hostLxcCommand(command); hostLxc {
+		if !strings.HasPrefix(targetSlug, "host:") && !strings.HasPrefix(targetSlug, "lxc:") {
+			return textResult(fmt.Sprintf("Cannot run %q on %s — %s only works on host:* or lxc:* targets.",
+				cmdPrefix, targetSlug, cmdPrefix))
+		}
+	}
+
 	// 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.
@@ -766,6 +978,19 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 	// Link classification to execution.
 	pool.Exec(ctx, `UPDATE executions SET classification_id = $2 WHERE entity_id = $1`, id, classID)

+	// Audit: record the execution creation with session_id for traceability.
+	// Every run call, whether auto-run or queued-for-approval, gets an audit
+	// entry so the agent's activity is traceable back to the originating session.
+	var auditSessionID *uuid.UUID
+	if sessionID != "" && sessionID != "ephemeral" {
+		if sid, serr := uuid.Parse(sessionID); serr == nil {
+			auditSessionID = &sid
+		}
+	}
+	_ = observability.Audit(ctx, sqlcgen.New(pool), "agent", "nomos", "run",
+		&id, "POST", "/mcp", correlationID, auditSessionID,
+		map[string]any{"command": command, "target": targetSlug, "risk_class": riskClass, "purpose": purpose})
+
 	// read_only and reversible_low both run unattended, as seeds/policy.yaml
 	// and .agents/OIKOS.md declare ("reversible_low — restart, cache clear,
 	// sync pull. Unattended + ledger.").
@@ -784,6 +1009,11 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 	// agent still cannot talk a command DOWN: declaring reversible_low on
 	// something computed as config_mutation keeps config_mutation.
 	if riskClass == policy.RiskReadOnly || riskClass == policy.RiskReversibleLow {
+		if isLongRunningCommand(command) {
+			autoRunAsync(ctx, pool, id, targetSlug, command)
+			return textResult(fmt.Sprintf("run on %s (%s, async): started — execution %s. Poll with get_execution_status(%s) for result.",
+				targetSlug, riskClass, id, id))
+		}
 		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
 		if xerr != nil {
 			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
@@ -802,6 +1032,12 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 	// consent. The assent window, opened only on operator approval, is the
 	// sole gate for config_mutation auto-run.)
 	if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID, sessionID) {
+		if isLongRunningCommand(command) {
+			autoRunAsync(ctx, pool, id, targetSlug, command)
+			slog.Info("mcp: run async via assent window", "target", targetSlug, "execution_id", id)
+			return textResult(fmt.Sprintf("run on %s (config_mutation, async via assent window): started — execution %s. Poll with get_execution_status(%s) for result.",
+				targetSlug, id, id))
+		}
 		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
 		if xerr != nil {
 			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
@@ -817,6 +1053,12 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
 	// 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) {
+		if isLongRunningCommand(command) {
+			autoRunAsync(ctx, pool, id, targetSlug, command)
+			slog.Info("mcp: run async via destructive window", "target", targetSlug, "execution_id", id)
+			return textResult(fmt.Sprintf("run on %s (destructive, async via confirmed-target window): started — execution %s. Poll with get_execution_status(%s) for result.",
+				targetSlug, id, id))
+		}
 		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
 		if xerr != nil {
 			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
@@ -887,9 +1129,16 @@ func executeApprovedViaAPI(ctx context.Context, execID uuid.UUID, targetSlug, ac
 // targets (not LXCs or VMs). Running these against an lxc: or vm: target
 // always fails with "command not found" and wastes a turn.
 var hostOnlyCommands = map[string]bool{
-	"qm":    true,
-	"pct":   true,
-	"pvesh": true,
+	"qm":      true,
+	"pct":     true,
+	"pvesh":   true,
+	"iptables": true,
+}
+
+// hostLxcCommands maps command prefixes valid on host:* and lxc:* but not vm:*.
+var hostLxcCommands = map[string]bool{
+	"systemctl": true,
+	"docker":    true,
 }

 // hostOnlyCommand checks whether the leading word of cmd is a host-only
@@ -918,22 +1167,70 @@ func hostOnlyCommand(cmd string) (string, bool) {
 	return first, hostOnlyCommands[first]
 }

+// hostLxcCommand checks whether the leading word of cmd is a command valid on
+// host:* and lxc:* targets but not vm:*. Returns the command word and true if
+// the command is restricted to host/lxc.
+func hostLxcCommand(cmd string) (string, bool) {
+	trimmed := strings.TrimSpace(cmd)
+	parts := strings.Fields(trimmed)
+	if len(parts) == 0 {
+		return "", false
+	}
+	first := parts[0]
+	if idx := strings.LastIndexByte(first, '/'); idx >= 0 {
+		first = first[idx+1:]
+	}
+	return first, hostLxcCommands[first]
+}
+
+// validateCommandSyntax checks for common LLM-generated bash errors that always
+// fail at the shell. Returns an error message or "" if the command looks valid.
+func validateCommandSyntax(cmd string) string {
+	// Reject literal \n (the LLM sometimes writes `echo "---" && \n curl ...`
+	// — the \n is literal in the command string, not an actual newline).
+	if strings.Contains(cmd, "\\n") {
+		return fmt.Sprintf("Command contains literal '\\n' — use ';' or '&&' between commands, not a literal backslash-n. Command: %q", cmd)
+	}
+
+	// Reject `&& \n` patterns (the LLM writes `cmd1 && \n cmd2` — the \n is
+	// a literal newline that bash interprets as a command separator, but the
+	// leading backslash makes it a syntax error).
+	if andBackslashRe.MatchString(cmd) {
+		return fmt.Sprintf("Command contains '&&' followed by a literal backslash-newline — remove the backslash or use ';' instead. Command: %q", cmd)
+	}
+
+	// Reject `\` at end of command with no continuation (last line ends with
+	// backslash but there's nothing after it).
+	trimmed := strings.TrimSpace(cmd)
+	if strings.HasSuffix(trimmed, "\\") {
+		return fmt.Sprintf("Command ends with a backslash but has nothing after it to continue. Remove the trailing '\\'. Command: %q", cmd)
+	}
+
+	// Warn on common flag typos: `head - n`, `grep - i`, `tail - n`, etc.
+	// These are space-between-flag-and-value errors the LLM produces.
+	if flagSpaceRe.MatchString(cmd) {
+		return fmt.Sprintf("Command has a space between a flag and its value (e.g. 'head - n' instead of 'head -n'). Remove the space. Command: %q", cmd)
+	}
+
+	return ""
+}
+
+var andBackslashRe = regexp.MustCompile(`&&\s*\\\s*\n`)
+var flagSpaceRe = regexp.MustCompile(`\b(head|tail|grep|sed|awk|sort|uniq|wc)\s+(-\w)\s+\w`)
+
 // sessionHasPlan reports whether this nomos session has any plan step on
-// record (any generation, any status). Used by the P1 plan-first gate in
-// classifyAndGate to refuse `run` before `propose_plan` has been called.
-// A `replaced` step (from a prior plan generation that was superseded by a
-// follow-up sub-task — see store.reopenSession) still counts: it proves the
-// agent once framed a plan for this session, and the reopen path guarantees a
-// fresh `propose_plan` will run before the next `run` anyway. Fails closed
-// (returns true) when the query errors so a transient DB issue doesn't block
-// an otherwise-valid run.
+// record that isn't `replaced`. Replaced steps (from session reopen via
+// store.reopenSession) don't count — the agent must propose fresh plan before
+// any `run`. Fails closed (returns true) when the query errors so a transient
+// DB issue doesn't block an otherwise-valid run.
 func sessionHasPlan(ctx context.Context, pool *db.Pool, sessionID string) bool {
 	if sessionID == "" {
 		return true // no session → no gate (direct MCP call from a script)
 	}
 	var count int
 	if err := pool.QueryRow(ctx,
-		`SELECT COUNT(*) FROM session_plan_steps WHERE session_id = $1`,
+		`SELECT COUNT(*) FROM session_plan_steps
+		 WHERE session_id = $1 AND status <> 'replaced'`,
 		sessionID).Scan(&count); err != nil {
 		return true // fail open on DB error — don't block work over a flake
 	}