mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 18:37:24 +00:00
fix(orchestration): unify staged mail submission and settlement
This commit is contained in:
@@ -34,13 +34,11 @@ export class MailboxDeliveryRecheck {
|
||||
}
|
||||
if (decision.foreground === false || decision.reset) {
|
||||
this.delays.delete(ptyId)
|
||||
this.clearTimer(ptyId)
|
||||
}
|
||||
if (decision.foreground === false) {
|
||||
this.foregroundPending.delete(ptyId)
|
||||
}
|
||||
if (this.read.inFlight(ptyId)) {
|
||||
this.clearTimer(ptyId)
|
||||
return
|
||||
}
|
||||
if (this.timers.has(ptyId) || this.decisions.has(ptyId)) {
|
||||
|
||||
+2
-1
@@ -230,7 +230,8 @@ describe('terminal-owned foreground work', () => {
|
||||
shell = false
|
||||
await vi.advanceTimersByTimeAsync(62000)
|
||||
expect(write).toHaveBeenCalled()
|
||||
await vi.advanceTimersByTimeAsync(20000 * mailboxes)
|
||||
// Each mailbox needs its 500 ms paste delay plus at most 2 s of screen quiescence.
|
||||
await vi.advanceTimersByTimeAsync(1000 + 2500 * mailboxes)
|
||||
expect(
|
||||
messages.every(
|
||||
(id) =>
|
||||
|
||||
+1
-1
@@ -182,7 +182,7 @@ it.each([
|
||||
`\x1b[2J\x1b[H${'─'.repeat(100)}\r\n❯ ${write.mock.calls[0][1].trim()}\r\n${'─'.repeat(100)}\r\n auto mode on\x1b[2;3H`,
|
||||
Date.now()
|
||||
)
|
||||
await vi.advanceTimersByTimeAsync(10000)
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([
|
||||
expect.stringContaining('You have 1 orchestration message'),
|
||||
'\r'
|
||||
|
||||
+31
-14
@@ -75,11 +75,15 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
write,
|
||||
writeWithSettlement: (ptyId, data) => {
|
||||
write(ptyId, data)
|
||||
return data === '\r' &&
|
||||
if (
|
||||
data === '\r' &&
|
||||
repair === 'refused' &&
|
||||
write.mock.calls.filter(([, text]) => text === '\r').length === 1
|
||||
? { outcome: 'refused', reason: 'provider_refused_write' }
|
||||
: { outcome: 'accepted' }
|
||||
) {
|
||||
unknown = true
|
||||
return { outcome: 'refused', reason: 'provider_refused_write' }
|
||||
}
|
||||
return { outcome: 'accepted' }
|
||||
},
|
||||
kill: vi.fn(),
|
||||
getSize: () => ({ cols: 100, rows: 30 }),
|
||||
@@ -199,6 +203,9 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
}
|
||||
if (repair === 'refused') {
|
||||
unknown = false
|
||||
}
|
||||
if (repair === 'unruled') {
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
@@ -207,14 +214,27 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
runtime.onPtyData(PTY_ID, '\x1b]0;grok working\x07', Date.now())
|
||||
db.markAsReadAndDelivered([message.id])
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
return
|
||||
}
|
||||
if (repair === 'emptied') {
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
db.markAsReadAndDelivered([message.id])
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
return
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([
|
||||
expect.stringContaining('You have 1 orchestration message'),
|
||||
...(repair === 'emptied' || repair === 'unruled'
|
||||
? [expect.stringContaining('You have 1 orchestration message')]
|
||||
: []),
|
||||
...(repair === 'refused' ? ['\r'] : []),
|
||||
'\r'
|
||||
])
|
||||
@@ -225,9 +245,7 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
})
|
||||
const attempts = vi.spyOn(runtime, 'deliverPendingMessagesForPty')
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(
|
||||
['emptied', 'unruled', 'refused'].includes(repair ?? '') ? 3 : 2
|
||||
)
|
||||
expect(write).toHaveBeenCalledTimes(repair === 'refused' ? 3 : 2)
|
||||
expect(attempts).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
finishSnapshot({ data: '', cols: 100, rows: 30 })
|
||||
@@ -671,11 +689,10 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
runtime.requestMailDecision(PTY_ID)
|
||||
}
|
||||
if (kind === 'owed-read-error') {
|
||||
const read = vi.spyOn(db, 'getMessageById').mockImplementationOnce(() => {
|
||||
const read = vi.spyOn(db, 'areUnreadMessages').mockImplementationOnce(() => {
|
||||
throw new Error('temporarily unavailable')
|
||||
})
|
||||
await runtime.readTerminalMailHold(`run:${run.id}`)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(read).toHaveBeenCalled()
|
||||
read.mockRestore()
|
||||
}
|
||||
@@ -703,7 +720,7 @@ describe('mail-local foreground recovery readiness', () => {
|
||||
recovering ? pointerComposer(write.mock.calls[0][1]) : COMPOSER,
|
||||
Date.now()
|
||||
)
|
||||
await vi.advanceTimersByTimeAsync(10000)
|
||||
await vi.advanceTimersByTimeAsync(recovering ? 65000 : 10000)
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([
|
||||
expect.stringContaining('You have 1 orchestration message'),
|
||||
'\r'
|
||||
|
||||
+475
@@ -0,0 +1,475 @@
|
||||
import { rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
readRuntimeFixture,
|
||||
readTimedRuntimeFixture
|
||||
} from '../../../terminal/wait/agent-transcript-replay-test-harness'
|
||||
import { settledWriteStub } from '../../../../providers/testing/settled-pty-write-stub'
|
||||
import {
|
||||
createBoundRun,
|
||||
createDatabase,
|
||||
createRuntime,
|
||||
insertDirectRunMessage,
|
||||
PTY_ID,
|
||||
temporaryDirectories
|
||||
} from './orchestration-mailbox-notification-test-harness'
|
||||
import { countForegroundSqlWork } from './mailbox-foreground-work.test-fixture'
|
||||
import type { WriteSettlement } from '../../../../../shared/terminal/pty/lifecycle/pty-write-settlement'
|
||||
|
||||
vi.mock('electron', () => ({
|
||||
app: { getPath: vi.fn(() => tmpdir()), isPackaged: false },
|
||||
BrowserWindow: { fromId: vi.fn(() => null) },
|
||||
ipcMain: { on: vi.fn(), removeListener: vi.fn() },
|
||||
webContents: { fromId: vi.fn(() => null) }
|
||||
}))
|
||||
|
||||
const READY = `\x1b[2J\x1b[H${readRuntimeFixture('claude-2-1-316-default-unlabelled-empty')}`
|
||||
const DIALOG =
|
||||
'\x1b[2J\x1b[HDo you want to create hello.txt?\r\n❯ 1. Yes\r\n 2. No\r\nEsc to cancel · Tab to amend'
|
||||
function composer(text: string) {
|
||||
return `\x1b[2J\x1b[H${'─'.repeat(100)}\r\n❯ ${text.trim()}\r\n${'─'.repeat(100)}\r\n auto mode on\x1b[2;3H`
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
vi.useRealTimers()
|
||||
for (const directory of temporaryDirectories.splice(0)) {
|
||||
rmSync(directory, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
describe('one mailbox submission owner', () => {
|
||||
it.each([
|
||||
'refused',
|
||||
'manual Enter',
|
||||
'deleted',
|
||||
'consumed',
|
||||
'unverifiable',
|
||||
'throw after handoff',
|
||||
'late acknowledgement'
|
||||
])(
|
||||
'revalidates an ordinary flight that becomes recovery across delayed checks (%s)',
|
||||
async (intervention) => {
|
||||
vi.useFakeTimers()
|
||||
const db = createDatabase('orca-mail-late-recovery-')
|
||||
const { runtime, write } = createRuntime(db, { launchAgent: 'claude' })
|
||||
let process = 'claude'
|
||||
let staged = ''
|
||||
let releaseAbsence!: (live: boolean) => void
|
||||
const absence = new Promise<boolean>((resolve) => {
|
||||
releaseAbsence = resolve
|
||||
})
|
||||
let delayAbsence = false
|
||||
let refuse = intervention === 'refused'
|
||||
let finishWrite!: (settlement: WriteSettlement) => void
|
||||
const acknowledgement = new Promise<WriteSettlement>((resolve) => {
|
||||
finishWrite = resolve
|
||||
})
|
||||
runtime.setPtyController({
|
||||
write,
|
||||
writeWithSettlement: (ptyId, data) => {
|
||||
write(ptyId, data)
|
||||
if (data === '\r' && intervention === 'late acknowledgement') {
|
||||
return acknowledgement
|
||||
}
|
||||
if (data === '\r' && refuse) {
|
||||
refuse = false
|
||||
runtime.onPtyData(PTY_ID, DIALOG, Date.now())
|
||||
return { outcome: 'refused', reason: 'provider_refused_write' }
|
||||
}
|
||||
if (data === '\r' && intervention === 'throw after handoff') {
|
||||
return Promise.reject(new Error('lost write acknowledgement'))
|
||||
}
|
||||
return data === '\r' && intervention === 'unverifiable'
|
||||
? {
|
||||
outcome: 'unverifiable',
|
||||
reason: 'transport_settlement_lost',
|
||||
bytesHandedToTransport: true
|
||||
}
|
||||
: { outcome: 'accepted' }
|
||||
},
|
||||
kill: vi.fn(),
|
||||
getSize: () => ({ cols: 100, rows: 30 }),
|
||||
probePtyLiveness: () => (delayAbsence ? absence : Promise.resolve(true)),
|
||||
getForegroundProcess: async () => process,
|
||||
supportsForegroundObservation: () => true,
|
||||
inspectProcess: async () => ({
|
||||
foregroundProcess: process,
|
||||
hasChildProcesses: process !== 'zsh',
|
||||
foregroundObservation: {
|
||||
ptyId: PTY_ID,
|
||||
ptyIncarnationId: 'mailbox-incarnation',
|
||||
authorityGeneration: 'test',
|
||||
observationEpoch: 1,
|
||||
capturedAgeMs: 0,
|
||||
verdict: 'observed',
|
||||
source: 'pty-master',
|
||||
foregroundKind: process === 'zsh' ? 'shell' : 'non-shell',
|
||||
processName: process
|
||||
}
|
||||
})
|
||||
})
|
||||
try {
|
||||
await runtime.listTerminals()
|
||||
runtime.onPtyData(PTY_ID, READY, Date.now())
|
||||
const run = createBoundRun(db, 'Late recovery')
|
||||
const message = insertDirectRunMessage(db, run.id, 'One staged obligation')
|
||||
write.mockImplementation((_ptyId, data) => {
|
||||
if (data !== '\r') {
|
||||
staged = data
|
||||
delayAbsence = true
|
||||
runtime.onPtyData(PTY_ID, composer(staged), Date.now())
|
||||
}
|
||||
return true
|
||||
})
|
||||
runtime.notifyMessageArrived(`run:${run.id}`, 'status')
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
expect(staged).toContain('You have 1 orchestration message')
|
||||
if (intervention !== 'late acknowledgement') {
|
||||
process = 'zsh'
|
||||
expect(await runtime.readTerminalMailHold(`run:${run.id}`)).toMatchObject({
|
||||
kind: 'not-foreground'
|
||||
})
|
||||
}
|
||||
if (intervention === 'manual Enter' || intervention === 'deleted') {
|
||||
runtime.onPtyData(PTY_ID, `${READY}\x1b]0;Claude working\x07`, Date.now())
|
||||
}
|
||||
if (intervention === 'consumed') {
|
||||
db.markAsRead([message.id])
|
||||
}
|
||||
// Quiescence must not mask the ownership transition while absence is unresolved.
|
||||
await vi.advanceTimersByTimeAsync(2500)
|
||||
process = 'claude'
|
||||
delayAbsence = false
|
||||
releaseAbsence(true)
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
if (intervention === 'late acknowledgement') {
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r'])
|
||||
process = 'zsh'
|
||||
expect(await runtime.readTerminalMailHold(`run:${run.id}`)).toMatchObject({
|
||||
kind: 'not-foreground'
|
||||
})
|
||||
process = 'claude'
|
||||
finishWrite({
|
||||
outcome: 'unverifiable',
|
||||
reason: 'transport_settlement_lost',
|
||||
bytesHandedToTransport: true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
}
|
||||
if (
|
||||
['unverifiable', 'throw after handoff', 'late acknowledgement'].includes(intervention)
|
||||
) {
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r'])
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 3
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(125000)
|
||||
expect(write).toHaveBeenCalledTimes(2)
|
||||
runtime.onPtyData(PTY_ID, READY, Date.now())
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(2)
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 3
|
||||
})
|
||||
db.markAsReadAndDelivered([message.id])
|
||||
await vi.advanceTimersByTimeAsync(125000)
|
||||
expect(write).toHaveBeenCalledTimes(2)
|
||||
const attempts = vi.spyOn(runtime, 'deliverPendingMessagesForPty')
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(attempts).not.toHaveBeenCalled()
|
||||
return
|
||||
}
|
||||
if (intervention === 'consumed') {
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r'])
|
||||
expect(db.getMessageById(message.id)).toMatchObject({ read: 1, pointer_enter_pending: 0 })
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(2)
|
||||
return
|
||||
}
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
if (intervention === 'refused') {
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r'])
|
||||
runtime.onPtyData(PTY_ID, DIALOG, Date.now())
|
||||
} else {
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
runtime.onPtyData(PTY_ID, READY, Date.now())
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(125000)
|
||||
expect(write).toHaveBeenCalledTimes(intervention === 'refused' ? 2 : 1)
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
if (intervention === 'refused') {
|
||||
runtime.onPtyData(PTY_ID, composer(staged), Date.now())
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r', '\r'])
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: expect.any(String),
|
||||
pointer_enter_pending: 0
|
||||
})
|
||||
} else {
|
||||
db.markAsReadAndDelivered([message.id])
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
}
|
||||
const attempts = vi.spyOn(runtime, 'deliverPendingMessagesForPty')
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(attempts).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
releaseAbsence(true)
|
||||
finishWrite({ outcome: 'accepted' })
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['claude', 'codex', 'opencode2'] as const)(
|
||||
'recognises agent-rendered wrapped pointer rows in %s without accepting edits or dialogs',
|
||||
async (agent) => {
|
||||
vi.useFakeTimers()
|
||||
const db = createDatabase('orca-mail-wrapped-pointer-')
|
||||
const { runtime, write } = createRuntime(db, { launchAgent: agent })
|
||||
let cols = agent === 'codex' ? 120 : agent === 'claude' ? 100 : 40
|
||||
const rows = agent === 'claude' ? 30 : 40
|
||||
const timed = readTimedRuntimeFixture('opencode-2-0-21-timed-narrow-pane')
|
||||
const ready =
|
||||
agent === 'claude'
|
||||
? readRuntimeFixture('claude-2-1-316-labelled-empty')
|
||||
: agent === 'codex'
|
||||
? readRuntimeFixture('codex-0157-plain-ready')
|
||||
: timed.chunks.filter((_, i) => timed.times[i] < timed.promptSentAtMs! - 600).join('')
|
||||
let shell = true
|
||||
let staged = ''
|
||||
runtime.setPtyController({
|
||||
write,
|
||||
writeWithSettlement: settledWriteStub(write),
|
||||
kill: vi.fn(),
|
||||
getSize: () => ({ cols, rows }),
|
||||
getForegroundProcess: async () => null,
|
||||
supportsForegroundObservation: () => true,
|
||||
inspectProcess: async () => ({
|
||||
foregroundProcess: shell ? 'zsh' : agent,
|
||||
hasChildProcesses: !shell,
|
||||
foregroundObservation: {
|
||||
ptyId: PTY_ID,
|
||||
ptyIncarnationId: 'mailbox-incarnation',
|
||||
authorityGeneration: 'test',
|
||||
observationEpoch: 1,
|
||||
capturedAgeMs: 0,
|
||||
verdict: 'observed',
|
||||
source: 'pty-master',
|
||||
foregroundKind: shell ? 'shell' : 'non-shell',
|
||||
processName: shell ? 'zsh' : agent
|
||||
}
|
||||
})
|
||||
})
|
||||
// Recorded empty frames above; pointer contents and hard-row redraws below are synthetic.
|
||||
const wrapped = (text: string) => {
|
||||
const parts = text.trim().match(/.{1,28}/g) ?? ['']
|
||||
if (agent === 'opencode2') {
|
||||
return `\x1b[2J\x1b[HOpenCode\r\n ┃\r\n${parts.map((part) => ` ┃ ${part}`).join('\r\n')}\r\n ┃\r\n ┃ Build · gpt-6.1\r\n ╹▀▀▀▀▀▀▀▀\x1b[3;7H\x1b[?25h`
|
||||
}
|
||||
if (agent === 'codex') {
|
||||
return `\x1b[2J\x1b[HOpenAI Codex\r\n${'─'.repeat(40)}\r\n\x1b[1m›\x1b[0m ${parts.join('\r\n ')}\r\n\r\n\x1b[2mgpt-6.1 · ~/repo\r\n? for shortcuts\x1b[0m\x1b[3;3H\x1b[?25h`
|
||||
}
|
||||
return `\x1b[2J\x1b[H${'─'.repeat(28)} ultracode ─\r\n❯ ${parts.join('\r\n ')}\r\n${'─'.repeat(40)}\r\n bypass permissions on\x1b[2;3H\x1b[?25h`
|
||||
}
|
||||
try {
|
||||
await runtime.listTerminals()
|
||||
runtime.onPtyData(PTY_ID, ready, Date.now())
|
||||
// Give the recorded frame its normal explicit rest marker, independent of custom labels.
|
||||
runtime.onPtyData(PTY_ID, `\x1b]0;${agent} done\x07`, Date.now())
|
||||
const run = createBoundRun(db, 'Agent wrapped pointer')
|
||||
const message = insertDirectRunMessage(db, run.id, 'One owned composer')
|
||||
runtime.notifyMessageArrived(`run:${run.id}`, 'status')
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
expect(write).not.toHaveBeenCalled()
|
||||
shell = false
|
||||
write.mockImplementation((_ptyId, data) => {
|
||||
if (data !== '\r') {
|
||||
staged = data
|
||||
runtime.onPtyData(PTY_ID, wrapped(`${data.trim()} edited`), Date.now())
|
||||
}
|
||||
return true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: null,
|
||||
pointer_enter_pending: 2
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
if (agent !== 'opencode2') {
|
||||
cols = agent === 'claude' ? 40 : 120
|
||||
runtime.resizeHeadlessTerminalToPtyGrid(PTY_ID, cols, rows)
|
||||
runtime.onPtyData(
|
||||
PTY_ID,
|
||||
`\x1b[2J\x1b[H${readRuntimeFixture(
|
||||
agent === 'claude'
|
||||
? 'claude-2-1-316-labelled-wrapped'
|
||||
: 'codex-fullscreen-multiline-early-input'
|
||||
)}`,
|
||||
Date.now()
|
||||
)
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
}
|
||||
runtime.onPtyData(PTY_ID, DIALOG, Date.now())
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
runtime.onPtyData(PTY_ID, wrapped(staged), Date.now())
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write.mock.calls.map(([, data]) => data)).toEqual([staged, '\r'])
|
||||
expect(db.getMessageById(message.id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: expect.any(String),
|
||||
pointer_enter_pending: 0
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(65000)
|
||||
expect(write).toHaveBeenCalledTimes(2)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
it.each([
|
||||
{ chunks: 12, m: 2, replay: false },
|
||||
{ chunks: 120, m: 20, replay: false },
|
||||
{ chunks: 12, m: 2, replay: true },
|
||||
{ chunks: 120, m: 20, replay: true }
|
||||
])(
|
||||
'keeps an owed wake under separate-turn output ($chunks chunks/s, $m mailboxes, replay=$replay)',
|
||||
async ({ chunks, m, replay }) => {
|
||||
vi.useFakeTimers()
|
||||
const db = createDatabase('orca-mail-replay-cost-')
|
||||
const { runtime, write } = createRuntime(db, { launchAgent: 'claude' })
|
||||
let shell = true
|
||||
let inspections = 0
|
||||
let unknown = false
|
||||
let staged = ''
|
||||
let hydrate = false
|
||||
let coveredThrough = 0
|
||||
let sequence = 0
|
||||
runtime.setPtyController({
|
||||
write,
|
||||
writeWithSettlement: settledWriteStub(write),
|
||||
kill: vi.fn(),
|
||||
getSize: () => ({ cols: 100, rows: 30 }),
|
||||
hasRendererSerializer: () => hydrate,
|
||||
serializeBuffer: async () => ({
|
||||
data: composer(staged),
|
||||
cols: 100,
|
||||
rows: 30,
|
||||
seq: coveredThrough
|
||||
}),
|
||||
getForegroundProcess: async () => null,
|
||||
supportsForegroundObservation: () => true,
|
||||
inspectProcess: async () => {
|
||||
inspections++
|
||||
if (unknown) {
|
||||
return { foregroundProcess: null, hasChildProcesses: false }
|
||||
}
|
||||
return {
|
||||
foregroundProcess: shell ? 'zsh' : 'claude',
|
||||
hasChildProcesses: !shell,
|
||||
foregroundObservation: {
|
||||
ptyId: PTY_ID,
|
||||
ptyIncarnationId: 'mailbox-incarnation',
|
||||
authorityGeneration: 'test',
|
||||
observationEpoch: 1,
|
||||
capturedAgeMs: 0,
|
||||
verdict: 'observed',
|
||||
source: 'pty-master',
|
||||
foregroundKind: shell ? 'shell' : 'non-shell',
|
||||
processName: shell ? 'zsh' : 'claude'
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
try {
|
||||
await runtime.listTerminals()
|
||||
sequence = runtime.onPtyData(PTY_ID, READY, Date.now())
|
||||
const messages: string[] = []
|
||||
for (let i = 0; i < m; i++) {
|
||||
const run = createBoundRun(db, 'Streaming recovery')
|
||||
messages.push(insertDirectRunMessage(db, run.id, 'Owed').id)
|
||||
runtime.notifyMessageArrived(`run:${run.id}`, 'status')
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
expect(write).not.toHaveBeenCalled()
|
||||
shell = false
|
||||
write.mockImplementation((_ptyId, data) => {
|
||||
if (data !== '\r') {
|
||||
staged = data
|
||||
sequence = runtime.onPtyData(PTY_ID, composer(data), Date.now())
|
||||
unknown = true
|
||||
}
|
||||
return true
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(6000)
|
||||
expect(write).toHaveBeenCalledTimes(1)
|
||||
if (replay) {
|
||||
hydrate = true
|
||||
coveredThrough = sequence + chunks * '\x1b[0m\x1b[?25h'.length + 60 * '\x1b[0m'.length
|
||||
runtime.notePtyDataGap(PTY_ID)
|
||||
await runtime.maybeHydrateHeadlessFromRenderer(PTY_ID)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
}
|
||||
const work = countForegroundSqlWork(db)
|
||||
const attempts = vi.spyOn(runtime, 'deliverPendingMessagesForPty')
|
||||
const before = inspections
|
||||
// Every chunk arrives in a different turn; no title/status/hook observation is injected.
|
||||
for (let i = 0; i < chunks; i++) {
|
||||
runtime.onPtyData(PTY_ID, '\x1b[0m\x1b[?25h', Date.now())
|
||||
await vi.advanceTimersByTimeAsync(1000 / chunks)
|
||||
}
|
||||
expect(work.counts).toEqual({ statements: 0, owners: 0, messages: 0 })
|
||||
expect(inspections).toBe(before)
|
||||
expect(attempts).not.toHaveBeenCalled()
|
||||
work.stop()
|
||||
// Steady traffic must not cancel the owed deadline or cause a catch-up burst.
|
||||
for (let second = 0; second < 60; second++) {
|
||||
runtime.onPtyData(PTY_ID, '\x1b[0m', Date.now())
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
}
|
||||
expect(attempts.mock.calls.length).toBeGreaterThanOrEqual(2)
|
||||
expect(attempts.mock.calls.length).toBeLessThanOrEqual(4)
|
||||
expect(inspections - before).toBeLessThanOrEqual(4)
|
||||
unknown = false
|
||||
write.mockImplementation((_ptyId, data) => {
|
||||
runtime.onPtyData(PTY_ID, data === '\r' ? READY : composer(data), Date.now())
|
||||
return true
|
||||
})
|
||||
runtime.onPtyData(PTY_ID, composer(staged), Date.now())
|
||||
await vi.advanceTimersByTimeAsync(62000 + 2500 * m)
|
||||
expect(write.mock.calls.filter(([, text]) => text === '\r')).toHaveLength(m)
|
||||
for (const id of messages) {
|
||||
expect(db.getMessageById(id)).toMatchObject({
|
||||
read: 0,
|
||||
delivered_at: expect.any(String),
|
||||
pointer_enter_pending: 0
|
||||
})
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
)
|
||||
})
|
||||
+9
-14
@@ -1,4 +1,3 @@
|
||||
import { recoveryFlightDecision } from './mailbox-recovery-flight'
|
||||
import { resumeDeferredMailboxPointer } from './mailbox-pointer-submit'
|
||||
import { MailboxForegroundDelivery } from './mailbox-foreground-delivery'
|
||||
import type {
|
||||
@@ -178,10 +177,9 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
|
||||
}
|
||||
|
||||
hasPendingFlightMail(terminal: OrchestrationMailboxTerminal): boolean {
|
||||
return hasPendingMailboxPointerFlight(
|
||||
this.deps,
|
||||
terminal,
|
||||
this.state.flightForPty(terminal.ptyId)
|
||||
return (
|
||||
this.hasRecoveryFlight(terminal.ptyId) ||
|
||||
hasPendingMailboxPointerFlight(this.deps, terminal, this.state.flightForPty(terminal.ptyId))
|
||||
)
|
||||
}
|
||||
|
||||
@@ -195,13 +193,11 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
|
||||
}
|
||||
|
||||
hasRecoveryFlight(ptyId: string): boolean {
|
||||
return recoveryFlightDecision(this.state, ptyId, this.state.flightForPty(ptyId)) !== 'ordinary'
|
||||
}
|
||||
|
||||
observeOutput(ptyId: string): void {
|
||||
if (this.hasRecoveryFlight(ptyId)) {
|
||||
this.deps.onDeliveryDecision(ptyId, { reset: true })
|
||||
}
|
||||
const flight = this.state.flightForPty(ptyId)
|
||||
return (
|
||||
!!flight?.processIncarnation &&
|
||||
this.state.foreground.screenRecovery(ptyId) === flight.processIncarnation
|
||||
)
|
||||
}
|
||||
|
||||
isDeliveryInFlight(ptyId: string): boolean {
|
||||
@@ -225,7 +221,6 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
|
||||
}
|
||||
|
||||
observeTerminal(ptyId: string, status?: string): void {
|
||||
this.observeOutput(ptyId)
|
||||
if (this.state.foreground.observeTerminal(ptyId, status)) {
|
||||
this.deps.onDeliveryDecision(ptyId, { reset: true })
|
||||
queueMicrotask(() => this.deps.redriveTerminal(ptyId))
|
||||
@@ -303,7 +298,7 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
|
||||
|
||||
private settle(ptyId: string, flight: OrchestrationMailboxDeliveryFlight): void {
|
||||
const parked = this.state.settleFlight(ptyId, flight)
|
||||
this.deps.onDeliveryDecision(ptyId, { settled: true })
|
||||
this.deps.onDeliveryDecision(ptyId, { settled: true, foreground: false })
|
||||
const current = this.deps.getTerminal(ptyId)
|
||||
for (const [mailboxHandle, reservedTypes] of parked ?? []) {
|
||||
if (current) {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { recoveryFlightDecision, submitRecoveryMailboxPointer } from './mailbox-recovery-flight'
|
||||
import { terminalComposerTextOwnsPointer } from '../../../terminal/wait/terminal-delivery-refusal'
|
||||
import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-contract'
|
||||
import type { OrchestrationMessageWaiter } from './mailbox-pointer-eligibility'
|
||||
import type { MailboxForeground } from '../../../terminal/wait/launched-agent-foreground'
|
||||
@@ -60,7 +60,7 @@ export function resumeDeferredMailboxPointer<T extends OrchestrationMessageWaite
|
||||
}
|
||||
if (
|
||||
!state.hasActiveFlight(ptyId) &&
|
||||
(recoveryFlightDecision(state, ptyId, state.flightForPty(ptyId)) !== 'ordinary' ||
|
||||
(state.foreground.screenRecovery(ptyId) === state.flightForPty(ptyId)?.processIncarnation ||
|
||||
((!coldParked || terminal.lastAgentStatus === 'idle') &&
|
||||
deps.isAgentSettledForDelivery(ptyId)))
|
||||
) {
|
||||
@@ -77,24 +77,23 @@ export type PointerSubmitInput = {
|
||||
flight: OrchestrationMailboxDeliveryFlight
|
||||
expectedTarget: OrchestrationMailboxPointerSubmitTarget
|
||||
stagedText?: string
|
||||
foregroundFollowup?: boolean
|
||||
enterUncertain?: boolean
|
||||
}
|
||||
|
||||
export function submitOrchestrationMailboxPointer(
|
||||
deps: PointerSubmitDependencies,
|
||||
input: PointerSubmitInput
|
||||
): void {
|
||||
if (submitRecoveryMailboxPointer(deps, input)) {
|
||||
return
|
||||
}
|
||||
let clearAndRedrive = false
|
||||
let redriveClearedPointer = true
|
||||
let submitted = false
|
||||
let releaseWithoutRedrive = false
|
||||
let finalizeReservation = true
|
||||
let preserveAmbiguousDelivery = false
|
||||
let preserveAmbiguousDelivery = input.enterUncertain === true
|
||||
let preserveFlight = false
|
||||
let expectedPhase = MAILBOX_POINTER_WRITE_ATTEMPTED
|
||||
const isRecovering = () =>
|
||||
deps.state.foreground.screenRecovery(input.ptyId) === input.expectedTarget.processIncarnation
|
||||
const messageIds = input.messages.map((message) => message.id)
|
||||
const reservationTarget = {
|
||||
ptyId: input.ptyId,
|
||||
@@ -103,10 +102,37 @@ export function submitOrchestrationMailboxPointer(
|
||||
void deps
|
||||
.isLeafPtyProvenAbsent(input.ptyId)
|
||||
.then(async (absent) => {
|
||||
if (!deps.state.isCurrentFlight(input.ptyId, input.flight)) {
|
||||
finalizeReservation = false
|
||||
return
|
||||
}
|
||||
if (absent) {
|
||||
clearAndRedrive = true
|
||||
return
|
||||
}
|
||||
if (input.enterUncertain) {
|
||||
if (messageIds.every((id) => deps.getDb()?.getMessageById(id)?.read === 1)) {
|
||||
clearAndRedrive = true
|
||||
} else {
|
||||
preserveAmbiguousDelivery = true
|
||||
preserveFlight = true
|
||||
}
|
||||
return
|
||||
}
|
||||
const composerBeforeProbe = deps.readComposerText?.(input.ptyId)
|
||||
if (
|
||||
isRecovering() &&
|
||||
!terminalComposerTextOwnsPointer(composerBeforeProbe, input.stagedText) &&
|
||||
deps.getDb()?.areUnreadMessages(input.mailboxHandle, messageIds) === false &&
|
||||
messageIds.every((id) => deps.getDb()?.getMessageById(id)?.read === 1)
|
||||
) {
|
||||
clearAndRedrive = true
|
||||
return
|
||||
}
|
||||
if (isRecovering() && deps.isAgentSettledForDelivery?.(input.ptyId) === false) {
|
||||
preserveFlight = true
|
||||
return
|
||||
}
|
||||
const foreground = await deps.state.foreground.probeForeground(
|
||||
input.ptyId,
|
||||
input.expectedTarget.processIncarnation,
|
||||
@@ -116,7 +142,11 @@ export function submitOrchestrationMailboxPointer(
|
||||
finalizeReservation = false
|
||||
return
|
||||
}
|
||||
const target = deps.resolveSubmitTarget(input.ptyId, foreground.foreground, input.stagedText)
|
||||
const target = deps.resolveSubmitTarget(
|
||||
input.ptyId,
|
||||
isRecovering() && foreground.foreground !== 'agent' ? undefined : foreground.foreground,
|
||||
input.stagedText
|
||||
)
|
||||
const terminal =
|
||||
target?.terminal.handle === input.expectedTarget.terminal.handle &&
|
||||
target.processIncarnation === input.expectedTarget.processIncarnation
|
||||
@@ -124,13 +154,22 @@ export function submitOrchestrationMailboxPointer(
|
||||
: null
|
||||
const sameMailbox =
|
||||
terminal && deps.mailboxOwner.resolve(terminal, input.mailboxHandle) === input.mailboxHandle
|
||||
if (!terminal || !sameMailbox) {
|
||||
clearAndRedrive = true
|
||||
return
|
||||
}
|
||||
if (isRecovering() && foreground.foreground !== 'agent') {
|
||||
deps.state.foreground.applyForeground(input.ptyId, foreground, foreground.foreground)
|
||||
preserveFlight = true
|
||||
return
|
||||
}
|
||||
const queueSafe = !!terminal && isMailboxTerminalTakingInput(terminal)
|
||||
const settled =
|
||||
deps.isAgentSettledForDelivery?.(input.ptyId, foreground.foreground, input.stagedText) !==
|
||||
false
|
||||
if (!terminal || !sameMailbox) {
|
||||
clearAndRedrive = true
|
||||
} else if (
|
||||
const composer = deps.readComposerText?.(input.ptyId)
|
||||
const ownPointer = terminalComposerTextOwnsPointer(composer, input.stagedText)
|
||||
if (
|
||||
!deps.state.foreground.applyForeground(
|
||||
input.ptyId,
|
||||
foreground,
|
||||
@@ -138,31 +177,25 @@ export function submitOrchestrationMailboxPointer(
|
||||
queueSafe && settled && !deps.isPtyAwaitingUserInput(input.ptyId)
|
||||
)
|
||||
) {
|
||||
if (!input.foregroundFollowup) {
|
||||
submitOrchestrationMailboxPointer(deps, { ...input, foregroundFollowup: true })
|
||||
} else {
|
||||
deps.state.deferFlightUntilIdle(input.ptyId)
|
||||
input.flight.submitEnter = () =>
|
||||
submitOrchestrationMailboxPointer(deps, { ...input, foregroundFollowup: false })
|
||||
deps.onDeliveryDecision?.(input.ptyId)
|
||||
}
|
||||
preserveFlight = true
|
||||
} else if (
|
||||
isRecovering() &&
|
||||
(foreground.foreground !== 'agent' ||
|
||||
!terminal.lastAgentStatusObservedLive ||
|
||||
!settled ||
|
||||
!ownPointer)
|
||||
) {
|
||||
// Missing text cannot distinguish deletion from a user-submitted pointer.
|
||||
preserveFlight = true
|
||||
} else if (deps.state.foreground.foregroundHold(input.ptyId)) {
|
||||
if (submitRecoveryMailboxPointer(deps, input)) {
|
||||
preserveFlight = true
|
||||
} else {
|
||||
clearAndRedrive = true
|
||||
redriveClearedPointer = false
|
||||
}
|
||||
preserveFlight = true
|
||||
} else if (
|
||||
(terminal.lastAgentStatusObservedLive && terminal.lastAgentStatus === null) ||
|
||||
(!isRecovering() &&
|
||||
terminal.lastAgentStatusObservedLive &&
|
||||
terminal.lastAgentStatus === null) ||
|
||||
(queueSafe && !settled) ||
|
||||
deps.isPtyAwaitingUserInput(input.ptyId)
|
||||
) {
|
||||
// Keep staged text pending until admission is safe; Enter could answer a dialog.
|
||||
deps.state.deferFlightUntilIdle(input.ptyId)
|
||||
input.flight.submitEnter = () => submitOrchestrationMailboxPointer(deps, input)
|
||||
deps.onDeliveryDecision?.(input.ptyId, { foreground: false })
|
||||
preserveFlight = true
|
||||
} else if (!queueSafe) {
|
||||
releaseWithoutRedrive = true
|
||||
@@ -172,29 +205,53 @@ export function submitOrchestrationMailboxPointer(
|
||||
if (db?.markMailboxPointerEnterAttempted(messageIds, reservationTarget)) {
|
||||
expectedPhase = MAILBOX_POINTER_ENTER_ATTEMPTED
|
||||
} else if (db?.areUnreadMessages(input.mailboxHandle, messageIds) !== false) {
|
||||
// Unread mail lost its claim to an earlier Enter or a release; read mail still gets its Enter.
|
||||
// A prior attempted Enter cannot authorize replay; read mail still gets its Enter.
|
||||
preserveFlight = isRecovering()
|
||||
return
|
||||
}
|
||||
const enterSettlement = await deps.writePty(input.ptyId, '\r')
|
||||
submitted = enterSettlement.outcome === 'accepted'
|
||||
if (isRecovering() && enterSettlement.outcome === 'unverifiable') {
|
||||
preserveFlight = true
|
||||
}
|
||||
if (!deps.state.isCurrentFlight(input.ptyId, input.flight)) {
|
||||
finalizeReservation = false
|
||||
return
|
||||
}
|
||||
// Unverifiable Enter stays attempted: neither delivery nor safe resend is proven.
|
||||
if (enterSettlement.outcome === 'refused') {
|
||||
releaseWithoutRedrive = true
|
||||
if (isRecovering()) {
|
||||
preserveAmbiguousDelivery = false
|
||||
preserveFlight = true
|
||||
if (
|
||||
expectedPhase === MAILBOX_POINTER_ENTER_ATTEMPTED &&
|
||||
db?.markMailboxPointerWriteAttempted(messageIds, reservationTarget, expectedPhase)
|
||||
) {
|
||||
expectedPhase = MAILBOX_POINTER_WRITE_ATTEMPTED
|
||||
}
|
||||
} else {
|
||||
releaseWithoutRedrive = true
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
.catch(() => {
|
||||
if (!preserveAmbiguousDelivery) {
|
||||
if (deps.state.isCurrentFlight(input.ptyId, input.flight) && isRecovering()) {
|
||||
preserveFlight = true
|
||||
} else if (!preserveAmbiguousDelivery) {
|
||||
clearAndRedrive = true
|
||||
redriveClearedPointer = false
|
||||
}
|
||||
})
|
||||
.finally(() => {
|
||||
if (preserveFlight) {
|
||||
if (preserveFlight && deps.state.isCurrentFlight(input.ptyId, input.flight)) {
|
||||
deps.state.deferFlightUntilIdle(input.ptyId)
|
||||
input.flight.submitEnter = () =>
|
||||
submitOrchestrationMailboxPointer(deps, {
|
||||
...input,
|
||||
enterUncertain: preserveAmbiguousDelivery
|
||||
})
|
||||
deps.onDeliveryDecision?.(input.ptyId, isRecovering() ? undefined : { foreground: false })
|
||||
return
|
||||
}
|
||||
let released = false
|
||||
|
||||
@@ -1,167 +0,0 @@
|
||||
import type { MailboxForeground } from '../../../terminal/wait/launched-agent-foreground'
|
||||
import type { WriteSettlement } from '../../../../../shared/terminal/pty/lifecycle/pty-write-settlement'
|
||||
import type {
|
||||
OrchestrationMailboxDeliveryFlight,
|
||||
OrchestrationMailboxPointerState
|
||||
} from './mailbox-pointer-state'
|
||||
import type { PointerSubmitDependencies, PointerSubmitInput } from './mailbox-pointer-submit'
|
||||
import {
|
||||
MAILBOX_POINTER_ENTER_ATTEMPTED,
|
||||
MAILBOX_POINTER_WRITE_ATTEMPTED
|
||||
} from '../../db/messages/mailbox-pointer-enter-state'
|
||||
|
||||
type RecoveryEvidence = {
|
||||
targetCurrent?: boolean
|
||||
proofCurrent?: boolean
|
||||
foreground?: MailboxForeground
|
||||
ready?: boolean
|
||||
composer?: string | null
|
||||
write?: WriteSettlement
|
||||
}
|
||||
|
||||
/** A recovery flight owes its pointer and Enter until transport acceptance or explicit release. */
|
||||
export function recoveryFlightDecision(
|
||||
state: OrchestrationMailboxPointerState,
|
||||
ptyId: string,
|
||||
flight: OrchestrationMailboxDeliveryFlight | undefined,
|
||||
evidence?: RecoveryEvidence
|
||||
): 'ordinary' | 'revalidate' | 'retry' | 'release' | 'enter' | 'settle' | 'uncertain' {
|
||||
if (
|
||||
!flight?.processIncarnation ||
|
||||
state.foreground.screenRecovery(ptyId) !== flight.processIncarnation
|
||||
) {
|
||||
return 'ordinary'
|
||||
}
|
||||
if (!evidence) {
|
||||
return 'revalidate'
|
||||
}
|
||||
if (evidence.targetCurrent === false) {
|
||||
return 'release'
|
||||
}
|
||||
if (evidence.write) {
|
||||
return evidence.write.outcome === 'refused'
|
||||
? 'retry'
|
||||
: evidence.write.outcome === 'accepted'
|
||||
? 'settle'
|
||||
: 'uncertain'
|
||||
}
|
||||
if (!evidence.proofCurrent || evidence.foreground !== 'agent' || !evidence.ready) {
|
||||
return 'retry'
|
||||
}
|
||||
if (evidence.composer === '') {
|
||||
return 'release'
|
||||
}
|
||||
return evidence.composer === flight.stagedPointerText ? 'enter' : 'retry'
|
||||
}
|
||||
|
||||
export function submitRecoveryMailboxPointer(
|
||||
deps: PointerSubmitDependencies,
|
||||
input: PointerSubmitInput
|
||||
): boolean {
|
||||
const { state, getDb } = deps
|
||||
const { ptyId, flight } = input
|
||||
if (recoveryFlightDecision(state, ptyId, flight) === 'ordinary') {
|
||||
return false
|
||||
}
|
||||
const ids = input.messages.map((message) => message.id)
|
||||
const reservation = { ptyId, processIncarnation: input.expectedTarget.processIncarnation }
|
||||
let phase = MAILBOX_POINTER_WRITE_ATTEMPTED
|
||||
let decision: ReturnType<typeof recoveryFlightDecision> = 'retry'
|
||||
let attempted = false
|
||||
void (async () => {
|
||||
const absent = await deps.isLeafPtyProvenAbsent(ptyId)
|
||||
const before = deps.resolveSubmitTarget(ptyId)
|
||||
if (absent || before?.processIncarnation !== reservation.processIncarnation) {
|
||||
decision = recoveryFlightDecision(state, ptyId, flight, { targetCurrent: false })
|
||||
return
|
||||
}
|
||||
// Avoid host inspections for an unchanged blocked screen; admission is checked again after await.
|
||||
if (!deps.isAgentSettledForDelivery?.(ptyId)) {
|
||||
return
|
||||
}
|
||||
const proof = await state.foreground.probeForeground(
|
||||
ptyId,
|
||||
reservation.processIncarnation,
|
||||
() => deps.readForeground(ptyId)
|
||||
)
|
||||
if (!state.isCurrentFlight(ptyId, flight)) {
|
||||
return
|
||||
}
|
||||
const target = deps.resolveSubmitTarget(ptyId)
|
||||
const current =
|
||||
target?.processIncarnation === reservation.processIncarnation &&
|
||||
target.terminal.handle === input.expectedTarget.terminal.handle &&
|
||||
deps.mailboxOwner.resolve(target.terminal, input.mailboxHandle) === input.mailboxHandle
|
||||
const ready =
|
||||
current &&
|
||||
proof.foreground === 'agent' &&
|
||||
target.terminal.lastAgentStatusObservedLive &&
|
||||
deps.isAgentSettledForDelivery?.(ptyId) !== false
|
||||
const applied = state.foreground.applyForeground(ptyId, proof, proof.foreground, ready)
|
||||
decision = recoveryFlightDecision(state, ptyId, flight, {
|
||||
targetCurrent: !!current,
|
||||
proofCurrent: applied,
|
||||
foreground: proof.foreground,
|
||||
ready,
|
||||
composer: ready ? deps.readComposerText?.(ptyId) : null
|
||||
})
|
||||
if (decision !== 'enter') {
|
||||
return
|
||||
}
|
||||
const db = getDb()
|
||||
if (db?.areUnreadMessages(input.mailboxHandle, ids) === false) {
|
||||
decision = 'release'
|
||||
return
|
||||
}
|
||||
if (!db?.markMailboxPointerEnterAttempted(ids, reservation)) {
|
||||
return
|
||||
}
|
||||
phase = MAILBOX_POINTER_ENTER_ATTEMPTED
|
||||
attempted = true
|
||||
const write = await deps.writePty(ptyId, '\r')
|
||||
decision = recoveryFlightDecision(state, ptyId, flight, { write })
|
||||
if (decision === 'retry') {
|
||||
// Refusal proves no Enter byte left; restore the same claim instead of duplicating the pointer.
|
||||
if (!db.markMailboxPointerWriteAttempted(ids, reservation, MAILBOX_POINTER_ENTER_ATTEMPTED)) {
|
||||
decision = 'uncertain'
|
||||
} else {
|
||||
phase = MAILBOX_POINTER_WRITE_ATTEMPTED
|
||||
}
|
||||
}
|
||||
})()
|
||||
.catch(() => {
|
||||
decision = attempted ? 'uncertain' : 'retry'
|
||||
})
|
||||
.finally(() => {
|
||||
if (!state.isCurrentFlight(ptyId, flight)) {
|
||||
return
|
||||
}
|
||||
if (decision === 'retry' || decision === 'enter') {
|
||||
state.deferFlightUntilIdle(ptyId)
|
||||
flight.submitEnter = () => submitRecoveryMailboxPointer(deps, input)
|
||||
deps.onDeliveryDecision?.(ptyId)
|
||||
return
|
||||
}
|
||||
try {
|
||||
if (decision === 'settle') {
|
||||
getDb()?.settleMailboxPointerEnter(ids, reservation, [phase])
|
||||
} else if (decision === 'release') {
|
||||
getDb()?.releaseMailboxPointerEnter(ids, reservation, [phase])
|
||||
deps.onDeliveryDecision?.(ptyId, { foreground: true })
|
||||
}
|
||||
} catch {
|
||||
// Keep ambiguous reservations durable when the database becomes unavailable.
|
||||
decision = 'uncertain'
|
||||
}
|
||||
const released =
|
||||
decision === 'settle' || decision === 'release'
|
||||
? state.clearWatermark(input.mailboxHandle, input.newestSequence, ptyId)
|
||||
: state.deactivateWatermark(input.mailboxHandle, input.newestSequence, ptyId)
|
||||
deps.settle(ptyId, flight)
|
||||
if (released) {
|
||||
deps.redrive(input.mailboxHandle, decision === 'release')
|
||||
}
|
||||
})
|
||||
.catch(() => undefined)
|
||||
return true
|
||||
}
|
||||
+10
-7
@@ -1,4 +1,4 @@
|
||||
import { readTerminalComposerTypedText } from '../../../../shared/terminal/input/terminal-composer-draft'
|
||||
import { readAgentTerminalComposerTypedText } from '../../terminal/wait/terminal-delivery-refusal'
|
||||
import type { MailboxForeground } from '../../terminal/wait/launched-agent-foreground'
|
||||
import type { TerminalIdentityReadContext } from '../../terminal/query/terminal-identity-read-context'
|
||||
import { isRemoteRuntimePtyId } from '../../../../shared/remote-runtime/remote-runtime-pty-id'
|
||||
@@ -24,6 +24,15 @@ import { startedMainTerminalSurfaces } from '../../terminal/surface/main-termina
|
||||
import { mainPaneVerdict } from '../../terminal/surface/main-pane-verdict'
|
||||
|
||||
export abstract class OrcaRuntimeWithWriteOrchestrationPointerPty extends OrcaRuntimeWithRefreshFloatingWorkspacePtyLiveness {
|
||||
protected abstract mailDeliveryAgent(ptyId: string): string | null | undefined
|
||||
|
||||
protected readMailboxComposerText(ptyId: string): string | null {
|
||||
return readAgentTerminalComposerTypedText(
|
||||
this.mailDeliveryAgent(ptyId) ?? null,
|
||||
this.readWholeScreenModel(ptyId)?.emulator.getCursorLineContext()
|
||||
)
|
||||
}
|
||||
|
||||
protected abstract issueHandle(
|
||||
leaf: RuntimeLeafRecord,
|
||||
context?: TerminalIdentityReadContext
|
||||
@@ -40,12 +49,6 @@ export abstract class OrcaRuntimeWithWriteOrchestrationPointerPty extends OrcaRu
|
||||
})
|
||||
}
|
||||
|
||||
protected readMailboxComposerText(ptyId: string): string | null {
|
||||
return readTerminalComposerTypedText(
|
||||
this.readWholeScreenModel(ptyId)?.emulator.getCursorLineContext()
|
||||
)
|
||||
}
|
||||
|
||||
/** The terminal a staged pointer's Enter goes to, re-read so a replaced process gets none. */
|
||||
protected resolveOrchestrationPointerSubmitTarget(
|
||||
ptyId: string,
|
||||
|
||||
@@ -238,7 +238,6 @@ export abstract class OrcaRuntimeWithMaybeHydrateHeadlessFromRenderer extends Or
|
||||
data
|
||||
)
|
||||
state.outputSequence = outputSequence
|
||||
this.orchestrationMailboxPointerDelivery.observeOutput(ptyId)
|
||||
return
|
||||
}
|
||||
const escapeTail = state.rendererSeedEscapeTail
|
||||
@@ -288,7 +287,6 @@ export abstract class OrcaRuntimeWithMaybeHydrateHeadlessFromRenderer extends Or
|
||||
}
|
||||
}
|
||||
state.outputSequence = outputSequence
|
||||
this.orchestrationMailboxPointerDelivery.observeOutput(ptyId)
|
||||
})
|
||||
state.writeChain = completion.catch(() => {})
|
||||
}
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import { agentStatusTakesTypedInput } from '../../orchestration/mailbox/routing/mailbox-owner'
|
||||
import { openCodeComposerHoldsDraft } from '../../../../shared/terminal/input/opencode-composer'
|
||||
import {
|
||||
openCodeComposerHoldsDraft,
|
||||
readOpenCodeComposerTypedText
|
||||
} from '../../../../shared/terminal/input/opencode-composer'
|
||||
import {
|
||||
readTerminalComposerTypedText,
|
||||
type TerminalCursorContext
|
||||
@@ -23,7 +26,11 @@ export function mailboxDeliveryRefusal(
|
||||
recognizedAgent: boolean,
|
||||
stagedText?: string
|
||||
) {
|
||||
const ownPointer = terminalComposerTextOwnsPointer(readComposerText(), stagedText)
|
||||
const admissionSource = source
|
||||
if (recovering && ownPointer) {
|
||||
source = { ...source, quiescenceMs: 0 }
|
||||
}
|
||||
const screen = recovering ? source.readScreenLines(pty.ptyId) : null
|
||||
if (recovering) {
|
||||
// An execution identity cannot renew an old hook's readiness or a tail's composer.
|
||||
@@ -55,7 +62,7 @@ export function mailboxDeliveryRefusal(
|
||||
() => {
|
||||
const text = readComposerText()
|
||||
return (
|
||||
text === stagedText ||
|
||||
terminalComposerTextOwnsPointer(text, stagedText) ||
|
||||
(text === '' &&
|
||||
(pty.lastAgentStatus === null ||
|
||||
agentStatusTakesTypedInput(pty.lastAgentStatus)))
|
||||
@@ -118,3 +125,36 @@ export function mailboxComposerHoldsDraft(
|
||||
? openCodeComposerHoldsDraft(context)
|
||||
: !!readTerminalComposerTypedText(context)
|
||||
}
|
||||
|
||||
export function readAgentTerminalComposerTypedText(
|
||||
agent: string | null,
|
||||
context: TerminalCursorContext | null | undefined
|
||||
): string | null {
|
||||
if (context && (agent === 'opencode' || agent === 'opencode2')) {
|
||||
return readOpenCodeComposerTypedText(context)
|
||||
}
|
||||
return readTerminalComposerTypedText(context)
|
||||
}
|
||||
|
||||
/** Agents paint hard-wrapped rows themselves, including breaks inside a word. */
|
||||
export function terminalComposerTextOwnsPointer(
|
||||
shown: string | null | undefined,
|
||||
pointer: string | undefined
|
||||
): boolean {
|
||||
if (!shown || !pointer) {
|
||||
return false
|
||||
}
|
||||
const expected = pointer.replace(/\s+/g, ' ').trim()
|
||||
let offset = 0
|
||||
for (const line of shown.split('\n')) {
|
||||
const row = line.trim().replace(/\s+/g, ' ')
|
||||
if (!expected.startsWith(row, offset)) {
|
||||
return false
|
||||
}
|
||||
offset += row.length
|
||||
if (expected[offset] === ' ') {
|
||||
offset++
|
||||
}
|
||||
}
|
||||
return offset === expected.length
|
||||
}
|
||||
|
||||
@@ -9,11 +9,14 @@ import type { AgentStatus } from '../../../../shared/agent-status/title/agent-ti
|
||||
import { detectExplicitIdleStatusFromTitle } from './terminal-wait-detection'
|
||||
import { buildTerminalWaitText } from './terminal-wait-tail-state'
|
||||
import { evaluateTuiIdle, type TuiIdleVerdict } from './tui-idle-evidence'
|
||||
import { mailboxDeliveryRefusal, mailboxComposerHoldsDraft } from './terminal-delivery-refusal'
|
||||
import {
|
||||
mailboxDeliveryRefusal,
|
||||
mailboxComposerHoldsDraft,
|
||||
terminalComposerTextOwnsPointer
|
||||
} from './terminal-delivery-refusal'
|
||||
import { ptyTuiIdleEvidence } from './tui-idle-evidence-source'
|
||||
import { TUI_IDLE_QUIESCENCE_MS } from './terminal-idle-policy'
|
||||
import { USER_INPUT_HOLD_CAP_MS } from './claude-open-dialog-hold'
|
||||
import { readTerminalComposerTypedText } from '../../../../shared/terminal/input/terminal-composer-draft'
|
||||
import {
|
||||
MailboxDeliveryRecheck,
|
||||
type MailboxDeliveryDecision
|
||||
@@ -252,7 +255,7 @@ export abstract class OrcaRuntimeWithResolveExitWaiters extends OrcaRuntimeWithB
|
||||
)
|
||||
}
|
||||
|
||||
private mailDeliveryAgent(ptyId: string) {
|
||||
protected mailDeliveryAgent(ptyId: string) {
|
||||
const agent = this.getPaneTerminalAgent(ptyId)
|
||||
if (agent) {
|
||||
return agent
|
||||
@@ -275,11 +278,9 @@ export abstract class OrcaRuntimeWithResolveExitWaiters extends OrcaRuntimeWithB
|
||||
const facts = this.terminalRunFacts
|
||||
const state = this.readWholeScreenModel(ptyId)
|
||||
const context = state?.emulator.getCursorLineContext()
|
||||
const ownPointer =
|
||||
stagedText !== undefined && readTerminalComposerTypedText(context) === stagedText
|
||||
const draft =
|
||||
!!context &&
|
||||
!ownPointer &&
|
||||
!terminalComposerTextOwnsPointer(this.readMailboxComposerText(ptyId), stagedText) &&
|
||||
mailboxComposerHoldsDraft(this.getPaneTerminalAgent(ptyId), context)
|
||||
const drivingInputAt = facts.readLastDrivingInputAt(ptyId)
|
||||
const unattributed = drivingInputAt === null && facts.readProcessStartedAt(ptyId) === null
|
||||
|
||||
@@ -8,9 +8,8 @@ const OPENCODE_BOX_GUTTER = '┃'
|
||||
// A stock prompt, then a quoted example that a narrow box may cut short.
|
||||
const OPENCODE_PLACEHOLDER = /^(?:Ask anything\u2026|Run a command\u2026)(?: "[^"]*"?)?$/
|
||||
|
||||
/** Whether OpenCode's composer shows a typed draft, so mail must wait. False when the cursor is not
|
||||
* in a box laid out as OpenCode's. Callers check that OpenCode is the pane's agent. */
|
||||
export function openCodeComposerHoldsDraft(context: TerminalCursorContext): boolean {
|
||||
/** Typed textarea, empty placeholder, or null when the cursor is outside OpenCode's composer. */
|
||||
export function readOpenCodeComposerTypedText(context: TerminalCursorContext): string | null {
|
||||
const cursorIndex = context.rows.length - 1
|
||||
const column = (context.rows[cursorIndex] ?? '').indexOf(OPENCODE_BOX_GUTTER)
|
||||
const inBox = (row: string | undefined): boolean =>
|
||||
@@ -19,7 +18,7 @@ export function openCodeComposerHoldsDraft(context: TerminalCursorContext): bool
|
||||
row.startsWith(OPENCODE_BOX_GUTTER, column) &&
|
||||
!row.slice(0, column).trim()
|
||||
if (context.cursorHidden || !inBox(context.rows[cursorIndex])) {
|
||||
return false
|
||||
return null
|
||||
}
|
||||
let top = cursorIndex
|
||||
while (inBox(context.rows[top - 1])) {
|
||||
@@ -38,7 +37,7 @@ export function openCodeComposerHoldsDraft(context: TerminalCursorContext): bool
|
||||
const spacer = rows.findLastIndex((row) => !row)
|
||||
// Not a padding row, textarea, spacer and footer around the cursor: read as any other prompt.
|
||||
if (cursorLine === 0 || rows[0] || spacer <= cursorLine || spacer === rows.length - 1) {
|
||||
return false
|
||||
return null
|
||||
}
|
||||
const textarea = rows.slice(1, spacer)
|
||||
// The placeholder wraps onto more rows in a narrow box.
|
||||
@@ -48,5 +47,9 @@ export function openCodeComposerHoldsDraft(context: TerminalCursorContext): bool
|
||||
textarea.every(Boolean) &&
|
||||
!context.beforeCursor.slice(column + OPENCODE_BOX_GUTTER.length).trim() &&
|
||||
OPENCODE_PLACEHOLDER.test(shown)
|
||||
return !!shown && !placeholder
|
||||
return placeholder ? '' : textarea.join('\n').trim()
|
||||
}
|
||||
|
||||
export function openCodeComposerHoldsDraft(context: TerminalCursorContext): boolean {
|
||||
return !!readOpenCodeComposerTypedText(context)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user