From 3823a8241735afa51ca7d2979687153b445af232 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 7 Jul 2026 16:07:08 +0200 Subject: [PATCH] =?UTF-8?q?phase=204:=20hermes=20agent=20=E2=80=94=20MCP?= =?UTF-8?q?=20tools,=20activity=20logging,=20'all'=20role,=20wiring=20fixe?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - mcp/server.go: 7 new tools (get_signal_history, get_patterns, get_skills, request_execution, get_trend, get_event_timeline, get_agent_activity), agent_activity logging middleware on every tool call. - phase3.go: QueryAgentActivity REST handler implemented (was stub). Fixed scan count mismatch in ListSkills/PatchSkill/ListSkillVersions (13 cols → 12 targets). Fixed AgentActivity cursor pagination (lexicographic → integer comparison). Fixed s.Slug → s.Name in log. - cmd/oikos/main.go: 'all' role now runs api + scheduler + notifier in one process. Replaced nil SchedulerRunner/NotifierRunner with direct scheduler.RunnerForMain() / notifier.RunnerForMain() imports. Added runWithPool helper for standalone scheduler/notifier roles. - internal/config/config.go: added HermesAgentID env var. - internal/httpapi/server.go: pass HermesAgentID to MCP handler. - docker-compose.yml: added scheduler and notifier services (dev profile). - hermes/: config.yaml, SOUL.md, skills/homelab-ops/SKILL.md. - Cleaned up: scheduler/init.go dead code, mcp/server.go pgx import guard. --- cmd/oikos/main.go | 59 ++++--- docker-compose.yml | 35 ++++ hermes/SOUL.md | 50 ++++++ hermes/config.yaml | 26 +++ hermes/skills/homelab-ops/SKILL.md | 45 ++++++ internal/config/config.go | 6 + internal/httpapi/phase3.go | 96 ++++++++++- internal/httpapi/server.go | 9 +- internal/mcp/server.go | 252 ++++++++++++++++++++++++++--- internal/mcp/server_test.go | 4 +- internal/notifier/notifier.go | 5 + internal/scheduler/init.go | 16 -- 12 files changed, 542 insertions(+), 61 deletions(-) create mode 100644 hermes/SOUL.md create mode 100644 hermes/config.yaml create mode 100644 hermes/skills/homelab-ops/SKILL.md diff --git a/cmd/oikos/main.go b/cmd/oikos/main.go index 2c0d0c6..a177fd2 100644 --- a/cmd/oikos/main.go +++ b/cmd/oikos/main.go @@ -13,15 +13,14 @@ import ( "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/httpapi" + "github.com/dtoro/oikos/internal/notifier" "github.com/dtoro/oikos/internal/observability" + "github.com/dtoro/oikos/internal/scheduler" "github.com/jackc/pgx/v5" ) -// SchedulerRunner is set by the scheduler init() to avoid circular imports. -var SchedulerRunner func(context.Context, *db.Pool, config.Config) - -// NotifierRunner is set by the notifier init() to avoid circular imports. -var NotifierRunner func(context.Context, *db.Pool, config.Config) +var schedulerRunner = scheduler.RunnerForMain() +var notifierRunner = notifier.RunnerForMain() func main() { if len(os.Args) < 2 { @@ -64,22 +63,30 @@ func main() { os.Exit(1) } case "scheduler": - if SchedulerRunner != nil { - SchedulerRunner(ctx, nil, cfg) - } else { - slog.Error("scheduler not compiled in (import internal/scheduler)") - os.Exit(1) - } + runWithPool(ctx, cfg, "scheduler", schedulerRunner) case "notifier": - if NotifierRunner != nil { - NotifierRunner(ctx, nil, cfg) - } else { - slog.Error("notifier not compiled in (import internal/notifier)") + runWithPool(ctx, cfg, "notifier", notifierRunner) + case "all": + pool, err := db.New(ctx, cfg.DatabaseURL) + if err != nil { + slog.Error("connect db", "error", err) + os.Exit(1) + } + defer pool.Close() + + if err := pool.Migrate(ctx); err != nil { + slog.Error("migrate", "error", err) + os.Exit(1) + } + + go schedulerRunner(ctx, pool, cfg) + go notifierRunner(ctx, pool, cfg) + + slog.Info("all: starting api with scheduler + notifier in background") + if err := httpapi.ListenAndServe(ctx, pool, cfg); err != nil { + slog.Error("api failed", "error", err) os.Exit(1) } - case "all": - slog.Info("all role not yet implemented (runs api + scheduler + notifier in one process)") - os.Exit(1) case "version": fmt.Println("oikos dev (Phase 1)") case "help", "--help", "-h": @@ -231,6 +238,22 @@ func runAPI(ctx context.Context, cfg config.Config) error { return err } +func runWithPool(ctx context.Context, cfg config.Config, name string, fn func(context.Context, *db.Pool, config.Config)) { + pool, err := db.New(ctx, cfg.DatabaseURL) + if err != nil { + slog.Error("connect db", "role", name, "error", err) + os.Exit(1) + } + defer pool.Close() + + if err := pool.Migrate(ctx); err != nil { + slog.Error("migrate", "role", name, "error", err) + os.Exit(1) + } + + fn(ctx, pool, cfg) +} + func runExport(ctx context.Context, cfg config.Config) error { pool, err := db.New(ctx, cfg.DatabaseURL) if err != nil { diff --git a/docker-compose.yml b/docker-compose.yml index d62f9fa..bdb1abb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -60,11 +60,46 @@ services: OIKOS_API_LISTEN: ":8090" OIKOS_ENV: dev OIKOS_DEBUG: "true" + OIKOS_HERMES_AGENT_ID: ${OIKOS_HERMES_AGENT_ID:-} ports: - "8090:8090" command: ["api"] stop_signal: SIGTERM stop_grace_period: 30s + # Scheduler (Phase 3) — observe loop + scheduler: + build: + context: . + dockerfile: compose/oikos/Dockerfile + profiles: ["dev", "full"] + depends_on: + seed: + condition: service_completed_successfully + environment: + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable + OIKOS_DEBUG: "true" + OIKOS_SCHEDULER_INTERVAL: "30s" + command: ["scheduler"] + stop_signal: SIGTERM + stop_grace_period: 30s + + # Notifier (Phase 3) — Matrix alerts + notifier: + build: + context: . + dockerfile: compose/oikos/Dockerfile + profiles: ["dev", "full"] + depends_on: + seed: + condition: service_completed_successfully + environment: + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable + OIKOS_DEBUG: "true" + OIKOS_APPROVAL_HMAC_SECRET: ${OIKOS_APPROVAL_HMAC_SECRET:-dev-secret} + command: ["notifier"] + stop_signal: SIGTERM + stop_grace_period: 30s + volumes: pg-data: diff --git a/hermes/SOUL.md b/hermes/SOUL.md new file mode 100644 index 0000000..032b4bd --- /dev/null +++ b/hermes/SOUL.md @@ -0,0 +1,50 @@ +# SOUL.md — Hermes agent persona (Phase 4, container runtime) + +You are **Hermes**, the homelab AI agent running in a Docker container on +mac-mini. You operate in **gateway mode** on mesh-only port 8092. + +## Source of truth + +The Oikos DB is the authoritative source for topology, service state, policy, +and agent activity. The homelab-context repo at `/opt/homelab-context/` backs +the human-facing wiki. When they disagree, the DB wins. + +## Interaction model + +| Tool | Route | +|---|---| +| Read state | MCP tools (query DB directly) | +| Request action | `request_execution` MCP tool (routes through policy gating) | +| Escalate | Matrix notification to operator | +| Self-inspect | `get_agent_activity` MCP tool | + +You have **no SSH access**. All mutations flow through `/executions`, which +the actuator (a separate container with restricted SSH key) picks up. + +## Key MCP tools + +- `get_entity`, `list_entities` — resolve slugs to state +- `get_blast_radius` — understand impact before requesting action +- `get_health_summary` — fleet status at a glance +- `get_signal_history` — open alerts +- `get_trend` — metric trends for decisions +- `request_execution` — the ONLY mutation path +- `get_agent_activity` — your own behavior log + +## Policy awareness + +Before calling `request_execution`: +- Check risk class via `get_entity` on the target +- If `destructive` or `config_mutation`: escalate to operator +- If `reversible_low` with validated pattern: auto-act allowed + +## Token efficiency + +Use MCP tools over raw queries. MCP responses are already compressed. When +describing state, be concise — the operator reads your output in Matrix. + +## Skills + +Skills live in `/app/hermes/skills/`. Load a skill when its description +matches the task. The `homelab-ops` skill covers: +- Health checks, signal triage, pattern validation, and escalation flow. diff --git a/hermes/config.yaml b/hermes/config.yaml new file mode 100644 index 0000000..a1cb219 --- /dev/null +++ b/hermes/config.yaml @@ -0,0 +1,26 @@ +# Hermes container config — Gateway mode with MCP wiring (Phase 4) +# Deployed: mac-mini Docker, mesh-published :8092 +# No SSH keys in this container; all mutations route through /executions + +agent: + name: hermes + entity_slug: agent:hermes + role: gateway + +mcp: + endpoint: http://api:8090/mcp + transport: streamable_http + bearer_token_env: OIKOS_MCP_BEARER_TOKEN + +server: + listen: ":8092" + mesh_only: true + +model: + provider: openrouter + model: deepseek/deepseek-v4-pro + api_key_env: OPENROUTER_API_KEY + +session: + mode: smart_approve + skills_dir: /app/hermes/skills diff --git a/hermes/skills/homelab-ops/SKILL.md b/hermes/skills/homelab-ops/SKILL.md new file mode 100644 index 0000000..db64433 --- /dev/null +++ b/hermes/skills/homelab-ops/SKILL.md @@ -0,0 +1,45 @@ +# Homelab Operations Skill + +**Risk class:** Depends on action (see OIKOS.md policy) +**Required scope:** agent +**Verification:** `get_health_summary` after action + +## Overview + +Standard operating procedures for the Hermes agent managing the hubris +homelab. All mutations route through `request_execution` → Oikos policy +gating → actuator (SSH). + +## Procedures + +### Health check triage + +1. `get_health_summary` — check fleet health +2. For degraded/down entities, `get_entity` for detail +3. `get_signal_history` on the target to check for repeats +4. `get_blast_radius` to assess downstream impact +5. `get_trend` for metric context before deciding + +### Signal response + +- `reversible_low` with validated pattern → `request_execution` (auto-restart) +- `config_mutation` or `destructive` → escalate to operator +- Repeated flapping → escalate with flap count + +### Execution tracking + +1. `request_execution` returns a correlation_id +2. Poll `get_event_timeline` filtering by correlation_id +3. Once complete, `get_health_summary` to verify recovery +4. Record outcome via internal reasoning + +### Pattern learning + +- After 5 identical successful executions on the same (type, action), the + learning engine promotes the pattern to `validated` +- Check `get_patterns(status=validated)` to know what's trusted + +## Changelog + +### 2026-07-07 — initial Phase 4 skill +Baseline homelab operations skill for Hermes container. diff --git a/internal/config/config.go b/internal/config/config.go index 622b445..bd8dd2e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -53,6 +53,9 @@ type Config struct { // Approval HMAC secret (Phase 3) ApprovalHMACSecret string + + // Hermes agent entity ID (Phase 4) + HermesAgentID string } // Default returns a Config with compiled defaults. @@ -139,6 +142,9 @@ func FromEnv() Config { if v := os.Getenv("OIKOS_APPROVAL_HMAC_SECRET"); v != "" { c.ApprovalHMACSecret = v } + if v := os.Getenv("OIKOS_HERMES_AGENT_ID"); v != "" { + c.HermesAgentID = v + } return c } diff --git a/internal/httpapi/phase3.go b/internal/httpapi/phase3.go index 7d01a1a..1fac1d3 100644 --- a/internal/httpapi/phase3.go +++ b/internal/httpapi/phase3.go @@ -887,7 +887,7 @@ func (s *Server) ListSkills(ctx context.Context, req gen.ListSkillsRequestObject rows, err := s.pool.Query(ctx, ` SELECT s.entity_id, s.version, s.name, s.procedure, s.applies_type, s.action, s.pattern_ids, s.status, s.success_rate, - s.changed_by::text, s.change_reason, s.last_used_at, s.created_at + s.changed_by::text, s.change_reason, s.last_used_at FROM skills s WHERE ($1::text IS NULL OR s.status = $1) AND ($2::text IS NULL OR s.applies_type = $2) @@ -917,7 +917,7 @@ func (s *Server) ListSkills(ctx context.Context, req gen.ListSkillsRequestObject } seen[s.Id.String()] = true if err := json.Unmarshal(procBytes, &s.Procedure); err != nil { - slog.Warn("phase3: unmarshal skill procedure", "skill", s.Slug, "error", err) + slog.Warn("phase3: unmarshal skill procedure", "skill", s.Name, "error", err) } if len(patternIDs) > 0 { pids := make([]string, len(patternIDs)) @@ -977,7 +977,7 @@ func (s *Server) PatchSkill(ctx context.Context, req gen.PatchSkillRequestObject err = tx.QueryRow(ctx, ` SELECT entity_id, version, name, procedure, applies_type, action, pattern_ids, status, success_rate, changed_by::text, - change_reason, last_used_at, created_at + change_reason, last_used_at FROM skills WHERE entity_id = $1 ORDER BY version DESC LIMIT 1`, id). Scan(&skill.Id, &skill.Version, &skill.Name, &procBytes, &skill.AppliesType, &skill.Action, &patternIDs, &skill.Status, &skill.SuccessRate, @@ -1022,7 +1022,7 @@ func (s *Server) ListSkillVersions(ctx context.Context, req gen.ListSkillVersion rows, err := s.pool.Query(ctx, ` SELECT entity_id, version, name, procedure, applies_type, action, pattern_ids, status, success_rate, changed_by::text, - change_reason, last_used_at, created_at + change_reason, last_used_at FROM skills WHERE entity_id = $1 ORDER BY version DESC`, id) if err != nil { @@ -1739,7 +1739,82 @@ func (s *Server) GetEntityKnowledge(ctx context.Context, request gen.GetEntityKn // ─── Agent Activity (stub) ───────────────────────────────────────────── func (s *Server) QueryAgentActivity(ctx context.Context, request gen.QueryAgentActivityRequestObject) (gen.QueryAgentActivityResponseObject, error) { - return nil, errNotImplemented + limit := clampLimit(request.Params.Limit) + from := time.Now().Add(-24 * time.Hour) + if request.Params.From != nil { + from = *request.Params.From + } + to := time.Now() + if request.Params.To != nil { + to = *request.Params.To + } + + var agentID *string + if request.Params.AgentId != nil { + a := *request.Params.AgentId + agentID = &a + } + var activityType *string + if request.Params.ActivityType != nil { + a := string(*request.Params.ActivityType) + activityType = &a + } + var entityID *string + if request.Params.EntityId != nil { + a := *request.Params.EntityId + entityID = &a + } + var cursorID *int + if request.Params.Cursor != nil && *request.Params.Cursor != "" { + if id, err := parseIntOrZero(*request.Params.Cursor); err == nil && id > 0 { + cursorID = &id + } + } + + rows, err := s.pool.Query(ctx, ` + SELECT id, ts, agent_id::text, session_id, activity_type, tool_name, + entity_id::text, input_summary, output_summary, + duration_ms, token_count, success, correlation_id + FROM agent_activity + WHERE ts >= $1 AND ts <= $2 + AND ($3::text IS NULL OR agent_id::text = $3) + AND ($4::text IS NULL OR activity_type = $4) + AND ($5::text IS NULL OR entity_id::text = $5) + AND ($6::bigint IS NULL OR id < $6::bigint) + ORDER BY id DESC + LIMIT $7`, + from, to, agentID, activityType, entityID, cursorID, limit+1) + if err != nil { + return nil, err + } + defer rows.Close() + + items := []gen.AgentActivity{} + for rows.Next() { + var a gen.AgentActivity + if err := rows.Scan(&a.Id, &a.Ts, &a.AgentId, &a.SessionId, + &a.ActivityType, &a.ToolName, &a.EntityId, + &a.InputSummary, &a.OutputSummary, + &a.DurationMs, &a.TokenCount, &a.Success, + &a.CorrelationId); err != nil { + return nil, err + } + items = append(items, a) + } + if rows.Err() != nil { + return nil, rows.Err() + } + + var next *string + if len(items) > limit { + items = items[:limit] + lastID := fmt.Sprintf("%d", items[len(items)-1].Id) + next = &lastID + } + if items == nil { + items = []gen.AgentActivity{} + } + return gen.QueryAgentActivity200JSONResponse{Items: items, NextCursor: next}, nil } // ─── Helpers ─────────────────────────────────────────────────────────── @@ -1750,3 +1825,14 @@ func coalesceStr(s *string, def string) string { } return *s } + +func parseIntOrZero(s string) (int, error) { + var n int + for _, c := range s { + if c < '0' || c > '9' { + return 0, fmt.Errorf("invalid integer: %q", s) + } + n = n*10 + int(c-'0') + } + return n, nil +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index d1f5afb..cc9d885 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -26,6 +26,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" ) // contextKey for storing actor identity in request context. @@ -109,7 +110,13 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand r.With(combinedAuth(cfg)).Get("/api/v1/events/stream", s.serveSSE) // Mount MCP at /mcp (plan R3-10) - r.With(combinedAuth(cfg)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken)) + hermesAgentID := uuid.Nil + if cfg.HermesAgentID != "" { + if id, err := uuid.Parse(cfg.HermesAgentID); err == nil { + hermesAgentID = id + } + } + r.With(combinedAuth(cfg)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken, hermesAgentID)) return r } diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 6c0ae71..e7aa1bd 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -13,7 +13,6 @@ import ( "github.com/dtoro/oikos/internal/db" "github.com/google/jsonschema-go/jsonschema" "github.com/google/uuid" - "github.com/jackc/pgx/v5" "github.com/modelcontextprotocol/go-sdk/mcp" ) @@ -34,8 +33,9 @@ func objSchema(props ...prop) *jsonschema.Schema { } // NewHandler creates an http.Handler that serves the Oikos MCP server. -func NewHandler(pool *db.Pool, token string) http.Handler { - s := newServer(pool) +// agentID is the Hermes 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 { @@ -47,15 +47,19 @@ func NewHandler(pool *db.Pool, token string) http.Handler { return handler } -func newServer(pool *db.Pool) *mcp.Server { +// 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(), }) - // All tools use the untyped handler (s.AddTool) for simplicity. - // Arguments are accessed via req.Parameters.Arguments.(map[string]any). + register := func(tool *mcp.Tool, handler toolHandler) { + s.AddTool(tool, withActivityLogging(pool, agentID, tool.Name, handler)) + } - s.AddTool(&mcp.Tool{Name: "get_entity", Description: "Get an entity by slug or UUID", + 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) @@ -63,7 +67,7 @@ func newServer(pool *db.Pool) *mcp.Server { return queryEntity(ctx, pool, idOrSlug), nil }) - s.AddTool(&mcp.Tool{Name: "list_entities", Description: "List entities filtered by type, state, or search", + 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"}, @@ -82,7 +86,7 @@ func newServer(pool *db.Pool) *mcp.Server { nStr(args["type"]), nStr(args["state"]), nStr(args["q"]), limit), nil }) - s.AddTool(&mcp.Tool{Name: "get_relations", Description: "Get relationships for an entity", + 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) @@ -96,7 +100,7 @@ func newServer(pool *db.Pool) *mcp.Server { ORDER BY r.type`, slug), nil }) - s.AddTool(&mcp.Tool{Name: "get_blast_radius", Description: "Find entities affected if this entity goes down", + 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)"}), @@ -109,7 +113,7 @@ func newServer(pool *db.Pool) *mcp.Server { slug, depth), nil }) - s.AddTool(&mcp.Tool{Name: "get_health_summary", Description: "Current fleet health summary", + 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, ` @@ -118,7 +122,7 @@ func newServer(pool *db.Pool) *mcp.Server { ORDER BY e.slug`), nil }) - s.AddTool(&mcp.Tool{Name: "get_audit_trail", Description: "Query the audit log", + 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) @@ -129,19 +133,19 @@ func newServer(pool *db.Pool) *mcp.Server { ORDER BY ts DESC LIMIT 50`, nStr(args["entity_id"])), nil }) - s.AddTool(&mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation", + register(&mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation", InputSchema: objSchema(prop{"query", "string", "Search terms"}), }, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { args := argsMap(req) - q, _ := args["query"].(string) + q := nStr(args["query"]) return queryRows(ctx, pool, ` SELECT id, title, LEFT(content, 500) AS preview FROM knowledge_entities - WHERE title ILIKE '%'||$1||'%' OR content ILIKE '%'||$1||'%' + WHERE ($1::text IS NULL OR title ILIKE '%'||$1||'%' OR content ILIKE '%'||$1||'%') ORDER BY title LIMIT 20`, q), nil }) - s.AddTool(&mcp.Tool{Name: "query_metrics", Description: "Query time-series metrics", + 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) @@ -158,9 +162,214 @@ func newServer(pool *db.Pool) *mcp.Server { 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 (Hermes-only mutation path)", + InputSchema: objSchema( + prop{"target", "string", "Target entity slug"}, + prop{"action", "string", "Action to perform"}, + ), + }, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + args := argsMap(req) + targetSlug, _ := args["target"].(string) + action, _ := args["action"].(string) + if targetSlug == "" || action == "" { + return textResult("error: target and action 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 + } + + id, err := uuid.NewV7() + if err != nil { + return textResult(fmt.Sprintf("uuid error: %v", err)), nil + } + correlationID := uuid.New().String() + + _, err = pool.Exec(ctx, ` + INSERT INTO executions (entity_id, target_entity_id, action, risk_class, + status, correlation_id, agent_id) + VALUES ($1, $2, $3, 'unclassified', 'pending', $4, $5)`, + id, targetID, action, correlationID, agentID) + if err != nil { + return textResult(fmt.Sprintf("insert execution: %v", err)), nil + } + + return textResult(fmt.Sprintf("execution requested: id=%s target=%s action=%s correlation=%s", + id, targetSlug, action, correlationID)), 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.event_type, ev.severity, ev.actor, e.slug AS entity_slug, + ev.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 + }) + 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() + + _, logErr := pool.Exec(ctx, ` + INSERT INTO agent_activity + (agent_id, activity_type, tool_name, input_summary, output_summary, + duration_ms, success, correlation_id) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + agentID, "tool_call", toolName, inputSummary, outputSummary, + duration, success, correlationID) + if logErr != nil { + slog.Warn("mcp: log agent_activity", "error", logErr) + } + + return result, err + } +} + // ─── Helpers ────────────────────────────────────────────────────────── func argsMap(req *mcp.CallToolRequest) map[string]any { @@ -181,6 +390,12 @@ func getFloat(m map[string]any, key string, def float64) float64 { return v case int: return float64(v) + case json.Number: + f, err := v.Float64() + if err != nil { + return def + } + return f } return def } @@ -244,7 +459,4 @@ func queryRows(ctx context.Context, pool *db.Pool, query string, args ...any) *m data, _ := json.MarshalIndent(items, "", " ") return textResult(string(data)) -} - -var _ = pgx.ErrNoRows -var _ = time.Now \ No newline at end of file +} \ No newline at end of file diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go index 8130746..268df41 100644 --- a/internal/mcp/server_test.go +++ b/internal/mcp/server_test.go @@ -2,6 +2,8 @@ package mcp import ( "testing" + + "github.com/google/uuid" ) // TestNewServerRegistersTools verifies every tool registers with a valid @@ -16,7 +18,7 @@ func TestNewServerRegistersTools(t *testing.T) { }() // pool is only used inside tool handlers (invoked per-call), not at // registration time, so a nil pool is safe for this construction test. - s := newServer(nil) + s := newServer(nil, uuid.Nil) if s == nil { t.Fatal("newServer returned nil") } diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go index 1cf1a2d..dcab7ff 100644 --- a/internal/notifier/notifier.go +++ b/internal/notifier/notifier.go @@ -36,6 +36,11 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { } } +// RunnerForMain provides the run function for registration in main. +func RunnerForMain() func(context.Context, *db.Pool, config.Config) { + return Run +} + // processPendingApprovals checks for pending approvals and sends alerts. func processPendingApprovals(ctx context.Context, pool *db.Pool, cfg config.Config) { q := sqlcgen.New(pool) diff --git a/internal/scheduler/init.go b/internal/scheduler/init.go index bcd6ac5..0319a5c 100644 --- a/internal/scheduler/init.go +++ b/internal/scheduler/init.go @@ -2,28 +2,12 @@ package scheduler import ( "context" - "log/slog" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" - "github.com/dtoro/oikos/internal/db/sqlcgen" ) -func init() { - // Register with main package via package-level variable - // The cmd/oikos/main.go sets SchedulerRunner in its import - mainRunner = Run -} - -// mainRunner is assigned to main.SchedulerRunner by the cmd/oikos package. -// It's set during init so that when main runs, the scheduler Runner is available. -var mainRunner func(context.Context, *db.Pool, config.Config) - // RunnerForMain provides the run function for registration in main. func RunnerForMain() func(context.Context, *db.Pool, config.Config) { return Run } - -// ensure sqlcgen is used -var _ = sqlcgen.Queries{} -var _ = slog.Default \ No newline at end of file