fix(connectors): a connector backlog cannot fill orchestration's receipt ledger (#32130)

* fix(connectors): a connector backlog cannot fill orchestration's receipt ledger

Connector deliveries now refuse with delivery_backlog_full once 1,000 are
unsettled, and a new seam settles a previous process's pending connector
receipts as completed-unknown so the ledger can prune them.

* fix(connectors): settle only a previous process's receipts, not a re-enable's

The delivery receipt owner now lives for the install (one per process) and
is shared by every enabled-period seam, so toggling connectors off and on
no longer completes this process's still-running deliveries as unknown.

src/main/external-conversations/external-conversation-delivery-receipts.test.ts

@@ -0,0 +1,111 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db'
+import { MUTATION_RECEIPT_MAX_ROWS } from '../runtime/orchestration/mutation-receipt-capacity'
+import {
+  CONNECTOR_PENDING_RECEIPT_LIMIT,
+  DeliveryReceipts
+} from './external-conversation-delivery-receipts'
+import type { ConversationKey, Delivery } from './external-conversation-types'
+
+const KEY: ConversationKey = { source: 'dev', externalId: 'thread' }
+const ORDINARY = {
+  callerFingerprint: 'ordinary-user',
+  requestId: 'send',
+  method: 'orchestration.send',
+  payloadHash: 'hash'
+}
+const never = (): Promise<Delivery> => new Promise(() => {})
+
+const directories: string[] = []
+function fileDb(): { path: string; db: OrchestrationDb } {
+  const directory = mkdtempSync(join(tmpdir(), 'connector-receipts-'))
+  directories.push(directory)
+  const path = join(directory, 'orchestration.db')
+  return { path, db: new OrchestrationDb(path) }
+}
+
+afterEach(() => {
+  for (const directory of directories.splice(0)) {
+    rmSync(directory, { recursive: true, force: true })
+  }
+})
+
+describe('DeliveryReceipts ledger capacity', () => {
+  it('cannot fill orchestration’s ledger with a backlog that outlives the process', async () => {
+    const { path, db } = fileDb()
+    const receipts = new DeliveryReceipts(() => db)
+    const answers: Promise<Delivery>[] = []
+    for (let i = 0; i < MUTATION_RECEIPT_MAX_ROWS; i++) {
+      answers.push(receipts.run(KEY, String(i), 'connector.send', ['hello'], never))
+    }
+    await expect(answers[CONNECTOR_PENDING_RECEIPT_LIMIT]).resolves.toEqual({
+      outcome: 'refused',
+      reason: 'delivery_backlog_full'
+    })
+    expect(db.countPendingMutationReceipts(['connector:dev'])).toBe(CONNECTOR_PENDING_RECEIPT_LIMIT)
+    db.close()
+
+    // Process loss, then a restart that never starts connectors.
+    const reopened = new OrchestrationDb(path)
+    expect(reopened.beginMutationReceipt(ORDINARY).disposition).toBe('started')
+    reopened.close()
+  })
+
+  it('answers a recorded delivery even when the backlog is full', async () => {
+    const { db } = fileDb()
+    const receipts = new DeliveryReceipts(() => db)
+    for (let i = 0; i < CONNECTOR_PENDING_RECEIPT_LIMIT; i++) {
+      void receipts.run(KEY, String(i), 'connector.send', ['hello'], never)
+    }
+    const joined = receipts.run(KEY, '0', 'connector.send', ['hello'], never)
+    const raced = await Promise.race([joined, Promise.resolve('waiting')])
+    expect(raced).toBe('waiting')
+    db.close()
+  })
+
+  it('settles a previous process’s pending deliveries as unknown, without replaying them', async () => {
+    const { path, db } = fileDb()
+    const before = new DeliveryReceipts(() => db)
+    for (let i = 0; i < CONNECTOR_PENDING_RECEIPT_LIMIT; i++) {
+      void before.run(KEY, String(i), 'connector.send', ['hello'], never)
+    }
+    db.close()
+
+    const reopened = new OrchestrationDb(path)
+    const after = new DeliveryReceipts(() => reopened)
+    let executed = 0
+    const execute = async (): Promise<Delivery> => {
+      executed++
+      return { outcome: 'accepted' }
+    }
+    await expect(after.run(KEY, '0', 'connector.send', ['hello'], execute)).resolves.toEqual({
+      outcome: 'unknown',
+      reason: 'replayed_after_restart'
+    })
+    expect(
+      reopened.getMutationReceipt('connector:dev', JSON.stringify(['thread', '0']))?.state
+    ).toBe('completed')
+    // The settled rows no longer hold the connector's budget.
+    await expect(after.run(KEY, 'new', 'connector.send', ['hello'], execute)).resolves.toEqual({
+      outcome: 'accepted'
+    })
+    expect(executed).toBe(1)
+    expect(reopened.countPendingMutationReceipts(['connector:dev'])).toBe(0)
+    reopened.close()
+  })
+
+  it('leaves other callers’ pending receipts alone', async () => {
+    const { db } = fileDb()
+    db.beginMutationReceipt(ORDINARY)
+    await new DeliveryReceipts(() => db).run(KEY, 'd1', 'connector.send', ['hello'], async () => ({
+      outcome: 'accepted'
+    }))
+    expect(db.getMutationReceipt(ORDINARY.callerFingerprint, ORDINARY.requestId)?.state).toBe(
+      'pending'
+    )
+    db.close()
+  })
+})

