---
phase: 20-real-time-subscriptions
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - src/trpc/router.ts
  - src/trpc/context.ts
  - src/server/trpc-adapter.ts
  - src/trpc/subscriptions.ts
autonomous: true
---

<objective>
Add tRPC subscription procedures and enable SSE streaming on the backend server.

Purpose: The existing tRPC HTTP adapter buffers entire responses before sending (`fetchResponse.text()` + `res.end()`), which kills streaming. This plan fixes the adapter to support SSE and adds subscription procedures that bridge EventBus domain events into tRPC async generator subscriptions. Uses tRPC v11's native SSE support via `fetchRequestHandler` — no WebSocket dependency needed.

Output: Backend ready to stream real-time events via SSE to any tRPC client using `httpSubscriptionLink`.
</objective>

<execution_context>
@~/.claude/get-shit-done/workflows/execute-plan.md
@~/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md

# Key source files:
@src/trpc/router.ts
@src/trpc/context.ts
@src/server/trpc-adapter.ts
@src/events/types.ts
@src/events/bus.ts
</context>

<tasks>

<task type="auto">
  <name>Task 1: Fix HTTP adapter for SSE streaming and add subscription procedures</name>
  <files>src/server/trpc-adapter.ts, src/trpc/subscriptions.ts, src/trpc/router.ts, src/trpc/context.ts</files>
  <action>
**1. Fix the tRPC HTTP adapter to support streaming responses (`src/server/trpc-adapter.ts`):**

The current adapter does `const responseBody = await fetchResponse.text(); res.end(responseBody)` which buffers the entire response and breaks SSE. Replace this with stream piping:

```
// Instead of: const responseBody = await fetchResponse.text(); res.end(responseBody);
// Do: Check if response body is a ReadableStream and pipe it
if (fetchResponse.body) {
  // For SSE, the response body is a ReadableStream that stays open
  const reader = fetchResponse.body.getReader();
  const pump = async () => {
    while (true) {
      const { done, value } = await reader.read();
      if (done) { res.end(); return; }
      res.write(value);
    }
  };
  pump().catch(() => res.end());
} else {
  const responseBody = await fetchResponse.text();
  res.end(responseBody);
}
```

Also set response headers before streaming (move the `fetchResponse.headers.forEach` BEFORE the body handling).

**2. Create subscription procedures module (`src/trpc/subscriptions.ts`):**

Create a new file with helper functions for bridging EventBus events to tRPC async generators. The pattern:

```typescript
import { tracked } from '@trpc/server';
import type { EventBus, DomainEventMap } from '../events/types.js';

// Helper: Create an async iterable from EventBus events
export function eventBusIterable<T extends DomainEventMap>(
  eventBus: EventBus,
  eventTypes: T['type'][],
  signal: AbortSignal,
): AsyncGenerator<{ id: string; type: string; payload: unknown; timestamp: Date }> {
  // Use a queue + promise pattern to bridge push (EventBus) to pull (async generator)
}
```

Use a simple queue-based approach: EventBus handlers push events into a queue array, the async generator yields from the queue, and waits on a promise when the queue is empty. Clean up handlers on AbortSignal abort.

Generate event IDs using a counter (e.g., `${Date.now()}-${counter++}`). Use `tracked(id, data)` for each yielded event to enable client-side reconnection.

**3. Add subscription procedures to the router (`src/trpc/router.ts`):**

Add these subscription procedures to `appRouter`:

- `onEvent` — Subscribe to ALL domain events. Input: optional `lastEventId` for reconnection. Yields `{ id, type, payload, timestamp }` for every event emitted on the EventBus. This is the general-purpose firehose for the frontend.

- `onAgentUpdate` — Subscribe to agent-specific events only (`agent:spawned`, `agent:stopped`, `agent:crashed`, `agent:resumed`, `agent:waiting`). Input: optional `lastEventId`. More targeted than `onEvent` for the inbox page.

- `onTaskUpdate` — Subscribe to task and phase events (`task:queued`, `task:dispatched`, `task:completed`, `task:blocked`, `phase:queued`, `phase:started`, `phase:completed`, `phase:blocked`). Input: optional `lastEventId`. For the initiative detail page.

Each subscription procedure follows this pattern:
```typescript
onEvent: publicProcedure
  .input(z.object({ lastEventId: z.string().nullish() }).optional())
  .subscription(async function* (opts) {
    const eventBus = opts.ctx.eventBus;
    // Use eventBusIterable helper
    yield* eventBusIterable(eventBus, ALL_EVENT_TYPES, opts.signal);
  }),
```

