[codex] Add k6 SSE stream soak test

agent/server.ts

@@ -28,9 +28,11 @@ import {
   metricsHandler,
   agentRunsTotal,
   agentToolCallsTotal,
+  agentLlmErrorTotal,
   agentLlmTokensTotal,
   agentLlmIterationTokens,
   agentLlmContextUsageRatio,
+  agentSseClients,
 } from "../shared/metrics.ts";
 import {
   comparePharmacyPrices,
@@ -395,6 +397,41 @@ let toolCallCapHitsTotal = 0;
 
 let agentPaused = false;
 
+// SSE client registry — one entry per open /agent/stream connection
+const sseClients = new Set<express.Response>();
+
+function serializeSSEData(event: string, data: unknown): string {
+  try {
+    return JSON.stringify(data) ?? "null";
+  } catch (err) {
+    logger.warn({ err, event }, "[sse] Failed to serialize payload");
+    return JSON.stringify({
+      error: "sse_payload_not_serializable",
+      event,
+    });
+  }
+}
+
+function writeSSEEvent(client: express.Response, event: string, data: unknown): void {
+  client.write(`event: ${event}\ndata: ${serializeSSEData(event, data)}\n\n`);
+}
+
+function refreshSSEClientGauge(): void {
+  agentSseClients.set(sseClients.size);
+}
+
+function broadcastSSE(event: string, data: unknown): void {
+  const payload = `event: ${event}\ndata: ${serializeSSEData(event, data)}\n\n`;
+  for (const client of sseClients) {
+    try {
+      client.write(payload);
+    } catch {
+      sseClients.delete(client);
+      refreshSSEClientGauge();
+    }
+  }
+}
+
 // In-memory cache for wallet balances (5s TTL)
 interface WalletCacheEntry {
   data: { usdc: string; xlm: string; address: string };
@@ -505,15 +542,53 @@ app.post("/agent/pause", (_req, res) => {
   agentPaused = true;
   logger.info("agent paused by caregiver");
   notify({ level: "warning", title: "Agent Paused", description: "CareGuard agent has been paused by the caregiver. No payments or actions will be processed until resumed." });
+  broadcastSSE("status", { paused: true });
   res.json({ paused: true });
 });
 app.post("/agent/resume", (_req, res) => {
   agentPaused = false;
   logger.info("agent resumed by caregiver");
   notify({ level: "info", title: "Agent Resumed", description: "CareGuard agent has been resumed and is now processing actions." });
+  broadcastSSE("status", { paused: false });
   res.json({ paused: false });
 });
 
+// SSE stream: pushes spending, transactions, and agent status on state changes.
+// Clients reconnect automatically via the EventSource API; heartbeats keep the
+// connection alive through proxies that close idle connections.
+app.get("/agent/stream", (req, res) => {
+  res.setHeader("Content-Type", "text/event-stream");
+  res.setHeader("Cache-Control", "no-cache");
+  res.setHeader("Connection", "keep-alive");
+  res.flushHeaders();
+
+  const recipientId = (req.query.recipient_id as string) || "rosa";
+  setCurrentRecipient(recipientId);
+
+  writeSSEEvent(res, "spending", getSpendingSummary());
+  writeSSEEvent(res, "status", { paused: agentPaused });
+  writeSSEEvent(res, "transactions", getSpendingTracker());
+
+  sseClients.add(res);
+  refreshSSEClientGauge();
+
+  const heartbeat = setInterval(() => {
+    try {
+      res.write(": heartbeat\n\n");
+    } catch {
+      sseClients.delete(res);
+      refreshSSEClientGauge();
+      clearInterval(heartbeat);
+    }
+  }, 30_000);
+
+  req.on("close", () => {
+    sseClients.delete(res);
+    refreshSSEClientGauge();
+    clearInterval(heartbeat);
+  });
+});
+
 app.post("/agent/run", async (req, res) => {
   const validation = validateTask(req.body?.task);
   if (!validation.ok) { res.status(400).json({ error: validation.error }); return; }
@@ -570,7 +645,7 @@ function validatePolicyPayload(body: any): { ok: true; policy: any } | { ok: fal
 
 app.post("/agent/policy", (req, res) => {
   const result = validatePolicyPayload(req.body);
-  if (!result.ok) return res.status(400).json({ error: "Invalid policy", details: result.errors });
+  if (result.ok === false) return res.status(400).json({ error: "Invalid policy", details: result.errors });
   const recipientId = (req.query.recipient_id as string) || "rosa";
   setCurrentRecipient(recipientId);
   setSpendingPolicy(result.policy);

docs/load-testing.md

@@ -37,6 +37,47 @@ To target a different server:
 BASE_URL=https://your-app.onrender.com k6 run load/agent-run.js
 ```
 
+## Running the SSE stream soak test
+
+`load/agent-stream.js` verifies that `/agent/stream` can hold many long-lived
+SSE connections while agent status broadcasts are delivered to every connected
+client. It uses the community SSE extension import `k6/x/sse`.
+
+```bash
+# From the careguard/ root, with the agent server running
+BASE_URL=http://localhost:3004 AGENT_API_KEY=dev-secret pnpm load:stream
+
+# Or directly
+BASE_URL=http://localhost:3004 AGENT_API_KEY=dev-secret k6 run load/agent-stream.js
+```
+
+Configurable environment variables:
+
+| Variable | Default | Description |
+|---|---:|---|
+| `STREAM_CONNECTIONS` | `50` | Number of concurrent `/agent/stream` clients. |
+| `STREAM_HOLD_SECONDS` | `30` | Minimum time each client should keep the SSE connection open. |
+| `STREAM_BROADCASTS` | `4` | Number of pause/resume broadcast requests sent during the soak. |
+| `BROADCAST_DELAY_SECONDS` | `5` | Delay before the broadcaster starts, giving clients time to connect. |
+| `MIN_EVENTS_PER_CLIENT` | `3 + STREAM_BROADCASTS` | Minimum events each client must see. The initial stream emits spending, status, and transactions. |
+| `EXPECTED_STATUS_EVENTS` | `1 + STREAM_BROADCASTS` | Initial status event plus broadcast status events expected per client. |
+| `METRICS_TOKEN` | unset | Bearer token for `/metrics` when metrics are protected. |
+| `MAX_RSS_DELTA_BYTES` | `67108864` | Maximum allowed RSS growth after all clients disconnect. |
+
+The SSE soak fails when:
+
+- any SSE connection errors,
+- fewer than 95% of clients receive the expected initial and broadcast events,
+- fewer than 95% of clients disconnect cleanly,
+- a pause/resume broadcast fails or exceeds the latency threshold,
+- `agent_sse_clients` does not return to its starting value,
+- process RSS grows beyond `MAX_RSS_DELTA_BYTES` after disconnect.
+
+The agent exposes the `agent_sse_clients` Prometheus gauge so the test can
+confirm long-lived connections are removed from the server registry after the
+soak. SSE payloads are also serialized defensively, so an unexpected circular
+payload is reported as an SSE error object instead of crashing the broadcaster.
+
 ## What it checks
 
 | Check | Threshold |