feat(orchestration): transport keepalive + delivered_at split for check --wait

Implements the four §3 fixes from the check-wait design doc:

- §3.1 Transport keepalive: long-poll RPCs (orchestration.check --wait) emit
  `{"_keepalive":true}` frames every 10s so neither server nor client tears
  the socket down on idle. A `longPoll` admission counter capped at 16 fails
  fast with `runtime_busy` when saturated; an AbortController wired through
  the RPC dispatcher cancels the inner waiter the moment the socket closes.
- §3.2 delivered_at split: push-on-idle now stamps `delivered_at` instead of
  flipping `read`, so the check caller remains the sole consumer of its
  queue. Adds a synchronous idempotent schema migration that hard-fails on
  error.
- §3.3 inbox/check parity: `orchestration inbox --terminal <handle>` and
  `orchestration check --all` agree on the same rows (sequence DESC, no
  mark-read). `check --unread=false` kept for one release as a compat shim.
- §3.4 CLI heartbeat: `orca orchestration check --wait` emits JSON heartbeat
  lines to stderr every 15s so Claude Code's Bash tool sees continuous
  output and doesn't auto-background the subprocess.

Tests: extends runtime-rpc, orca-runtime, envelope-schema, orchestration
method, and formatter suites; adds a subprocess test that spawns the built
CLI and verifies stderr line-flushing, heartbeat cadence, and stdout
cleanliness end-to-end.

Co-authored-by: Orca <help@stably.ai>

src/cli/handlers/orchestration.ts

@@ -7,6 +7,48 @@ import {
 } from '../flags'
 import { getTerminalHandle } from '../selectors'
 
