# Ticket 1.2 — SSE Event Consumer

**Phase**: 1 — OpenCode Bridge (Core)
**Priority**: Critical (blocking)
**Depends on**: 0.2

## Description

Connect to OpenCode's Server-Sent Events (SSE) endpoint (`GET /event/`) and consume the real-time event stream. This is the primary mechanism for receiving agent responses, tool executions, permission requests, and status updates from OpenCode.

OpenCode also exposes a `GET /global/event` endpoint for instance-level events (not scoped to a session). Both need to be consumed.

## Acceptance Criteria

### AC1: Connects to instance-scoped SSE stream

```gherkin
Given the OpenCode server is running on port 4096
When the SSE consumer connects to GET /event
Then it establishes an EventSource connection
And it receives a { type: "server.connected", properties: {} } event immediately
And it begins receiving events as they occur
And each event is JSON with the shape { type: string, properties: object }
```

### AC2: Connects to global SSE stream

```gherkin
Given the OpenCode server is running on port 4096
When the SSE consumer connects to GET /global/event
Then it establishes a separate EventSource connection
And it receives instance-level events wrapped as { directory: string, payload: { type, properties } }
And it receives heartbeat events every 10 seconds ({ type: "server.heartbeat", properties: {} })
```

### AC3: Reconnects on connection drop

```gherkin
Given an active SSE connection
When the connection drops unexpectedly
Then the consumer detects the disconnection within 5 seconds
And it attempts to reconnect with exponential backoff (1s, 2s, 4s, 8s, max 30s)
And it logs each reconnection attempt
And it resumes receiving events after successful reconnection
```

### AC4: Filters events by session ID

```gherkin
Given events are streaming for multiple sessions
When the consumer receives an event with a session ID
Then it routes the event only to the project/session that owns it
And it drops events for sessions not registered with the relay
```

### AC5: Handles malformed events gracefully

```gherkin
Given an active SSE connection
When a malformed (non-JSON) event is received
Then the consumer logs a warning with the raw event data
And it continues processing subsequent events
And it does NOT crash or close the connection
```

### AC6: Emits typed events to internal listeners

```gherkin
Given the SSE consumer is connected and receiving events
When an event is received
Then it emits a typed internal event based on the event's type field
And listeners registered for that event type are called
And for events containing sessionID in properties, the session ID is available for routing

The key event types to handle are:
  - message.part.updated — Part created or fully updated (contains full Part object)
  - message.part.delta — Incremental text/reasoning content (sessionID, messageID, partID, field, delta)
  - message.part.removed — Part deleted
  - message.updated — Message metadata update (cost, tokens, status)
  - message.removed — Message deleted
  - session.status — Session state change (idle/busy/retry)
  - permission.asked — Permission request
  - permission.replied — Permission resolved
  - question.asked — Question prompt
  - question.replied — Question answered
  - question.rejected — Question skipped
  - pty.created/updated/exited/deleted — PTY lifecycle
  - file.edited, file.watcher.updated — File change notifications
  - server.connected, server.heartbeat — Connection lifecycle
```

### AC7: Tracks connection health

```gherkin
Given the SSE consumer is running
When a caller checks connection health
Then it reports: connected (boolean), lastEventAt (timestamp), reconnectCount (number)
And if no event has been received in 60 seconds, it reports as potentially stale
```

## Testable Components

| Component | Test Type | What to Assert |
|-----------|-----------|----------------|
| `connect(url)` | Unit | Opens connection to correct URL |
| `onEvent(rawData)` | Unit | Parses JSON, emits typed event, routes by session |
| `onError()` | Unit | Triggers reconnect with backoff |
| `reconnect()` | Unit | Applies exponential backoff, max cap, attempt counting |
| `filterBySession(event, sessionId)` | Unit | Routes correctly, drops unowned events |
| `handleMalformed(data)` | Unit | Logs warning, doesn't throw |
| `getHealth()` | Unit | Returns correct status object |
| Full SSE lifecycle | Fixture | Connect → receive → disconnect → reconnect → receive |

## API Shape

```ts
// src/lib/opencode-bridge.ts (SSE consumer section)
export class SSEConsumer extends EventEmitter {
  constructor(options = {})        // { baseUrl, auth?, reconnectMaxDelay, staleThreshold }
                                   // auth: { username, password } — forwarded as HTTP Basic Auth
                                   // from OPENCODE_SERVER_USERNAME / OPENCODE_SERVER_PASSWORD env vars
  async connect()                  // Start consuming both /event/ and /global/event
  async disconnect()               // Close all SSE connections
  getHealth()                      // { connected, lastEventAt, reconnectCount }

  // Events emitted (matching OpenCode's event types):
  // 'message.part.updated'    — Part created or fully updated
  // 'message.part.delta'      — Incremental text/reasoning content
  // 'message.part.removed'    — Part deleted
  // 'message.updated'         — Message metadata update
  // 'message.removed'         — Message deleted
  // 'session.status'          — Session state (idle/busy/retry)
  // 'permission.asked'        — Permission request
  // 'permission.replied'      — Permission resolved
  // 'question.asked'          — Question prompt
  // 'question.replied'        — Question answered
  // 'question.rejected'       — Question skipped
  // 'pty.created'             — PTY created
  // 'pty.updated'             — PTY updated
  // 'pty.exited'              — PTY exited
  // 'pty.deleted'             — PTY deleted
  // 'file.edited'             — File edited
  // 'file.watcher.updated'    — File changed externally
  // 'server.connected'        — Initial connection
  // 'server.heartbeat'        — 10s heartbeat
  // 'error'                   — Connection error (internal)
}
```

### AC8: Emits reconnection event for downstream state recovery

```gherkin
Given the SSE consumer reconnects after a connection drop
When the connection is re-established
Then it emits a 'reconnected' event to internal listeners
And downstream consumers (e.g., the event translator) can use this signal
  to reset or rebuild their internal state
And the reconnection event includes { reconnectCount, downtime }
```

## Notes

### EventSource and HTTP Basic Auth

OpenCode's SSE endpoints may require HTTP Basic Auth (when `OPENCODE_SERVER_PASSWORD` is set). The browser's native `EventSource` API does **not** support custom headers. Node.js SSE options:

- **Node.js 22+ native `EventSource`** (via undici) — supports headers natively
- **`eventsource` npm package** — supports custom headers via `headers` option
- **Fetch-based SSE** (using `fetch` + `eventsource-parser`) — full control over headers, recommended if targeting Node.js 18+

**Spike task**: Before implementing the full SSE consumer, test connecting to `opencode serve` with auth enabled using the chosen SSE library. If the library doesn't support auth headers, switch to fetch-based SSE (e.g., `eventsource-parser` + native `fetch`). This should take < 1 hour and prevents a blocker mid-implementation.

## Reference

- OpenCode SSE: `../opencode/packages/opencode/src/server/server.ts` (GET /event route, lines ~486-541)
- OpenCode global SSE: `../opencode/packages/opencode/src/server/routes/global.ts` (GET /global/event)
- OpenCode event bus: `../opencode/packages/opencode/src/bus/` (discriminated union with Zod schemas)
- All events have shape: `{ type: string, properties: object }`
- Global events wrap payload: `{ directory: string, payload: { type, properties } }`
- Heartbeat every 10s, initial server.connected event on connection
- Unlike claude-relay's async iterable from SDK, this is standard SSE (EventSource API)