Define event type arrays as constants:
- `AGENT_EVENT_TYPES = ['agent:spawned', 'agent:stopped', 'agent:crashed', 'agent:resumed', 'agent:waiting']`
- `TASK_EVENT_TYPES = ['task:queued', 'task:dispatched', 'task:completed', 'task:blocked', 'phase:queued', 'phase:started', 'phase:completed', 'phase:blocked']`
- `ALL_EVENT_TYPES` = union of all DomainEventType values

Do NOT add output validation with `zAsyncIterable` — keep it simple, trust the EventBus types. The subscription output shape is `{ id: string, type: string, payload: unknown, timestamp: string }`.

**Important:** The `tracked` import is from `@trpc/server`. No new npm packages needed — tRPC v11.9.0 already includes async generator subscription support and SSE handling in `fetchRequestHandler`.
  </action>
  <verify>
1. `npm run build` succeeds (TypeScript compiles)
2. `npm run test` passes (existing tests not broken)
3. Start server with `npx tsx src/bin/cw.ts --server`, then `curl -N http://127.0.0.1:3847/trpc/onEvent.subscribe` returns SSE-formatted response with `content-type: text/event-stream` header (connection stays open)
  </verify>
  <done>
- tRPC HTTP adapter streams responses instead of buffering
- Three subscription procedures (onEvent, onAgentUpdate, onTaskUpdate) available on the router
- EventBus events bridge into SSE streams via async generators
- Existing HTTP queries/mutations still work (no regression)
  </done>
</task>

<task type="auto">
  <name>Task 2: Update shared types and verify SSE integration</name>
  <files>packages/shared/src/types.ts, packages/shared/src/index.ts</files>
  <action>
**1. Export subscription event type from shared package:**

Add a `SubscriptionEvent` type to `packages/shared/src/types.ts`:

```typescript
export interface SubscriptionEvent {
  id: string;
  type: string;
  payload: unknown;
  timestamp: string;
}
```

Export it from `packages/shared/src/index.ts`.

This type is what the frontend will use when consuming subscription data in `onData` callbacks.

**2. Verify the full SSE flow works end-to-end:**

Write a quick integration check (not a persistent test file — just verify manually or via curl):
- Start the server
- Open an SSE connection to `onEvent.subscribe`
- In another terminal, trigger a health check or any mutation
- Verify the SSE stream receives the event

If the `fetchRequestHandler` does not natively produce SSE for subscription procedures over HTTP, check if the `@trpc/server` package needs the response to include specific headers. The tRPC fetch handler should set `Content-Type: text/event-stream` automatically for subscription responses.

**3. Verify Vite dev proxy passes SSE correctly:**

The Vite proxy at `/trpc` must not buffer SSE responses. Check that the existing Vite proxy config in `packages/web/vite.config.ts` works for SSE. If it buffers, add `configure: (proxy) => { proxy.on('proxyRes', (proxyRes) => { proxyRes.headers['cache-control'] = 'no-cache'; }); }` or set `ws: false` explicitly (SSE is NOT WebSocket — it's plain HTTP with `Transfer-Encoding: chunked`).

Verify by running `npm run dev` in `packages/web/` and curling `http://localhost:5173/trpc/onEvent.subscribe` — should stream SSE events through the Vite proxy.
  </action>
  <verify>
1. `npm run build` succeeds across all packages (root + web + shared)
2. `SubscriptionEvent` type is importable from `@codewalk-district/shared`
3. SSE stream works through both direct backend (port 3847) and Vite proxy (port 5173)
  </verify>
  <done>
- SubscriptionEvent type exported from shared package
- SSE streaming verified end-to-end through both direct and proxied connections
- Vite dev proxy confirmed to pass SSE without buffering
  </done>
</task>

</tasks>

<verification>
Before declaring plan complete:
- [ ] `npm run build` succeeds without errors
- [ ] `npm run test` passes all existing tests
- [ ] SSE subscription endpoint responds with `text/event-stream` content type
- [ ] EventBus events appear in SSE stream when triggered
- [ ] Existing tRPC queries/mutations still work (no regression from adapter changes)
</verification>

<success_criteria>

- All tasks completed
- All verification checks pass
- No errors or warnings introduced
- tRPC router exposes 3 subscription procedures (onEvent, onAgentUpdate, onTaskUpdate)
- HTTP adapter supports streaming responses for SSE
- Shared SubscriptionEvent type available for frontend
</success_criteria>

<output>
After completion, create `.planning/phases/20-real-time-subscriptions/20-01-SUMMARY.md`
</output>