src/main/external-conversations/external-conversation-delivery-receipts.ts

@@ -1,4 +1,5 @@
 import { z } from 'zod'
+import { CONNECTOR_SOURCES, type ConnectorSource } from '../../shared/external-conversation-link'
 import type { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db'
 import { OrchestrationError } from '../runtime/orchestration/orchestration-error'
 import { hashCanonical } from '../runtime/rpc/orchestration-mutation-receipt'
@@ -16,14 +17,22 @@ const RecordedDelivery = z.discriminatedUnion('outcome', [
 
 const LOST: Delivery = { outcome: 'unknown', reason: 'replayed_after_restart' }
 
+/** Unsettled connector deliveries the shared ledger holds at once, so orchestration keeps the rest. */
+export const CONNECTOR_PENDING_RECEIPT_LIMIT = 1_000
+
+const callerFingerprint = (source: ConnectorSource): string => `connector:${source}`
+const CONNECTOR_FINGERPRINTS = CONNECTOR_SOURCES.map(callerFingerprint)
+
 /**
  * A connector's delivery ids on orchestration's durable mutation receipts, under one caller
  * identity per source: the same id and payload share one answer, across restarts too, and another
  * payload is refused. A duplicate of one still running waits for it; a running one a restart
  * interrupted reads `unknown`. Refusals and throws wrote nothing, so they leave no receipt.
+ * A new delivery past `CONNECTOR_PENDING_RECEIPT_LIMIT` unsettled ones is refused unrecorded.
  */
 export class DeliveryReceipts {
   private readonly inFlight = new Map<string, Promise<Delivery>>()
+  private leftoversSettled = false
 
   constructor(private readonly db: () => DeliveryReceiptDb) {}
 
@@ -35,13 +44,25 @@ export class DeliveryReceipts {
     execute: () => Promise<Delivery>
   ): Promise<Delivery> {
     const identity = {
-      callerFingerprint: `connector:${key.source}`,
+      callerFingerprint: callerFingerprint(key.source),
       requestId: JSON.stringify([key.externalId, deliveryId]),
       method,
       payloadHash: hashCanonical(payload)
     }
     const flightKey = JSON.stringify([identity.callerFingerprint, identity.requestId])
     const db = this.db()
+    if (!this.leftoversSettled) {
+      // Pending rows from a previous process already answer only unknown; record that, never delete
+      // (they may have written), so the ledger can prune them and they leave the limit.
+      db.settlePendingMutationReceipts(CONNECTOR_FINGERPRINTS, JSON.stringify(LOST))
+      this.leftoversSettled = true
+    }
+    if (
+      !db.getMutationReceipt(identity.callerFingerprint, identity.requestId) &&
+      db.countPendingMutationReceipts(CONNECTOR_FINGERPRINTS) >= CONNECTOR_PENDING_RECEIPT_LIMIT
+    ) {
+      return Promise.resolve(refused('delivery_backlog_full'))
+    }
     let begun: ReturnType<DeliveryReceiptDb['beginMutationReceipt']>
     try {
       begun = db.beginMutationReceipt(identity)

src/main/external-conversations/external-conversation-seam.test-fixture.ts

@@ -1,5 +1,6 @@
 import type { GlobalSettings } from '../../shared/global-settings-types'
 import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db'
+import { DeliveryReceipts } from './external-conversation-delivery-receipts'
 import { ExternalConversationSeam } from './external-conversation-seam'
 import type { SessionRows } from './external-conversation-session-capture'
 import { fakeLinkStore } from './external-conversation.test-fixture'
@@ -19,7 +20,7 @@ export function harness(
     new ExternalConversationSeam({
       store: records.store,
       readSettings: () => settings,
-      receiptDb: () => db,
+      receipts: new DeliveryReceipts(() => db),
       sessionRows,
       onWorkspaceRemoved: (listener) => {
         removed.add(listener)

src/main/external-conversations/external-conversation-seam.ts

@@ -3,7 +3,7 @@ import type { ExternalConversationLink } from '../../shared/external-conversatio
 import type { OrcaRuntimeService } from '../runtime/orca-runtime'
 import { admitConnectorAgent } from './connector-agents'
 import { ConversationChain } from './external-conversation-chain'
-import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts'
+import type { DeliveryReceipts } from './external-conversation-delivery-receipts'
 import {
   findLink,
   writeLink,
@@ -44,7 +44,8 @@ export function runtimeConversationHost(runtime: OrcaRuntimeService): ExternalCo
 export type ExternalConversationSeamDeps = {
   store: ExternalConversationLinkStore
   readSettings: () => AgentStatusHooksSettings
-  receiptDb: () => DeliveryReceiptDb
+  /** Outlives the seam: a re-enable must not take this process's running deliveries for leftovers. */
+  receipts: DeliveryReceipts
   /** Worktree removals; returns the release. */
   onWorkspaceRemoved: (listener: (worktreeId: string) => void) => () => void
   /** The hook rows linked agents' sessions are read from. */
@@ -65,7 +66,6 @@ type StartInput = Omit<ConversationLaunchInput, 'agent'> & { agent: string }
 export class ExternalConversationSeam {
   private readonly abort = new AbortController()
   private readonly chain = new ConversationChain(this.abort.signal)
-  private readonly receipts: DeliveryReceipts
   /** Conversations this period started, so a second start for one shares its creation. */
   private readonly startedHere = new Set<string>()
   /** Conversations with work sent here, by the workspace their link was on. */
@@ -80,7 +80,6 @@ export class ExternalConversationSeam {
   private readonly releaseSessions: () => void
 
   constructor(private readonly deps: ExternalConversationSeamDeps) {
-    this.receipts = new DeliveryReceipts(deps.receiptDb)
     this.sessions = new LinkedSessionCapture(deps.store, deps.sessionRows)
     // A removed workspace may keep its PTY registered a while; its held and queued sends end now.
     // The event names no host, so a same-path twin's removal is told apart by the link surviving.
@@ -116,7 +115,7 @@ export class ExternalConversationSeam {
     deliveryId: string,
     input: StartInput
   ): Promise<ConversationStart> {
-    const initial = await this.receipts.run(
+    const initial = await this.deps.receipts.run(
       key,
       deliveryId,
       'connector.start',
@@ -168,7 +167,7 @@ export class ExternalConversationSeam {
     text: string
   ): Promise<Delivery> {
     const id = conversationId(key)
-    return this.receipts.run(key, deliveryId, 'connector.send', [text], () => {
+    return this.deps.receipts.run(key, deliveryId, 'connector.send', [text], () => {
       const queued = findLink(this.deps.store, key)
       if (queued) {
         this.linked.set(id, { key, worktreeId: queued.workspace.worktreeId })

src/main/external-conversations/external-conversation-types.ts

@@ -8,6 +8,7 @@ export const CONNECTOR_REFUSALS = [
   'agent_not_reachable',
   'unsupported_agent',
   'delivery_id_reused',
+  'delivery_backlog_full',
   // The three a closed agent's reopen ends on; each asks the user to open it in Orca.
   'agent_asleep_in_window',
   'no_session_to_reopen',

src/main/external-conversations/external-conversations-install.test.ts

@@ -5,7 +5,10 @@ import type { RuntimeWorktreeLifecycleEvent } from '../runtime/orca-runtime-core
 import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db'
 import { installExternalConversations } from './external-conversations-install'
 import {
+  fakeLauncher,
   fakeLinkStore,
+  fakeTerminal,
+  launchResult,
   PANE,
   SESSION,
   sessionRow,
@@ -98,6 +101,44 @@ describe('installExternalConversations', () => {
     expect(host.lifecycle.size).toBe(1)
   })
 
+  it('leaves a delivery still running across a disable and re-enable to its owner', async () => {
+    const { store, update } = fakeStore({ experimentalConnectors: true })
+    const host = runtime()
+    const db = host.getOrchestrationDb()
+    const connectors = installExternalConversations(store, host)
+    let release = (): void => {}
+    const held = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    const { launcher, launch } = fakeLauncher(async () => {
+      await held
+      return launchResult('handed-to-terminal')
+    })
+    const conversation = { launcher, terminal: fakeTerminal().terminal }
+    const key = { source: 'dev', externalId: 'thread' } as const
+    const input = {
+      agent: 'claude',
+      target: { kind: 'existing' as const, worktree: `id:${WORKTREE}` },
+      prompt: 'hi'
+    }
+    const live = connectors.current()?.start(conversation, key, 'd1', input)
+    await vi.waitFor(() => expect(launch).toHaveBeenCalledTimes(1))
+
+    update({ experimentalConnectors: false })
+    update({ experimentalConnectors: true })
+    const retry = connectors.current()?.start(conversation, key, 'd1', input)
+    const requestId = JSON.stringify(['thread', 'd1'])
+    expect(db.getMutationReceipt('connector:dev', requestId)?.state).toBe('pending')
+
+    release()
+    const [first, second] = await Promise.all([live, retry])
+    expect(second?.initial).toEqual(first?.initial)
+    expect(db.getMutationReceipt('connector:dev', requestId)?.receipt).toBe(
+      JSON.stringify(first?.initial)
+    )
+    expect(launch).toHaveBeenCalledTimes(1)
+  })
+
   it('keeps the seam through unrelated settings changes', () => {
     const { store, update } = fakeStore({ experimentalConnectors: true })
     const connectors = installExternalConversations(store, runtime())

src/main/external-conversations/external-conversations-install.ts

@@ -1,6 +1,6 @@
 import type { GlobalSettings } from '../../shared/global-settings-types'
 import type { RuntimeWorktreeLifecycleEvent } from '../runtime/orca-runtime-core'
-import type { DeliveryReceiptDb } from './external-conversation-delivery-receipts'
+import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts'
 import type { ExternalConversationLinkStore } from './external-conversation-links'
 import {
   ExternalConversationSeam,
@@ -34,6 +34,8 @@ export function installExternalConversations(
   if (!store) {
     return { current: () => null }
   }
+  // One per process, so leftover settlement only reaches a previous process's receipts.
+  const receipts = new DeliveryReceipts(() => runtime.getOrchestrationDb())
   let seam: ExternalConversationSeam | null = null
   let reconciled = false
   const reconcile = (): void => {
@@ -43,7 +45,7 @@ export function installExternalConversations(
       seam = new ExternalConversationSeam({
         store,
         readSettings: () => store.getSettings(),
-        receiptDb: () => runtime.getOrchestrationDb(),
+        receipts,
         onWorkspaceRemoved: (listener) =>
           runtime.onWorktreeLifecycle((event) => {
             if (event.kind === 'removed') {

src/main/runtime/orchestration/db/mutation-receipts/mutation-receipt-store.ts

@@ -157,6 +157,34 @@ export function discardPendingMutationReceipt(
     .run(callerFingerprint, requestId)
 }
 
+export function countPendingMutationReceipts(
+  this: OrchestrationDb,
+  callerFingerprints: readonly string[]
+): number {
+  const count = this.db
+    .prepare(
+      `SELECT COUNT(*) AS count FROM mutation_receipts
+       WHERE caller_fingerprint IN (SELECT value FROM json_each(?)) AND state = 'pending'`
+    )
+    .get(JSON.stringify(callerFingerprints))?.count
+  return Number(count ?? 0)
+}
+
+/** Completes these callers' pending rows with `receipt`, for operations no running work owns. */
+export function settlePendingMutationReceipts(
+  this: OrchestrationDb,
+  callerFingerprints: readonly string[],
+  receipt: string
+): void {
+  this.db
+    .prepare(
+      `UPDATE mutation_receipts
+       SET state = 'completed', receipt = ?, updated_at = datetime('now')
+       WHERE caller_fingerprint IN (SELECT value FROM json_each(?)) AND state = 'pending'`
+    )
+    .run(receipt, JSON.stringify(callerFingerprints))
+}
+
 export function getMutationReceipt(
   this: OrchestrationDb,
   callerFingerprint: string,
@@ -176,6 +204,8 @@ export type MutationReceiptStoreMethods = {
   completeMutationReceipt: typeof completeMutationReceipt
   checkpointPendingMutationReceipt: typeof checkpointPendingMutationReceipt
   discardPendingMutationReceipt: typeof discardPendingMutationReceipt
+  countPendingMutationReceipts: typeof countPendingMutationReceipts
+  settlePendingMutationReceipts: typeof settlePendingMutationReceipts
   getMutationReceipt: typeof getMutationReceipt
 }
 
@@ -186,6 +216,8 @@ export function attachMutationReceiptStore(ctor: { prototype: object }): void {
     completeMutationReceipt,
     checkpointPendingMutationReceipt,
     discardPendingMutationReceipt,
+    countPendingMutationReceipts,
+    settlePendingMutationReceipts,
     getMutationReceipt
   })
 }