feat(connector): inbox with held long poll, two confirmations and expiry, on connector tokens (#33734)

.github/workflows/cloud-verify.yml

@@ -189,6 +189,7 @@ jobs:
       ORCA_BACKGROUND_LAUNCH: '1'
       ORCA_PUSH_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
       ORCA_CONNECTOR_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
+      ORCA_CLOUD_DATABASE_TEST_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
       ORCA_RELAY_TEST_POSTGRES_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
     steps:
       - uses: actions/checkout@v4

cloud/apps/connector/Dockerfile

@@ -4,14 +4,19 @@ ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
 COPY package.json pnpm-lock.yaml pnpm-workspace.yaml tsconfig.base.json ./
 RUN npm install --prefix /opt/pnpm --ignore-scripts --no-audit --no-fund "$(node -p 'require("./package.json").packageManager.split("+")[0]')" \
   && npm audit signatures --prefix /opt/pnpm
+COPY packages/push-contract/package.json packages/push-contract/package.json
+COPY packages/connector-contract/package.json packages/connector-contract/package.json
 COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
 COPY packages/cloud-database/package.json packages/cloud-database/package.json
 COPY apps/connector/package.json apps/connector/package.json
 RUN pnpm install --frozen-lockfile
+COPY packages/push-contract packages/push-contract
+COPY packages/connector-contract packages/connector-contract
 COPY packages/postgres-schema packages/postgres-schema
 COPY packages/cloud-database packages/cloud-database
 COPY apps/connector apps/connector
-RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build \
+RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build \
+  && pnpm --filter @orca-cloud/connector-contract build && pnpm --filter @orca-cloud/cloud-database build \
   && pnpm --filter @orca-cloud/connector build
 
 FROM node:24.22.1-alpine@sha256:feb76650b7195319268c59cb2d419c732d6c499b13bf5296f16ab59eb1f0a4f3 AS runtime
@@ -21,9 +26,13 @@ WORKDIR /app
 COPY --from=build /opt/pnpm /opt/pnpm
 ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
 COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./
+COPY packages/push-contract/package.json packages/push-contract/package.json
+COPY packages/connector-contract/package.json packages/connector-contract/package.json
 COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
 COPY packages/cloud-database/package.json packages/cloud-database/package.json
 COPY apps/connector/package.json apps/connector/package.json
+COPY --from=build /app/packages/push-contract/dist packages/push-contract/dist
+COPY --from=build /app/packages/connector-contract/dist packages/connector-contract/dist
 COPY --from=build /app/packages/postgres-schema/dist packages/postgres-schema/dist
 COPY --from=build /app/packages/cloud-database/dist packages/cloud-database/dist
 COPY --from=build /app/apps/connector/dist apps/connector/dist

cloud/apps/connector/package.json

@@ -9,7 +9,7 @@
     "clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"",
     "dev": "tsx watch src/index.ts",
     "lint": "tsc -p tsconfig.json --noEmit",
-    "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build",
+    "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build && pnpm --filter @orca-cloud/connector-contract build && pnpm --filter @orca-cloud/cloud-database build",
     "start": "node dist/index.js",
     "test": "vitest run",
     "typecheck": "tsc -p tsconfig.json --noEmit"
@@ -17,6 +17,7 @@
   "dependencies": {
     "@hono/node-server": "^2.1.5",
     "@orca-cloud/cloud-database": "workspace:*",
+    "@orca-cloud/connector-contract": "workspace:*",
     "hono": "^4.13.13",
     "jose": "^6.2.13",
     "zod": "^4.6.7"

cloud/apps/connector/src/connector-background.ts

@@ -0,0 +1,58 @@
+// Chains each sweep after the previous one settles, so a slow database never stacks them.
+export function every(label: string, intervalMs: number, run: () => Promise<unknown>): () => void {
+  let timer: NodeJS.Timeout | undefined
+  let stopped = false
+  const schedule = (): void => {
+    if (stopped) {
+      return
+    }
+    timer = setTimeout(() => {
+      void run()
+        .catch((error: unknown) => {
+          console.warn(
+            JSON.stringify({
+              event: 'orca_connector_sweep_failed',
+              target: label,
+              error: error instanceof Error ? error.name : 'unknown'
+            })
+          )
+        })
+        .finally(schedule)
+    }, intervalMs)
+    timer.unref()
+  }
+  schedule()
+  return () => {
+    stopped = true
+    clearTimeout(timer)
+  }
+}
+
+const EXPIRY_SWEEP_INTERVAL_MS = 15_000
+const INBOX_PRUNE_INTERVAL_MS = 60_000
+// Settled rows keep only ids and results; this bounds how long even those stay.
+const INBOX_RETENTION_MS = 7 * 24 * 60 * 60_000
+
+export function startConnectorBackground(runtime: {
+  inbox: {
+    expireDue(): Promise<{ hostId: string; externalId: string }[]>
+    pruneSettled(retentionMs: number): Promise<number>
+  }
+}): () => void {
+  const stops = [
+    every('inbox_expiry', EXPIRY_SWEEP_INTERVAL_MS, async () => {
+      const threads = await runtime.inbox.expireDue()
+      if (threads.length > 0) {
+        console.log(JSON.stringify({ event: 'orca_connector_deliveries_expired', threads: threads.length }))
+      }
+    }),
+    every('inbox_retention', INBOX_PRUNE_INTERVAL_MS, () =>
+      runtime.inbox.pruneSettled(INBOX_RETENTION_MS)
+    )
+  ]
+  return () => {
+    for (const stop of stops) {
+      stop()
+    }
+  }
+}

cloud/apps/connector/src/connector-database.ts

@@ -4,8 +4,7 @@ import {
   type CloudDatabase,
   type OpenCloudDatabaseInput
 } from '@orca-cloud/cloud-database'
-
-const CONNECTOR_SCHEMA_STATEMENTS: string[] = []
+import { connectorSchemaStatements } from './connector-schema.js'
 
 export async function openConnectorDatabase(
   input: Omit<OpenCloudDatabaseInput, 'service' | 'schemaStatements'>
@@ -13,10 +12,10 @@ export async function openConnectorDatabase(
   return await openCloudDatabase({
     ...input,
     service: 'connector',
-    schemaStatements: CONNECTOR_SCHEMA_STATEMENTS
+    schemaStatements: connectorSchemaStatements()
   })
 }
 
 export async function openInMemoryConnectorDatabase(): Promise<CloudDatabase> {
-  return await openInMemoryCloudDatabase(CONNECTOR_SCHEMA_STATEMENTS)
+  return await openInMemoryCloudDatabase(connectorSchemaStatements())
 }

cloud/apps/connector/src/connector-events.test.ts

@@ -0,0 +1,295 @@
+import { CONNECTOR_PATHS, ConnectorPollResponseSchema } from '@orca-cloud/connector-contract'
+import { afterEach, describe, expect, it } from 'vitest'
+import { listenHostname } from './connector-server.js'
+import {
+  cleanupConnectorHarnesses,
+  createConnectorHarness,
+  pauseBefore
+} from './connector-server-harness.test-fixture.js'
+
+afterEach(cleanupConnectorHarnesses)
+
+const conversation = { source: 'slack', externalId: 'T1:C1:1.0:U1' }
+
+// A loopback origin turns on the dev enqueue route, which also creates the thread.
+async function setup(options: Parameters<typeof createConnectorHarness>[1] = {}) {
+  const harness = await createConnectorHarness({ publicUrl: 'http://127.0.0.1:8080' }, options)
+  const machine = await harness.machine()
+  await machine.hello()
+  const enqueue = (hostId: string) =>
+    harness.json('/dev/enqueue', {
+      hostId,
+      externalId: conversation.externalId,
+      delivery: { kind: 'start', agent: 'claude', placeId: 'pl_1', prompt: 'fix it' }
+    })
+  let nextSeq = 0
+  const events = (list: unknown[], seq = nextSeq++, helloRev = 1) =>
+    machine.call(CONNECTOR_PATHS.events, { v: 1, helloRev, seq, events: list })
+  const thread = async () =>
+    (
+      await harness.database.query(
+        `SELECT pane_key, agent, open_turn_id, turn_state, turn_stop_reason,
+           pending_prompt FROM connector_threads`
+      )
+    )[0]
+  return { harness, machine, enqueue, events, thread }
+}
+
+describe('connector turn events', () => {
+  it('queues through the dev route only on a loopback origin', async () => {
+    const remote = await createConnectorHarness()
+    expect((await remote.json('/dev/enqueue', {})).status).toBe(404)
+    const { machine, enqueue } = await setup()
+    const queued = await enqueue(machine.hostId)
+    expect(await queued.json()).toMatchObject({ status: 'queued' })
+    const reply = ConnectorPollResponseSchema.parse((await machine.poll()).body)
+    expect(reply.deliveries[0]).toMatchObject({ kind: 'start', conversation })
+  })
+
+  it('listens only on loopback when the dev route is on', async () => {
+    expect(listenHostname('http://127.0.0.1:8080')).toBe('127.0.0.1')
+    expect(listenHostname('http://[::1]:8080')).toBe('::1')
+    expect(listenHostname('https://connector.onorca.dev')).toBeUndefined()
+    const { harness } = await setup()
+    const { server } = harness.runtime
+    await new Promise<void>((resolve) => server.listen(0, listenHostname('http://127.0.0.1:8080'), resolve))
+    expect(server.address()).toMatchObject({ address: '127.0.0.1' })
+    await new Promise((resolve) => server.close(resolve))
+  })
+
+  it('applies turn events once, whatever order and repeats they arrive in', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const linked = { conversation, paneKey: 'pane-1', kind: 'linked', agent: 'claude' }
+    const started = { conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }
+    const ended = {
+      conversation,
+      paneKey: 'pane-1',
+      kind: 'turn-ended',
+      turnId: 100,
+      at: 160,
+      stopReason: 'end_turn',
+      lastAssistantMessage: 'done'
+    }
+    expect((await events([linked, started, ended])).status).toBe(204)
+    const settled = {
+      pane_key: 'pane-1',
+      agent: 'claude',
+      open_turn_id: 100,
+      turn_state: 'ended',
+      turn_stop_reason: 'end_turn',
+      pending_prompt: null
+    }
+    const row = await thread()
+    // Postgres returns BIGINT as text.
+    expect({ ...row, open_turn_id: Number(row?.open_turn_id) }).toEqual(settled)
+    // A resent start for the same turn changes nothing.
+    await events([started, ended, started])
+    expect(await thread()).toMatchObject({ turn_state: 'ended', turn_stop_reason: 'end_turn' })
+  })
+
+  it('reads an open turn the snapshot no longer shows as unverifiable, never as done', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }])
+    const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] }
+    await events([
+      {
+        kind: 'snapshot',
+        conversations: [{ conversation, paneKey: 'pane-1', agent: 'claude', working: false, prompt }]
+      }
+    ])
+    expect(await thread()).toMatchObject({
+      turn_state: 'unverifiable',
+      pending_prompt: JSON.stringify(prompt)
+    })
+  })
+
+  it('keeps a replayed turn end of an old pane off the pane a snapshot rebound', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await events([{ conversation, paneKey: 'old-pane', kind: 'turn-started', at: 100 }])
+    await events([
+      {
+        kind: 'snapshot',
+        conversations: [
+          { conversation, paneKey: 'new-pane', agent: 'claude', working: false, prompt: null }
+        ]
+      }
+    ])
+    const oldEnd = { conversation, paneKey: 'old-pane', kind: 'turn-ended', turnId: 100, at: 160 }
+    expect((await events([{ ...oldEnd, stopReason: 'end_turn' }])).status).toBe(204)
+    expect(await thread()).toMatchObject({ pane_key: 'new-pane', turn_state: 'unverifiable' })
+    // The new pane's own turn applies, whatever its start time against the old pane's.
+    await events([{ conversation, paneKey: 'new-pane', kind: 'turn-started', at: 90 }])
+    expect(await thread()).toMatchObject({ pane_key: 'new-pane', turn_state: 'working' })
+  })
+
+  it("drops the old pane's prompt when a conversation links to another pane", async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] }
+    await events([{ conversation, paneKey: 'old-pane', kind: 'prompt', prompt }])
+    await events([{ conversation, paneKey: 'old-pane', kind: 'linked', agent: 'claude' }])
+    expect(await thread()).toMatchObject({ pending_prompt: JSON.stringify(prompt) })
+    await events([{ conversation, paneKey: 'new-pane', kind: 'linked', agent: 'claude' }])
+    expect(await thread()).toMatchObject({ pane_key: 'new-pane', pending_prompt: null })
+  })
+
+  it("drops the old pane's completed turn when a conversation moves to another pane", async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const ended = { conversation, paneKey: 'old-pane', kind: 'turn-ended', turnId: 100, at: 160 }
+    await events([{ conversation, paneKey: 'old-pane', kind: 'turn-started', at: 100 }])
+    await events([{ ...ended, stopReason: 'end_turn' }])
+    await events([{ conversation, paneKey: 'old-pane', kind: 'linked', agent: 'claude' }])
+    expect(await thread()).toMatchObject({ turn_state: 'ended', turn_stop_reason: 'end_turn' })
+    await events([{ conversation, paneKey: 'new-pane', kind: 'linked', agent: 'claude' }])
+    const cleared = { open_turn_id: null, turn_state: null, turn_stop_reason: null }
+    expect(await thread()).toMatchObject({ pane_key: 'new-pane', ...cleared })
+    await events([{ conversation, paneKey: 'new-pane', kind: 'turn-started', at: 200 }])
+    await events([
+      {
+        kind: 'snapshot',
+        conversations: [
+          { conversation, paneKey: 'newer-pane', agent: 'claude', working: true, prompt: null }
+        ]
+      }
+    ])
+    // The snapshot moved it off a working turn: unverifiable, never the old verdict.
+    expect(await thread()).toMatchObject({
+      pane_key: 'newer-pane',
+      open_turn_id: null,
+      turn_state: 'unverifiable',
+      turn_stop_reason: null
+    })
+  })
+
+  it('clears the prompt of a conversation the snapshot no longer lists', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] }
+    await events([{ conversation, paneKey: 'pane-1', kind: 'prompt', prompt }])
+    await events([{ kind: 'snapshot', conversations: [] }])
+    expect(await thread()).toMatchObject({ pending_prompt: null })
+  })
+
+  it('reads an open turn missing from the snapshot as unverifiable', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }])
+    await events([{ kind: 'snapshot', conversations: [] }])
+    expect(await thread()).toMatchObject({ turn_state: 'unverifiable' })
+  })
+
+  it("drops a batch at or below the host's high-water mark", async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] }
+    const promptEvent = { conversation, paneKey: 'pane-1', kind: 'prompt', prompt }
+    const ended = {
+      conversation,
+      paneKey: 'pane-1',
+      kind: 'turn-ended',
+      turnId: 100,
+      at: 160,
+      stopReason: 'end_turn'
+    }
+    // Batch 1 (the prompt) lands late, after batch 2 cleared it.
+    expect((await events([ended], 2)).status).toBe(204)
+    expect((await events([promptEvent], 1)).status).toBe(204)
+    expect(await thread()).toMatchObject({ pending_prompt: null, turn_state: 'ended' })
+    // A retry of the applied batch changes nothing either.
+    await events([promptEvent], 2)
+    expect(await thread()).toMatchObject({ pending_prompt: null })
+  })
+
+  it('keeps the high-water mark across tokens for the same machine', async () => {
+    const { harness, machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await events([{ conversation, paneKey: 'pane-1', kind: 'turn-started', at: 100 }], 5)
+    const refreshed = await harness.machine('user-1', { machineId: machine.machineId })
+    const batch = (seq: number, paneKey: string) =>
+      refreshed.call(CONNECTOR_PATHS.events, {
+        v: 1,
+        helloRev: 1,
+        seq,
+        events: [{ conversation, paneKey, kind: 'linked', agent: 'claude' }]
+      })
+    expect((await batch(5, 'pane-2')).status).toBe(204)
+    expect(await thread()).toMatchObject({ pane_key: 'pane-1' })
+    await batch(6, 'pane-3')
+    expect(await thread()).toMatchObject({ pane_key: 'pane-3' })
+  })
+
+  it('applies events only for the current hello and restarts the mark with a new one', async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    const linked = (paneKey: string) => [{ conversation, paneKey, kind: 'linked', agent: 'claude' }]
+    await events(linked('pane-1'), 5)
+    expect((await events(linked('pane-x'), 0, 2)).status).toBe(409)
+    await machine.hello(2)
+    // The new hello's batch 0 applies; a late batch from the old hello changes nothing.
+    expect((await events(linked('pane-2'), 0, 2)).status).toBe(204)
+    expect((await events(linked('pane-old'), 6, 1)).status).toBe(204)
+    expect(await thread()).toMatchObject({ pane_key: 'pane-2' })
+    // A repeated hello for the same revision keeps the mark.
+    await machine.hello(2)
+    await events(linked('pane-retry'), 0, 2)
+    expect(await thread()).toMatchObject({ pane_key: 'pane-2' })
+  })
+
+  it('keeps the newer hello when an older one arrives late', async () => {
+    const { harness, machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await machine.hello(2)
+    const linked = (paneKey: string) => [{ conversation, paneKey, kind: 'linked', agent: 'claude' }]
+    await events(linked('new-pane'), 3, 2)
+    expect((await machine.hello(1)).status).toBe(204)
+    expect((await events(linked('old-pane'), 0, 1)).status).toBe(204)
+    expect(await thread()).toMatchObject({ pane_key: 'new-pane' })
+    const [host] = await harness.database.query('SELECT hello_rev, events_seq FROM connector_hosts')
+    expect({ helloRev: Number(host?.hello_rev), eventsSeq: Number(host?.events_seq) }).toEqual({
+      helloRev: 2,
+      eventsSeq: 3
+    })
+    expect((await machine.poll(2)).status).toBe(200)
+  })
+
+  // SQLite runs one transaction at a time, so only Postgres can race a batch with a hello.
+  it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)(
+    'ignores a batch from the old hello that waited on a newer hello',
+    async () => {
+      let gate!: ReturnType<typeof pauseBefore>
+      const { machine, enqueue, events, thread } = await setup({
+        wrapDatabase: (database) => {
+          gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1)
+          return gate.database
+        }
+      })
+      await enqueue(machine.hostId)
+      const hello = machine.hello(2)
+      await gate.atStatement
+      const late = events([{ conversation, paneKey: 'pane-old', kind: 'linked', agent: 'claude' }], 0, 1)
+      await new Promise((resolve) => setTimeout(resolve, 200))
+      gate.release()
+      expect((await hello).status).toBe(204)
+      expect((await late).status).toBe(204)
+      expect(await thread()).toMatchObject({ pane_key: null })
+    }
+  )
+
+  it("ignores events for another host's thread", async () => {
+    const { harness, machine, enqueue, thread } = await setup()
+    await enqueue(machine.hostId)
+    const other = await harness.machine('user-1')
+    await other.hello()
+    await other.call(CONNECTOR_PATHS.events, {
+      v: 1,
+      helloRev: 1,
+      seq: 0,
+      events: [{ conversation, paneKey: 'pane-9', kind: 'turn-started', at: 100 }]
+    })
+    expect(await thread()).toMatchObject({ pane_key: null, turn_state: null })
+  })
+})