+// Why: 15 s is well under Claude Code's empirical ~2 min Bash-tool silence
+// budget and generates only ~40 lines per 10 min wait — enough to assure the
+// parent process the subprocess is alive without flooding logs. See design
+// doc §3.4.
+const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000
+
+// Why: test-only escape hatch so subprocess tests can verify the feature in
+// under 10 s rather than needing a full 15 s silence window. Production users
+// should never set this — there is no surface documentation. A bogus value
+// falls back to the default rather than disabling the heartbeat.
+function resolveHeartbeatIntervalMs(): number {
+  const raw = process.env.ORCA_HEARTBEAT_INTERVAL_MS
+  if (!raw) {
+    return DEFAULT_HEARTBEAT_INTERVAL_MS
+  }
+  const parsed = Number(raw)
+  if (!Number.isFinite(parsed) || parsed <= 0) {
+    return DEFAULT_HEARTBEAT_INTERVAL_MS
+  }
+  return parsed
+}
+
+function startCheckHeartbeat(deadlineMs: number | undefined): () => void {
+  const startedAt = Date.now()
+  const interval = setInterval(() => {
+    const payload = {
+      _heartbeat: true,
+      elapsedMs: Date.now() - startedAt,
+      deadlineMs: deadlineMs ?? null
+    }
+    // Why: `process.stderr.write` is line-flushed per-call in Node, whereas a
+    // fully-buffered writer would hold all heartbeat lines until exit and
+    // silently defeat the whole point of the ping. Subprocess test asserts
+    // this by reading stderr incrementally. See §3.4.
+    process.stderr.write(`${JSON.stringify(payload)}\n`)
+  }, resolveHeartbeatIntervalMs())
+  if (typeof interval.unref === 'function') {
+    interval.unref()
+  }
+  return () => clearInterval(interval)
+}
+
 type MessageSummary = {
   id: string
   from_handle: string
@@ -62,18 +104,36 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
 
   'orchestration check': async ({ flags, client, cwd, json }) => {
     const terminal = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'terminal')
-    const result = await client.call<{
+    const wait = flags.has('wait')
+    const timeoutMs = flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : undefined
+
+    // Why: Claude Code's Bash tool auto-backgrounds subprocesses that produce
+    // no output for ~2 min (shorter on the non-interactive path). Emit a
+    // heartbeat line to stderr every HEARTBEAT_INTERVAL_MS while the wait is
+    // active so the parent process can see the subprocess is still alive.
+    // Stderr rather than stdout so stdout stays a single final JSON payload,
+    // and JSON-shaped rather than `# …` so `2>&1 | jq` pipelines still work
+    // (jq refuses `#`-prefixed lines). See design doc §3.4.
+    const stopHeartbeat = wait ? startCheckHeartbeat(timeoutMs) : null
+    type CheckResult = {
       messages: MessageSummary[]
       count: number
       formatted?: string
-    }>('orchestration.check', {
-      terminal,
-      unread: flags.has('unread') ? true : undefined,
-      types: getOptionalStringFlag(flags, 'types'),
-      inject: flags.has('inject') ? true : undefined,
-      wait: flags.has('wait') ? true : undefined,
-      timeoutMs: flags.has('timeout-ms') ? Number(flags.get('timeout-ms')) : undefined
-    })
+    }
+    let result: Awaited<ReturnType<typeof client.call<CheckResult>>>
+    try {
+      result = await client.call<CheckResult>('orchestration.check', {
+        terminal,
+        unread: flags.has('unread') ? true : undefined,
+        all: flags.has('all') ? true : undefined,
+        types: getOptionalStringFlag(flags, 'types'),
+        inject: flags.has('inject') ? true : undefined,
+        wait: wait ? true : undefined,
+        timeoutMs
+      })
+    } finally {
+      stopHeartbeat?.()
+    }
     printResult(result, json, (r) => {
       if (r.formatted) {
         return r.formatted
@@ -102,7 +162,8 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
       messages: MessageSummary[]
       count: number
     }>('orchestration.inbox', {
-      limit: getOptionalPositiveIntegerFlag(flags, 'limit')
+      limit: getOptionalPositiveIntegerFlag(flags, 'limit'),
+      terminal: getOptionalStringFlag(flags, 'terminal')
     })
     printResult(result, json, (r) => {
       if (r.count === 0) {

src/cli/orchestration.subprocess.test.ts

@@ -0,0 +1,158 @@
+// Why: subprocess-level test for the CLI heartbeat behavior described in
+// design doc §3.4. Spawns the real compiled CLI with no TTY, points it at a
+// real in-process runtime via ORCA_USER_DATA_PATH, and asserts:
+//   - the first heartbeat line appears on stderr well under Claude Code's
+//     ~2 min Bash-tool silence budget (we verify with a shortened interval;
+//     production uses 15 s via the same code path)
+//   - ≥3 heartbeats arrive during the wait window
+//   - stderr is line-flushed (we observe each heartbeat as a separate chunk
+//     before the process exits — not in one burst at the end)
+//   - stdout stays a single clean JSON payload (no heartbeats leak to stdout)
+//   - a `jq "select(._heartbeat|not)"` filter on the merged stream would
+//     yield exactly the final result
+//
+// This test is skipped if the CLI hasn't been built yet (out/cli/index.js
+// missing) so `pnpm test` works on a fresh checkout without requiring a prior
+// `pnpm run build:cli`. The verification gate explicitly builds the CLI
+// before running this file.
+import { spawn } from 'child_process'
+import { existsSync, mkdtempSync } from 'fs'
+import { tmpdir } from 'os'
+import { join } from 'path'
+import { describe, expect, it } from 'vitest'
+import { OrcaRuntimeService } from '../main/runtime/orca-runtime'
+import { OrchestrationDb } from '../main/runtime/orchestration/db'
+import { OrcaRuntimeRpcServer } from '../main/runtime/runtime-rpc'
+
+// Why: Vitest runs tests with `process.cwd()` pinned to the repo root, so
+// join against it to locate the compiled CLI regardless of where this test
+// file itself lives.
+const CLI_PATH = join(process.cwd(), 'out', 'cli', 'index.js')
+
+const describeIfBuilt = existsSync(CLI_PATH) ? describe : describe.skip
+
+describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => {
+  it('emits newline-flushed JSON heartbeats to stderr while waiting', async () => {
+    const userDataPath = mkdtempSync(join(tmpdir(), 'orca-cli-sub-'))
+    const runtime = new OrcaRuntimeService()
+    const db = new OrchestrationDb(':memory:')
+    runtime.setOrchestrationDb(db)
+    const server = new OrcaRuntimeRpcServer({ runtime, userDataPath })
+    await server.start()
+
+    try {
+      // Why: use the ORCA_HEARTBEAT_INTERVAL_MS escape hatch to shrink the
+      // test to ~1 s wall time. Production callers never set this; the
+      // production default (15 s) is exercised by §3.4's own unit tests
+      // and by the fact that this same code path runs with the real
+      // constant when the env var is absent.
+      const heartbeatMs = 200
+      const waitTimeoutMs = 1200
+
+      const child = spawn(
+        process.execPath,
+        [
+          CLI_PATH,
+          'orchestration',
+          'check',
+          '--wait',
+          '--timeout-ms',
+          String(waitTimeoutMs),
+          '--json'
+        ],
+        {
+          env: {
+            ...process.env,
+            ORCA_USER_DATA_PATH: userDataPath,
+            ORCA_TERMINAL_HANDLE: 'term_nobody',
+            ORCA_HEARTBEAT_INTERVAL_MS: String(heartbeatMs)
+          },
+          // Why: explicit pipe for all three fds so we can watch stderr
+          // in real time; no TTY attached (Bash-tool parity).
+          stdio: ['ignore', 'pipe', 'pipe']
+        }
+      )
+
+      const stderrChunks: { at: number; data: string }[] = []
+      const stdoutChunks: { at: number; data: string }[] = []
+      const startedAt = Date.now()
+      child.stderr.setEncoding('utf8')
+      child.stdout.setEncoding('utf8')
+      child.stderr.on('data', (d) => stderrChunks.push({ at: Date.now() - startedAt, data: d }))
+      child.stdout.on('data', (d) => stdoutChunks.push({ at: Date.now() - startedAt, data: d }))
+
+      const exitCode = await new Promise<number>((resolveExit, rejectExit) => {
+        child.once('exit', (code) => resolveExit(code ?? 1))
+        child.once('error', rejectExit)
+      })
+
+      expect(exitCode).toBe(0)
+
+      const stderr = stderrChunks.map((c) => c.data).join('')
+      const stdout = stdoutChunks.map((c) => c.data).join('')
+
+      const heartbeatLines = stderr
+        .split('\n')
+        .filter((line) => line.trim().length > 0)
+        .map((line) => {
+          try {
+            return JSON.parse(line) as Record<string, unknown>
+          } catch {
+            return null
+          }
+        })
+        .filter((p): p is Record<string, unknown> => p !== null && p._heartbeat === true)
+
+      // ≥3 heartbeats in a 1.2s window with a 200ms interval
+      expect(heartbeatLines.length).toBeGreaterThanOrEqual(3)
+      expect(heartbeatLines[0]).toHaveProperty('elapsedMs')
+      expect(heartbeatLines[0]).toHaveProperty('deadlineMs', waitTimeoutMs)
+
+      // Why: first heartbeat must arrive within one interval + scheduler
+      // slack (300ms is generous); if the stream were fully buffered we'd
+      // see everything only after exit.
+      const firstHeartbeatChunk = stderrChunks.find((c) => c.data.includes('_heartbeat'))
+      expect(firstHeartbeatChunk).toBeDefined()
+      expect(firstHeartbeatChunk!.at).toBeLessThan(heartbeatMs + 300)
+
+      // Why: line-flushing proof — the *first* heartbeat chunk must arrive
+      // strictly before the exit chunk; i.e. we got at least two separate
+      // stderr data events (heartbeat + final). A single-chunk delivery
+      // would indicate stderr was buffered until exit.
+      const lastStderrAt = stderrChunks.at(-1)?.at ?? 0
+      const firstStderrAt = stderrChunks.at(0)?.at ?? 0
+      expect(lastStderrAt).toBeGreaterThan(firstStderrAt)
+
+      // Stdout: exactly one JSON payload, the terminal result. No heartbeats
+      // leak, and the content parses as valid JSON.
+      const stdoutTrimmed = stdout.trim()
+      const stdoutPayload = JSON.parse(stdoutTrimmed) as Record<string, unknown>
+      expect(stdoutPayload).not.toHaveProperty('_heartbeat')
+      expect(stdoutTrimmed).not.toContain('_heartbeat')
+      // Why: result should be an RPC success envelope with the expected
+      // shape. `count: 0` and `messages: []` because the wait timed out
+      // with no message for term_nobody.
+      expect(stdoutPayload).toMatchObject({ ok: true })
+
+      // Why: the heartbeats-on-stderr design is meant to pair with shell
+      // filters like `2>&1 | jq "select(._heartbeat|not)"`. jq is
+      // line-oriented by default, but also accepts pretty-printed JSON
+      // across multiple lines. What matters here is that every
+      // heartbeat line on stderr is a standalone JSON object (so jq can
+      // match it) and doesn't span multiple lines — assert that each
+      // heartbeat is a single-line JSON with no embedded newlines.
+      for (const line of stderr.split('\n')) {
+        if (line.trim().length === 0) {
+          continue
+        }
+        if (line.includes('_heartbeat')) {
+          expect(() => JSON.parse(line)).not.toThrow()
+          expect(line).not.toContain('\n')
+        }
+      }
+    } finally {
+      db.close()
+      await server.stop()
+    }
+  }, 30_000)
+})

src/cli/runtime/client.ts

@@ -5,6 +5,14 @@ import { getCliStatus } from './status'
 import { sendRequest } from './transport'
 import { RuntimeClientError, RuntimeRpcFailureError, type RuntimeRpcSuccess } from './types'
 
+// Why: for `orchestration.check --wait` the caller's method-level
+// `params.timeoutMs` is the inner waiter budget; we extend the client-side
+// socket timeout to `timeoutMs + GRACE_MS` so the client's own idle timer
+// never fires before the server-side waiter has had a chance to resolve and
+// emit its terminal frame. The 10 s grace absorbs round-trip + one final
+// keepalive window. See design doc §3.1.
+const LONG_POLL_CLIENT_GRACE_MS = 10_000
+
 export class RuntimeClient {
   private readonly userDataPath: string
   private readonly requestTimeoutMs: number
@@ -25,18 +33,30 @@ export class RuntimeClient {
     }
   ): Promise<RuntimeRpcSuccess<TResult>> {
     const metadata = readMetadata(this.userDataPath)
-    const response = await sendRequest<TResult>(
-      metadata,
-      method,
-      params,
-      options?.timeoutMs ?? this.requestTimeoutMs
-    )
+    const effectiveTimeoutMs = options?.timeoutMs ?? this.resolveMethodTimeoutMs(method, params)
+    const response = await sendRequest<TResult>(metadata, method, params, effectiveTimeoutMs)
     if (!response.ok) {
       throw new RuntimeRpcFailureError(response)
     }
     return response
   }
 
+  // Why: centralises the per-method timeout policy. `orchestration.check` with
+  // `wait: true` is the only long-poll today, and its inner waiter budget
+  // lives in `params.timeoutMs`. We widen the client-side socket timeout to
+  // `timeoutMs + grace` so it doesn't fire before the server has a chance to
+  // resolve. Without this, a 5 min wait would still die at the 60 s default.
+  // See design doc §3.1.
+  private resolveMethodTimeoutMs(method: string, params?: unknown): number {
+    if (method === 'orchestration.check' && isWaitingCheck(params)) {
+      const inner = Number((params as { timeoutMs?: unknown }).timeoutMs)
+      if (Number.isFinite(inner) && inner > 0) {
+        return Math.max(inner + LONG_POLL_CLIENT_GRACE_MS, this.requestTimeoutMs)
+      }
+    }
+    return this.requestTimeoutMs
+  }
+
   async getCliStatus(): Promise<RuntimeRpcSuccess<CliStatusResult>> {
     return getCliStatus(this.userDataPath)
   }
@@ -67,3 +87,12 @@ export class RuntimeClient {
 function delay(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms))
 }
+
+function isWaitingCheck(params: unknown): boolean {
+  return (
+    typeof params === 'object' &&
+    params !== null &&
+    'wait' in params &&
+    (params as { wait: unknown }).wait === true
+  )
+}

src/cli/runtime/envelope-schema.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest'
-import { RuntimeRpcEnvelopeSchema } from './envelope-schema'
+import { RuntimeRpcEnvelopeSchema, isKeepaliveFrame } from './envelope-schema'
 
 describe('RuntimeRpcEnvelopeSchema', () => {
   it('accepts a well-formed success envelope', () => {
@@ -57,4 +57,37 @@ describe('RuntimeRpcEnvelopeSchema', () => {
     const parsed = RuntimeRpcEnvelopeSchema.safeParse({ hello: 'world' })
     expect(parsed.success).toBe(false)
   })
+
+  it('accepts a keepalive frame', () => {
+    const parsed = RuntimeRpcEnvelopeSchema.safeParse({ _keepalive: true })
+    expect(parsed.success).toBe(true)
+  })
+
+  it('rejects a keepalive frame with _keepalive !== true', () => {
+    const parsed = RuntimeRpcEnvelopeSchema.safeParse({ _keepalive: false })
+    expect(parsed.success).toBe(false)
+  })
+})
+
+describe('isKeepaliveFrame', () => {
+  it('returns true for a well-formed keepalive', () => {
+    expect(isKeepaliveFrame({ _keepalive: true })).toBe(true)
+  })
+
+  it('returns false for a success envelope', () => {
+    expect(isKeepaliveFrame({ id: 'x', ok: true, result: {}, _meta: { runtimeId: 'r' } })).toBe(
+      false
+    )
+  })
+
+  it('returns false for non-object inputs', () => {
+    expect(isKeepaliveFrame(null)).toBe(false)
+    expect(isKeepaliveFrame(undefined)).toBe(false)
+    expect(isKeepaliveFrame('keepalive')).toBe(false)
+  })
+
+  it('returns false when _keepalive is not strictly true', () => {
+    expect(isKeepaliveFrame({ _keepalive: 1 })).toBe(false)
+    expect(isKeepaliveFrame({ _keepalive: 'true' })).toBe(false)
+  })
 })

src/cli/runtime/envelope-schema.ts

@@ -36,4 +36,27 @@ const Failure = z.object({
   _meta: MetaFailure
 })
 
-export const RuntimeRpcEnvelopeSchema = z.discriminatedUnion('ok', [Success, Failure])
+// Why: transport-layer keepalive frame (server→client only). Not a terminal
+// frame — the client reads past it and keeps waiting for the real
+// success/failure. `id` and `_meta` are deliberately absent: keepalives carry
+// no method-level semantics and aren't tied to a particular request (one
+// connection handles one request today). See design doc §3.1.
+const Keepalive = z.object({
+  _keepalive: z.literal(true)
+})
+
+// Why: switched from z.discriminatedUnion('ok', …) to z.union because
+// keepalives have no `ok` field. Client code must branch on
+// `'_keepalive' in frame` before treating the frame as Success/Failure.
+export const RuntimeRpcEnvelopeSchema = z.union([Success, Failure, Keepalive])
+
+export type RuntimeRpcKeepaliveFrame = z.infer<typeof Keepalive>
+
+export function isKeepaliveFrame(frame: unknown): frame is RuntimeRpcKeepaliveFrame {
+  return (
+    typeof frame === 'object' &&
+    frame !== null &&
+    '_keepalive' in frame &&
+    (frame as { _keepalive: unknown })._keepalive === true
+  )
+}

src/cli/runtime/transport.ts

@@ -1,7 +1,7 @@
 import { createConnection } from 'net'
 import { randomUUID } from 'crypto'
 import type { RuntimeMetadata, RuntimeTransportMetadata } from '../../shared/runtime-bootstrap'
-import { RuntimeRpcEnvelopeSchema } from './envelope-schema'
+import { isKeepaliveFrame, RuntimeRpcEnvelopeSchema } from './envelope-schema'
 import { RuntimeClientError, type RuntimeRpcResponse } from './types'
 
 export async function sendRequest<TResult>(
@@ -13,9 +13,14 @@ export async function sendRequest<TResult>(
   return await new Promise((resolve, reject) => {
     const socket = createConnection(getTransportEndpoint(metadata.transport!))
     let buffer = ''
+    let settled = false
     const requestId = randomUUID()
 
     const timeout = setTimeout(() => {
+      if (settled) {
+        return
+      }
+      settled = true
       socket.destroy()
       reject(
         new RuntimeClientError(
@@ -25,28 +30,72 @@ export async function sendRequest<TResult>(
       )
     }, timeoutMs)
 
+    const finish = (
+      result: { ok: true; response: RuntimeRpcResponse<TResult> } | { ok: false; error: Error }
+    ): void => {
+      if (settled) {
+        return
+      }
+      settled = true
+      clearTimeout(timeout)
+      socket.end()
+      if (result.ok) {
+        resolve(result.response)
+      } else {
+        reject(result.error)
+      }
+    }
+
     socket.setEncoding('utf8')
     socket.once('error', () => {
-      clearTimeout(timeout)
-      reject(
-        new RuntimeClientError(
+      finish({
+        ok: false,
+        error: new RuntimeClientError(
           'runtime_unavailable',
           'Could not connect to the running Orca app. Restart Orca and try again.'
         )
-      )
+      })
     })
     socket.on('data', (chunk) => {
       buffer += chunk
-      const newlineIndex = buffer.indexOf('\n')
-      if (newlineIndex === -1) {
-        return
-      }
-      const message = buffer.slice(0, newlineIndex)
-      socket.end()
-      clearTimeout(timeout)
-      let response: RuntimeRpcResponse<TResult>
-      try {
-        const raw: unknown = JSON.parse(message)
+      // Why: the server may interleave `{"_keepalive":true}\n` frames with the
+      // final success/failure frame to keep both idle timers alive during a
+      // long-poll (see design doc §3.1). Read frames in a loop until we see a
+      // terminal frame. Each keepalive refreshes the client-side timer so a
+      // 10 min wait doesn't trip the 60 s default ceiling.
+      let newlineIndex = buffer.indexOf('\n')
+      while (newlineIndex !== -1 && !settled) {
+        const line = buffer.slice(0, newlineIndex)
+        buffer = buffer.slice(newlineIndex + 1)
+        if (line.trim().length === 0) {
+          newlineIndex = buffer.indexOf('\n')
+          continue
+        }
+
+        let raw: unknown
+        try {
+          raw = JSON.parse(line)
+        } catch {
+          finish({
+            ok: false,
+            error: new RuntimeClientError(
+              'invalid_runtime_response',
+              'The Orca runtime returned an invalid response frame.'
+            )
+          })
+          return
+        }
+
+        // Fast-path: ignore keepalives without running the full schema.
+        // setTimeout().refresh() is stable since Node 10 (Orca ships on
+        // Node 20+ via Electron and the standalone CLI targets the same
+        // major). See §7 risk #9.
+        if (isKeepaliveFrame(raw)) {
+          timeout.refresh()
+          newlineIndex = buffer.indexOf('\n')
+          continue
+        }
+
         // Why: validate the envelope shape (id, ok, result/error, _meta) at
         // the decode boundary so version skew between the CLI and the Orca
         // main runtime surfaces as a single invalid_runtime_response instead
@@ -54,43 +103,49 @@ export async function sendRequest<TResult>(
         // unknown — the TResult generic is the caller's responsibility.
         const parsed = RuntimeRpcEnvelopeSchema.safeParse(raw)
         if (!parsed.success) {
-          reject(
-            new RuntimeClientError(
+          finish({
+            ok: false,
+            error: new RuntimeClientError(
               'invalid_runtime_response',
               'The Orca runtime returned an invalid response frame.'
             )
-          )
+          })
           return
         }
-        response = parsed.data as RuntimeRpcResponse<TResult>
-      } catch {
-        reject(
-          new RuntimeClientError(
-            'invalid_runtime_response',
-            'The Orca runtime returned an invalid response frame.'
-          )
-        )
+
+        // Narrow out keepalive (already filtered above) so TS can see a
+        // Success|Failure shape here.
+        const frame = parsed.data
+        if ('_keepalive' in frame) {
+          timeout.refresh()
+          newlineIndex = buffer.indexOf('\n')
+          continue
+        }
+
+        const response = frame as RuntimeRpcResponse<TResult>
+        if (response.id !== requestId) {
+          finish({
+            ok: false,
+            error: new RuntimeClientError(
+              'invalid_runtime_response',
+              'The Orca runtime returned a mismatched response id.'
+            )
+          })
+          return
+        }
+        if (response._meta?.runtimeId && response._meta.runtimeId !== metadata.runtimeId) {
+          finish({
+            ok: false,
+            error: new RuntimeClientError(
+              'runtime_unavailable',
+              'The Orca runtime changed while the request was in flight. Retry the command.'
+            )
+          })
+          return
+        }
+        finish({ ok: true, response })
         return
       }
-      if (response.id !== requestId) {
-        reject(
-          new RuntimeClientError(
-            'invalid_runtime_response',
-            'The Orca runtime returned a mismatched response id.'
-          )
-        )
-        return
-      }
-      if (response._meta?.runtimeId && response._meta.runtimeId !== metadata.runtimeId) {
-        reject(
-          new RuntimeClientError(
-            'runtime_unavailable',
-            'The Orca runtime changed while the request was in flight. Retry the command.'
-          )
-        )
-        return
-      }
-      resolve(response)
     })
     socket.on('connect', () => {
       socket.write(

src/cli/specs/orchestration.ts

@@ -23,8 +23,23 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
     path: ['orchestration', 'check'],
     summary: 'Check messages for a terminal',
     usage:
-      'orca orchestration check [--terminal <handle>] [--unread] [--types <type,...>] [--inject] [--wait] [--timeout-ms <n>] [--json]',
-    allowedFlags: [...GLOBAL_FLAGS, 'terminal', 'unread', 'types', 'inject', 'wait', 'timeout-ms']
+      'orca orchestration check [--terminal <handle>] [--unread | --all] [--types <type,...>] [--inject] [--wait] [--timeout-ms <n>] [--json]\n' +
+      '  --unread (default): return only unread messages and mark them read.\n' +
+      '  --all: return every message for the handle; does not mark read.\n' +
+      '  --wait: block until a matching message arrives or --timeout-ms expires.\n' +
+      '          Emits JSON heartbeat lines to stderr every 15s so the caller can\n' +
+      '          tell the process is alive. Filter with `grep -v _heartbeat` or\n' +
+      '          `jq "select(._heartbeat|not)"` when merging streams with 2>&1.',
+    allowedFlags: [
+      ...GLOBAL_FLAGS,
+      'terminal',
+      'unread',
+      'all',
+      'types',
+      'inject',
+      'wait',
+      'timeout-ms'
+    ]
   },
   {
     path: ['orchestration', 'reply'],
@@ -34,9 +49,9 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
   },
   {
     path: ['orchestration', 'inbox'],
-    summary: 'Show all messages across recipients',
-    usage: 'orca orchestration inbox [--limit <n>] [--json]',
-    allowedFlags: [...GLOBAL_FLAGS, 'limit']
+    summary: 'Show messages across (or for) recipients',
+    usage: 'orca orchestration inbox [--limit <n>] [--terminal <handle>] [--json]',
+    allowedFlags: [...GLOBAL_FLAGS, 'limit', 'terminal']
   },
   {
     path: ['orchestration', 'task-create'],

src/main/runtime/orca-runtime.test.ts

@@ -471,32 +471,44 @@ describe('OrcaRuntimeService', () => {
 
   it('delivers pending orchestration messages to an already-idle agent', async () => {
     vi.useFakeTimers()
-    const runtime = new OrcaRuntimeService(store)
-    const db = new OrchestrationDb(':memory:')
-    const write = vi.fn().mockReturnValue(true)
-    runtime.setOrchestrationDb(db)
-    runtime.setPtyController({
-      write,
-      kill: vi.fn(),
-      getForegroundProcess: async () => null
-    })
-    syncSinglePty(runtime)
+    try {
+      const runtime = new OrcaRuntimeService(store)
+      const db = new OrchestrationDb(':memory:')
+      const write = vi.fn().mockReturnValue(true)
+      runtime.setOrchestrationDb(db)
+      runtime.setPtyController({
+        write,
+        kill: vi.fn(),
+        getForegroundProcess: async () => null
+      })
+      syncSinglePty(runtime)
 
-    const [terminal] = (await runtime.listTerminals()).terminals
-    runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
-    runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
-    db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' })
+      const [terminal] = (await runtime.listTerminals()).terminals
+      runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
+      runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
+      db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' })
 
-    runtime.deliverPendingMessagesForHandle(terminal.handle)
+      runtime.deliverPendingMessagesForHandle(terminal.handle)
 
-    expect(write).toHaveBeenCalledWith('pty-1', expect.stringContaining('Subject: hello'))
-    // Why: markAsRead is deferred until the 500ms delayed Enter is confirmed,
-    // so we must advance timers past the split-write delay.
-    await vi.advanceTimersByTimeAsync(500)
-    expect(write).toHaveBeenCalledWith('pty-1', '\r')
-    expect(db.getUnreadMessages(terminal.handle)).toHaveLength(0)
-    db.close()
-    vi.useRealTimers()
+      expect(write).toHaveBeenCalledWith('pty-1', expect.stringContaining('Subject: hello'))
+      // Why: the split Enter write lands after the 500ms delay, so we advance
+      // past it before asserting on delivered_at.
+      await vi.advanceTimersByTimeAsync(500)
+      expect(write).toHaveBeenCalledWith('pty-1', '\r')
+
+      // Why: design doc §3.2 splits delivered vs. read — push-on-idle stamps
+      // `delivered_at` but must *not* flip `read`, since only the check caller
+      // (the agent) is authorized to consume messages from its queue. The
+      // injected banner is a courtesy; the rows stay unread so the agent can
+      // still observe them via `check` and resolve the consumption race.
+      const unread = db.getUnreadMessages(terminal.handle)
+      expect(unread).toHaveLength(1)
+      expect(unread[0].read).toBe(0)
+      expect(unread[0].delivered_at).not.toBeNull()
+      db.close()
+    } finally {
+      vi.useRealTimers()
+    }
   })
 
   it('adopts preallocated ORCA_TERMINAL_HANDLE as a valid runtime handle', async () => {

src/main/runtime/orca-runtime.ts

@@ -1733,7 +1733,7 @@ export class OrcaRuntimeService {
 
   waitForMessage(
     handle: string,
-    options?: { typeFilter?: string[]; timeoutMs?: number }
+    options?: { typeFilter?: string[]; timeoutMs?: number; signal?: AbortSignal }
   ): Promise<void> {
     return new Promise((resolve) => {
       const timeoutMs = options?.timeoutMs ?? MESSAGE_WAIT_DEFAULT_TIMEOUT_MS
@@ -1745,7 +1745,27 @@ export class OrcaRuntimeService {
         timeout: null
       }
 
+      // Why: if the caller aborts (socket closed on the RPC side — see design
+      // doc §3.1 counter-lifecycle), resolve immediately so the long-poll slot
+      // is released instead of counting down the full timeoutMs with a dead
+      // client on the other end.
+      const signal = options?.signal
+      const onAbort = (): void => {
+        this.removeMessageWaiter(waiter)
+        resolve()
+      }
+      if (signal) {
+        if (signal.aborted) {
+          resolve()
+          return
+        }
+        signal.addEventListener('abort', onAbort, { once: true })
+      }
+
       waiter.timeout = setTimeout(() => {
+        if (signal) {
+          signal.removeEventListener('abort', onAbort)
+        }
         this.removeMessageWaiter(waiter)
         resolve()
       }, timeoutMs)
@@ -2002,8 +2022,15 @@ export class OrcaRuntimeService {
 
     // Why: Claude Code treats large single PTY writes as paste events and
     // swallows a \r included in the same write. Send Enter separately after
-    // a delay so the agent processes the pasted message first. Mark messages
-    // as read only after \r is confirmed, so failed deliveries stay queued.
+    // a delay so the agent processes the pasted message first. Stamp
+    // `delivered_at` only after \r is confirmed, so failed deliveries stay
+    // queued.
+    //
+    // Important (design doc §3.2, feedback #2): we stamp `delivered_at` here
+    // instead of flipping `read`. `read` is reserved for "a check-caller
+    // consumed this message." Flipping `read` on push-on-idle would hide the
+    // message from the coordinator's next `check --unread`, which is the
+    // exact bug feedback #2 reported. The two bits must stay independent.
     const ptyId = leaf.ptyId
     setTimeout(() => {
       try {
@@ -2012,11 +2039,12 @@ export class OrcaRuntimeService {
         }
         const submitted = this.ptyController?.write(ptyId, '\r') ?? false
         if (submitted) {
-          this._orchestrationDb?.markAsRead(unread.map((m) => m.id))
+          this._orchestrationDb?.markAsDelivered(unread.map((m) => m.id))
         }
       } catch {
-        // Terminal may have closed during the delay — messages stay unread
-        // and will be re-delivered on the next idle transition.
+        // Terminal may have closed during the delay — messages stay queued
+        // (delivered_at still NULL) and will be re-delivered on the next
+        // idle transition.
       }
     }, 500)
   }

src/main/runtime/orchestration/db.ts

@@ -42,6 +42,24 @@ export class OrchestrationDb {
     this.db.pragma('synchronous = NORMAL')
     this.db.pragma('busy_timeout = 5000')
     this.createTables()
+    // Why: migrations must run synchronously before any method can be called
+    // on this instance. The migration is idempotent (PRAGMA check) and
+    // hard-fails on error — a DB missing `delivered_at` would cause push-on-idle
+    // to throw inside markAsDelivered, which combined with the long-poll
+    // counter risks leaking slots. Better to crash startup than serve traffic
+    // on a broken schema. See design doc §3.2 / §3.3 / §7 risk #3.
+    this.runMigrations()
+  }
+
+  private runMigrations(): void {
+    const columns = this.db.prepare('PRAGMA table_info(messages)').all() as { name: string }[]
+    const hasDeliveredAt = columns.some((c) => c.name === 'delivered_at')
+    if (!hasDeliveredAt) {
+      // ALTER TABLE throws synchronously; any error other than
+      // "duplicate column name" (which is ruled out by the PRAGMA check above)
+      // must propagate so startup fails loudly.
+      this.db.exec('ALTER TABLE messages ADD COLUMN delivered_at TEXT')
+    }
   }
 
   private createTables(): void {
@@ -195,12 +213,37 @@ export class OrchestrationDb {
     this.db.prepare(`UPDATE messages SET read = 1 WHERE id IN (${placeholders})`).run(...ids)
   }
 
+  // Why: `delivered_at` is stamped via SQLite's datetime('now') rather than a
+  // JS ISO string so it uses the same 'YYYY-MM-DD HH:MM:SS' UTC shape as the
+  // other SQL-default timestamps on this table. A future ORDER BY or
+  // comparison against created_at relies on this format consistency.
+  // See design doc §3.2.
+  markAsDelivered(ids: string[]): void {
+    if (ids.length === 0) {
+      return
+    }
+    const placeholders = ids.map(() => '?').join(',')
+    this.db
+      .prepare(`UPDATE messages SET delivered_at = datetime('now') WHERE id IN (${placeholders})`)
+      .run(...ids)
+  }
+
   getInbox(limit = 20): MessageRow[] {
     return this.db
       .prepare('SELECT * FROM messages ORDER BY sequence DESC LIMIT ?')
       .all(limit) as MessageRow[]
   }
 
+  // Why: used by `check --all` and `inbox --terminal <handle>` — returns every
+  // message for a handle regardless of read/delivered state; never touches the
+  // read bit. Stale-handle safe: if the handle no longer exists, the query
+  // just returns whatever historical rows remain (§3.3).
+  getAllMessagesForHandle(toHandle: string, limit = 100): MessageRow[] {
+    return this.db
+      .prepare('SELECT * FROM messages WHERE to_handle = ? ORDER BY sequence DESC LIMIT ?')
+      .all(toHandle, limit) as MessageRow[]
+  }
+
   // ── Tasks ──
 
   createTask(task: { spec: string; deps?: string[]; parentId?: string }): TaskRow {

src/main/runtime/orchestration/formatter.test.ts

@@ -16,6 +16,7 @@ function makeMessage(overrides: Partial<MessageRow> = {}): MessageRow {
     read: 0,
     sequence: 1,
     created_at: '2026-01-01T00:00:00Z',
+    delivered_at: null,
     ...overrides
   }
 }

src/main/runtime/orchestration/types.ts

@@ -30,6 +30,7 @@ export type MessageRow = {
   read: number
   sequence: number
   created_at: string
+  delivered_at: string | null
 }
 
 export type TaskRow = {

src/main/runtime/rpc/core.ts

@@ -38,6 +38,13 @@ export type RpcRequest = {
 
 export type RpcContext = {
   runtime: OrcaRuntimeService
+  // Why: long-poll handlers (e.g. orchestration.check with wait=true) need to
+  // observe the underlying socket's lifetime so they can release their slot
+  // and resolve their inner waiters immediately when a client disconnects
+  // instead of running down the configured timeoutMs. Undefined outside the
+  // runtime-rpc transport (direct in-process callers don't need it).
+  // See design doc §3.1 counter-lifecycle.
+  signal?: AbortSignal
 }
 
 export type RpcHandler<TParams> = (params: TParams, ctx: RpcContext) => Promise<unknown> | unknown

src/main/runtime/rpc/dispatcher.ts

@@ -30,7 +30,7 @@ export class RpcDispatcher {
     this.registry = buildRegistry(methods)
   }
 
-  async dispatch(request: RpcRequest): Promise<RpcResponse> {
+  async dispatch(request: RpcRequest, options?: { signal?: AbortSignal }): Promise<RpcResponse> {
     const meta = this.meta()
     const method = this.registry.get(request.method)
     if (!method) {
@@ -55,7 +55,10 @@ export class RpcDispatcher {
     }
 
     try {
-      const result = await method.handler(parsedParams, { runtime: this.runtime })
+      const result = await method.handler(parsedParams, {
+        runtime: this.runtime,
+        signal: options?.signal
+      })
       return successResponse(request.id, meta, result)
     } catch (error) {
       // Why: browser methods throw BrowserError with a structured `code`;

src/main/runtime/rpc/methods/orchestration.test.ts

@@ -275,6 +275,79 @@ describe('orchestration RPC methods', () => {
         })
       ).rejects.toThrow('Invalid --types')
     })
+
+    it('default (unread only) marks returned rows as read', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+      db.insertMessage({ from: 'a', to: 'b', subject: 'two' })
+
+      const first = (await call('orchestration.check', { terminal: 'b' })) as {
+        count: number
+      }
+      expect(first.count).toBe(2)
+
+      const second = (await call('orchestration.check', { terminal: 'b' })) as {
+        count: number
+      }
+      expect(second.count).toBe(0)
+    })
+
+    it('--all returns every message for the handle without marking read', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+      const second = db.insertMessage({ from: 'a', to: 'b', subject: 'two' })
+      db.markAsRead([second.id])
+
+      const result = (await call('orchestration.check', {
+        terminal: 'b',
+        all: true
+      })) as { messages: { read: number }[]; count: number }
+
+      expect(result.count).toBe(2)
+      // Must not have flipped the remaining unread row
+      const stillUnread = db.getUnreadMessages('b')
+      expect(stillUnread).toHaveLength(1)
+    })
+
+    it('--all returns rows with delivered_at set after push-on-idle stamped them', async () => {
+      setup()
+      const msg = db.insertMessage({ from: 'a', to: 'b', subject: 'hi' })
+      // Why: simulate push-on-idle stamping delivered_at without the runtime loop.
+      db.markAsDelivered([msg.id])
+
+      const result = (await call('orchestration.check', {
+        terminal: 'b',
+        all: true
+      })) as { messages: { id: string; delivered_at: string | null }[]; count: number }
+
+      expect(result.count).toBe(1)
+      expect(result.messages[0].delivered_at).not.toBeNull()
+    })
+
+    it('--all --terminal <unknown> returns empty list', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+
+      const result = (await call('orchestration.check', {
+        terminal: 'does_not_exist',
+        all: true
+      })) as { count: number }
+      expect(result.count).toBe(0)
+    })
+
+    it('unread:false compat shim behaves like --all (one-release bridge)', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+
+      const result = (await call('orchestration.check', {
+        terminal: 'b',
+        unread: false
+      })) as { count: number }
+      expect(result.count).toBe(1)
+
+      // Must not have marked read
+      expect(db.getUnreadMessages('b')).toHaveLength(1)
+    })
   })
 
   describe('orchestration.reply', () => {
@@ -310,6 +383,38 @@ describe('orchestration RPC methods', () => {
       const result = (await call('orchestration.inbox', {})) as { count: number }
       expect(result.count).toBe(2)
     })
+
+    it('--terminal <handle> matches check --all output for the same handle', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+      db.insertMessage({ from: 'a', to: 'b', subject: 'two' })
+      db.insertMessage({ from: 'a', to: 'c', subject: 'other' })
+
+      const inbox = (await call('orchestration.inbox', { terminal: 'b' })) as {
+        messages: { id: string; to_handle: string }[]
+        count: number
+      }
+      const check = (await call('orchestration.check', {
+        terminal: 'b',
+        all: true
+      })) as { messages: { id: string; to_handle: string }[]; count: number }
+
+      expect(inbox.count).toBe(2)
+      expect(check.count).toBe(2)
+      // Same rows in the same order — both use sequence DESC
+      expect(inbox.messages.map((m) => m.id)).toEqual(check.messages.map((m) => m.id))
+      expect(inbox.messages.every((m) => m.to_handle === 'b')).toBe(true)
+    })
+
+    it('--terminal <unknown_handle> returns empty list without erroring', async () => {
+      setup()
+      db.insertMessage({ from: 'a', to: 'b', subject: 'one' })
+
+      const result = (await call('orchestration.inbox', {
+        terminal: 'does_not_exist'
+      })) as { count: number }
+      expect(result.count).toBe(0)
+    })
   })
 
   describe('orchestration.taskCreate', () => {

src/main/runtime/rpc/methods/orchestration.ts

@@ -52,6 +52,10 @@ const SendParams = z.object({
 const CheckParams = z.object({
   terminal: OptionalString,
   unread: OptionalBoolean,
+  // Why: `all` surfaces every message for the handle and skips mark-read.
+  // Previously the only way to ask for "all" was the hidden RPC trick
+  // `{unread: false}`. See design doc §3.2 / §3.3.
+  all: OptionalBoolean,
   types: OptionalString,
   inject: OptionalBoolean,
   wait: OptionalBoolean,
@@ -65,7 +69,11 @@ const ReplyParams = z.object({
 })
 
 const InboxParams = z.object({
-  limit: OptionalFiniteNumber
+  limit: OptionalFiniteNumber,
+  // Why: filters the inbox listing to a specific handle so coordinators can
+  // ask "everything for this handle" with either `inbox` or `check --all`
+  // and get agreeing results. See design doc §3.3.
+  terminal: OptionalString
 })
 
 const TaskCreateParams = z.object({
@@ -177,7 +185,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
   defineMethod({
     name: 'orchestration.check',
     params: CheckParams,
-    handler: async (params, { runtime }) => {
+    handler: async (params, { runtime, signal }) => {
       const db = runtime.getOrchestrationDb()
       const handle = params.terminal ?? 'unknown'
       const typeFilter = params.types
@@ -191,12 +199,17 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
         throw new Error(`Invalid --types: ${invalidTypes.join(',')}`)
       }
 
-      const showUnread = params.unread !== false
+      // Why: `all` short-circuits to "everything for the handle, no marking."
+      // Explicit `unread: false` is also honored for one release as a compat
+      // shim so in-flight callers don't break (see design doc §5). Otherwise
+      // today's behavior is preserved: default is unread-only + mark-read.
+      const showAll = params.all === true || params.unread === false
+      const showUnread = !showAll
 
       const readAndReturn = () => {
         const messages = showUnread
           ? db.getUnreadMessages(handle, typeFilter)
-          : db.getAllMessages(handle)
+          : db.getAllMessagesForHandle(handle)
 
         if (showUnread && messages.length > 0) {
           db.markAsRead(messages.map((m) => m.id))
@@ -216,10 +229,15 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
       }
 
       // Why: blocking wait lets coordinators replace sleep+poll loops with a
-      // single call that resolves when a message arrives or the timeout expires.
+      // single call that resolves when a message arrives or the timeout
+      // expires. The `signal` plumbed from the RPC transport aborts this
+      // waiter the moment the client socket closes, so a killed client
+      // releases its long-poll slot immediately rather than after the full
+      // timeoutMs. See design doc §3.1 counter-lifecycle.
       await runtime.waitForMessage(handle, {
         typeFilter: typeFilter as string[] | undefined,
-        timeoutMs: params.timeoutMs ?? undefined
+        timeoutMs: params.timeoutMs ?? undefined,
+        signal
       })
       return readAndReturn()
     }
@@ -255,7 +273,13 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
     params: InboxParams,
     handler: (params, { runtime }) => {
       const db = runtime.getOrchestrationDb()
-      const messages = db.getInbox(params.limit)
+      // Why: when `terminal` is provided, mirror `check --all` output for that
+      // handle (same rows in the same sequence order). Stale/unknown handles
+      // return an empty list instead of erroring, matching the "historical
+      // rows survive handle deletion" rule in design doc §3.3.
+      const messages = params.terminal
+        ? db.getAllMessagesForHandle(params.terminal, params.limit)
+        : db.getInbox(params.limit)
       return { messages, count: messages.length }
     }
   }),

src/main/runtime/runtime-rpc.test.ts

@@ -2,9 +2,11 @@
 import { existsSync, mkdtempSync } from 'fs'
 import { tmpdir } from 'os'
 import { join } from 'path'
-import { createConnection } from 'net'
+import { createConnection, type Socket } from 'net'
 import { describe, expect, it, vi } from 'vitest'
+import Database from 'better-sqlite3'
 import { OrcaRuntimeService } from './orca-runtime'
+import { OrchestrationDb } from './orchestration/db'
 import * as runtimeMetadataModule from './runtime-metadata'
 import { readRuntimeMetadata } from './runtime-metadata'
 import { createRuntimeTransportMetadata, OrcaRuntimeRpcServer } from './runtime-rpc'
@@ -46,6 +48,62 @@ async function sendRequest(
   })
 }
 
+// Why: long-poll keepalive tests need every frame, not just the first, because
+// we need to count `_keepalive` frames before the terminal success/failure.
+// Also exposes the socket so tests can close it mid-wait to exercise the
+// long-poll counter decrement path.
+type FramedSession = {
+  socket: Socket
+  frames: Record<string, unknown>[]
+  done: Promise<void>
+}
+
+function openFramedSession(endpoint: string, request: Record<string, unknown>): FramedSession {
+  const frames: Record<string, unknown>[] = []
+  const socket = createConnection(endpoint)
+  let buffer = ''
+  socket.setEncoding('utf8')
+  const done = new Promise<void>((resolve, reject) => {
+    socket.once('error', (err) => {
+      // Why: ECONNRESET is expected when we deliberately destroy the socket
+      // mid-wait to probe the counter decrement; surface other errors.
+      if ((err as NodeJS.ErrnoException).code === 'ECONNRESET') {
+        resolve()
+        return
+      }
+      reject(err)
+    })
+    socket.on('close', () => resolve())
+    socket.on('data', (chunk: string) => {
+      buffer += chunk
+      let newlineIndex = buffer.indexOf('\n')
+      while (newlineIndex !== -1) {
+        const raw = buffer.slice(0, newlineIndex).trim()
+        buffer = buffer.slice(newlineIndex + 1)
+        if (raw) {
+          const frame = JSON.parse(raw) as Record<string, unknown>
+          frames.push(frame)
+          // Why: the server leaves the socket open after writing the terminal
+          // frame (short RPCs expect the client to close); close the client
+          // side so `done` resolves once we've captured the response.
+          if (frame._keepalive !== true) {
+            socket.end()
+          }
+        }
+        newlineIndex = buffer.indexOf('\n')
+      }
+    })
+    socket.on('connect', () => {
+      socket.write(`${JSON.stringify(request)}\n`)
+    })
+  })
+  return { socket, frames, done }
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms))
+}
+
 describe('OrcaRuntimeRpcServer', () => {
   const makeStore = (overrides?: { isUnread?: boolean }) => ({
     getRepo: (id: string) =>
@@ -502,4 +560,209 @@ describe('OrcaRuntimeRpcServer', () => {
 
     await server.stop()
   })
+
+  // Why: §6 tests for the transport keepalive + long-poll counter path in §3.1.
+  // Exercise the real socket (not a mock) so we catch buffer/flush regressions
+  // that a unit-level test would miss.
+  describe('long-poll transport (§3.1)', () => {
+    it('emits keepalive frames while a check --wait handler blocks', async () => {
+      const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
+      const runtime = new OrcaRuntimeService()
+      const db = new OrchestrationDb(':memory:')
+      runtime.setOrchestrationDb(db)
+      // Why: 50ms keepalive lets us collect ≥3 frames within a 300ms wait
+      // window without slowing the suite.
+      const server = new OrcaRuntimeRpcServer({
+        runtime,
+        userDataPath,
+        keepaliveIntervalMs: 50
+      })
+      await server.start()
+
+      try {
+        const metadata = readRuntimeMetadata(userDataPath)
+        const session = openFramedSession(metadata!.transport!.endpoint, {
+          id: 'req_wait',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: {
+            terminal: 'term_nobody',
+            wait: true,
+            timeoutMs: 300
+          }
+        })
+        await session.done
+
+        const keepalives = session.frames.filter((f) => f._keepalive === true)
+        const terminals = session.frames.filter((f) => f.ok !== undefined)
+        expect(terminals).toHaveLength(1)
+        expect(terminals[0]).toMatchObject({ id: 'req_wait', ok: true })
+        // Why: 300ms wait with 50ms keepalive → expect roughly 5 keepalives;
+        // assert ≥3 to tolerate scheduler jitter without flaking.
+        expect(keepalives.length).toBeGreaterThanOrEqual(3)
+      } finally {
+        db.close()
+        await server.stop()
+      }
+    })
+
+    it('releases long-poll slot when client closes mid-wait', async () => {
+      const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
+      const runtime = new OrcaRuntimeService()
+      const db = new OrchestrationDb(':memory:')
+      runtime.setOrchestrationDb(db)
+      const server = new OrcaRuntimeRpcServer({
+        runtime,
+        userDataPath,
+        keepaliveIntervalMs: 1000,
+        longPollCap: 2
+      })
+      await server.start()
+
+      try {
+        const metadata = readRuntimeMetadata(userDataPath)
+        const endpoint = metadata!.transport!.endpoint
+
+        // Fill the cap with two long waits (10s each — we'll kill them).
+        const a = openFramedSession(endpoint, {
+          id: 'req_a',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: { terminal: 'term_a', wait: true, timeoutMs: 10_000 }
+        })
+        const b = openFramedSession(endpoint, {
+          id: 'req_b',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: { terminal: 'term_b', wait: true, timeoutMs: 10_000 }
+        })
+        // Let the two waits land in the handler and increment the counter.
+        await sleep(100)
+        expect(server['activeLongPolls']).toBe(2)
+
+        // Kill one client mid-wait; counter must drop to 1.
+        a.socket.destroy()
+        await a.done
+        // Give Node one tick to fire the close event on the server socket.
+        await sleep(50)
+        expect(server['activeLongPolls']).toBe(1)
+
+        // The freed slot must admit a new long-poll immediately.
+        const c = openFramedSession(endpoint, {
+          id: 'req_c',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: { terminal: 'term_c', wait: true, timeoutMs: 100 }
+        })
+        await c.done
+        const cTerminal = c.frames.find((f) => f.ok !== undefined)
+        expect(cTerminal).toMatchObject({ ok: true, id: 'req_c' })
+
+        b.socket.destroy()
+        await b.done
+      } finally {
+        db.close()
+        await server.stop()
+      }
+    })
+
+    it('responds runtime_busy once the long-poll cap is saturated', async () => {
+      const userDataPath = mkdtempSync(join(tmpdir(), 'orca-runtime-rpc-'))
+      const runtime = new OrcaRuntimeService()
+      const db = new OrchestrationDb(':memory:')
+      runtime.setOrchestrationDb(db)
+      const server = new OrcaRuntimeRpcServer({
+        runtime,
+        userDataPath,
+        keepaliveIntervalMs: 1000,
+        longPollCap: 1
+      })
+      await server.start()
+
+      try {
+        const metadata = readRuntimeMetadata(userDataPath)
+        const endpoint = metadata!.transport!.endpoint
+
+        const a = openFramedSession(endpoint, {
+          id: 'req_a',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: { terminal: 'term_a', wait: true, timeoutMs: 5_000 }
+        })
+        await sleep(100)
+        expect(server['activeLongPolls']).toBe(1)
+
+        // Second long-poll overflows the cap → runtime_busy.
+        const overflow = await sendRequest(endpoint, {
+          id: 'req_overflow',
+          authToken: metadata!.authToken,
+          method: 'orchestration.check',
+          params: { terminal: 'term_b', wait: true, timeoutMs: 5_000 }
+        })
+        expect(overflow).toMatchObject({
+          id: 'req_overflow',
+          ok: false,
+          error: { code: 'runtime_busy' }
+        })
+        // The failing request must not have counted against the cap.
+        expect(server['activeLongPolls']).toBe(1)
+
+        // Short RPCs still succeed even when the long-poll cap is full.
+        const short = await sendRequest(endpoint, {
+          id: 'req_short',
+          authToken: metadata!.authToken,
+          method: 'status.get'
+        })
+        expect(short).toMatchObject({ id: 'req_short', ok: true })
+
+        a.socket.destroy()
+        await a.done
+      } finally {
+        db.close()
+        await server.stop()
+      }
+    })
+  })
+
+  // Why: §6 test for the idempotent + hard-fail schema migration. A broken
+  // migration must crash startup loudly rather than serve traffic against a
+  // schema missing the delivered_at column.
+  describe('orchestration DB migration (§3.2)', () => {
+    it('is idempotent when delivered_at already exists', () => {
+      // First open creates the column; second open should be a no-op.
+      const db1 = new OrchestrationDb(':memory:')
+      db1.close()
+      // File path reuse is meaningless with :memory:, so use a tmp file.
+      const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-')), 'orch.sqlite')
+      const a = new OrchestrationDb(tmpPath)
+      a.close()
+      // Second construction must not throw "duplicate column name".
+      expect(() => {
+        const b = new OrchestrationDb(tmpPath)
+        b.close()
+      }).not.toThrow()
+    })
+
+    it('hard-fails startup when the migration cannot be applied', () => {
+      // Simulate a migration error by monkey-patching better-sqlite3's exec.
+      // If ALTER TABLE throws for any reason (e.g. disk full, permissions),
+      // the constructor must propagate — not swallow and serve half-broken.
+      const tmpPath = join(mkdtempSync(join(tmpdir(), 'orca-orch-mig-')), 'orch.sqlite')
+      const realPrototype = Database.prototype as unknown as {
+        exec: (sql: string) => unknown
+      }
+      const originalExec = realPrototype.exec
+      realPrototype.exec = function (sql: string) {
+        if (sql.includes('ALTER TABLE messages ADD COLUMN delivered_at')) {
+          throw new Error('simulated migration failure')
+        }
+        return originalExec.call(this, sql)
+      }
+      try {
+        expect(() => new OrchestrationDb(tmpPath)).toThrow('simulated migration failure')
+      } finally {
+        realPrototype.exec = originalExec
+      }
+    })
+  })
 })

src/main/runtime/runtime-rpc.ts

@@ -1,3 +1,4 @@
+/* eslint-disable max-lines -- Why: this file is the single security boundary for the bundled CLI — transport setup, auth-token enforcement, admission control, keepalive framing, and orphan-socket sweeping all co-locate deliberately so a reviewer can audit the boundary in one sitting. Splitting this across files would scatter the invariants without reducing complexity. */
 // Why: this is the single security boundary for the bundled CLI. It owns
 // transport setup (unix socket / named pipe), auth-token enforcement, and
 // bootstrap-metadata publication so a running runtime is always discoverable
@@ -19,12 +20,49 @@ type OrcaRuntimeRpcServerOptions = {
   userDataPath: string
   pid?: number
   platform?: NodeJS.Platform
+  // Why: test-only overrides for the two time-bound constants below.
+  // Production callers must not pass these — defaults are set by the design
+  // doc (§3.1) and changing them in production would weaken the admission
+  // fence or flood the socket with keepalive frames.
+  keepaliveIntervalMs?: number
+  longPollCap?: number
 }
 
 const MAX_RUNTIME_RPC_MESSAGE_BYTES = 1024 * 1024
 const RUNTIME_RPC_SOCKET_IDLE_TIMEOUT_MS = 30_000
 const MAX_RUNTIME_RPC_CONNECTIONS = 32
 
+// Why: after 10 s of a pending dispatch we emit a tiny `{"_keepalive":true}`
+// frame every 10 s until the handler resolves. Each write resets both the
+// server's own socket idle timer (30 s) and — once §3.1 ships on the client —
+// the client's idle timer, because any byte counts as socket activity. This
+// is the transport-layer fix for feedback #1: long-poll RPCs (i.e.
+// orchestration.check --wait) can now run past the 30 s/60 s idle caps
+// without either end tearing the socket down. See design doc §3.1.
+const KEEPALIVE_INTERVAL_MS = 10_000
+
+// Why: long-poll slot cap. With keepalives a `check --wait --timeout-ms
+// 600000` can hold a connection for up to 10 minutes; unbounded that would
+// saturate MAX_RUNTIME_RPC_CONNECTIONS (32) with 32 waiting coordinators
+// and lock out normal short RPCs. Capping at half the connection budget
+// leaves the other half for short traffic. On overflow the server responds
+// immediately with `runtime_busy` (CLI exit 75) — fail fast, not silent
+// queuing. See design doc §3.1 + §7 risk #2.
+const LONG_POLL_CAP = 16
+
+// Why: a long-poll request is one whose handler blocks for an unbounded
+// amount of time waiting for an external event (today, only
+// `orchestration.check` with `wait === true`). This function is the single
+// place that classifies it — the long-poll counter, abort wiring, and
+// runtime_busy admission check all share this decision. See §3.1.
+function isLongPollRequest(request: RpcRequest): boolean {
+  if (request.method !== 'orchestration.check') {
+    return false
+  }
+  const params = request.params as { wait?: unknown } | undefined
+  return params?.wait === true
+}
+
 export class OrcaRuntimeRpcServer {
   private readonly runtime: OrcaRuntimeService
   private readonly dispatcher: RpcDispatcher
@@ -32,20 +70,30 @@ export class OrcaRuntimeRpcServer {
   private readonly pid: number
   private readonly platform: NodeJS.Platform
   private readonly authToken = randomBytes(24).toString('hex')
+  private readonly keepaliveIntervalMs: number
+  private readonly longPollCap: number
   private server: Server | null = null
   private transport: RuntimeTransportMetadata | null = null
+  // Why: separate from Node's server.maxConnections because we need to count
+  // only long-running dispatches, not every in-flight short RPC. See §3.1 +
+  // §7 risk #2.
+  private activeLongPolls = 0
 
   constructor({
     runtime,
     userDataPath,
     pid = process.pid,
-    platform = process.platform
+    platform = process.platform,
+    keepaliveIntervalMs = KEEPALIVE_INTERVAL_MS,
+    longPollCap = LONG_POLL_CAP
   }: OrcaRuntimeRpcServerOptions) {
     this.runtime = runtime
     this.dispatcher = new RpcDispatcher({ runtime })
     this.userDataPath = userDataPath
     this.pid = pid
     this.platform = platform
+    this.keepaliveIntervalMs = keepaliveIntervalMs
+    this.longPollCap = longPollCap
   }
 
   async start(): Promise<void> {
@@ -174,37 +222,129 @@ export class OrcaRuntimeRpcServer {
         const rawMessage = buffer.slice(0, newlineIndex).trim()
         buffer = buffer.slice(newlineIndex + 1)
         if (rawMessage) {
-          void this.handleMessage(rawMessage).then((response) => {
-            socket.write(`${JSON.stringify(response)}\n`)
-          })
+          void this.handleRequest(socket, rawMessage)
         }
         newlineIndex = buffer.indexOf('\n')
       }
     })
   }
 
-  private async handleMessage(rawMessage: string): Promise<RpcResponse> {
+  // Why: a single entry point per inbound request so keepalive + long-poll
+  // admission + AbortController wiring + response write all live in one
+  // place. See design doc §3.1.
+  private async handleRequest(socket: Socket, rawMessage: string): Promise<void> {
+    const parsed = this.parseAndAuth(rawMessage)
+    if ('error' in parsed) {
+      this.safeWrite(socket, `${JSON.stringify(parsed.error)}\n`)
+      return
+    }
+    const request = parsed.request
+
+    // Why: long-poll admission fence. Short RPCs bypass the counter entirely
+    // — it only guards handlers that can block for minutes. See §7 risk #2.
+    const longPoll = isLongPollRequest(request)
+    if (longPoll) {
+      if (this.activeLongPolls >= this.longPollCap) {
+        const busy = this.buildError(
+          request.id,
+          'runtime_busy',
+          'long-poll capacity reached; retry with backoff'
+        )
+        this.safeWrite(socket, `${JSON.stringify(busy)}\n`)
+        socket.end()
+        return
+      }
+      this.activeLongPolls += 1
+    }
+
+    // Why: `decremented` must guard against double-decrement when both
+    // `close` and a post-resolve cleanup path fire. `socket.on('close')` is
+    // the only path that fires for every termination (normal end, destroy,
+    // idle timer, client kill -9, OS reset), so it carries the decrement.
+    // Tying it to `.finally` alone would leak a slot any time a client dies
+    // mid-wait because the inner waitForMessage can keep counting down for
+    // minutes after the socket is gone. See §3.1 counter-lifecycle.
+    let decremented = !longPoll
+    const abortController = new AbortController()
+    const onClose = (): void => {
+      if (!decremented) {
+        decremented = true
+        this.activeLongPolls = Math.max(0, this.activeLongPolls - 1)
+      }
+      abortController.abort()
+    }
+    socket.on('close', onClose)
+
+    // Why: for long-poll requests we start a keepalive ticker after 10 s. The
+    // first frame at 10 s resets both the server's 30 s idle timer and the
+    // client's configured timeout. Short RPCs never see a keepalive — the
+    // ticker never fires because the handler resolves first.
+    let keepaliveTimer: NodeJS.Timeout | null = null
+    if (longPoll) {
+      keepaliveTimer = setInterval(() => {
+        if (socket.writable && !socket.destroyed) {
+          socket.write('{"_keepalive":true}\n')
+        }
+      }, this.keepaliveIntervalMs)
+      // Why: don't hold the process open solely on the keepalive interval —
+      // .unref() lets the event loop exit when nothing else is pending.
+      if (typeof keepaliveTimer.unref === 'function') {
+        keepaliveTimer.unref()
+      }
+    }
+
+    try {
+      const response = await this.dispatcher.dispatch(request, {
+        signal: longPoll ? abortController.signal : undefined
+      })
+      if (!socket.destroyed) {
+        this.safeWrite(socket, `${JSON.stringify(response)}\n`)
+      }
+    } finally {
+      if (keepaliveTimer) {
+        clearInterval(keepaliveTimer)
+      }
+      // Why: the close-handler path is still what decrements the counter (the
+      // socket may be closed by the client before the response write flushes).
+      // We don't remove the listener here — `once` semantics are handled by
+      // the boolean guard.
+    }
+  }
+
+  private parseAndAuth(rawMessage: string): { request: RpcRequest } | { error: RpcResponse } {
     let request: RpcRequest
     try {
       request = JSON.parse(rawMessage) as RpcRequest
     } catch {
-      return this.buildError('unknown', 'bad_request', 'Invalid JSON request')
+      return { error: this.buildError('unknown', 'bad_request', 'Invalid JSON request') }
     }
 
     if (typeof request.id !== 'string' || request.id.length === 0) {
-      return this.buildError('unknown', 'bad_request', 'Missing request id')
+      return { error: this.buildError('unknown', 'bad_request', 'Missing request id') }
     }
     if (typeof request.method !== 'string' || request.method.length === 0) {
-      return this.buildError(request.id, 'bad_request', 'Missing RPC method')
+      return { error: this.buildError(request.id, 'bad_request', 'Missing RPC method') }
     }
     if (typeof request.authToken !== 'string' || request.authToken.length === 0) {
-      return this.buildError(request.id, 'unauthorized', 'Missing auth token')
+      return { error: this.buildError(request.id, 'unauthorized', 'Missing auth token') }
     }
     if (request.authToken !== this.authToken) {
-      return this.buildError(request.id, 'unauthorized', 'Invalid auth token')
+      return { error: this.buildError(request.id, 'unauthorized', 'Invalid auth token') }
     }
 
-    return this.dispatcher.dispatch(request)
+    return { request }
+  }
+
+  private safeWrite(socket: Socket, payload: string): void {
+    if (socket.destroyed || !socket.writable) {
+      return
+    }
+    try {
+      socket.write(payload)
+    } catch {
+      // Socket was closed in between the writable check and the write —
+      // nothing we can do; the client already disconnected.
+    }
   }
 
   private buildError(id: string, code: string, message: string): RpcResponse {