Add cmd/agent-probe — worked example consuming the agent substrate

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

@@ -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 "<api-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)
+}

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)
+			}
+		})
+	}
+}

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