cloud/apps/connector/src/connector-hosts.ts

@@ -0,0 +1,114 @@
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import type { ConnectorHelloRequest } from '@orca-cloud/connector-contract'
+import type { ConnectorInbox } from './connector-inbox.js'
+
+// Coarse on purpose: last-seen is written at most once a minute per host.
+const LAST_SEEN_RESOLUTION_MS = 60_000
+
+// Every machine request and queueing take this, so a newer hello lands wholly before or after each.
+export async function lockConnectorHost(transaction: CloudDatabase, hostId: string): Promise<void> {
+  await transaction.lockQuotaScope(`orca-connector-host:${hostId}`)
+}
+
+// A host is one machine of one account, (sub, mid) from its token; its first hello records it.
+export class ConnectorHosts {
+  constructor(
+    private readonly database: CloudDatabase,
+    private readonly inbox: ConnectorInbox,
+    private readonly now: () => number
+  ) {}
+
+  // Under the host lock. The highest revision wins (a delayed older hello changes nothing); held polls
+  // wake to find theirs stale, the next poll lists what was received but never settled, and a new
+  // revision restarts the events mark.
+  async recordHello(
+    transaction: CloudDatabase,
+    host: { hostId: string; accountId: string },
+    hello: ConnectorHelloRequest
+  ): Promise<void> {
+    const now = this.now()
+    const [recorded] = await transaction.query(
+      `INSERT INTO connector_hosts
+       (host_id, account_id, host_name, platform, runtime, orca_version, hello_rev, hello,
+        unsettled_due, first_seen_at, last_seen_at)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)
+       ON CONFLICT (host_id) DO UPDATE SET host_name = excluded.host_name,
+         platform = excluded.platform, runtime = excluded.runtime,
+         orca_version = excluded.orca_version, hello_rev = excluded.hello_rev,
+         hello = excluded.hello, unsettled_due = 1, last_seen_at = excluded.last_seen_at,
+         events_seq = CASE WHEN connector_hosts.hello_rev = excluded.hello_rev
+           THEN connector_hosts.events_seq END
+       WHERE connector_hosts.hello_rev <= excluded.hello_rev`,
+      [
+        host.hostId,
+        host.accountId,
+        hello.hostName,
+        hello.platform,
+        hello.runtime,
+        hello.orcaVersion,
+        hello.helloRev,
+        JSON.stringify(hello),
+        now,
+        now
+      ]
+    )
+    if (Number(recorded?.changes ?? 0) > 0) {
+      await this.inbox.wake(transaction, host.hostId)
+    }
+  }
+
+  // Under the host lock: a request's hello is the machine's current one, an older one, or one the
+  // cloud has no record of (the machine must send it).
+  async helloStatus(
+    transaction: CloudDatabase,
+    hostId: string,
+    helloRev: number
+  ): Promise<'current' | 'older' | 'unknown'> {
+    const [row] = await transaction.query(
+      'SELECT hello_rev FROM connector_hosts WHERE host_id = ?',
+      [hostId]
+    )
+    const current = row ? Number(row.hello_rev) : null
+    if (current === helloRev) {
+      return 'current'
+    }
+    return current !== null && helloRev < current ? 'older' : 'unknown'
+  }
+
+  // Under the host lock: true once per hello, for the poll that lists unsettled deliveries.
+  async takeUnsettledDue(transaction: CloudDatabase, hostId: string): Promise<boolean> {
+    const [taken] = await transaction.query(
+      'UPDATE connector_hosts SET unsettled_due = 0 WHERE host_id = ? AND unsettled_due = 1',
+      [hostId]
+    )
+    return Number(taken?.changes ?? 0) > 0
+  }
+
+  async accountOf(transaction: CloudDatabase, hostId: string): Promise<string | null> {
+    const [row] = await transaction.query(
+      'SELECT account_id FROM connector_hosts WHERE host_id = ?',
+      [hostId]
+    )
+    return row ? String(row.account_id) : null
+  }
+
+  // Under the host lock, for the current hello: events apply only above its high-water mark, so a
+  // late original never undoes newer state.
+  async advanceEventsSeq(transaction: CloudDatabase, hostId: string, seq: number): Promise<boolean> {
+    const [advanced] = await transaction.query(
+      `UPDATE connector_hosts SET events_seq = ?
+       WHERE host_id = ? AND (events_seq IS NULL OR events_seq < ?)`,
+      [seq, hostId, seq]
+    )
+    return Number(advanced?.changes ?? 0) > 0
+  }
+
+  async touch(hostId: string): Promise<void> {
+    const now = this.now()
+    await this.database.query(
+      `UPDATE connector_hosts SET last_seen_at = ?
+       WHERE host_id = ? AND last_seen_at < ?`,
+      [now, hostId, now - LAST_SEEN_RESOLUTION_MS]
+    )
+  }
+}

cloud/apps/connector/src/connector-inbox.test.ts

