phase 4: hermes agent — MCP tools, activity logging, 'all' role, wiring fixes
- 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.
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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:
|
||||
|
||||
50
hermes/SOUL.md
Normal file
50
hermes/SOUL.md
Normal file
@@ -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.
|
||||
26
hermes/config.yaml
Normal file
26
hermes/config.yaml
Normal file
@@ -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
|
||||
45
hermes/skills/homelab-ops/SKILL.md
Normal file
45
hermes/skills/homelab-ops/SKILL.md
Normal file
@@ -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.
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user