From 956646a5c1aa76ab26d5f9b178bf43bb37c3a7d2 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sat, 9 May 2026 22:28:00 +0100 Subject: [PATCH] =?UTF-8?q?Add=20cmd/agent-probe=20=E2=80=94=20worked=20ex?= =?UTF-8?q?ample=20consuming=20the=20agent=20substrate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The substrate's read and write surfaces are end-to-end-tested internally; this slice answers the harder question — "is the substrate actually usable from the outside?" — by writing the smallest standalone program that consumes it. agent-probe walks the discovery → triage → depth → push flow against a running Pulse instance using only the Go standard library, so it doubles as a reference implementation for anyone building MCP servers, Claude Code integrations, or custom agents on top of Pulse. It resolves every path from the manifest rather than hardcoding them — if discovery moves a path, the probe follows automatically — and branches on the stable error envelope's "error" code field, never on human-readable messages. The focus rule (severity-lex-ordered) is intentionally simple so a reader can predict what the probe will pick; real agents will have richer policies. This is documentation as code: the program is short enough to read top-to-bottom and reads like the agent's own narration of what it's doing. The unit test pins the focus rule's lex ordering so a refactor that swaps it for a weighted score (which allowed many warnings to outrank one critical) cannot regress silently. --- cmd/agent-probe/main.go | 383 ++++++++++++++++++ cmd/agent-probe/main_test.go | 83 ++++ .../v6/internal/subsystems/api-contracts.md | 10 + 3 files changed, 476 insertions(+) create mode 100644 cmd/agent-probe/main.go create mode 100644 cmd/agent-probe/main_test.go diff --git a/cmd/agent-probe/main.go b/cmd/agent-probe/main.go new file mode 100644 index 000000000..93c8512c7 --- /dev/null +++ b/cmd/agent-probe/main.go @@ -0,0 +1,383 @@ +// Command agent-probe is a worked example of an external agent +// consuming Pulse's agent-paradigm read substrate. It is not a +// production tool — it exists as the smallest legible reference +// implementation so anyone building MCP servers, Claude Code +// integrations, or custom agents on top of Pulse can see the +// discovery → triage → depth → push flow as one short program. +// +// What it does, in order: +// +// 1. Fetches /api/agent/capabilities (unauthenticated) and prints +// the declared capabilities. This is "discovery" — the agent +// learns what's available from the wire, not from documentation. +// +// 2. Resolves the get_fleet_context capability from the manifest +// and calls it (authenticated). This is "triage" — one read for +// "where do I focus?". +// +// 3. Picks a focus resource: critical findings first, then warning, +// then first in the list. Calls get_resource_context for it. +// This is "depth" — the situated picture for the chosen target. +// +// 4. Subscribes to /api/agent/events (SSE). Prints the next event +// it sees and exits. This is "push" — proof the doorbell wires +// up correctly. A short timeout guards against an idle stream. +// +// Hard constraints honored on purpose: +// +// - Standard library only. No internal/ai imports, no Pulse types +// reused. A real external agent has no privileged access. +// - Branches on stable error codes from the manifest, never on +// human-readable error messages. +// - Resolves paths from the manifest rather than hardcoding them. +// If discovery moves a path, this probe follows automatically. +// +// Run it against a local instance: +// +// go run ./cmd/agent-probe \ +// --base-url http://localhost:7655 \ +// --token "" \ +// --event-timeout 5s +// +// The token needs the monitoring:read scope. Discovery itself is +// public, but the rest of the substrate is gated. +package main + +import ( + "bufio" + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" +) + +// agentCapability mirrors the api package's AgentCapability wire +// shape — defined inline so this program depends on nothing in the +// pulse module. If the manifest's shape evolves, this struct +// follows; the JSON tags are the contract. +type agentCapability struct { + Name string `json:"name"` + Description string `json:"description"` + Category string `json:"category"` + Method string `json:"method"` + Path string `json:"path"` + Scope string `json:"scope"` + ResponseShape string `json:"responseShape,omitempty"` + ErrorCodes []string `json:"errorCodes,omitempty"` + RequestBodyShape string `json:"requestBodyShape,omitempty"` +} + +type agentCapabilitiesManifest struct { + Version string `json:"version"` + Capabilities []agentCapability `json:"capabilities"` +} + +// fleetResource is the per-resource thin rollup the fleet endpoint +// returns. We only depend on the fields the probe actually needs to +// pick a focus and print one-line summaries. +type fleetResource struct { + CanonicalID string `json:"canonicalId"` + ResourceType string `json:"resourceType"` + ResourceName string `json:"resourceName"` + IntentionallyOffline bool `json:"intentionallyOffline"` + NeverAutoRemediate bool `json:"neverAutoRemediate"` + MaintenanceWindowActive bool `json:"maintenanceWindowActive"` + Findings struct { + Total int `json:"total"` + Critical int `json:"critical"` + Warning int `json:"warning"` + Info int `json:"info"` + } `json:"findings"` + PendingApprovalCount int `json:"pendingApprovalCount"` +} + +type fleetContext struct { + Resources []fleetResource `json:"resources"` + GeneratedAt time.Time `json:"generatedAt"` +} + +// errorEnvelope is the shared shape every agent-surface endpoint +// uses on failure. The "error" field is a stable snake_case code +// (e.g. resource_not_found, operator_state_not_set); "message" is +// human text agents may surface but must not branch on. +type errorEnvelope struct { + Error string `json:"error"` + Message string `json:"message"` +} + +func main() { + baseURL := flag.String("base-url", "http://localhost:7655", "Pulse base URL") + token := flag.String("token", "", "API token with monitoring:read scope (required for triage and depth steps)") + eventTimeout := flag.Duration("event-timeout", 5*time.Second, "How long to wait for the first SSE event before giving up") + flag.Parse() + + if strings.TrimSpace(*token) == "" { + fmt.Fprintln(os.Stderr, "agent-probe: --token is required for triage/depth/push steps; discovery alone works without it") + os.Exit(2) + } + + client := &http.Client{Timeout: 15 * time.Second} + + // --- 1. Discovery. Public, no token needed. --- + manifest, err := fetchManifest(client, *baseURL) + if err != nil { + exitf("discovery failed: %v", err) + } + fmt.Printf("discovered manifest %s with %d capabilities\n", manifest.Version, len(manifest.Capabilities)) + for _, cap := range manifest.Capabilities { + fmt.Printf(" %-22s %s %s (%s)\n", cap.Name, cap.Method, cap.Path, cap.Scope) + } + + byName := map[string]agentCapability{} + for _, c := range manifest.Capabilities { + byName[c.Name] = c + } + + fleetCap, ok := byName["get_fleet_context"] + if !ok { + exitf("manifest missing required capability get_fleet_context") + } + contextCap, ok := byName["get_resource_context"] + if !ok { + exitf("manifest missing required capability get_resource_context") + } + streamCap, ok := byName["subscribe_events"] + if !ok { + exitf("manifest missing required capability subscribe_events") + } + + // --- 2. Triage. Authenticated. --- + fleet, err := fetchFleet(client, *baseURL, fleetCap.Path, *token) + if err != nil { + exitf("triage failed: %v", err) + } + fmt.Printf("\nfleet sweep — %d resources at %s\n", len(fleet.Resources), fleet.GeneratedAt.Format(time.RFC3339)) + for _, r := range fleet.Resources { + flags := []string{} + if r.IntentionallyOffline { + flags = append(flags, "offline") + } + if r.NeverAutoRemediate { + flags = append(flags, "locked") + } + if r.MaintenanceWindowActive { + flags = append(flags, "maintenance") + } + flagStr := "" + if len(flags) > 0 { + flagStr = " [" + strings.Join(flags, ",") + "]" + } + fmt.Printf(" %-30s %-12s findings: %d (c=%d w=%d i=%d) pending: %d%s\n", + r.CanonicalID, r.ResourceType, + r.Findings.Total, r.Findings.Critical, r.Findings.Warning, r.Findings.Info, + r.PendingApprovalCount, flagStr) + } + + focus := pickFocus(fleet.Resources) + if focus == nil { + fmt.Println("\nno resources visible to this token; skipping depth step") + } else { + // --- 3. Depth. --- + depthPath := strings.Replace(contextCap.Path, "{resourceId}", focus.CanonicalID, 1) + body, err := fetchAuthenticatedRaw(client, *baseURL+depthPath, *token) + if err != nil { + exitf("depth failed for %s: %v", focus.CanonicalID, err) + } + fmt.Printf("\nresource-context for %s:\n", focus.CanonicalID) + // Pretty-print the body so a reader sees the substrate's + // shape without us redefining every type. Real agents would + // decode against typed structs. + var pretty map[string]any + if err := json.Unmarshal(body, &pretty); err == nil { + out, _ := json.MarshalIndent(pretty, " ", " ") + fmt.Println(" " + string(out)) + } else { + fmt.Println(string(body)) + } + } + + // --- 4. Push. --- + fmt.Printf("\nsubscribing to %s (waiting up to %s for the first event)\n", + streamCap.Path, *eventTimeout) + ctx, cancel := context.WithTimeout(context.Background(), *eventTimeout) + defer cancel() + if err := readOneSSEEvent(ctx, *baseURL+streamCap.Path, *token); err != nil { + // Timeout is the boring case (no events fired); not a hard + // failure for a probe. + if strings.Contains(err.Error(), "deadline exceeded") { + fmt.Println(" (no event in window — stream is healthy if the connect succeeded; idle is normal)") + } else { + exitf("push failed: %v", err) + } + } + fmt.Println("\nagent-probe done — discovery, triage, depth, and push all walked the substrate cleanly.") +} + +func fetchManifest(client *http.Client, baseURL string) (*agentCapabilitiesManifest, error) { + resp, err := client.Get(baseURL + "/api/agent/capabilities") + if err != nil { + return nil, fmt.Errorf("GET capabilities: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("GET capabilities: status %d", resp.StatusCode) + } + var manifest agentCapabilitiesManifest + if err := json.NewDecoder(resp.Body).Decode(&manifest); err != nil { + return nil, fmt.Errorf("decode manifest: %w", err) + } + return &manifest, nil +} + +func fetchFleet(client *http.Client, baseURL, path, token string) (*fleetContext, error) { + body, err := fetchAuthenticatedRaw(client, baseURL+path, token) + if err != nil { + return nil, err + } + var fleet fleetContext + if err := json.Unmarshal(body, &fleet); err != nil { + return nil, fmt.Errorf("decode fleet: %w", err) + } + return &fleet, nil +} + +// fetchAuthenticatedRaw is the shared GET helper for any +// authenticated agent-surface endpoint. Branches on the stable +// error envelope when a non-2xx comes back so the caller sees the +// agent-stable code, not just an HTTP status. +func fetchAuthenticatedRaw(client *http.Client, fullURL, token string) ([]byte, error) { + parsed, err := url.Parse(fullURL) + if err != nil { + return nil, fmt.Errorf("parse %q: %w", fullURL, err) + } + req, err := http.NewRequest(http.MethodGet, parsed.String(), nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + req.Header.Set("X-API-Token", token) + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("GET %s: %w", parsed.Path, err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read body: %w", err) + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return body, nil + } + // Non-2xx: try to surface the stable error code from the + // envelope. Unauthenticated rejection comes through as plain + // text from the auth middleware (not the envelope), so fall back + // to the body verbatim if the JSON decode fails. + var env errorEnvelope + if err := json.Unmarshal(body, &env); err == nil && env.Error != "" { + return nil, fmt.Errorf("GET %s: %d %s (%s)", parsed.Path, resp.StatusCode, env.Error, env.Message) + } + return nil, fmt.Errorf("GET %s: %d %s", parsed.Path, resp.StatusCode, strings.TrimSpace(string(body))) +} + +// pickFocus chooses the most "interesting" resource from a fleet +// sweep. The triage rule is lexicographic — severity dominates +// count: any resource with a critical finding outranks every +// resource with only warnings, regardless of warning count; same +// for warning over info, and findings over pending approvals. The +// rule is intentionally simple so a reader can predict what the +// probe will pick. Real agents will have richer policies; this is +// legible default behavior for a probe. +func pickFocus(resources []fleetResource) *fleetResource { + if len(resources) == 0 { + return nil + } + best := 0 + for i := 1; i < len(resources); i++ { + if focusLess(resources[best], resources[i]) { + best = i + } + } + return &resources[best] +} + +// focusLess returns true if `a` is less interesting than `b` under +// the probe's lex ordering. Comparison cascades: critical count, +// then warning count, then info count, then pending approvals. +func focusLess(a, b fleetResource) bool { + if a.Findings.Critical != b.Findings.Critical { + return a.Findings.Critical < b.Findings.Critical + } + if a.Findings.Warning != b.Findings.Warning { + return a.Findings.Warning < b.Findings.Warning + } + if a.Findings.Info != b.Findings.Info { + return a.Findings.Info < b.Findings.Info + } + return a.PendingApprovalCount < b.PendingApprovalCount +} + +// readOneSSEEvent opens the SSE stream, reads up to the first +// non-keepalive event payload, prints it, and returns. SSE is line- +// based with empty lines as record separators; this implementation +// is deliberately minimal — a few hundred bytes of stdlib code is +// enough to consume the substrate's push channel. +func readOneSSEEvent(ctx context.Context, fullURL, token string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, nil) + if err != nil { + return err + } + req.Header.Set("X-API-Token", token) + req.Header.Set("Accept", "text/event-stream") + // No timeout on this client — context cancels the read. + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("subscribe: status %d", resp.StatusCode) + } + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 64*1024), 1<<20) + + var event, data string + skippedConnected := false + for scanner.Scan() { + line := scanner.Text() + switch { + case strings.HasPrefix(line, "event: "): + event = strings.TrimPrefix(line, "event: ") + case strings.HasPrefix(line, "data: "): + data = strings.TrimPrefix(line, "data: ") + case line == "": + // End of event record. The first event the server + // always emits is "stream.connected" — skip it once so + // the probe surfaces the first *real* event (or a + // heartbeat). + if event == "stream.connected" && !skippedConnected { + skippedConnected = true + event, data = "", "" + continue + } + if event != "" || data != "" { + fmt.Printf(" event: %s\n data: %s\n", event, data) + return nil + } + } + } + if err := scanner.Err(); err != nil { + return err + } + return ctx.Err() +} + +func exitf(format string, args ...any) { + fmt.Fprintf(os.Stderr, "agent-probe: "+format+"\n", args...) + os.Exit(1) +} diff --git a/cmd/agent-probe/main_test.go b/cmd/agent-probe/main_test.go new file mode 100644 index 000000000..909081fe1 --- /dev/null +++ b/cmd/agent-probe/main_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "testing" +) + +// TestPickFocus_PrefersCriticalThenWarningThenInfo pins the focus +// rule the probe uses for "where do I look first?". The unit test +// is here rather than against the substrate because the rule is a +// probe-side triage convention, not a contract Pulse owns; agents +// building on the substrate can implement their own ordering. +func TestPickFocus_PrefersCriticalThenWarningThenInfo(t *testing.T) { + mk := func(id string, c, w, i, p int) fleetResource { + r := fleetResource{CanonicalID: id, PendingApprovalCount: p} + r.Findings.Critical = c + r.Findings.Warning = w + r.Findings.Info = i + r.Findings.Total = c + w + i + return r + } + + cases := []struct { + name string + resources []fleetResource + want string + }{ + { + name: "single critical beats many warnings", + resources: []fleetResource{ + mk("vm:noisy", 0, 50, 0, 0), + mk("vm:critical", 1, 0, 0, 0), + mk("vm:quiet", 0, 0, 0, 0), + }, + want: "vm:critical", + }, + { + name: "tie on severity broken by count", + resources: []fleetResource{ + mk("vm:warm", 0, 1, 0, 0), + mk("vm:warmer", 0, 3, 0, 0), + }, + want: "vm:warmer", + }, + { + name: "no findings — pending approvals as tiebreaker", + resources: []fleetResource{ + mk("vm:idle", 0, 0, 0, 0), + mk("vm:waiting", 0, 0, 0, 2), + }, + want: "vm:waiting", + }, + { + name: "no findings or approvals — first wins so depth step still runs", + resources: []fleetResource{ + mk("vm:first", 0, 0, 0, 0), + mk("vm:second", 0, 0, 0, 0), + }, + want: "vm:first", + }, + { + name: "empty fleet returns nil", + resources: nil, + want: "", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := pickFocus(tc.resources) + if tc.want == "" { + if got != nil { + t.Fatalf("expected nil for empty fleet; got %+v", got) + } + return + } + if got == nil { + t.Fatalf("expected %q; got nil", tc.want) + } + if got.CanonicalID != tc.want { + t.Errorf("focus = %q; want %q", got.CanonicalID, tc.want) + } + }) + } +} diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 1222dcc44..d6001bbce 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -1463,6 +1463,16 @@ body-supplied id (preventing scope-confusion writes). Together the two tests are the substantive proof for the agent surface — read, write, push — as one substrate. +The companion worked example lives at `cmd/agent-probe/main.go` — +a small standalone Go program that walks the same discovery → +triage → depth → push flow against a running Pulse instance. +The probe depends only on the standard library so it doubles as +a reference implementation: anyone building MCP servers, Claude +Code integrations, or custom agents on top of Pulse can read it +top-to-bottom to see how the substrate fits together. The probe +resolves paths from the manifest rather than hardcoding them, so +discovery moves automatically follow. + `/api/agent/resource-context/{id}` is the agent-consumable bundled context endpoint. One read returns the full situated picture of a resource — identity, operator-set state (with server-computed