@@ -0,0 +1,432 @@
+import { CONNECTOR_LIMITS, ConnectorPollResponseSchema } from '@orca-cloud/connector-contract'
+import { afterEach, describe, expect, it } from 'vitest'
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import type { DeliveryBody } from './connector-inbox.js'
+import {
+  cleanupConnectorHarnesses,
+  createConnectorHarness,
+  pauseBefore
+} from './connector-server-harness.test-fixture.js'
+
+afterEach(cleanupConnectorHarnesses)
+
+const START: DeliveryBody = { kind: 'start', agent: 'claude', placeId: 'pl_1', prompt: 'fix it' }
+const SEND: DeliveryBody = { kind: 'send', text: 'and the tests' }
+
+// Fails the first statement starting with `prefix` once armed, as a transient database fault.
+function failOnce(database: CloudDatabase, prefix: string) {
+  let armed = false
+  const wrap = (inner: CloudDatabase): CloudDatabase => ({
+    dialect: inner.dialect,
+    query: async (sql, params) => {
+      if (armed && sql.trimStart().startsWith(prefix)) {
+        armed = false
+        throw new Error('injected')
+      }
+      return await inner.query(sql, params)
+    },
+    transaction: (run) => inner.transaction((transaction) => run(wrap(transaction))),
+    lockQuotaScope: (key) => inner.lockQuotaScope(key),
+    tryLockScope: (key) => inner.tryLockScope(key),
+    tryLockSharedScope: (key) => inner.tryLockSharedScope(key),
+    close: () => inner.close()
+  })
+  return { database: wrap(database), arm: () => (armed = true) }
+}
+
+async function setup(
+  options: Parameters<typeof createConnectorHarness>[1] = {},
+  machineOptions: Parameters<Awaited<ReturnType<typeof createConnectorHarness>>['machine']>[1] = {}
+) {
+  const harness = await createConnectorHarness({}, options)
+  const machine = await harness.machine('user-1', machineOptions)
+  expect((await machine.hello()).status).toBe(204)
+  const enqueue = (externalId: string, deliveryId: string, body: DeliveryBody) =>
+    harness.runtime.queueDelivery({
+      thread: {
+        externalId,
+        teamId: 'T1',
+        channelId: 'C1',
+        threadTs: externalId,
+        ownerUserId: 'U1',
+        hostId: machine.hostId,
+        placeId: 'pl_1'
+      },
+      deliveryId,
+      body
+    })
+  const poll = async () => {
+    const { status, body } = await machine.poll()
+    expect(status).toBe(200)
+    return ConnectorPollResponseSchema.parse(body)
+  }
+  const ids = async () => (await poll()).deliveries.map((delivery) => delivery.deliveryId)
+  return { harness, machine, enqueue, poll, ids }
+}
+
+describe('connector inbox', () => {
+  it('refuses a request without a connector token', async () => {
+    const harness = await createConnectorHarness()
+    expect((await harness.json('/v1/connector/poll', { v: 1, helloRev: 1 })).status).toBe(401)
+  })
+
+  it('names a host by account and machine, whichever token it presents', async () => {
+    const { harness, machine, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    const sameMachine = await harness.machine('user-1', { machineId: machine.machineId })
+    expect(ConnectorPollResponseSchema.parse((await sameMachine.poll()).body).deliveries).toHaveLength(1)
+    const otherAccount = await harness.machine('user-2', { machineId: machine.machineId })
+    expect((await otherAccount.poll()).status).toBe(409)
+    expect(await ids()).toEqual([])
+  })
+
+  it('requires a hello for the revision a poll names', async () => {
+    const harness = await createConnectorHarness()
+    const machine = await harness.machine()
+    expect((await machine.poll()).status).toBe(409)
+    await machine.hello(1)
+    expect((await machine.poll(2)).status).toBe(409)
+    expect((await machine.poll(1)).status).toBe(200)
+  })
+
+  it('hands out a queued delivery with its envelope', async () => {
+    const { harness, enqueue, poll } = await setup()
+    await expect(enqueue('T1:C1:1.0:U1', 'd1', START)).resolves.toBe('queued')
+    const reply = await poll()
+    expect(reply.holdMs).toBe(0)
+    expect(reply.serverNow).toBe(harness.now())
+    expect(reply.deliveries).toEqual([
+      {
+        ...START,
+        deliveryId: 'd1',
+        seq: 1,
+        expiresAt: harness.now() + CONNECTOR_LIMITS.deliveryTtlMs,
+        conversation: { source: 'slack', externalId: 'T1:C1:1.0:U1' }
+      }
+    ])
+    await expect(enqueue('T1:C1:1.0:U1', 'd1', START)).resolves.toBe('duplicate')
+  })
+
+  it('wakes a held poll as soon as a delivery is queued', async () => {
+    const { enqueue, poll } = await setup({ pollHoldMs: 10_000 })
+    const held = poll()
+    await new Promise((resolve) => setTimeout(resolve, 100))
+    const startedAt = Date.now()
+    await enqueue('T1:C1:1.0:U1', 'd1', START)
+    expect((await held).deliveries.map((delivery) => delivery.deliveryId)).toEqual(['d1'])
+    expect(Date.now() - startedAt).toBeLessThan(5_000)
+  })
+
+  it('sends one delivery per conversation until it is received', async () => {
+    const { machine, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    await enqueue('a', 'a2', SEND)
+    await enqueue('b', 'b1', START)
+    expect(await ids()).toEqual(['a1', 'b1'])
+    expect(await ids()).toEqual([])
+    await machine.ack([{ deliveryId: 'a1', step: 'received' }])
+    expect(await ids()).toEqual(['a2'])
+  })
+
+  it('resends a delivery with the same id once its lease runs out unreceived', async () => {
+    const { harness, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    expect(await ids()).toEqual(['a1'])
+    harness.advance(CONNECTOR_LIMITS.leaseMs - 1)
+    expect(await ids()).toEqual([])
+    harness.advance(1)
+    expect(await ids()).toEqual(['a1'])
+  })
+
+  it('lists received-but-not-done deliveries on the first poll after each hello only', async () => {
+    const { machine, enqueue, poll } = await setup()
+    await enqueue('a', 'a1', START)
+    await enqueue('b', 'b1', START)
+    await poll()
+    await machine.ack([
+      { deliveryId: 'a1', step: 'received' },
+      { deliveryId: 'b1', step: 'received' },
+      { deliveryId: 'b1', step: 'done', result: { outcome: 'accepted' } }
+    ])
+    expect((await poll()).unsettled).toBeUndefined()
+
+    // The machine restarts: a new hello, then its first poll.
+    await machine.hello(2)
+    const first = ConnectorPollResponseSchema.parse((await machine.poll(2)).body)
+    expect(first.unsettled).toEqual([
+      { deliveryId: 'a1', conversation: { source: 'slack', externalId: 'a' } }
+    ])
+    const second = ConnectorPollResponseSchema.parse((await machine.poll(2)).body)
+    expect(second.unsettled).toBeUndefined()
+  })
+
+  // SQLite runs one transaction at a time, so only Postgres can interleave the sweep with a claim.
+  it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)(
+    'keeps an expiry that lands while a resend is being claimed',
+    async () => {
+      let gate!: ReturnType<typeof pauseBefore>
+      const { harness, enqueue, poll } = await setup({
+        wrapDatabase: (database) => {
+          gate = pauseBefore(database, "UPDATE connector_inbox SET state = 'leased'", 1)
+          return gate.database
+        }
+      })
+      await enqueue('a', 'a1', START)
+      await poll()
+      harness.advance(CONNECTOR_LIMITS.leaseMs)
+      const resend = poll()
+      await gate.atStatement
+      // The sweep takes no host lock, so it can expire the row between the claim's read and write.
+      harness.advance(CONNECTOR_LIMITS.deliveryTtlMs)
+      expect(await harness.runtime.inbox.expireDue()).toHaveLength(1)
+      gate.release()
+      expect((await resend).deliveries).toEqual([])
+      const [row] = await harness.database.query(
+        "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'"
+      )
+      expect(row?.state).toBe('expired')
+    }
+  )
+
+  it('keeps recovery for the next poll when the first one fails', async () => {
+    let fault!: ReturnType<typeof failOnce>
+    const { machine, enqueue, poll } = await setup({
+      wrapDatabase: (database) => {
+        fault = failOnce(database, 'SELECT delivery_id, external_id, seq')
+        return fault.database
+      }
+    })
+    await enqueue('a', 'a1', START)
+    await poll()
+    await machine.ack([{ deliveryId: 'a1', step: 'received' }])
+    await machine.hello(2)
+    fault.arm()
+    expect((await machine.poll(2)).status).toBe(500)
+    const retry = ConnectorPollResponseSchema.parse((await machine.poll(2)).body)
+    expect(retry.unsettled).toEqual([
+      { deliveryId: 'a1', conversation: { source: 'slack', externalId: 'a' } }
+    ])
+  })
+
+  it('sends a poll held across a newer hello back to hello', async () => {
+    const { machine } = await setup({ pollHoldMs: 10_000 })
+    const held = machine.poll(1)
+    await new Promise((resolve) => setTimeout(resolve, 100))
+    const startedAt = Date.now()
+    expect((await machine.hello(2)).status).toBe(204)
+    expect((await held).status).toBe(409)
+    expect(Date.now() - startedAt).toBeLessThan(5_000)
+  })
+
+  it('ignores acks from an older hello and asks for an unrecorded one', async () => {
+    const { harness, machine, enqueue, poll } = await setup()
+    await enqueue('a', 'a1', START)
+    await poll()
+    expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }], 2)).status).toBe(409)
+    await machine.hello(2)
+    expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }], 1)).status).toBe(204)
+    const [row] = await harness.database.query(
+      "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'"
+    )
+    expect(row?.state).toBe('leased')
+  })
+
+  // SQLite runs one transaction at a time, so only Postgres can race an ack with a hello.
+  it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)(
+    'ignores an ack from the old hello that waited on a newer hello; the lease resends it',
+    async () => {
+      let gate!: ReturnType<typeof pauseBefore>
+      const { harness, machine, enqueue, poll } = await setup({
+        wrapDatabase: (database) => {
+          gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1)
+          return gate.database
+        }
+      })
+      await enqueue('a', 'a1', START)
+      await poll()
+      const hello = machine.hello(2)
+      await gate.atStatement
+      const late = machine.ack([{ deliveryId: 'a1', step: 'received' }], 1)
+      await new Promise((resolve) => setTimeout(resolve, 200))
+      gate.release()
+      expect((await hello).status).toBe(204)
+      expect((await late).status).toBe(204)
+      harness.advance(CONNECTOR_LIMITS.leaseMs)
+      const first = ConnectorPollResponseSchema.parse((await machine.poll(2)).body)
+      expect(first.deliveries.map((delivery) => delivery.deliveryId)).toEqual(['a1'])
+    }
+  )
+
+  // SQLite runs one transaction at a time, so only Postgres can race two hellos.
+  it.skipIf(!process.env.ORCA_CONNECTOR_TEST_DATABASE_URL)(
+    'keeps the newer hello when an older one waited on it',
+    async () => {
+      let gate!: ReturnType<typeof pauseBefore>
+      const { harness, machine } = await setup({
+        wrapDatabase: (database) => {
+          gate = pauseBefore(database, 'INSERT INTO connector_hosts', 1)
+          return gate.database
+        }
+      })
+      const newer = machine.hello(3)
+      await gate.atStatement
+      const older = machine.hello(2)
+      await new Promise((resolve) => setTimeout(resolve, 200))
+      gate.release()
+      expect([(await newer).status, (await older).status]).toEqual([204, 204])
+      const [host] = await harness.database.query('SELECT hello_rev FROM connector_hosts')
+      expect(Number(host?.hello_rev)).toBe(3)
+      expect((await machine.poll(3)).status).toBe(200)
+    }
+  )
+
+  it('keeps the first done', async () => {
+    const { harness, machine, enqueue, poll } = await setup()
+    await enqueue('a', 'a1', START)
+    await poll()
+    await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'accepted' } }])
+    await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'unsupported' } }])
+    const [row] = await harness.database.query(
+      "SELECT state, result, payload FROM connector_inbox WHERE delivery_id = 'a1'"
+    )
+    expect(row).toEqual({ state: 'done', result: '{"outcome":"accepted"}', payload: null })
+  })
+
+  it('drops a start the machine never received when it is cancelled', async () => {
+    const { enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    await enqueue('a', 'a2', SEND)
+    await expect(enqueue('a', 'a3', { kind: 'cancel' })).resolves.toBe('dropped')
+    await expect(enqueue('a', 'a3', { kind: 'cancel' })).resolves.toBe('duplicate')
+    expect(await ids()).toEqual([])
+  })
+
+  it('lets a cancel skip the queue once its start is received', async () => {
+    const { machine, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    expect(await ids()).toEqual(['a1'])
+    await machine.ack([{ deliveryId: 'a1', step: 'received' }])
+    await enqueue('a', 'a2', SEND)
+    expect(await ids()).toEqual(['a2'])
+    await enqueue('a', 'a3', SEND)
+    await expect(enqueue('a', 'a4', { kind: 'cancel' })).resolves.toBe('queued')
+    expect(await ids()).toEqual(['a4'])
+  })
+
+  it('still lets a cancel skip the queue after its settled start is pruned', async () => {
+    const { harness, machine, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    await ids()
+    await machine.ack([{ deliveryId: 'a1', step: 'done', result: { outcome: 'accepted' } }])
+    // What retention leaves of a long conversation: its settled start is gone.
+    await harness.database.query("DELETE FROM connector_inbox WHERE delivery_id = 'a1'")
+    await enqueue('a', 'a2', SEND)
+    expect(await ids()).toEqual(['a2'])
+    await enqueue('a', 'a3', { kind: 'cancel' })
+    expect(await ids()).toEqual(['a3'])
+  })
+
+  it('queues nothing for a machine that never said hello', async () => {
+    const { harness, enqueue } = await setup()
+    const silent = await harness.machine('user-1')
+    await expect(
+      harness.runtime.queueDelivery({
+        thread: {
+          externalId: 'a',
+          teamId: 'T1',
+          channelId: 'C1',
+          threadTs: 'a',
+          ownerUserId: 'U1',
+          hostId: silent.hostId,
+          placeId: 'pl_1'
+        },
+        deliveryId: 'a1',
+        body: START
+      })
+    ).resolves.toBe('unknown_host')
+    await expect(enqueue('b', 'b1', START)).resolves.toBe('queued')
+  })
+
+  it('expires pending work at its deadline and leased work 30 s after it', async () => {
+    const { harness, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    expect(await ids()).toEqual(['a1'])
+    await enqueue('b', 'b1', START)
+    harness.advance(CONNECTOR_LIMITS.deliveryTtlMs)
+    expect(await harness.runtime.inbox.expireDue()).toEqual([
+      { hostId: expect.any(String), externalId: 'b' }
+    ])
+    expect(await ids()).toEqual([])
+    harness.advance(CONNECTOR_LIMITS.leasedExpiryGraceMs)
+    expect(await harness.runtime.inbox.expireDue()).toEqual([
+      { hostId: expect.any(String), externalId: 'a' }
+    ])
+  })
+
+  it('accepts a late received for expired work', async () => {
+    const { harness, machine, enqueue, ids } = await setup()
+    await enqueue('a', 'a1', START)
+    await ids()
+    harness.advance(CONNECTOR_LIMITS.deliveryTtlMs + CONNECTOR_LIMITS.leasedExpiryGraceMs)
+    await harness.runtime.inbox.expireDue()
+    expect((await machine.ack([{ deliveryId: 'a1', step: 'received' }])).status).toBe(204)
+    const [row] = await harness.database.query(
+      "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'"
+    )
+    expect(row?.state).toBe('received')
+  })
+
+  it('ends a held poll by its token expiry', async () => {
+    const { machine, poll } = await setup({ pollHoldMs: 10_000 }, { ttlSeconds: 2 })
+    const reply = await poll()
+    expect(reply.deliveries).toEqual([])
+    expect(Date.now()).toBeGreaterThanOrEqual(machine.expiresAt)
+    expect(Date.now() - machine.expiresAt).toBeLessThan(1_000)
+  })
+
+  it('rolls back a poll pass whose token expired while it ran', async () => {
+    let gate!: ReturnType<typeof pauseBefore>
+    const { harness, machine, enqueue } = await setup(
+      {
+        wrapDatabase: (database) => {
+          gate = pauseBefore(database, 'SELECT delivery_id, external_id, seq')
+          return gate.database
+        }
+      },
+      { ttlSeconds: 2 }
+    )
+    await enqueue('a', 'a1', START)
+    const reply = machine.poll()
+    await gate.atStatement
+    await new Promise((resolve) => setTimeout(resolve, machine.expiresAt - Date.now() + 10))
+    gate.release()
+    expect(ConnectorPollResponseSchema.parse((await reply).body).deliveries).toEqual([])
+    const [row] = await harness.database.query(
+      "SELECT state FROM connector_inbox WHERE delivery_id = 'a1'"
+    )
+    expect(row?.state).toBe('pending')
+    const [host] = await harness.database.query('SELECT unsettled_due FROM connector_hosts')
+    expect(Number(host?.unsettled_due)).toBe(1)
+  })
+
+  it('wakes a held poll when a lease runs out', async () => {
+    const { harness, enqueue, ids, poll } = await setup({ pollHoldMs: 10_000 })
+    await enqueue('a', 'a1', START)
+    expect(await ids()).toEqual(['a1'])
+    harness.advance(CONNECTOR_LIMITS.leaseMs - 100)
+    const startedAt = Date.now()
+    const held = poll()
+    setTimeout(() => harness.advance(100), 50)
+    expect((await held).deliveries.map((delivery) => delivery.deliveryId)).toEqual(['a1'])
+    expect(Date.now() - startedAt).toBeLessThan(5_000)
+  })
+
+  it('answers held polls at once with a reconnect when the instance drains', async () => {
+    const { harness, machine } = await setup({ pollHoldMs: 10_000 })
+    const held = machine.poll()
+    await new Promise((resolve) => setTimeout(resolve, 100))
+    harness.runtime.waker.drain()
+    const reply = ConnectorPollResponseSchema.parse((await held).body)
+    expect(reply.control).toEqual({ reconnect: true })
+  })
+})

cloud/apps/connector/src/connector-inbox.ts

@@ -0,0 +1,209 @@
+import type { CloudDatabase, CloudNotifications } from '@orca-cloud/cloud-database'
+import {
+  CONNECTOR_LIMITS,
+  ConnectorDeliverySchema,
+  type ConnectorAcksRequest,
+  type ConnectorDelivery
+} from '@orca-cloud/connector-contract'
+
+// NOTIFY channel; the payload is the host id whose inbox changed.
+export const INBOX_CHANNEL = 'connector_inbox'
+
+type DeliveryEnvelope = 'deliveryId' | 'seq' | 'expiresAt' | 'conversation'
+// The kind-specific part of a delivery; the inbox assigns the envelope.
+export type DeliveryBody = ConnectorDelivery extends infer D
+  ? D extends ConnectorDelivery
+    ? Omit<D, DeliveryEnvelope>
+    : never
+  : never
+
+export type EnqueueResult = 'queued' | 'duplicate' | 'dropped'
+export type Unsettled = { deliveryId: string; conversation: { source: 'slack'; externalId: string } }
+
+const conversation = (externalId: string) => ({ source: 'slack' as const, externalId })
+
+// States: pending -> leased -> received -> done, or expired / dropped. Text leaves with settlement.
+export class ConnectorInbox {
+  constructor(
+    private readonly database: CloudDatabase,
+    private readonly notifications: CloudNotifications,
+    private readonly now: () => number
+  ) {}
+
+  // Under the host lock, after the caller found the host and bound the thread (queueDelivery).
+  async enqueue(
+    transaction: CloudDatabase,
+    input: { hostId: string; externalId: string; deliveryId: string; body: DeliveryBody }
+  ): Promise<EnqueueResult> {
+    const { hostId, externalId, deliveryId, body } = input
+    const now = this.now()
+    const { kind, ...payload } = body
+    const [inserted] = await transaction.query(
+      `INSERT INTO connector_inbox
+       (host_id, delivery_id, external_id, seq, kind, payload, state, expires_at, created_at)
+       SELECT ?, ?, ?, COALESCE(MAX(seq), 0) + 1, ?, ?, 'pending', ?, ?
+       FROM connector_inbox WHERE host_id = ?
+       ON CONFLICT (host_id, delivery_id) DO NOTHING`,
+      [
+        hostId,
+        deliveryId,
+        externalId,
+        kind,
+        JSON.stringify(payload),
+        now + CONNECTOR_LIMITS.deliveryTtlMs,
+        now,
+        hostId
+      ]
+    )
+    if (Number(inserted?.changes ?? 0) === 0) {
+      return 'duplicate'
+    }
+    if (body.kind === 'cancel') {
+      // A start the machine never received is dropped, with everything queued behind it, this
+      // cancel included, so a retried cancel reads as a duplicate.
+      const [dropped] = await transaction.query(
+        `UPDATE connector_inbox SET state = 'dropped', payload = NULL, settled_at = ?
+         WHERE host_id = ? AND external_id = ? AND state = 'pending'
+           AND EXISTS (SELECT 1 FROM connector_inbox s WHERE s.host_id = connector_inbox.host_id
+             AND s.external_id = connector_inbox.external_id AND s.kind = 'start'
+             AND s.state = 'pending')`,
+        [now, hostId, externalId]
+      )
+      if (Number(dropped?.changes ?? 0) > 0) {
+        return 'dropped'
+      }
+    }
+    await this.wake(transaction, hostId)
+    return 'queued'
+  }
+
+  // Under the host lock (the poll's pass). Each conversation's head goes out; the next waits
+  // until the head is received. A cancel skips the queue unless its start is still unreceived.
+  async claim(
+    transaction: CloudDatabase,
+    hostId: string
+  ): Promise<{ deliveries: ConnectorDelivery[]; nextLeaseDueAt: number | null }> {
+    const now = this.now()
+    const rows = await transaction.query(
+      `SELECT delivery_id, external_id, seq, kind, payload, expires_at FROM connector_inbox i
+       WHERE i.host_id = ? AND i.expires_at > ?
+         AND (i.state = 'pending' OR (i.state = 'leased' AND i.lease_until <= ?))
+         AND (
+           NOT EXISTS (SELECT 1 FROM connector_inbox e WHERE e.host_id = i.host_id
+             AND e.external_id = i.external_id AND e.seq < i.seq
+             AND e.state IN ('pending', 'leased'))
+           OR (i.kind = 'cancel' AND NOT EXISTS (SELECT 1 FROM connector_inbox s
+             WHERE s.host_id = i.host_id AND s.external_id = i.external_id
+               AND s.kind = 'start' AND s.state IN ('pending', 'leased')))
+         )
+       ORDER BY i.seq LIMIT ?`,
+      [hostId, now, now, CONNECTOR_LIMITS.maxDeliveriesPerPoll]
+    )
+    const deliveries: ConnectorDelivery[] = []
+    for (const row of rows) {
+      // The expiry sweep doesn't take the host lock; the state guard keeps its verdict.
+      const [leased] = await transaction.query(
+        `UPDATE connector_inbox SET state = 'leased', lease_until = ?
+         WHERE host_id = ? AND delivery_id = ?
+           AND (state = 'pending' OR (state = 'leased' AND lease_until <= ?))`,
+        [now + CONNECTOR_LIMITS.leaseMs, hostId, row.delivery_id, now]
+      )
+      if (Number(leased?.changes ?? 0) === 0) {
+        continue
+      }
+      deliveries.push(
+        ConnectorDeliverySchema.parse({
+          ...JSON.parse(String(row.payload)),
+          kind: row.kind,
+          deliveryId: row.delivery_id,
+          seq: Number(row.seq),
+          expiresAt: Number(row.expires_at),
+          conversation: conversation(String(row.external_id))
+        })
+      )
+    }
+    const [next] = await transaction.query(
+      `SELECT MIN(lease_until) AS due FROM connector_inbox
+       WHERE host_id = ? AND state = 'leased' AND expires_at > ?`,
+      [hostId, now]
+    )
+    return { deliveries, nextLeaseDueAt: next?.due === null ? null : Number(next?.due) }
+  }
+
+  // What an earlier process received but never settled; the first poll after each hello lists it.
+  async listUnsettled(transaction: CloudDatabase, hostId: string): Promise<Unsettled[]> {
+    const rows = await transaction.query(
+      `SELECT delivery_id, external_id FROM connector_inbox
+       WHERE host_id = ? AND state = 'received' ORDER BY seq`,
+      [hostId]
+    )
+    return rows.map((row) => ({
+      deliveryId: String(row.delivery_id),
+      conversation: conversation(String(row.external_id))
+    }))
+  }
+
+  async wake(transaction: CloudDatabase, hostId: string): Promise<void> {
+    await this.notifications.notify(transaction, INBOX_CHANNEL, hostId)
+  }
+
+  // A late received is accepted (it did arrive); the first done wins.
+  async acknowledge(
+    transaction: CloudDatabase,
+    hostId: string,
+    acks: ConnectorAcksRequest['acks']
+  ): Promise<void> {
+    const now = this.now()
+    let changed = 0
+    for (const ack of acks) {
+      const [result] =
+        ack.step === 'received'
+          ? await transaction.query(
+              `UPDATE connector_inbox SET state = 'received'
+               WHERE host_id = ? AND delivery_id = ? AND state IN ('pending', 'leased', 'expired')`,
+              [hostId, ack.deliveryId]
+            )
+          : await transaction.query(
+              `UPDATE connector_inbox SET state = 'done', result = ?, payload = NULL, settled_at = ?
+               WHERE host_id = ? AND delivery_id = ? AND state IN ('pending', 'leased', 'received', 'expired')`,
+              [JSON.stringify(ack.result), now, hostId, ack.deliveryId]
+            )
+      changed += Number(result?.changes ?? 0)
+    }
+    // A received head lets the conversation's next delivery go out.
+    if (changed > 0) {
+      await this.wake(transaction, hostId)
+    }
+  }
+
+  // Pending past its deadline, or leased and never received 30 s after it. Returns the threads hit.
+  async expireDue(): Promise<{ hostId: string; externalId: string }[]> {
+    const now = this.now()
+    return await this.database.transaction(async (transaction) => {
+      const rows = await transaction.query(
+        `UPDATE connector_inbox SET state = 'expired', payload = NULL, settled_at = ?
+         WHERE (state = 'pending' AND expires_at <= ?) OR (state = 'leased' AND expires_at <= ?)
+         RETURNING host_id, external_id`,
+        [now, now, now - CONNECTOR_LIMITS.leasedExpiryGraceMs]
+      )
+      const threads = new Map<string, { hostId: string; externalId: string }>()
+      for (const row of rows) {
+        const thread = { hostId: String(row.host_id), externalId: String(row.external_id) }
+        threads.set(`${thread.hostId}\0${thread.externalId}`, thread)
+      }
+      for (const hostId of new Set([...threads.values()].map((thread) => thread.hostId))) {
+        await this.wake(transaction, hostId)
+      }
+      return [...threads.values()]
+    })
+  }
+
+  async pruneSettled(retentionMs: number): Promise<number> {
+    const [result] = await this.database.query(
+      `DELETE FROM connector_inbox
+       WHERE state IN ('done', 'expired', 'dropped') AND settled_at < ?`,
+      [this.now() - retentionMs]
+    )
+    return Number(result?.changes ?? 0)
+  }
+}

cloud/apps/connector/src/connector-schema.ts

@@ -0,0 +1,65 @@
+// Applied at startup for both dialects, so every column type reads the same in SQLite and PostgreSQL.
+const CONNECTOR_SCHEMA = `
+CREATE TABLE IF NOT EXISTS connector_hosts (
+  host_id TEXT PRIMARY KEY,
+  account_id TEXT NOT NULL,
+  host_name TEXT NOT NULL,
+  platform TEXT NOT NULL,
+  runtime TEXT NOT NULL,
+  orca_version TEXT NOT NULL,
+  hello_rev BIGINT NOT NULL,
+  hello TEXT NOT NULL,
+  unsettled_due INTEGER NOT NULL,
+  events_seq BIGINT,
+  first_seen_at BIGINT NOT NULL,
+  last_seen_at BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS connector_hosts_account ON connector_hosts(account_id);
+
+CREATE TABLE IF NOT EXISTS connector_threads (
+  external_id TEXT PRIMARY KEY,
+  team_id TEXT NOT NULL,
+  channel_id TEXT NOT NULL,
+  thread_ts TEXT NOT NULL,
+  owner_user_id TEXT NOT NULL,
+  account_id TEXT NOT NULL,
+  host_id TEXT NOT NULL,
+  place_id TEXT NOT NULL,
+  pane_key TEXT,
+  agent TEXT,
+  open_turn_id BIGINT,
+  turn_state TEXT,
+  turn_stop_reason TEXT,
+  pending_prompt TEXT,
+  created_at BIGINT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS connector_threads_slack
+  ON connector_threads(team_id, channel_id, thread_ts, owner_user_id);
+CREATE INDEX IF NOT EXISTS connector_threads_host ON connector_threads(host_id);
+
+CREATE TABLE IF NOT EXISTS connector_inbox (
+  host_id TEXT NOT NULL,
+  delivery_id TEXT NOT NULL,
+  external_id TEXT NOT NULL,
+  seq BIGINT NOT NULL,
+  kind TEXT NOT NULL,
+  payload TEXT,
+  state TEXT NOT NULL,
+  lease_until BIGINT,
+  expires_at BIGINT NOT NULL,
+  result TEXT,
+  created_at BIGINT NOT NULL,
+  settled_at BIGINT,
+  PRIMARY KEY (host_id, delivery_id)
+);
+CREATE INDEX IF NOT EXISTS connector_inbox_host_state ON connector_inbox(host_id, state, seq);
+CREATE INDEX IF NOT EXISTS connector_inbox_conversation
+  ON connector_inbox(host_id, external_id, seq);
+CREATE INDEX IF NOT EXISTS connector_inbox_expiry ON connector_inbox(state, expires_at);
+`
+
+export function connectorSchemaStatements(): string[] {
+  return CONNECTOR_SCHEMA.split(';')
+    .map((statement) => statement.trim())
+    .filter((statement) => statement.length > 0)
+}

cloud/apps/connector/src/connector-server-harness.test-fixture.ts

@@ -1,12 +1,16 @@
-import { randomUUID } from 'node:crypto'
-import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import { randomBytes, randomUUID } from 'node:crypto'
+import { openCloudNotifications, type CloudDatabase } from '@orca-cloud/cloud-database'
+import { CONNECTOR_PATHS } from '@orca-cloud/connector-contract'
+import { createLocalJWKSet, exportJWK, generateKeyPair, SignJWT } from 'jose'
 import pg from 'pg'
 import type { ConnectorConfig } from './config.js'
 import { openConnectorDatabase, openInMemoryConnectorDatabase } from './connector-database.js'
 import { createConnectorServer } from './connector-server.js'
+import { connectorHostId } from './connector-token-verifier.js'
 
 export const PUBLIC_URL = 'https://connector.onorca.dev'
 export const AUTH_ISSUER = 'https://login.onorca.dev'
+export const HOST_IP = '203.0.113.7'
 
 // Real Postgres when the CI service is present; each harness gets its own schema.
 const testDatabaseUrl = process.env.ORCA_CONNECTOR_TEST_DATABASE_URL
@@ -29,11 +33,11 @@ export async function cleanupConnectorHarnesses(): Promise<void> {
   await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()))
 }
 
