diff --git a/.gitignore b/.gitignore index b23dac9..b6e8b82 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,4 @@ __pycache__/ # health-probe cache. See oikos/scheduler.py. oikos/state.json -.worktrees/ \ No newline at end of file +.worktrees/bin/ diff --git a/bin/hermes b/bin/hermes index 596029a..86366b1 100755 Binary files a/bin/hermes and b/bin/hermes differ diff --git a/cmd/hermes/main.go b/cmd/hermes/main.go new file mode 100644 index 0000000..d78e8cf --- /dev/null +++ b/cmd/hermes/main.go @@ -0,0 +1,341 @@ +package main + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" +) + +func main() { + if len(os.Args) < 2 { + fmt.Fprintln(os.Stderr, "usage: hermes serve") + os.Exit(1) + } + + mcpURL := os.Getenv("HERMES_MCP_URL") + if mcpURL == "" { + mcpURL = "http://localhost:8090/mcp" + } + + agentSlug := os.Getenv("HERMES_AGENT_SLUG") + if agentSlug == "" { + agentSlug = "agent:hermes" + } + + switch os.Args[1] { + case "serve": + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) + defer cancel() + + client, err := newMCPClient(mcpURL) + if err != nil { + slog.Error("hermes: mcp connect", "url", mcpURL, "error", err) + os.Exit(1) + } + + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + w.Write([]byte("ok")) + }) + mux.HandleFunc("/query", func(w http.ResponseWriter, r *http.Request) { + handleQuery(w, r, client, agentSlug, mcpURL) + }) + + addr := os.Getenv("HERMES_LISTEN") + if addr == "" { + addr = ":8092" + } + + srv := &http.Server{Addr: addr, Handler: mux} + go func() { + slog.Info("hermes: gateway listening", "addr", addr, "mcp", mcpURL) + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + slog.Error("hermes: serve", "error", err) + } + }() + + <-ctx.Done() + slog.Info("hermes: shutting down") + srv.Shutdown(context.Background()) + client.close() + + default: + fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1]) + os.Exit(1) + } +} + +// handleQuery maps structured queries to MCP tool calls. +func handleQuery(w http.ResponseWriter, r *http.Request, client *mcpClient, agentSlug, mcpURL string) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + + var req struct { + Query string `json:"query"` + Tool string `json:"tool"` + Args map[string]any `json:"args"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "bad request: "+err.Error(), 400) + return + } + + start := time.Now() + var result any + var err error + + // Direct tool call (structured) + if req.Tool != "" { + result, err = client.callTool(req.Tool, req.Args) + } else { + // Natural-language-ish query routing + q := strings.ToLower(req.Query) + result, err = routeQuery(client, q, agentSlug) + } + + duration := time.Since(start).Milliseconds() + + if err != nil { + slog.Error("hermes: query failed", "query", req.Query, "error", err) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "error": err.Error(), + "elapsed_ms": duration, + "agent_slug": agentSlug, + }) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "result": result, + "elapsed_ms": duration, + "agent_slug": agentSlug, + "mcp_url": mcpURL, + }) +} + +// routeQuery maps natural-language-style queries to MCP tool calls. +func routeQuery(client *mcpClient, query, agentSlug string) (any, error) { + switch { + case strings.Contains(query, "depends on") || strings.Contains(query, "depend on"): + entity := extractEntity(query) + if entity == "" { + return nil, fmt.Errorf("no entity found in query: %s", query) + } + return client.callTool("get_blast_radius", map[string]any{ + "entity_id": entity, + }) + + case strings.Contains(query, "what is") || strings.Contains(query, "describe"): + entity := extractEntity(query) + if entity == "" { + entity = query + } + return client.callTool("get_entity", map[string]any{ + "slug_or_id": entity, + }) + + case strings.Contains(query, "health") || strings.Contains(query, "status"): + return client.callTool("get_health_summary", map[string]any{}) + + case strings.Contains(query, "restart") || strings.Contains(query, "reload"): + entity := extractEntity(query) + if entity == "" { + return nil, fmt.Errorf("no entity found in query: %s", query) + } + return client.callTool("request_execution", map[string]any{ + "target": entity, + "action": "restart", + }) + + case strings.Contains(query, "what can you do") || strings.Contains(query, "help"): + return client.callTool("tools/list", nil) + + default: + return client.callTool("get_health_summary", map[string]any{}) + } +} + +// extractEntity guesses an entity slug from a query. +func extractEntity(query string) string { + for _, slug := range []string{"authentik", "caddy", "vaultwarden", "gitea", "immich"} { + if strings.Contains(query, slug) { + return "service:" + slug + } + } + if strings.Contains(query, "mac-mini") { + return "host:mac-mini" + } + if strings.Contains(query, "hubris") { + return "host:hubris" + } + return "" +} + +// ─── MCP Streamable HTTP client ──────────────────────────────────────── + +type mcpClient struct { + baseURL string + sessionID string + http *http.Client + nextID int +} + +func newMCPClient(baseURL string) (*mcpClient, error) { + c := &mcpClient{ + baseURL: baseURL, + http: &http.Client{Timeout: 30 * time.Second}, + } + + // Initialize session + resp, err := c.doRequest("initialize", map[string]any{ + "protocolVersion": "2024-11-05", + "capabilities": map[string]any{}, + "clientInfo": map[string]any{"name": "hermes", "version": "2.0"}, + }) + if err != nil { + return nil, fmt.Errorf("initialize: %w", err) + } + if resp.sessionID == "" { + return nil, fmt.Errorf("no session ID in initialize response") + } + c.sessionID = resp.sessionID + + // Send initialized notification + c.doRequest("notifications/initialized", map[string]any{}) + + slog.Info("hermes: mcp connected", "session", c.sessionID[:16]+"...") + return c, nil +} + +type mcpJSONRPCResponse struct { + sessionID string + Result json.RawMessage `json:"result"` + Error json.RawMessage `json:"error"` +} + +func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPCResponse, error) { + c.nextID++ + body, _ := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": c.nextID, + }) + + req, err := http.NewRequest(http.MethodPost, c.baseURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + if c.sessionID != "" { + req.Header.Set("Mcp-Session-Id", c.sessionID) + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + result := &mcpJSONRPCResponse{} + result.sessionID = resp.Header.Get("Mcp-Session-Id") + + // Parse SSE stream: "event: message\ndata: \n\n" + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data: ") { + data := line[6:] + if err := json.Unmarshal([]byte(data), result); err != nil { + return nil, fmt.Errorf("parse response: %w", err) + } + } + } + + if result.Error != nil { + return nil, fmt.Errorf("rpc error: %s", string(result.Error)) + } + + if result.sessionID != "" { + c.sessionID = result.sessionID + } + + return result, nil +} + +func (c *mcpClient) callTool(name string, args map[string]any) (any, error) { + resp, err := c.doRequest("tools/call", map[string]any{ + "name": name, + "arguments": args, + }) + if err != nil { + return nil, err + } + + // Parse MCP content: { "content": [{ "type": "text", "text": "..." }] } + var toolResult struct { + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"content"` + } + if err := json.Unmarshal(resp.Result, &toolResult); err != nil { + return string(resp.Result), nil + } + + var texts []string + for _, c := range toolResult.Content { + if c.Type == "text" { + // Try to parse as JSON for structured display + var parsed any + if json.Unmarshal([]byte(c.Text), &parsed) == nil { + return parsed, nil + } + texts = append(texts, c.Text) + } + } + if len(texts) == 1 { + return texts[0], nil + } + return texts, nil +} + +func (c *mcpClient) listTools() ([]string, error) { + resp, err := c.doRequest("tools/list", map[string]any{}) + if err != nil { + return nil, err + } + var tr struct { + Tools []struct { + Name string `json:"name"` + Description string `json:"description"` + } `json:"tools"` + } + if err := json.Unmarshal(resp.Result, &tr); err != nil { + return nil, err + } + var names []string + for _, t := range tr.Tools { + names = append(names, t.Name) + } + return names, nil +} + +func (c *mcpClient) close() { + // MCP sessions are ephemeral; no explicit close needed +} diff --git a/compose/hermes/Dockerfile b/compose/hermes/Dockerfile new file mode 100644 index 0000000..84727f0 --- /dev/null +++ b/compose/hermes/Dockerfile @@ -0,0 +1,25 @@ +# Hermes agent container — standalone MCP client gateway (Phase 4) +FROM golang:1.26-alpine AS builder + +RUN apk add --no-cache git ca-certificates + +WORKDIR /build +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +RUN CGO_ENABLED=0 go build -o /hermes -tags timetzdata -ldflags="-s -w" ./cmd/hermes + +FROM gcr.io/distroless/static:nonroot + +COPY --from=builder /hermes /hermes +COPY hermes/ /app/hermes/ + +ENV HERMES_MCP_URL=http://api:8090/mcp +ENV HERMES_AGENT_SLUG=agent:hermes +ENV HERMES_LISTEN=:8092 + +EXPOSE 8092 + +ENTRYPOINT ["/hermes", "serve"] diff --git a/docker-compose.yml b/docker-compose.yml index 2c152b4..b734246 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -60,7 +60,7 @@ services: OIKOS_API_LISTEN: ":8090" OIKOS_ENV: dev OIKOS_DEBUG: "true" - OIKOS_HERMES_AGENT_ID: ${OIKOS_HERMES_AGENT_ID:-} + OIKOS_HERMES_AGENT_SLUG: ${OIKOS_HERMES_AGENT_SLUG:-agent:hermes} ports: - "8090:8090" command: ["api"] @@ -101,5 +101,22 @@ services: stop_signal: SIGTERM stop_grace_period: 30s + # Hermes agent gateway (Phase 4) — mesh-published :8092 + hermes: + build: + context: . + dockerfile: compose/hermes/Dockerfile + profiles: ["full"] + depends_on: + api: + condition: service_started + environment: + HERMES_MCP_URL: http://api:8090/mcp + HERMES_AGENT_SLUG: agent:hermes + ports: + - "8092:8092" + stop_signal: SIGTERM + stop_grace_period: 10s + volumes: pg-data: diff --git a/hermes/config.yaml b/hermes/config.yaml index a1cb219..2afe1af 100644 --- a/hermes/config.yaml +++ b/hermes/config.yaml @@ -1,26 +1,33 @@ -# 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 +# Hermes agent config — standalone MCP client gateway (Phase 4) + +mcp: + endpoint: ${HERMES_MCP_URL}?session_id=${HERMES_SESSION_ID} + transport: streamable_http + +server: + listen: ${HERMES_LISTEN} + mesh_only: true agent: name: hermes - entity_slug: agent:hermes - role: gateway + slug: ${HERMES_AGENT_SLUG} -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 +query_routing: + # Maps natural-language query patterns to MCP tools + - pattern: "depends on" + tool: get_blast_radius + entity_param: entity_id + - pattern: "restart" + tool: request_execution + action: restart + - pattern: "health" + tool: get_health_summary + - pattern: "what is" + tool: get_entity + entity_param: slug_or_id + - pattern: "recent events" + tool: get_event_timeline + - pattern: "signals" + tool: get_signal_history + - pattern: "patterns" + tool: get_patterns diff --git a/internal/config/config.go b/internal/config/config.go index bd8dd2e..0290d5b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -55,7 +55,8 @@ type Config struct { ApprovalHMACSecret string // Hermes agent entity ID (Phase 4) - HermesAgentID string + HermesAgentID string + HermesAgentSlug string } // Default returns a Config with compiled defaults. @@ -145,6 +146,9 @@ func FromEnv() Config { if v := os.Getenv("OIKOS_HERMES_AGENT_ID"); v != "" { c.HermesAgentID = v } + if v := os.Getenv("OIKOS_HERMES_AGENT_SLUG"); v != "" { + c.HermesAgentSlug = v + } return c } diff --git a/internal/httpapi/api_test.go b/internal/httpapi/api_test.go index 2d70186..81f9e60 100644 --- a/internal/httpapi/api_test.go +++ b/internal/httpapi/api_test.go @@ -109,6 +109,17 @@ func get(t *testing.T, h http.Handler, path string, headers map[string]string) ( return rec, body } +func postJSON(t *testing.T, h http.Handler, path string, payload string) (*httptest.ResponseRecorder, map[string]any) { + t.Helper() + req := httptest.NewRequest("POST", path, strings.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + var body map[string]any + json.Unmarshal(rec.Body.Bytes(), &body) + return rec, body +} + func devConfig() config.Config { c := config.Default() c.APIEnv = "dev" // no tokens → dev-open auth diff --git a/internal/httpapi/phase3_test.go b/internal/httpapi/phase3_test.go index 3e9aef4..b7207d4 100644 --- a/internal/httpapi/phase3_test.go +++ b/internal/httpapi/phase3_test.go @@ -6,6 +6,8 @@ package httpapi import ( "encoding/json" + "net/http/httptest" + "strings" "testing" ) @@ -108,4 +110,59 @@ func TestPhase3JSONRoundTrip(t *testing.T) { if back["kind"] != "down" { t.Errorf("kind = %v, want down", back["kind"]) } +} + +// ─── Phase 4 tests ──────────────────────────────────────────────────── + +func TestPhase4AgentActivity(t *testing.T) { + h := newTestHandler(t, devConfig()) + + rec, body := get(t, h, "/api/v1/agent-activity", nil) + if rec.Code != 200 { + t.Fatalf("agent-activity status %d: %v", rec.Code, body) + } +} + +func TestPhase4CreateExecution(t *testing.T) { + h := newTestHandler(t, devConfig()) + + body := `{"target":"service:authentik","action":"health-check"}` + rec, resp := postJSON(t, h, "/api/v1/executions", body) + if rec.Code != 201 { + t.Fatalf("create execution status %d: %v", rec.Code, resp) + } + if resp["action"] != "health-check" { + t.Errorf("action = %v, want health-check", resp["action"]) + } +} + +func TestPhase4ExecutionList(t *testing.T) { + h := newTestHandler(t, devConfig()) + + rec, body := get(t, h, "/api/v1/executions", nil) + if rec.Code != 200 { + t.Fatalf("list executions status %d: %v", rec.Code, body) + } + items, _ := body["items"].([]any) + t.Logf("executions: %d rows", len(items)) +} + +func TestPhase4MCPEndpointAlive(t *testing.T) { + h := newTestHandler(t, devConfig()) + + body := `{"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test-oikos","version":"1.0"}},"id":1}` + req := httptest.NewRequest("POST", "/mcp", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if rec.Code != 200 { + t.Fatalf("mcp initialize status %d: %s", rec.Code, rec.Body.String()) + } + sid := rec.Header().Get("Mcp-Session-Id") + if sid == "" { + t.Error("no Mcp-Session-Id header") + } + t.Logf("session: %s", sid) } \ No newline at end of file diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index cc9d885..c7400bf 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -116,6 +116,9 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand hermesAgentID = id } } + if hermesAgentID == uuid.Nil && cfg.HermesAgentSlug != "" { + _ = pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", cfg.HermesAgentSlug).Scan(&hermesAgentID) + } 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 e92b440..18492a0 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -243,12 +243,13 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { } correlationID := uuid.New().String() + execName := action + " on " + targetSlug + " (" + id.String()[:8] + ")" execSlug := "exec:" + targetSlug + ":" + id.String()[:8] _, err = pool.Exec(ctx, ` INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`, - id, execSlug, action+" on "+targetSlug) + id, execSlug, execName) if err != nil { return textResult(fmt.Sprintf("insert entity: %v", err)), nil } diff --git a/seeds/inventory.yaml b/seeds/inventory.yaml index 5f20214..267c20e 100644 --- a/seeds/inventory.yaml +++ b/seeds/inventory.yaml @@ -323,8 +323,8 @@ entities: - {slug: "idp:authentik", type: identity-provider, name: authentik, attributes: {issuer: "https://auth.hubris.network", auth_mode: both}} - {slug: "agent:hermes", type: agent, name: hermes, - state: planned, - attributes: {gateway_port: 8092, note: "Oikos Phase 4 — Docker gateway mode"}} + state: active, + attributes: {gateway_port: 8092, session_mode: smart_approve, note: "Phase 4 — Docker gateway mode"}} - {slug: "agent:oikos", type: agent, name: oikos, state: planned, attributes: {note: "the OS control loop itself (scheduler/actuator) as an actor"}}