fix: persist mailbox prompt attempt ownership

This commit is contained in:
Jinwoo-H
2026-08-23 22:38:36 -04:00
parent 8e56f9670b
commit e2f655e77c
19 changed files with 433 additions and 119 deletions
+2 -1
View File
@@ -9747,7 +9747,7 @@
"invariant": "A public CLI send containing non-empty text plus Enter and no interrupt identifies agent-prompt intent. The provider PTY-input owner grants one generation-aware prompt transaction across renderer, RPC/mobile, orchestration, startup, and internal writers. Admitted user input has priority: it safely closes an open prompt or automation paste, invalidates the unacknowledged prompt, and proceeds without declaring the provider unavailable; unversioned large renderer input cannot stale-invalidate the owner. Complete query replies are deferred only until the paste frame closes, with overflow preempting while preserving reply order. The runtime revalidates the foreground target before at most one Enter, and only fresh lifecycle evidence acknowledges delivery. Any unacknowledged post-byte failure is operation_unknown, and orchestration retains the exact terminal in start_unknown rather than transferring or retrying it.",
"oracle": "Withhold OpenCode readiness beyond the old 500 ms deadline, emit a stale pre-attempt marker and then one split fresh marker, and require zero premature Enter plus one lifecycle-acknowledged Enter. A swallowed Enter followed by autonomous redraw output must remain operation_unknown. Missing readiness emits no Enter but remains operation_unknown because paste bytes reached the composer. Current foreground identity must override stale launch metadata, a switch to shell before submit must emit no Enter, and unavailable legacy inspection must retain bounded metadata fallback. Manual input must close an open paste, cancel prompt authority, and proceed without WriteUnavailable recovery; query replies must be delivered after the paste frame. Partial-paste, post-paste, and post-Enter generation reset must remain unknown; an unknown attempt must prevent already-queued prompt delivery. Local and federated receipts persist operation_unknown as start_unknown, and the exact local terminal is not transferable.",
"commands": [
"pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts",
"pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/mailbox-pointer-state.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts",
"pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/orca-runtime.test.ts src/shared/agent-prompt-injection.test.ts",
"pnpm run build:cli && pnpm exec playwright test tests/e2e/terminal-send-agent-prompt-submit.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1",
"pnpm.cmd exec playwright test tests/e2e/terminal-send-agent-prompt-submit.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1",
@@ -9950,6 +9950,7 @@
"commands": [
"pnpm exec vitest run --config config/vitest.config.ts src/shared/agent-prompt-injection.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/mailbox-pointer-state.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/main/runtime/rpc/methods/orchestration-tasks-dispatch.test.ts src/main/runtime/orchestration/coordinator.test.ts",
"pnpm exec vitest run --config config/vitest.config.ts src/shared/agent-prompt-injection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/rpc/methods/orchestration-tasks-dispatch.test.ts src/main/runtime/orchestration/coordinator.test.ts",
"pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/mailbox-pointer-state.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts",
"node tests/tools/repro-orchestration-long-prompt.mjs --cli out/bin/orca-dev --mode codex-like --size-kb 32 --timeout-ms 20000"
],
"testFiles": [
@@ -80,6 +80,13 @@ export function releaseOwnerToken<T extends GenerationToken>(
ownerByPtyId.delete(ptyId)
}
}
export function releaseDormantOwner<T extends GenerationToken>(
owner: OwnedTokenState<T>,
flush: (owner: OwnedTokenState<T>) => boolean
): boolean {
return owner.tokens.size === 0 && !owner.pasteOpen && !owner.pasteMutationPending && flush(owner)
}
export function invalidateOwner<T extends GenerationToken, R extends string>(
ownerByPtyId: Map<string, OwnedTokenState<T>>,
ptyId: string,
@@ -99,6 +106,17 @@ export function invalidateOwner<T extends GenerationToken, R extends string>(
export function createTransactionToken(generation: number | null): TransactionToken {
return { generation, active: true, reason: null, invalidate: () => undefined }
}
export function invalidateTransactionToken(
token: TransactionToken,
reason: PtyInputInvalidationReason
): void {
if (!token.active) {
return
}
token.active = false
token.reason = reason
token.invalidate(reason)
}
export function transactionForToken(
token: TransactionToken,
operations: { write: (data: string) => PtyInputWriteResult; release: () => void }
+19
View File
@@ -40,6 +40,25 @@ export function flushQueryReplies(
state.pendingQueryReplyCodeUnits = 0
return true
}
export function writeQueuedQueryReply(
state: PendingQueryReplyState,
data: string,
write: (data: string) => boolean
): boolean {
if (state.pendingQueryReplies.length > 0 && !flushQueryReplies(state, write)) {
if (canQueueQueryReply(state, data)) {
enqueueQueryReply(state, data)
return true
}
return false
}
const wrote = write(data)
if (!wrote && canQueueQueryReply(state, data)) {
enqueueQueryReply(state, data)
return true
}
return wrote
}
export function nextPasteState(initial: boolean, data: string): boolean {
let open = initial
let cursor = 0
+72
View File
@@ -212,6 +212,78 @@ describe('PTY input transaction owner', () => {
expect(prompt.active).toBe(true)
})
it('rejects a concurrent direct client while a renderer transaction owns the PTY', () => {
const writer = vi.fn(() => true)
const owner = new PtyInputTransactionOwner(writer)
const renderer = owner.beginUnversionedInteractive('pty-1')
expect(renderer.write('first chunk')).toBe(true)
expect(owner.write('pty-1', 'concurrent client')).toBe(false)
expect(writer).toHaveBeenCalledOnce()
})
it('queues terminal query replies behind an interactive paste transaction', () => {
const writes: string[] = []
const owner = new PtyInputTransactionOwner((_ptyId, data) => {
writes.push(data)
return true
})
const renderer = owner.beginUnversionedInteractive('pty-1')
expect(renderer.write(`${START}chunk`)).toBe(true)
expect(owner.write('pty-1', '\x1b[6n')).toBe(true)
expect(owner.write('pty-1', 'concurrent client')).toBe(false)
expect(renderer.write(END)).toBe(true)
renderer.release()
expect(writes).toEqual([`${START}chunk`, END, '\x1b[6n'])
})
it('retries a deferred query reply before admitting a replacement transaction', () => {
let failReply = true
const writer = vi.fn((_ptyId: string, data: string) => {
if (data === '\x1b[6n' && failReply) {
return false
}
return true
})
const owner = new PtyInputTransactionOwner(writer)
const prompt = owner.begin('pty-1', 1, 'agent-prompt')!
expect(prompt.write(`${START}prompt`)).toBe(true)
expect(owner.write('pty-1', '\x1b[6n')).toBe(true)
expect(prompt.write(END)).toBe(true)
prompt.release()
failReply = false
const replacement = owner.begin('pty-1', 1, 'agent-prompt')
expect(replacement).not.toBeNull()
expect(writer.mock.calls).toEqual([
['pty-1', `${START}prompt`],
['pty-1', END],
['pty-1', '\x1b[6n'],
['pty-1', '\x1b[6n'],
['pty-1', '\x1b[6n']
])
})
it('keeps a newly arrived query reply behind a deferred reply', () => {
let failReply = true
const writes: string[] = []
const owner = new PtyInputTransactionOwner((_ptyId, data) => {
writes.push(data)
return !failReply || data !== '\x1b[6n'
})
const prompt = owner.begin('pty-1', 1, 'agent-prompt')!
prompt.write(`${START}prompt`)
owner.write('pty-1', '\x1b[6n')
prompt.write(END)
prompt.release()
expect(owner.write('pty-1', '\x1b[5n')).toBe(true)
failReply = false
expect(owner.begin('pty-1', 1, 'agent-prompt')).not.toBeNull()
expect(writes.slice(-2)).toEqual(['\x1b[6n', '\x1b[5n'])
})
it('joins interactive transactions only when each attempts a write', () => {
const owner = new PtyInputTransactionOwner(() => true)
const first = owner.begin('pty-1', 1, 'interactive')!
+31 -26
View File
@@ -3,6 +3,7 @@ import {
canQueueQueryReply,
enqueueQueryReply,
flushQueryReplies,
writeQueuedQueryReply,
nextPasteState,
isPtyInputTransactionQueryReply
} from './pty-input-transaction-state'
@@ -14,7 +15,9 @@ import {
pruneLazyTokens,
releaseLazyToken,
releaseOwnerToken,
transactionForToken
releaseDormantOwner,
transactionForToken,
invalidateTransactionToken
} from './pty-input-transaction-lifecycle'
import type { TransactionToken } from './pty-input-transaction-lifecycle'
export type PtyInputTransactionKind = 'agent-prompt' | 'automation' | 'interactive'
@@ -29,11 +32,10 @@ export type PtyInputTransaction = Readonly<{
}>
type PtyInputWriter = (ptyId: string, data: string) => boolean
type SettledPtyInputWriter = (ptyId: string, data: string) => PtyInputWriteResult
type InputToken = TransactionToken
type InputOwner = {
generation: number | null
kind: PtyInputTransactionKind
tokens: Set<InputToken>
tokens: Set<TransactionToken>
pasteOpen: boolean
pasteMutationPending: boolean
pendingQueryReplies: string[]
@@ -41,25 +43,28 @@ type InputOwner = {
}
export class PtyInputTransactionOwner {
private readonly ownerByPtyId = new Map<string, InputOwner>()
private readonly lazyInteractiveByPtyId = new Map<string, Set<InputToken>>()
private readonly lazyInteractiveByPtyId = new Map<string, Set<TransactionToken>>()
constructor(
private readonly writeInput: PtyInputWriter,
private readonly writeInputWithSettlement: SettledPtyInputWriter = writeInput
) {}
write(ptyId: string, data: string): boolean {
const owner = this.ownerByPtyId.get(ptyId)
if (!owner || owner.kind === 'interactive') {
if (!owner) {
return this.writeInput(ptyId, data)
}
if (isPtyInputTransactionQueryReply(data)) {
if (!owner.pasteOpen && !owner.pasteMutationPending) {
return this.writeInput(ptyId, data)
return writeQueuedQueryReply(owner, data, (reply) => this.writeInput(ptyId, reply))
}
if (canQueueQueryReply(owner, data)) {
enqueueQueryReply(owner, data)
return true
}
}
if (owner.kind === 'interactive') {
return false
}
if (!this.interruptOwner(ptyId, owner)) {
return false
}
@@ -77,12 +82,21 @@ export class PtyInputTransactionOwner {
ptyId,
generation,
undefined,
(token, reason) => this.invalidateToken(token, reason),
invalidateTransactionToken,
'terminal_handle_stale'
)
}
return this.beginLazyInteractive(ptyId, generation)
}
const existing = this.ownerByPtyId.get(ptyId)
if (
existing &&
releaseDormantOwner(existing, (state) =>
flushQueryReplies(state, (reply) => this.writeInput(ptyId, reply))
)
) {
this.ownerByPtyId.delete(ptyId)
}
if (this.ownerByPtyId.has(ptyId)) {
return null
}
@@ -91,7 +105,7 @@ export class PtyInputTransactionOwner {
ptyId,
generation,
undefined,
(token, reason) => this.invalidateToken(token, reason),
invalidateTransactionToken,
'terminal_handle_stale'
)
const owner: InputOwner = {
@@ -113,19 +127,19 @@ export class PtyInputTransactionOwner {
const owner = this.ownerByPtyId.get(ptyId)
if (owner && (owner.generation === null || owner.generation !== generation)) {
invalidateOwner(this.ownerByPtyId, ptyId, owner, 'terminal_handle_stale', (token, reason) =>
this.invalidateToken(token, reason)
invalidateTransactionToken(token, reason)
)
}
for (const token of this.lazyInteractiveByPtyId.get(ptyId) ?? []) {
if (token.active && (token.generation === null || token.generation !== generation)) {
this.invalidateToken(token, 'terminal_handle_stale')
invalidateTransactionToken(token, 'terminal_handle_stale')
}
}
pruneLazyTokens(this.lazyInteractiveByPtyId, ptyId)
}
private beginLazyInteractive(ptyId: string, generation: number | null): PtyInputTransaction {
const token = createTransactionToken(generation)
const lazy = this.lazyInteractiveByPtyId.get(ptyId) ?? new Set<InputToken>()
const lazy = this.lazyInteractiveByPtyId.get(ptyId) ?? new Set<TransactionToken>()
lazy.add(token)
this.lazyInteractiveByPtyId.set(ptyId, lazy)
let joinedOwner: InputOwner | null = null
@@ -154,7 +168,7 @@ export class PtyInputTransactionOwner {
private claimInteractive(
ptyId: string,
generation: number | null,
lazyToken: InputToken
lazyToken: TransactionToken
): InputOwner | null {
const current = this.ownerByPtyId.get(ptyId)
if (
@@ -164,7 +178,7 @@ export class PtyInputTransactionOwner {
current.generation !== generation
) {
releaseLazyToken(this.lazyInteractiveByPtyId, ptyId, lazyToken, false)
this.invalidateToken(lazyToken, 'terminal_handle_stale')
invalidateTransactionToken(lazyToken, 'terminal_handle_stale')
return null
}
if (current && current.kind !== 'interactive' && !this.interruptOwner(ptyId, current)) {
@@ -188,7 +202,7 @@ export class PtyInputTransactionOwner {
ptyId,
generation,
lazyToken,
(token, reason) => this.invalidateToken(token, reason),
invalidateTransactionToken,
'terminal_handle_stale'
)
}
@@ -214,7 +228,7 @@ export class PtyInputTransactionOwner {
private writeOwned(
ptyId: string,
owner: InputOwner,
token: InputToken,
token: TransactionToken,
data: string
): PtyInputWriteResult {
if (!token.active || this.ownerByPtyId.get(ptyId) !== owner || !owner.tokens.has(token)) {
@@ -275,21 +289,12 @@ export class PtyInputTransactionOwner {
return false
}
invalidateOwner(this.ownerByPtyId, ptyId, owner, 'terminal_input_superseded', (token, reason) =>
this.invalidateToken(token, reason)
invalidateTransactionToken(token, reason)
)
return true
}
private releaseOwnerToken(ptyId: string, owner: InputOwner, token: InputToken): void {
private releaseOwnerToken = (ptyId: string, owner: InputOwner, token: TransactionToken): void =>
releaseOwnerToken(this.ownerByPtyId, ptyId, owner, token, (current) =>
flushQueryReplies(current, (reply) => this.writeInput(ptyId, reply))
)
}
private invalidateToken(token: InputToken, reason: PtyInputInvalidationReason): void {
if (!token.active) {
return
}
token.active = false
token.reason = reason
token.invalidate(reason)
}
}
@@ -218,10 +218,19 @@ describe('agent prompt delivery state machine', () => {
expect(writes).toEqual([])
})
it('rejects an authoritative missing foreground before writing', async () => {
it('rejects a null foreground observation instead of trusting launch metadata', async () => {
vi.useFakeTimers()
const { runtime, handle, writes } = await createPromptRuntime({
launchAgent: 'opencode',
inspectProcess: () => ({ foregroundProcess: null, hasChildProcesses: false })
inspectProcess: () => ({ foregroundProcess: null, hasChildProcesses: false }),
onWrite: (runtime, data) => {
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now())
}
if (data === '\r') {
runtime.onPtyData('pty-prompt', '\x1b]0;OpenCode working\x07', Date.now())
}
}
})
await expect(runtime.sendTerminalAgentPrompt(handle, 'review this')).rejects.toThrow(
@@ -230,6 +239,31 @@ describe('agent prompt delivery state machine', () => {
expect(writes).toEqual([])
})
it('uses launch metadata only when inspection is explicitly unavailable', async () => {
vi.useFakeTimers()
const { runtime, handle, writes } = await createPromptRuntime({
launchAgent: 'opencode',
inspectProcess: () => ({
foregroundProcess: null,
hasChildProcesses: true,
unavailable: true
}),
onWrite: (runtime, data) => {
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now())
}
if (data === '\r') {
runtime.onPtyData('pty-prompt', '\x1b]0;OpenCode working\x07', Date.now())
}
}
})
const submission = runtime.sendTerminalAgentPrompt(handle, 'review this')
await vi.runAllTimersAsync()
await expect(submission).resolves.toMatchObject({ accepted: true })
expect(writes).toContain('\r')
})
it('does not submit after the agent returns to a shell', async () => {
vi.useFakeTimers()
let foregroundProcess = 'opencode'
+15 -4
View File
@@ -3237,19 +3237,26 @@ export class OrcaRuntimeService {
getLiveLeafForHandle: (handle) => this.getLiveLeafForHandle(handle).leaf,
getMessageWaiters: (mailboxHandle) => this.messageWaitersByHandle.get(mailboxHandle),
getTabTitle: (tabId) => this.tabs.get(tabId)?.title,
getTerminalHandleForLeaf: (leaf) => this.issueHandle(leaf as RuntimeLeafRecord),
getTerminalHandleForLeaf: (leaf) =>
this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId)),
redriveMailbox: (mailboxHandle, reservedTypes) =>
this.deliverPendingMessagesForHandle(mailboxHandle, reservedTypes),
sendPrompt: async (handle, prompt, options) => {
let providerAttempted = false
try {
await this.sendTerminalAgentPrompt(handle, prompt, {
signal: options.signal,
beforeWrite: options.beforeWrite,
beforeAttempt: options.beforeAttempt
beforeAttempt: () => {
options.beforeAttempt()
providerAttempted = true
}
})
return 'delivered'
} catch (error) {
return isAgentPromptDeliveryUnknownError(error) ? 'unknown' : 'rejected'
return providerAttempted || isAgentPromptDeliveryUnknownError(error)
? 'unknown'
: 'rejected'
}
}
})
@@ -6209,7 +6216,10 @@ export class OrcaRuntimeService {
// Why: CLI terminal writes must go through the main-owned PTY registry
// instead of tunneling back through renderer IPC, or live handles could
// drift from the process they are supposed to control during reloads.
if (controller && !controller.beginInputTransaction) {
if (
controller &&
(!controller.beginInputTransaction || !controller.invalidateInputTransactions)
) {
const write = controller.write.bind(controller)
const writeWithSettlement = controller.writeWithSettlement?.bind(controller)
const inputOwner = new PtyInputTransactionOwner(
@@ -19807,6 +19817,7 @@ export class OrcaRuntimeService {
try {
const inspection = await controller.inspectProcess(ptyId)
return {
// Null is a valid idle-shell observation; only the explicit unavailable bit permits fallback.
available: inspection.unavailable !== true,
process: inspection.foregroundProcess
}
@@ -88,6 +88,7 @@ describe('orchestration mailbox prompt settlement', () => {
}
expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
expect(db.getMessageById(message.id)?.delivery_state).toBe('unknown')
const restarted = createRuntime(db)
await driveToLiveIdle(restarted.runtime)
expect(pointerCount(restarted.write)).toBe(0)
@@ -106,7 +107,7 @@ describe('orchestration mailbox prompt settlement', () => {
})
const run = createBoundRun(db, 'Staging failure Run')
const message = insertDirectRunMessage(db, run.id, 'Must remain explicit')
vi.spyOn(db, 'markAsDelivered').mockImplementation(() => {
vi.spyOn(db, 'markAsDeliveryStaged').mockImplementation(() => {
throw new Error('disk unavailable')
})
@@ -6,5 +6,5 @@ export const LEGACY_RUN_ID = ORCHESTRATION_LEGACY_RUN_ID
export const LEGACY_CONTRACT_VERSION = 0
export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION
// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity.
export const SCHEMA_VERSION = 29
// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 durable scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v30 durable mailbox delivery attempts.
export const SCHEMA_VERSION = 30
@@ -96,7 +96,7 @@ export function getUndeliveredUnreadMessages(
const conditions = [
'to_handle = ?',
'read = 0',
'delivered_at IS NULL',
"COALESCE(delivery_state, CASE WHEN delivered_at IS NULL THEN 'none' ELSE 'delivered' END) = 'none'",
"delivery_contract = 'current_delivery'"
]
const params: (string | number)[] = [toHandle]
@@ -128,7 +128,7 @@ export function getUndeliveredUnreadMailboxHandles(this: OrchestrationDb): strin
this.db
.prepare(
`SELECT DISTINCT to_handle FROM messages
WHERE read = 0 AND delivered_at IS NULL
WHERE read = 0 AND COALESCE(delivery_state, CASE WHEN delivered_at IS NULL THEN 'none' ELSE 'delivered' END) = 'none'
AND delivery_contract = 'current_delivery'`
)
.all() as { to_handle: string }[]
@@ -164,7 +164,7 @@ export function markAsDelivered(this: OrchestrationDb, ids: string[]): void {
this,
ids,
(placeholders) =>
`UPDATE messages SET delivered_at = datetime('now') WHERE id IN (${placeholders})`
`UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'delivered' WHERE id IN (${placeholders})`
)
}
@@ -173,11 +173,29 @@ export function markAsUndelivered(this: OrchestrationDb, ids: string[]): void {
this,
ids,
(placeholders) =>
`UPDATE messages SET delivered_at = NULL
`UPDATE messages SET delivered_at = NULL, delivery_state = 'none'
WHERE read = 0 AND id IN (${placeholders})`
)
}
export function markAsDeliveryStaged(this: OrchestrationDb, ids: string[]): void {
runBatchedMessageMutation(
this,
ids,
(placeholders) =>
`UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'staged' WHERE read = 0 AND id IN (${placeholders})`
)
}
export function markAsDeliveryUnknown(this: OrchestrationDb, ids: string[]): void {
runBatchedMessageMutation(
this,
ids,
(placeholders) =>
`UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'unknown' WHERE read = 0 AND id IN (${placeholders})`
)
}
export function areUnreadMessages(this: OrchestrationDb, toHandle: string, ids: string[]): boolean {
let matched = 0
for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {
@@ -201,7 +219,7 @@ export function markAsReadAndDelivered(this: OrchestrationDb, ids: string[]): vo
this,
ids,
(placeholders) =>
`UPDATE messages SET read = 1, delivered_at = COALESCE(delivered_at, datetime('now')) WHERE id IN (${placeholders})`
`UPDATE messages SET read = 1, delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'delivered' WHERE id IN (${placeholders})`
)
}
@@ -270,6 +288,8 @@ export type MessageInboxMethods = {
markAsRead: typeof markAsRead
markAsDelivered: typeof markAsDelivered
markAsUndelivered: typeof markAsUndelivered
markAsDeliveryStaged: typeof markAsDeliveryStaged
markAsDeliveryUnknown: typeof markAsDeliveryUnknown
areUnreadMessages: typeof areUnreadMessages
markAsReadAndDelivered: typeof markAsReadAndDelivered
getInbox: typeof getInbox
@@ -288,6 +308,8 @@ export function attachMessageInbox(ctor: { prototype: object }): void {
markAsRead,
markAsDelivered,
markAsUndelivered,
markAsDeliveryStaged,
markAsDeliveryUnknown,
areUnreadMessages,
markAsReadAndDelivered,
getInbox,
@@ -36,6 +36,8 @@ CREATE TABLE IF NOT EXISTS messages (
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
delivered_at TEXT,
delivery_state TEXT NOT NULL DEFAULT 'none'
CHECK(delivery_state IN ('none', 'staged', 'delivered', 'unknown')),
sender_pane_key TEXT
);
@@ -16,6 +16,23 @@ export function migrate(this: OrchestrationDb): void {
try {
applySchemaMigrationsV2ToV12.call(this, current)
applySchemaMigrationsV13ToV29.call(this, current)
if (current < 30) {
if (!this.hasColumn('messages', 'delivery_state')) {
this.db.exec(`ALTER TABLE messages ADD COLUMN delivery_state TEXT NOT NULL DEFAULT 'none'`)
}
this.db.exec(`
UPDATE messages
SET delivery_state = CASE
WHEN delivered_at IS NULL THEN 'none'
ELSE 'delivered'
END
WHERE delivery_state = 'none' AND delivered_at IS NOT NULL
`)
this.db.exec(`
CREATE INDEX IF NOT EXISTS idx_messages_delivery_state
ON messages(to_handle, read, delivery_state, sequence)
`)
}
this.db.pragma(`user_version = ${SCHEMA_VERSION}`)
this.db.exec('COMMIT')
} catch (err) {
@@ -12,7 +12,7 @@ export type PointerDeliveryDependencies<TWaiter extends OrchestrationMessageWait
getLiveLeafForHandle: (handle: string) => OrchestrationMailboxLeaf
getMessageWaiters: (mailboxHandle: string) => ReadonlySet<TWaiter> | undefined
getTabTitle: (tabId: string) => string | null | undefined
getTerminalHandleForLeaf: (leaf: OrchestrationMailboxLeaf) => string
getTerminalHandleForLeaf: (leaf: OrchestrationMailboxLeaf) => string | undefined
redriveMailbox: (mailboxHandle: string, reservedTypes?: ReadonlySet<string>) => void
sendPrompt: (
handle: string,
@@ -0,0 +1,50 @@
import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-contract'
import { shouldReleaseOrchestrationPointer } from './mailbox-pointer-eligibility'
import type { OrchestrationMailboxLeaf } from './mailbox-owner'
import type {
OrchestrationMailboxDeliveryFlight,
OrchestrationMailboxPointerState
} from './mailbox-pointer-state'
export function markMailboxDeliveryDelivered(
db: { markAsDelivered: (messageIds: string[]) => void },
messageIds: readonly string[],
usedLegacyDeliveredTransition: boolean
): void {
if (messageIds.length > 0 && !usedLegacyDeliveredTransition) {
db.markAsDelivered([...messageIds])
}
}
export function assertMailboxPointerDeliveryCurrent<
TWaiter extends { typeFilter: string[] | undefined }
>(
deps: PointerDeliveryDependencies<TWaiter>,
state: OrchestrationMailboxPointerState,
leaf: OrchestrationMailboxLeaf,
mailboxHandle: string,
unread: readonly { id: string; type: string; sequence: number }[],
ptyId: string,
flight: OrchestrationMailboxDeliveryFlight
): void {
const currentLeaf = deps.getLeaf(deps.getLeafKey(leaf.tabId, leaf.leafId))
const ownerChanged =
!state.isCurrentFlight(ptyId, flight) ||
currentLeaf?.ptyId !== ptyId ||
!currentLeaf.writable ||
deps.mailboxOwner.resolve(currentLeaf) !== mailboxHandle
const agentNoLongerIdle =
currentLeaf?.lastAgentStatus !== 'idle' || !currentLeaf.lastAgentStatusObservedLive
const messagesReleased = shouldReleaseOrchestrationPointer(
deps.getDb(),
mailboxHandle,
unread,
deps.getMessageWaiters(mailboxHandle)
)
if (!ownerChanged && !agentNoLongerIdle && !messagesReleased) {
return
}
if (!flight.mutated && ownerChanged) {
flight.redriveMailbox = mailboxHandle
}
throw new Error('orchestration_pointer_released')
}
@@ -5,7 +5,6 @@ import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-con
import {
hasUnfilteredOrchestrationWaiter,
messageTypeHasOrchestrationWaiter,
shouldReleaseOrchestrationPointer,
type OrchestrationMessageWaiter
} from './mailbox-pointer-eligibility'
import type { OrchestrationMailboxLeaf } from './mailbox-owner'
@@ -13,6 +12,11 @@ import {
OrchestrationMailboxPointerState,
type OrchestrationMailboxDeliveryFlight
} from './mailbox-pointer-state'
import { redriveMailboxPointer } from './mailbox-pointer-redrive'
import {
assertMailboxPointerDeliveryCurrent,
markMailboxDeliveryDelivered
} from './mailbox-pointer-delivery-validation'
export type { OrchestrationMessageWaiter } from './mailbox-pointer-eligibility'
export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMessageWaiter> {
private readonly state = new OrchestrationMailboxPointerState()
@@ -44,7 +48,7 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
const terminalHandle = this.deps.getTerminalHandleForLeaf(leaf)
const db = this.deps.getDb()
const mailboxHandle = options.mailboxHandle
if (!db || !mailboxHandle.startsWith('run:')) {
if (!db || !mailboxHandle.startsWith('run:') || !terminalHandle) {
return
}
if (db.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))) {
@@ -55,7 +59,7 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
return
}
if (this.state.hasActiveWatermark(mailboxHandle)) {
this.parkRedelivery(mailboxHandle, options.reservedTypes)
this.state.parkRedelivery(mailboxHandle, options.reservedTypes)
return
}
const waiters = this.deps.getMessageWaiters(mailboxHandle)
@@ -86,7 +90,14 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
if (newestSequence === undefined) {
return
}
if (!this.state.releaseSupersededWatermark(mailboxHandle, newestSequence)) {
if (
!this.state.releaseSupersededWatermark(
mailboxHandle,
newestSequence,
leaf.ptyId,
this.deps.getLeafKey(leaf.tabId, leaf.leafId)
)
) {
return
}
if (
@@ -102,9 +113,6 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
}
this.stagePointer(leaf, terminalHandle, mailboxHandle, unread, newestSequence)
}
parkRedelivery(mailboxHandle: string, reservedTypes?: ReadonlySet<string>): void {
this.state.parkRedelivery(mailboxHandle, reservedTypes)
}
retirePty(ptyId: string): void {
const { flight, releasedMailboxes } = this.state.retirePty(ptyId)
flight?.abortController.abort()
@@ -117,7 +125,7 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
ptyId: string,
mailboxHandle: string
): void {
const currentLeaf = this.deps.getLeaf(this.leafKey(leaf))
const currentLeaf = this.deps.getLeaf(this.deps.getLeafKey(leaf.tabId, leaf.leafId))
if (
currentLeaf?.ptyId === ptyId &&
currentLeaf.lastAgentStatus === 'idle' &&
@@ -142,7 +150,15 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
.sendPrompt(terminalHandle, formatMessagePointer(unread.length, mailboxHandle), {
signal: flight.abortController.signal,
beforeWrite: () =>
this.assertPromptDeliveryCurrent(leaf, mailboxHandle, unread, ptyId, flight),
assertMailboxPointerDeliveryCurrent(
this.deps,
this.state,
leaf,
mailboxHandle,
unread,
ptyId,
flight
),
beforeAttempt: () =>
this.stagePromptDeliveryAttempt(
leaf,
@@ -170,9 +186,16 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
outcome: 'delivered' | 'unknown' | 'rejected'
): void {
try {
if (outcome === 'rejected' || !flight.mutated || !this.state.isCurrentFlight(ptyId, flight)) {
if (!flight.mutated || !this.state.isCurrentFlight(ptyId, flight)) {
return
}
const messageIds = flight.messageIds
if (outcome === 'delivered' && messageIds.length > 0) {
const db = this.deps.getDb()
if (db?.markAsDelivered && !flight.usedLegacyDeliveredTransition) {
markMailboxDeliveryDelivered(db, messageIds, flight.usedLegacyDeliveredTransition)
}
}
if (
outcome === 'delivered' &&
[leaf.lastOscTitle, leaf.paneTitle, this.deps.getTabTitle(leaf.tabId)].some(
@@ -183,11 +206,19 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
this.redrive(mailboxHandle)
return
}
if (outcome !== 'delivered' && messageIds.length > 0) {
const db = this.deps.getDb()
if (outcome === 'rejected') {
db?.markAsUndelivered?.(messageIds)
} else {
;(db?.markAsDeliveryUnknown ?? db?.markAsDelivered)?.call(db, messageIds)
}
}
const released =
outcome === 'delivered'
? this.state.clearWatermark(mailboxHandle, newestSequence, ptyId)
: this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId)
if (released && outcome === 'delivered') {
outcome === 'unknown'
? this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId)
: this.state.clearWatermark(mailboxHandle, newestSequence, ptyId)
if (released && outcome !== 'unknown') {
this.redrive(mailboxHandle)
}
} finally {
@@ -197,35 +228,6 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
}
}
}
private assertPromptDeliveryCurrent(
leaf: OrchestrationMailboxLeaf,
mailboxHandle: string,
unread: readonly { id: string; type: string; sequence: number }[],
ptyId: string,
flight: OrchestrationMailboxDeliveryFlight
): void {
const currentLeaf = this.deps.getLeaf(this.leafKey(leaf))
const ownerChanged =
!this.state.isCurrentFlight(ptyId, flight) ||
currentLeaf?.ptyId !== ptyId ||
!currentLeaf.writable ||
this.deps.mailboxOwner.resolve(currentLeaf) !== mailboxHandle
const agentNoLongerIdle =
currentLeaf?.lastAgentStatus !== 'idle' || !currentLeaf.lastAgentStatusObservedLive
const messagesReleased = shouldReleaseOrchestrationPointer(
this.deps.getDb(),
mailboxHandle,
unread,
this.deps.getMessageWaiters(mailboxHandle)
)
if (!ownerChanged && !agentNoLongerIdle && !messagesReleased) {
return
}
if (!flight.mutated && ownerChanged) {
flight.redriveMailbox = mailboxHandle
}
throw new Error('orchestration_pointer_released')
}
private stagePromptDeliveryAttempt(
leaf: OrchestrationMailboxLeaf,
mailboxHandle: string,
@@ -238,13 +240,32 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
if (!db) {
throw new Error('orchestration_db_unavailable')
}
db.markAsDelivered(unread.map((message) => message.id))
this.state.setWatermark(mailboxHandle, newestSequence, ptyId, this.leafKey(leaf))
const stage = db.markAsDeliveryStaged ?? db.markAsDelivered
flight.usedLegacyDeliveredTransition = !db.markAsDeliveryStaged
stage.call(
db,
unread.map((message) => message.id)
)
flight.messageIds = unread.map((message) => message.id)
this.state.setWatermark(
mailboxHandle,
newestSequence,
ptyId,
this.deps.getLeafKey(leaf.tabId, leaf.leafId)
)
flight.mutated = true
if (!this.state.isCurrentFlight(ptyId, flight)) {
this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId)
}
this.assertPromptDeliveryCurrent(leaf, mailboxHandle, unread, ptyId, flight)
assertMailboxPointerDeliveryCurrent(
this.deps,
this.state,
leaf,
mailboxHandle,
unread,
ptyId,
flight
)
if (
[leaf.lastOscTitle, leaf.paneTitle, this.deps.getTabTitle(leaf.tabId)].some(
isCursorAgentTitle
@@ -259,12 +280,14 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
return
}
for (const [mailboxHandle, delivery] of parked) {
const currentLeaf = this.deps.getLeaf(this.leafKey(delivery.leaf))
const currentLeaf = this.deps.getLeaf(
this.deps.getLeafKey(delivery.leaf.tabId, delivery.leaf.leafId)
)
if (
currentLeaf?.ptyId !== ptyId ||
this.deps.mailboxOwner.resolve(currentLeaf, mailboxHandle) !== mailboxHandle
) {
this.parkRedelivery(mailboxHandle, delivery.reservedTypes)
this.state.parkRedelivery(mailboxHandle, delivery.reservedTypes)
this.redrive(mailboxHandle)
} else {
this.deliver(currentLeaf, { mailboxHandle, reservedTypes: delivery.reservedTypes })
@@ -272,17 +295,6 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
}
}
private redrive(mailboxHandle: string, force = false): void {
const parkedTypes = this.state.takeRedelivery(mailboxHandle, force)
if (parkedTypes === undefined) {
return
}
queueMicrotask(() => {
try {
this.deps.redriveMailbox(mailboxHandle, parkedTypes ?? undefined)
} catch {}
})
}
private leafKey(leaf: OrchestrationMailboxLeaf): string {
return this.deps.getLeafKey(leaf.tabId, leaf.leafId)
redriveMailboxPointer(this.state, this.deps, mailboxHandle, force)
}
}
@@ -0,0 +1,24 @@
import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-contract'
import type { OrchestrationMessageWaiter } from './mailbox-pointer-eligibility'
import type { OrchestrationMailboxPointerState } from './mailbox-pointer-state'
export function redriveMailboxPointer<TWaiter extends OrchestrationMessageWaiter>(
state: OrchestrationMailboxPointerState,
deps: PointerDeliveryDependencies<TWaiter>,
mailboxHandle: string,
force = false
): void {
const parkedTypes = state.takeRedelivery(mailboxHandle, force)
if (parkedTypes === undefined) {
return
}
queueMicrotask(() => {
try {
const db = deps.getDb()
if (db && db.getUndeliveredUnreadMessages(mailboxHandle).length === 0) {
return
}
deps.redriveMailbox(mailboxHandle, parkedTypes ?? undefined)
} catch {}
})
}
@@ -1,11 +1,27 @@
import { describe, expect, it } from 'vitest'
import { describe, expect, it, vi } from 'vitest'
import { OrchestrationMailboxPointerState } from './mailbox-pointer-state'
import { markMailboxDeliveryDelivered } from './mailbox-pointer-delivery-validation'
describe('OrchestrationMailboxPointerState', () => {
it('does not admit newer mail after an unknown prompt is deactivated', () => {
it('persists delivery before a Cursor fast-path redrive', () => {
const markAsDelivered = vi.fn()
markMailboxDeliveryDelivered({ markAsDelivered }, ['message-1'], false)
expect(markAsDelivered).toHaveBeenCalledWith(['message-1'])
})
it('keeps an unknown attempt fenced until its PTY incarnation retires', () => {
const state = new OrchestrationMailboxPointerState()
state.setWatermark('run:1', 1, 'pty-1', 'pane-1')
expect(state.deactivateWatermark('run:1', 1, 'pty-1')).toBe(true)
expect(state.releaseSupersededWatermark('run:1', 2)).toBe(false)
expect(state.releaseSupersededWatermark('run:1', 2, 'pty-1', 'pane-1')).toBe(false)
expect(state.releaseSupersededWatermark('run:1', 2, 'pty-2', 'pane-2')).toBe(true)
})
it('releases an unknown watermark with its retired PTY', () => {
const state = new OrchestrationMailboxPointerState()
state.setWatermark('run:1', 1, 'pty-1', 'pane-1')
state.deactivateWatermark('run:1', 1, 'pty-1')
expect(state.retirePty('pty-1').releasedMailboxes).toEqual(['run:1'])
expect(state.releaseSupersededWatermark('run:1', 2, 'pty-2', 'pane-2')).toBe(true)
})
})
@@ -3,6 +3,8 @@ import type { OrchestrationMailboxLeaf } from './mailbox-owner'
export type OrchestrationMailboxDeliveryFlight = {
abortController: AbortController
mutated: boolean
messageIds: string[]
usedLegacyDeliveredTransition: boolean
redriveMailbox: string | null
}
@@ -29,7 +31,13 @@ export class OrchestrationMailboxPointerState {
}
beginFlight(ptyId: string): OrchestrationMailboxDeliveryFlight {
const flight = { abortController: new AbortController(), mutated: false, redriveMailbox: null }
const flight = {
abortController: new AbortController(),
mutated: false,
messageIds: [],
usedLegacyDeliveredTransition: false,
redriveMailbox: null
}
this.flightsByPtyId.set(ptyId, flight)
return flight
}
@@ -75,15 +83,20 @@ export class OrchestrationMailboxPointerState {
return this.watermarkByMailbox.get(mailboxHandle)?.active === true
}
releaseSupersededWatermark(mailboxHandle: string, newestSequence: number): boolean {
releaseSupersededWatermark(
mailboxHandle: string,
newestSequence: number,
ptyId: string | undefined,
leafKey: string
): boolean {
const owner = this.watermarkByMailbox.get(mailboxHandle)
if (owner && !owner.active) {
return false
}
if (newestSequence > (owner?.sequence ?? -1)) {
if (!owner) {
return true
}
return false
if (newestSequence <= owner.sequence) {
return false
}
return owner.active || owner.ptyId !== ptyId || owner.leafKey !== leafKey
}
setWatermark(mailbox: string, sequence: number, ptyId: string, leafKey: string): void {
@@ -153,17 +166,11 @@ export class OrchestrationMailboxPointerState {
if (!owner || owner.ptyId !== ptyId) {
continue
}
if (flight?.mutated || !owner.active) {
owner.active = false
continue
}
if (this.clearWatermark(mailboxHandle, owner.sequence, ptyId)) {
releasedMailboxes.push(mailboxHandle)
}
}
if (!flight?.mutated) {
this.watermarkMailboxesByPtyId.delete(ptyId)
}
this.watermarkMailboxesByPtyId.delete(ptyId)
return { flight, releasedMailboxes }
}
+3
View File
@@ -17,6 +17,8 @@ export type MessagePriority = 'normal' | 'high' | 'urgent'
export type MessageDeliveryContract = 'legacy_direct' | 'current_delivery' | 'audit_only'
export type MessageDeliveryState = 'none' | 'staged' | 'delivered' | 'unknown'
export type TaskStatus = 'pending' | 'ready' | 'dispatched' | 'completed' | 'failed' | 'blocked'
export type DispatchStatus = 'pending' | 'dispatched' | 'completed' | 'failed' | 'circuit_broken'
@@ -240,6 +242,7 @@ export type MessageRow = {
sequence: number
created_at: string
delivered_at: string | null
delivery_state?: MessageDeliveryState
sender_pane_key: string | null
}