-async function openTestDatabase(): Promise<CloudDatabase> {
+async function openTestDatabase(): Promise<{ database: CloudDatabase; databaseUrl?: string }> {
   if (!testDatabaseUrl) {
     const database = await openInMemoryConnectorDatabase()
     cleanups.push(() => database.close())
-    return database
+    return { database }
   }
   const admin = new pg.Client({ connectionString: testDatabaseUrl })
   await admin.connect()
@@ -47,13 +51,140 @@ async function openTestDatabase(): Promise<CloudDatabase> {
     await admin.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`)
     await admin.end()
   })
-  return database
+  return { database, databaseUrl: url.toString() }
 }
 
-export async function createConnectorHarness(config: Partial<ConnectorConfig> = {}) {
-  const database = await openTestDatabase()
-  const runtime = createConnectorServer({ ...testConnectorConfig(), ...config }, database)
+// Stands in for the auth service's connector token route.
+async function testTokenIssuer() {
+  const { privateKey, publicKey } = await generateKeyPair('ES256')
+  const jwk = { ...(await exportJWK(publicKey)), kid: 'test-key', alg: 'ES256' }
+  return {
+    jwks: createLocalJWKSet({ keys: [jwk] }),
+    sign: async (claims: { sub: string; mid: string }, expiresAtSeconds: number) =>
+      await new SignJWT({ prof: 'profile-1', sid: 'family-1', mid: claims.mid, purpose: 'connector-host' })
+        .setProtectedHeader({ alg: 'ES256', kid: 'test-key' })
+        .setIssuer(AUTH_ISSUER)
+        .setAudience('orca-connector')
+        .setSubject(claims.sub)
+        .setIssuedAt()
+        .setExpirationTime(expiresAtSeconds)
+        .sign(privateKey)
+  }
+}
+
+// Holds the first transaction statement starting with `prefix` until release(), locks still held.
+// `skip` lets that many matching statements through first.
+export function pauseBefore(database: CloudDatabase, prefix: string, skip = 0) {
+  let reached!: () => void
+  const atStatement = new Promise<void>((resolve) => (reached = resolve))
+  let release!: () => void
+  const released = new Promise<void>((resolve) => (release = resolve))
+  let seen = 0
+  const wrap = (inner: CloudDatabase): CloudDatabase => ({
+    dialect: inner.dialect,
+    query: async (sql, params) => {
+      if (sql.trimStart().startsWith(prefix) && seen++ === skip) {
+        reached()
+        await released
+      }
+      return await inner.query(sql, params)
+    },
+    transaction: (run) => inner.transaction((transaction) => run(wrap(transaction))),
+    lockQuotaScope: (key) => inner.lockQuotaScope(key),
+    tryLockScope: (key) => inner.tryLockScope(key),
+    tryLockSharedScope: (key) => inner.tryLockSharedScope(key),
+    close: () => inner.close()
+  })
+  return { database: wrap(database), atStatement, release }
+}
+
+export async function createConnectorHarness(
+  config: Partial<ConnectorConfig> = {},
+  options: {
+    pollHoldMs?: number
+    wrapDatabase?: (database: CloudDatabase) => CloudDatabase
+  } = {}
+) {
+  let now = 1_800_000_000_000
+  const issuer = await testTokenIssuer()
+  const { database, databaseUrl } = await openTestDatabase()
+  const notifications = openCloudNotifications(databaseUrl ? { databaseUrl } : {})
+  const runtime = createConnectorServer(
+    { ...testConnectorConfig(), ...config },
+    options.wrapDatabase?.(database) ?? database,
+    notifications,
+    { now: () => now, jwks: issuer.jwks, pollHoldMs: options.pollHoldMs ?? 0 }
+  )
+  await runtime.waker.start()
+  cleanups.push(() => runtime.waker.close())
   const request = (path: string, init?: RequestInit): Promise<Response> =>
     Promise.resolve(runtime.app.request(path, init))
-  return { runtime, database, request }
+  const json = (path: string, body: unknown, headers: Record<string, string> = {}) =>
+    request(path, {
+      method: 'POST',
+      headers: { 'content-type': 'application/json', 'x-forwarded-for': HOST_IP, ...headers },
+      body: JSON.stringify(body)
+    })
+
+  // A signed-in machine holding a connector token, with helpers for the machine API.
+  async function machine(
+    account = 'user-1',
+    { machineId = randomBytes(16).toString('base64url'), ttlSeconds = 300 } = {}
+  ) {
+    const claims = { sub: account, mid: machineId }
+    const expiresAt = (Math.floor(Date.now() / 1000) + ttlSeconds) * 1000
+    const token = await issuer.sign(claims, expiresAt / 1000)
+    const call = (path: string, body: unknown, method = 'POST') =>
+      request(path, {
+        method,
+        headers: {
+          'content-type': 'application/json',
+          'x-forwarded-for': HOST_IP,
+          authorization: `Bearer ${token}`
+        },
+        body: JSON.stringify(body)
+      })
+    return {
+      hostId: connectorHostId(claims),
+      machineId,
+      expiresAt,
+      call,
+      hello: (helloRev = 1) =>
+        call(
+          CONNECTOR_PATHS.hello,
+          {
+            v: 1,
+            helloRev,
+            contract: 1,
+            hostName: 'build-box',
+            platform: 'linux',
+            runtime: 'serve',
+            orcaVersion: '1.4.230',
+            agents: ['claude'],
+            promptKinds: ['approval'],
+            features: ['answer', 'cancel'],
+            projects: []
+          },
+          'PUT'
+        ),
+      poll: async (helloRev = 1) => {
+        const response = await call(CONNECTOR_PATHS.poll, { v: 1, helloRev })
+        const body: unknown = await response.json()
+        return { status: response.status, body }
+      },
+      ack: (acks: unknown[], helloRev = 1) => call(CONNECTOR_PATHS.acks, { v: 1, helloRev, acks })
+    }
+  }
+
+  return {
+    runtime,
+    database,
+    machine,
+    request,
+    json,
+    advance: (ms: number) => {
+      now += ms
+    },
+    now: () => now
+  }
 }

cloud/apps/connector/src/connector-server.ts

@@ -1,28 +1,71 @@
 import { createAdaptorServer } from '@hono/node-server'
-import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import type { CloudDatabase, CloudNotifications } from '@orca-cloud/cloud-database'
+import { CONNECTOR_LIMITS } from '@orca-cloud/connector-contract'
+import type { JWTVerifyGetKey } from 'jose'
 import { Hono } from 'hono'
+import { bodyLimit } from 'hono/body-limit'
 import type { ConnectorConfig } from './config.js'
+import { ConnectorHosts } from './connector-hosts.js'
+import { ConnectorInbox } from './connector-inbox.js'
+import { ConnectorThreads } from './connector-threads.js'
+import { createConnectorTokenVerifier } from './connector-token-verifier.js'
+import { registerDevEnqueueRoute } from './dev-enqueue-route.js'
+import { InboxWaker } from './inbox-waker.js'
+import { machineAuth, type MachineEnv } from './machine-auth.js'
+import { registerMachineRoutes } from './machine-routes.js'
+import { queueDelivery } from './queue-delivery.js'
 import { limitByClientIp, RateLimiter } from './rate-limiter.js'
 import { RequestDrain } from './request-drain.js'
 
+const MAX_BODY_BYTES = 64 * 1024
 const UNAUTHENTICATED_PER_MINUTE_PER_IP = 30
+const AUTHENTICATED_PER_MINUTE_PER_IP = 6_000
+const AUTHENTICATED_PER_MINUTE_PER_HOST = 600
 
 export type ConnectorServerOptions = {
   now?: () => number
+  // The auth service's signing keys; tests pass a local set.
+  jwks?: JWTVerifyGetKey
+  pollHoldMs?: number
+}
+
+function isLoopbackOrigin(origin: string): boolean {
+  return ['127.0.0.1', 'localhost', '[::1]'].includes(new URL(origin).hostname)
+}
+
+// Local development listens on its loopback origin only, so the unauthenticated dev route can't be
+// reached from another machine; anything else listens on every interface, as Cloud Run needs.
+export function listenHostname(publicUrl: string): string | undefined {
+  return isLoopbackOrigin(publicUrl) ? new URL(publicUrl).hostname.replace(/^\[|\]$/g, '') : undefined
 }
 
 export function createConnectorServer(
   config: ConnectorConfig,
   database: CloudDatabase,
+  notifications: CloudNotifications,
   options: ConnectorServerOptions = {}
 ) {
   const now = options.now ?? Date.now
+  const inbox = new ConnectorInbox(database, notifications, now)
+  const hosts = new ConnectorHosts(database, inbox, now)
+  const threads = new ConnectorThreads(now)
+  const waker = new InboxWaker(notifications)
+  const queue = (input: Parameters<typeof queueDelivery>[1]) =>
+    queueDelivery({ database, inbox, threads, hosts }, input)
   const limitUnauthenticated = limitByClientIp(
     new RateLimiter(UNAUTHENTICATED_PER_MINUTE_PER_IP, now),
     config.trustedProxyHops
   )
+  const limitAuthenticated = limitByClientIp(
+    new RateLimiter(AUTHENTICATED_PER_MINUTE_PER_IP, now),
+    config.trustedProxyHops
+  )
+  const limitBody = bodyLimit({
+    maxSize: MAX_BODY_BYTES,
+    onError: (context) => context.json({ error: 'request_too_large' }, 413)
+  })
 
-  const app = new Hono()
+  const app = new Hono<MachineEnv>()
   const requestDrain = new RequestDrain()
   app.use('*', requestDrain.middleware)
   // Default logging can carry row detail; log only the error's name.
@@ -50,5 +93,35 @@ export function createConnectorServer(
     app.use('*', async (context) => context.json({ error: 'validation_only' }, 503))
   }
 
-  return { app, requestDrain, server: createAdaptorServer(app) }
+  app.use(
+    '/v1/connector/*',
+    limitAuthenticated,
+    machineAuth({
+      verifyToken: createConnectorTokenVerifier(config, options.jwks),
+      hostRequests: new RateLimiter(AUTHENTICATED_PER_MINUTE_PER_HOST, now)
+    })
+  )
+  registerMachineRoutes(app, {
+    database,
+    inbox,
+    threads,
+    hosts,
+    waker,
+    now,
+    holdMs: options.pollHoldMs ?? CONNECTOR_LIMITS.initialPollHoldMs,
+    limitBody
+  })
+  // Local development only: a stand-in for Slack ingress until it exists.
+  if (isLoopbackOrigin(config.publicUrl)) {
+    registerDevEnqueueRoute(app, { queue, limitBody })
+  }
+
+  return {
+    app,
+    requestDrain,
+    server: createAdaptorServer(app),
+    inbox,
+    waker,
+    queueDelivery: queue
+  }
 }

cloud/apps/connector/src/connector-threads.ts

@@ -0,0 +1,138 @@
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import type { ConnectorEvent } from '@orca-cloud/connector-contract'
+
+// Linked and snapshot (re)bind a conversation's pane. Moving to another pane forgets the old pane's
+// turn and its verdict; one it left open (working or unverifiable) stays unverifiable, never done.
+const REBIND_PANE = `pane_key = ?,
+  open_turn_id = CASE WHEN pane_key = ? THEN open_turn_id END,
+  turn_state = CASE WHEN pane_key IS NULL OR pane_key = ? THEN turn_state
+    WHEN turn_state IN ('working', 'unverifiable') THEN 'unverifiable' END,
+  turn_stop_reason = CASE WHEN pane_key IS NULL OR pane_key = ? THEN turn_stop_reason END`
+const rebindParams = (paneKey: string) => [paneKey, paneKey, paneKey, paneKey]
+// Other events only touch the pane the conversation is bound to.
+const ON_BOUND_PANE = 'host_id = ? AND external_id = ? AND (pane_key IS NULL OR pane_key = ?)'
+
+export type NewThread = {
+  externalId: string
+  teamId: string
+  channelId: string
+  threadTs: string
+  ownerUserId: string
+  accountId: string
+  hostId: string
+  placeId: string
+}
+
+// Each Slack thread is bound to one host for life; events only ever touch that host's threads.
+export class ConnectorThreads {
+  constructor(private readonly now: () => number) {}
+
+  // Binds the thread to its host on first sight; false when it already belongs to another host.
+  async bind(transaction: CloudDatabase, thread: NewThread): Promise<boolean> {
+    await transaction.query(
+      `INSERT INTO connector_threads
+       (external_id, team_id, channel_id, thread_ts, owner_user_id, account_id, host_id, place_id,
+        created_at)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (external_id) DO NOTHING`,
+      [
+        thread.externalId,
+        thread.teamId,
+        thread.channelId,
+        thread.threadTs,
+        thread.ownerUserId,
+        thread.accountId,
+        thread.hostId,
+        thread.placeId,
+        this.now()
+      ]
+    )
+    const [row] = await transaction.query(
+      'SELECT host_id FROM connector_threads WHERE external_id = ?',
+      [thread.externalId]
+    )
+    return row?.host_id === thread.hostId
+  }
+
+  // Idempotent by each event's identity: a resend writes the same state again or matches nothing.
+  async applyEvents(
+    transaction: CloudDatabase,
+    hostId: string,
+    events: ConnectorEvent[]
+  ): Promise<void> {
+    for (const event of events) {
+      if (event.kind === 'snapshot') {
+        // Not working never means done: an open turn the snapshot doesn't show working (or doesn't
+        // list; it lists every linked conversation) is unverifiable, and an unlisted one has no prompt.
+        const notIn = (ids: string[]) =>
+          ids.length > 0 ? ` AND external_id NOT IN (${ids.map(() => '?').join(', ')})` : ''
+        const working = event.conversations
+          .filter((entry) => entry.working)
+          .map((entry) => entry.conversation.externalId)
+        await transaction.query(
+          `UPDATE connector_threads SET turn_state = 'unverifiable'
+           WHERE host_id = ? AND turn_state = 'working'${notIn(working)}`,
+          [hostId, ...working]
+        )
+        const listed = event.conversations.map((entry) => entry.conversation.externalId)
+        await transaction.query(
+          `UPDATE connector_threads SET pending_prompt = NULL WHERE host_id = ?${notIn(listed)}`,
+          [hostId, ...listed]
+        )
+        for (const entry of event.conversations) {
+          await transaction.query(
+            `UPDATE connector_threads SET ${REBIND_PANE}, agent = ?, pending_prompt = ?
+             WHERE host_id = ? AND external_id = ?`,
+            [
+              ...rebindParams(entry.paneKey),
+              entry.agent,
+              entry.prompt ? JSON.stringify(entry.prompt) : null,
+              hostId,
+              entry.conversation.externalId
+            ]
+          )
+        }
+        continue
+      }
+      const where = [hostId, event.conversation.externalId]
+      const onPane = [...where, event.paneKey]
+      switch (event.kind) {
+        case 'linked':
+          await transaction.query(
+            `UPDATE connector_threads SET ${REBIND_PANE}, agent = ?,
+               pending_prompt = CASE WHEN pane_key = ? THEN pending_prompt END
+             WHERE host_id = ? AND external_id = ?`,
+            [...rebindParams(event.paneKey), event.agent, event.paneKey, ...where]
+          )
+          break
+        case 'prompt':
+          await transaction.query(
+            `UPDATE connector_threads SET pane_key = ?, pending_prompt = ? WHERE ${ON_BOUND_PANE}`,
+            [event.paneKey, JSON.stringify(event.prompt), ...onPane]
+          )
+          break
+        case 'turn-started':
+          await transaction.query(
+            `UPDATE connector_threads SET pane_key = ?, open_turn_id = ?, turn_state = 'working',
+               turn_stop_reason = NULL, pending_prompt = NULL
+             WHERE ${ON_BOUND_PANE} AND (open_turn_id IS NULL OR open_turn_id < ?)`,
+            [event.paneKey, event.at, ...onPane, event.at]
+          )
+          break
+        case 'turn-ended':
+          await transaction.query(
+            `UPDATE connector_threads SET pane_key = ?, open_turn_id = ?, turn_state = 'ended',
+               turn_stop_reason = ?, pending_prompt = NULL
+             WHERE ${ON_BOUND_PANE} AND (open_turn_id IS NULL OR open_turn_id <= ?)`,
+            [
+              event.paneKey,
+              event.turnId,
+              event.stopReason,
+              ...onPane,
+              event.turnId
+            ]
+          )
+          break
+      }
+    }
+  }
+}

cloud/apps/connector/src/dev-enqueue-route.ts

@@ -0,0 +1,64 @@
+import { randomUUID } from 'node:crypto'
+import { ConnectorDeliverySchema } from '@orca-cloud/connector-contract'
+import type { Env, Hono, MiddlewareHandler } from 'hono'
+import { z } from 'zod'
+import type { queueDelivery } from './queue-delivery.js'
+
+const DevEnqueueSchema = z.object({
+  hostId: z.string().min(1),
+  externalId: z.string().min(1),
+  placeId: z.string().min(1).default('dev-place'),
+  deliveryId: z.string().min(1).optional(),
+  // The kind-specific delivery fields; the envelope is filled in below and checked by the contract.
+  delivery: z.record(z.string(), z.unknown())
+})
+
+// Queues a delivery without Slack for a host that said hello; only on a loopback origin.
+export function registerDevEnqueueRoute<E extends Env>(
+  app: Hono<E>,
+  deps: {
+    queue: (input: Parameters<typeof queueDelivery>[1]) => ReturnType<typeof queueDelivery>
+    limitBody: MiddlewareHandler
+  }
+): void {
+  app.post('/dev/enqueue', deps.limitBody, async (context) => {
+    const body = DevEnqueueSchema.safeParse(await context.req.json().catch(() => null))
+    if (!body.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    const { hostId, externalId, placeId } = body.data
+    const deliveryId = body.data.deliveryId ?? randomUUID()
+    const delivery = ConnectorDeliverySchema.safeParse({
+      ...body.data.delivery,
+      deliveryId,
+      seq: 1,
+      expiresAt: 1,
+      conversation: { source: 'slack', externalId }
+    })
+    if (!delivery.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    const { deliveryId: _id, seq: _seq, expiresAt: _expiresAt, conversation: _c, ...rest } =
+      delivery.data
+    const status = await deps.queue({
+      thread: {
+        externalId,
+        teamId: 'dev',
+        channelId: 'dev',
+        threadTs: externalId,
+        ownerUserId: 'dev',
+        hostId,
+        placeId
+      },
+      deliveryId,
+      body: rest
+    })
+    if (status === 'unknown_host') {
+      return context.json({ error: status }, 404)
+    }
+    if (status === 'not_this_hosts_thread') {
+      return context.json({ error: status }, 409)
+    }
+    return context.json({ deliveryId, status })
+  })
+}

cloud/apps/connector/src/inbox-waker.test.ts

@@ -0,0 +1,43 @@
+import type { CloudNotifications, ListenHandlers } from '@orca-cloud/cloud-database'
+import { describe, expect, it } from 'vitest'
+import { InboxWaker } from './inbox-waker.js'
+
+function fakeNotifications() {
+  let handlers: ListenHandlers | null = null
+  const notifications: CloudNotifications = {
+    notify: async () => {},
+    listen: async (_channel, next) => {
+      handlers = next
+      return async () => {}
+    }
+  }
+  return { notifications, handlers: () => handlers! }
+}
+
+describe('inbox waker', () => {
+  it('wakes only the named host, and every waiter after the listener reconnects', async () => {
+    const fake = fakeNotifications()
+    const waker = new InboxWaker(fake.notifications)
+    await waker.start()
+    const woken: string[] = []
+    const a = waker.register('a').wait(5_000).then(() => woken.push('a'))
+    const b = waker.register('b').wait(5_000).then(() => woken.push('b'))
+    fake.handlers().onNotify('a')
+    await a
+    expect(woken).toEqual(['a'])
+    fake.handlers().onReconnect()
+    await b
+    expect(woken).toEqual(['a', 'b'])
+  })
+
+  it('keeps a notification that lands between registering and waiting', async () => {
+    const fake = fakeNotifications()
+    const waker = new InboxWaker(fake.notifications)
+    await waker.start()
+    const waiter = waker.register('a')
+    fake.handlers().onNotify('a')
+    const startedAt = Date.now()
+    await waiter.wait(5_000)
+    expect(Date.now() - startedAt).toBeLessThan(1_000)
+  })
+})

cloud/apps/connector/src/inbox-waker.ts

@@ -0,0 +1,73 @@
+import type { CloudNotifications } from '@orca-cloud/cloud-database'
+import { INBOX_CHANNEL } from './connector-inbox.js'
+
+export type InboxWaiter = { wait(ms: number): Promise<void>; cancel(): void }
+
+// Held polls wait here. One LISTEN per instance; a reconnect wakes everyone to re-read, since
+// notifications sent while the listener was down are lost.
+export class InboxWaker {
+  private readonly waiters = new Map<string, Set<() => void>>()
+  private stopListening: (() => Promise<void>) | null = null
+  private isDraining = false
+
+  constructor(private readonly notifications: CloudNotifications) {}
+
+  async start(): Promise<void> {
+    this.stopListening = await this.notifications.listen(INBOX_CHANNEL, {
+      onNotify: (hostId) => this.wake(hostId),
+      onReconnect: () => this.wakeAll()
+    })
+  }
+
+  // Register before reading the inbox, so a change between the read and the wait still wakes it.
+  register(hostId: string): InboxWaiter {
+    let wake!: () => void
+    const woken = new Promise<void>((resolve) => (wake = resolve))
+    const waiters = this.waiters.get(hostId) ?? new Set()
+    waiters.add(wake)
+    this.waiters.set(hostId, waiters)
+    const cancel = (): void => {
+      waiters.delete(wake)
+      if (waiters.size === 0 && this.waiters.get(hostId) === waiters) {
+        this.waiters.delete(hostId)
+      }
+    }
+    return {
+      wait: async (ms) => {
+        let timer: NodeJS.Timeout | undefined
+        await Promise.race([woken, new Promise((resolve) => (timer = setTimeout(resolve, ms)))])
+        clearTimeout(timer)
+        cancel()
+      },
+      cancel
+    }
+  }
+
+  // Shutdown: every held poll answers now and tells its machine to reconnect elsewhere.
+  drain(): void {
+    this.isDraining = true
+    this.wakeAll()
+  }
+
+  get draining(): boolean {
+    return this.isDraining
+  }
+
+  async close(): Promise<void> {
+    await this.stopListening?.()
+  }
+
+  private wake(hostId: string): void {
+    for (const wake of this.waiters.get(hostId) ?? []) {
+      wake()
+    }
+  }
+
+  private wakeAll(): void {
+    for (const waiters of this.waiters.values()) {
+      for (const wake of waiters) {
+        wake()
+      }
+    }
+  }
+}

cloud/apps/connector/src/index.ts

@@ -1,6 +1,8 @@
+import { openCloudNotifications } from '@orca-cloud/cloud-database'
 import { loadConnectorConfig } from './config.js'
+import { startConnectorBackground } from './connector-background.js'
 import { openConnectorDatabase } from './connector-database.js'
-import { createConnectorServer } from './connector-server.js'
+import { createConnectorServer, listenHostname } from './connector-server.js'
 
 const config = loadConnectorConfig()
 const database = await openConnectorDatabase({
@@ -10,9 +12,18 @@ const database = await openConnectorDatabase({
   applicationName: 'orca-connector',
   readOnly: config.mode === 'validation'
 })
-const runtime = createConnectorServer(config, database)
+const notifications = openCloudNotifications({
+  ...(config.databaseUrl === undefined ? {} : { databaseUrl: config.databaseUrl }),
+  applicationName: 'orca-connector'
+})
+const runtime = createConnectorServer(config, database, notifications)
+const active = config.mode === 'active'
+if (active) {
+  await runtime.waker.start()
+}
+const stopBackground = active ? startConnectorBackground(runtime) : () => {}
 
-runtime.server.listen(config.port, () => {
+runtime.server.listen(config.port, listenHostname(config.publicUrl), () => {
   console.log(`[orca-connector] listening on ${config.publicUrl} (port ${config.port})`)
 })
 
@@ -25,10 +36,14 @@ const shutdown = (): void => {
   // Cloud Run sends SIGKILL after ten seconds; leave time for explicit cleanup.
   const deadline = setTimeout(() => process.exit(1), 9_000)
   deadline.unref()
+  stopBackground()
+  // Held polls answer at once with a reconnect, so the drain is not held up by them.
+  runtime.waker.drain()
   const requests = runtime.requestDrain.begin()
   const connections = new Promise<void>((resolve) => runtime.server.close(() => resolve()))
   void Promise.all([requests, connections])
     .then(async () => {
+      await runtime.waker.close()
       await database.close()
       clearTimeout(deadline)
     })

cloud/apps/connector/src/machine-auth.ts

@@ -0,0 +1,30 @@
+import type { MiddlewareHandler } from 'hono'
+import {
+  connectorHostId,
+  readBearer,
+  type ConnectorTokenVerifier
+} from './connector-token-verifier.js'
+import type { RateLimiter } from './rate-limiter.js'
+
+// One machine of one account, as its connector token names it; expiresAt bounds what it may hold.
+export type MachineHost = { hostId: string; accountId: string; expiresAt: number }
+export type MachineEnv = { Variables: { host: MachineHost } }
+
+export function machineAuth(deps: {
+  verifyToken: ConnectorTokenVerifier
+  hostRequests: RateLimiter
+}): MiddlewareHandler<MachineEnv> {
+  return async (context, next) => {
+    const bearer = readBearer(context.req.header('authorization'))
+    const claims = bearer ? await deps.verifyToken(bearer) : null
+    if (!claims) {
+      return context.json({ error: 'invalid_token' }, 401)
+    }
+    const hostId = connectorHostId(claims)
+    if (!deps.hostRequests.allow(hostId)) {
+      return context.json({ error: 'rate_limited' }, 429)
+    }
+    context.set('host', { hostId, accountId: claims.sub, expiresAt: claims.exp * 1000 })
+    await next()
+  }
+}

cloud/apps/connector/src/machine-routes.ts

@@ -0,0 +1,164 @@
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import {
+  CONNECTOR_CONTRACT_VERSION,
+  CONNECTOR_PATHS,
+  ConnectorAcksRequestSchema,
+  ConnectorEventsRequestSchema,
+  ConnectorHelloRequestSchema,
+  ConnectorPollRequestSchema,
+  type ConnectorDelivery,
+  type ConnectorPollResponse
+} from '@orca-cloud/connector-contract'
+import type { Context, Hono, MiddlewareHandler } from 'hono'
+import { lockConnectorHost, type ConnectorHosts } from './connector-hosts.js'
+import type { ConnectorInbox, Unsettled } from './connector-inbox.js'
+import type { ConnectorThreads } from './connector-threads.js'
+import type { InboxWaker } from './inbox-waker.js'
+import type { MachineEnv } from './machine-auth.js'
+
+class TokenExpiredDuringPoll extends Error {}
+
+// The machine API behind a connector token: hello, the held poll, confirmations and turn events.
+export function registerMachineRoutes(
+  app: Hono<MachineEnv>,
+  deps: {
+    database: CloudDatabase
+    inbox: ConnectorInbox
+    threads: ConnectorThreads
+    hosts: ConnectorHosts
+    waker: InboxWaker
+    now: () => number
+    holdMs: number
+    limitBody: MiddlewareHandler
+  }
+): void {
+  const { database, inbox, threads, hosts, waker, now, holdMs, limitBody } = deps
+
+  // Acks and events from an older hello change nothing: the first poll after the newer hello lists
+  // what is unsettled, and its batches start a new events mark. An unrecorded hello must be sent.
+  async function onCurrentHello(
+    context: Context<MachineEnv>,
+    helloRev: number,
+    run: (transaction: CloudDatabase, hostId: string) => Promise<void>
+  ): Promise<Response> {
+    const { hostId } = context.get('host')
+    const status = await database.transaction(async (transaction) => {
+      await lockConnectorHost(transaction, hostId)
+      const status = await hosts.helloStatus(transaction, hostId, helloRev)
+      if (status === 'current') {
+        await run(transaction, hostId)
+      }
+      return status
+    })
+    return status === 'unknown'
+      ? context.json({ error: 'hello_required' }, 409)
+      : context.body(null, 204)
+  }
+
+  app.put(CONNECTOR_PATHS.hello, limitBody, async (context) => {
+    const body = ConnectorHelloRequestSchema.safeParse(await context.req.json().catch(() => null))
+    if (!body.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    if (body.data.contract < CONNECTOR_CONTRACT_VERSION) {
+      return context.json({ error: 'client_too_old' }, 426)
+    }
+    const host = context.get('host')
+    await database.transaction(async (transaction) => {
+      await lockConnectorHost(transaction, host.hostId)
+      await hosts.recordHello(transaction, host, body.data)
+    })
+    return context.body(null, 204)
+  })
+
+  app.post(CONNECTOR_PATHS.poll, limitBody, async (context) => {
+    const body = ConnectorPollRequestSchema.safeParse(await context.req.json().catch(() => null))
+    if (!body.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    const { hostId, expiresAt } = context.get('host')
+    const { helloRev } = body.data
+    await hosts.touch(hostId)
+    const holdUntil = Math.min(Date.now() + holdMs, expiresAt)
+    let deliveries: ConnectorDelivery[] = []
+    let unsettled: Unsettled[] = []
+    for (;;) {
+      const waiter = waker.register(hostId)
+      try {
+        // Each pass decides under the host lock, so a newer hello either lands before it (and this
+        // refuses) or after it has committed what it hands out.
+        const pass = await database
+          .transaction(async (transaction) => {
+            await lockConnectorHost(transaction, hostId)
+            if ((await hosts.helloStatus(transaction, hostId, helloRev)) !== 'current') {
+              return null
+            }
+            const recovered = (await hosts.takeUnsettledDue(transaction, hostId))
+              ? await inbox.listUnsettled(transaction, hostId)
+              : []
+            const claimed = await inbox.claim(transaction, hostId)
+            // Last moment before commit: a token past its expiry rolls the pass back, so nothing
+            // goes out on it and the next token's poll sees the same inbox.
+            if (Date.now() >= expiresAt) {
+              throw new TokenExpiredDuringPoll()
+            }
+            return { recovered, ...claimed }
+          })
+          .catch((error: unknown) => {
+            if (error instanceof TokenExpiredDuringPoll) {
+              return { recovered: [], deliveries: [], nextLeaseDueAt: null }
+            }
+            throw error
+          })
+        if (pass === null) {
+          return context.json({ error: 'hello_required' }, 409)
+        }
+        deliveries = pass.deliveries
+        unsettled = pass.recovered
+        const remaining = holdUntil - Date.now()
+        const ready = deliveries.length > 0 || unsettled.length > 0
+        if (ready || remaining <= 0 || waker.draining) {
+          break
+        }
+        // A lease running out sends no notification, so wake for the next one too.
+        const leaseDueMs =
+          pass.nextLeaseDueAt === null ? remaining : Math.max(0, pass.nextLeaseDueAt - now())
+        await waiter.wait(Math.min(remaining, leaseDueMs))
+      } finally {
+        waiter.cancel()
+      }
+    }
+    const response: ConnectorPollResponse = {
+      v: 1,
+      serverNow: now(),
+      holdMs,
+      ...(unsettled.length > 0 ? { unsettled } : {}),
+      deliveries,
+      ...(waker.draining ? { control: { reconnect: true as const } } : {})
+    }
+    return context.json(response)
+  })
+
+  app.post(CONNECTOR_PATHS.acks, limitBody, async (context) => {
+    const body = ConnectorAcksRequestSchema.safeParse(await context.req.json().catch(() => null))
+    if (!body.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    return await onCurrentHello(context, body.data.helloRev, (transaction, hostId) =>
+      inbox.acknowledge(transaction, hostId, body.data.acks)
+    )
+  })
+
+  app.post(CONNECTOR_PATHS.events, limitBody, async (context) => {
+    const body = ConnectorEventsRequestSchema.safeParse(await context.req.json().catch(() => null))
+    if (!body.success) {
+      return context.json({ error: 'invalid_request' }, 400)
+    }
+    // A stale batch still answers 204: the machine has nothing to retry.
+    return await onCurrentHello(context, body.data.helloRev, async (transaction, hostId) => {
+      if (await hosts.advanceEventsSeq(transaction, hostId, body.data.seq)) {
+        await threads.applyEvents(transaction, hostId, body.data.events)
+      }
+    })
+  })
+}

cloud/apps/connector/src/queue-delivery.ts

@@ -0,0 +1,36 @@
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import { lockConnectorHost, type ConnectorHosts } from './connector-hosts.js'
+import type { ConnectorInbox, DeliveryBody, EnqueueResult } from './connector-inbox.js'
+import type { ConnectorThreads, NewThread } from './connector-threads.js'
+
+export type QueueResult = EnqueueResult | 'unknown_host' | 'not_this_hosts_thread'
+
+// One host-locked transaction finds the host, binds the thread and queues, so per-host seq is
+// assigned in order.
+export async function queueDelivery(
+  deps: {
+    database: CloudDatabase
+    inbox: ConnectorInbox
+    threads: ConnectorThreads
+    hosts: ConnectorHosts
+  },
+  input: { thread: Omit<NewThread, 'accountId'>; deliveryId: string; body: DeliveryBody }
+): Promise<QueueResult> {
+  const { hostId, externalId } = input.thread
+  return await deps.database.transaction(async (transaction) => {
+    await lockConnectorHost(transaction, hostId)
+    const accountId = await deps.hosts.accountOf(transaction, hostId)
+    if (!accountId) {
+      return 'unknown_host'
+    }
+    if (!(await deps.threads.bind(transaction, { ...input.thread, accountId }))) {
+      return 'not_this_hosts_thread'
+    }
+    return await deps.inbox.enqueue(transaction, {
+      hostId,
+      externalId,
+      deliveryId: input.deliveryId,
+      body: input.body
+    })
+  })
+}

cloud/packages/cloud-database/src/cloud-notifications.test.ts

@@ -0,0 +1,97 @@
+import { randomUUID } from 'node:crypto'
+import { describe, expect, it } from 'vitest'
+import { openCloudDatabase, openInMemoryCloudDatabase } from './cloud-database.js'
+import { openCloudNotifications } from './cloud-notifications.js'
+
+const postgresUrl = process.env.ORCA_CLOUD_DATABASE_TEST_URL
+
+function collector() {
+  const payloads: string[] = []
+  let reconnects = 0
+  let wake: (() => void) | null = null
+  return {
+    payloads,
+    reconnects: () => reconnects,
+    handlers: {
+      onNotify: (payload: string) => {
+        payloads.push(payload)
+        wake?.()
+      },
+      onReconnect: () => {
+        reconnects++
+        wake?.()
+      }
+    },
+    until: async (done: () => boolean, timeoutMs = 10_000) => {
+      const deadline = Date.now() + timeoutMs
+      while (!done()) {
+        if (Date.now() > deadline) {
+          throw new Error('timed out')
+        }
+        await new Promise<void>((resolve) => {
+          wake = resolve
+          setTimeout(resolve, 50)
+        })
+      }
+    }
+  }
+}
+
+describe('cloud database notifications', () => {
+  it('wakes SQLite listeners in this process', async () => {
+    const database = await openInMemoryCloudDatabase([])
+    const notifications = openCloudNotifications({})
+    const seen = collector()
+    const stop = await notifications.listen('inbox', seen.handlers)
+    await database.transaction((tx) => notifications.notify(tx, 'inbox', 'host-1'))
+    await stop()
+    await notifications.notify(database, 'inbox', 'host-2')
+    expect(seen.payloads).toEqual(['host-1'])
+    await database.close()
+  })
+
+  it('refuses a channel that is not a plain identifier', async () => {
+    const database = await openInMemoryCloudDatabase([])
+    await expect(
+      openCloudNotifications({}).notify(database, 'inbox; DROP', 'x')
+    ).rejects.toThrow('invalid_listen_channel')
+    await database.close()
+  })
+
+  it.skipIf(!postgresUrl)(
+    'delivers a Postgres notification at commit and re-listens after losing its connection',
+    async () => {
+      const applicationName = `cloud-db-listen-${randomUUID()}`
+      const database = await openCloudDatabase({
+        service: 'test',
+        schemaStatements: [],
+        databaseUrl: postgresUrl,
+        dataDir: '',
+        applicationName
+      })
+      const notifications = openCloudNotifications({ databaseUrl: postgresUrl, applicationName })
+      const seen = collector()
+      const stop = await notifications.listen('inbox', seen.handlers)
+      try {
+        await database.transaction(async (tx) => {
+          await notifications.notify(tx, 'inbox', 'committed')
+          await new Promise((resolve) => setTimeout(resolve, 100))
+          expect(seen.payloads).toEqual([])
+        })
+        await seen.until(() => seen.payloads.length === 1)
+
+        await database.query(
+          'SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = ?',
+          [`${applicationName}/listen`]
+        )
+        await seen.until(() => seen.reconnects() === 1)
+        await notifications.notify(database, 'inbox', 'after-reconnect')
+        await seen.until(() => seen.payloads.length === 2)
+        expect(seen.payloads).toEqual(['committed', 'after-reconnect'])
+      } finally {
+        await stop()
+        await database.close()
+      }
+    }
+  )
+})

cloud/packages/cloud-database/src/cloud-notifications.ts

@@ -0,0 +1,138 @@
+import { EventEmitter } from 'node:events'
+import pg from 'pg'
+import type { CloudDatabase } from './cloud-database.js'
+
+const LISTEN_CONNECTION_TIMEOUT_MS = 2_000
+const LISTEN_RETRY_BASE_MS = 1_000
+const LISTEN_RETRY_MAX_MS = 30_000
+// LISTEN takes no bind parameters, so a channel is a plain identifier.
+const CHANNEL_NAME = /^[a-z][a-z0-9_]{0,62}$/
+
+export type ListenHandlers = { onNotify: (payload: string) => void; onReconnect: () => void }
+
+export type CloudNotifications = {
+  // Postgres: NOTIFY on the caller's connection, so inside a transaction it is sent at commit.
+  // SQLite: wakes this process's listeners at once; they re-read, so a rollback costs one query.
+  notify(database: CloudDatabase, channel: string, payload: string): Promise<void>
+  // Postgres holds one dedicated connection and calls onReconnect after listening again,
+  // because notifications sent while it was down are lost.
+  listen(channel: string, handlers: ListenHandlers): Promise<() => Promise<void>>
+}
+
+function requireChannel(channel: string): string {
+  if (!CHANNEL_NAME.test(channel)) {
+    throw new Error('invalid_listen_channel')
+  }
+  return channel
+}
+
+// Pass the same databaseUrl as openCloudDatabase; without one, notifications stay in this process.
+export function openCloudNotifications(input: {
+  databaseUrl?: string
+  applicationName?: string
+}): CloudNotifications {
+  const local = new EventEmitter()
+  const { databaseUrl } = input
+  return {
+    notify: async (database, channel, payload) => {
+      requireChannel(channel)
+      if (database.dialect === 'postgres') {
+        await database.query('SELECT pg_notify(?, ?)', [channel, payload])
+      } else {
+        local.emit(channel, payload)
+      }
+    },
+    listen: async (channel, handlers) => {
+      requireChannel(channel)
+      if (!databaseUrl) {
+        const onNotify = (payload: string): void => handlers.onNotify(payload)
+        local.on(channel, onNotify)
+        return async () => {
+          local.off(channel, onNotify)
+        }
+      }
+      const listener = new PostgresListener(
+        {
+          connectionString: databaseUrl,
+          application_name: input.applicationName ? `${input.applicationName}/listen` : undefined,
+          connectionTimeoutMillis: LISTEN_CONNECTION_TIMEOUT_MS
+        },
+        channel,
+        handlers
+      )
+      await listener.connect()
+      return () => listener.close()
+    }
+  }
+}
+
+// One LISTEN connection outside any pool, reconnected with doubling backoff until closed.
+class PostgresListener {
+  private client: pg.Client | null = null
+  private retry: NodeJS.Timeout | undefined
+  private closed = false
+  private retryMs = LISTEN_RETRY_BASE_MS
+
+  constructor(
+    private readonly connection: pg.ClientConfig,
+    private readonly channel: string,
+    private readonly handlers: ListenHandlers
+  ) {}
+
+  async connect(): Promise<void> {
+    const client = new pg.Client(this.connection)
+    client.on('notification', (message) => {
+      if (message.channel === this.channel) {
+        this.handlers.onNotify(message.payload ?? '')
+      }
+    })
+    client.on('error', () => this.drop(client))
+    client.on('end', () => this.drop(client))
+    try {
+      await client.connect()
+      await client.query(`LISTEN ${this.channel}`)
+    } catch (error) {
+      await client.end().catch(() => undefined)
+      throw error
+    }
+    this.client = client
+    this.retryMs = LISTEN_RETRY_BASE_MS
+  }
+
+  async close(): Promise<void> {
+    this.closed = true
+    clearTimeout(this.retry)
+    await this.client?.end().catch(() => undefined)
+  }
+
+  private drop(client: pg.Client): void {
+    if (this.client !== client) {
+      return
+    }
+    this.client = null
+    void client.end().catch(() => undefined)
+    this.scheduleReconnect()
+  }
+
+  private scheduleReconnect(): void {
+    if (this.closed) {
+      return
+    }
+    this.retry = setTimeout(() => {
+      this.connect().then(
+        () => {
+          if (this.closed) {
+            void this.client?.end().catch(() => undefined)
+          } else {
+            this.handlers.onReconnect()
+          }
+        },
+        () => {
+          this.retryMs = Math.min(this.retryMs * 2, LISTEN_RETRY_MAX_MS)
+          this.scheduleReconnect()
+        }
+      )
+    }, this.retryMs)
+    this.retry.unref()
+  }
+}

cloud/packages/cloud-database/src/index.ts

@@ -5,3 +5,8 @@ export {
   type OpenCloudDatabaseInput,
   type SqlRow
 } from './cloud-database.js'
+export {
+  openCloudNotifications,
+  type CloudNotifications,
+  type ListenHandlers
+} from './cloud-notifications.js'

cloud/pnpm-lock.yaml

@@ -193,6 +193,9 @@ importers:
       '@orca-cloud/cloud-database':
         specifier: workspace:*
         version: link:../../packages/cloud-database
+      '@orca-cloud/connector-contract':
+        specifier: workspace:*
+        version: link:../../packages/connector-contract
       hono:
         specifier: ^4.13.13
         version: 4.13.13