mirror of
https://github.com/stablyai/orca.git
synced 2026-10-16 16:03:26 +00:00
Rework Codex completion status repair
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import type { ParsedAgentStatusPayload } from '../../../../shared/agent-status-types'
|
||||
import type { AgentType, ParsedAgentStatusPayload } from '../../../../shared/agent-status-types'
|
||||
import type { RecognizedAgentProcess } from '../../../../shared/agent-process-recognition'
|
||||
import type { GlobalSettings } from '../../../../shared/types'
|
||||
import type { RuntimeTerminalProcessInspection } from '@/runtime/runtime-terminal-inspection'
|
||||
@@ -18,6 +18,18 @@ export type AgentAttentionDispatchMeta = {
|
||||
agentStatus: AgentCompletionStatusSnapshot
|
||||
}
|
||||
|
||||
export type AgentCompletionStatusRepairSignal =
|
||||
| {
|
||||
source: 'title'
|
||||
title: string
|
||||
agentType?: AgentType
|
||||
}
|
||||
| {
|
||||
source: 'process-exit'
|
||||
title: string
|
||||
agent: RecognizedAgentProcess
|
||||
}
|
||||
|
||||
export type AgentCompletionCoordinatorOptions = {
|
||||
paneKey: string
|
||||
getPtyId: () => string | null
|
||||
@@ -31,8 +43,8 @@ export type AgentCompletionCoordinatorOptions = {
|
||||
isLive: () => boolean
|
||||
shouldPollProcessCadence?: () => boolean
|
||||
shouldSuppressHookCompletion?: (payload: AgentCompletionStatusSnapshot) => boolean
|
||||
onProcessExitCompletion?: (
|
||||
agent: RecognizedAgentProcess
|
||||
onCompletionStatusRepair?: (
|
||||
signal: AgentCompletionStatusRepairSignal
|
||||
) => AgentCompletionStatusSnapshot | null | undefined
|
||||
}
|
||||
|
||||
|
||||
@@ -206,7 +206,7 @@ describe('agent completion coordinator', () => {
|
||||
it('passes synthesized status detail when the process-exit backstop completes a row', async () => {
|
||||
let foregroundProcess: string | null = 'codex'
|
||||
const dispatchCompletion = vi.fn()
|
||||
const onProcessExitCompletion = vi.fn(() => ({
|
||||
const onCompletionStatusRepair = vi.fn(() => ({
|
||||
state: 'done' as const,
|
||||
prompt: 'Fix the spinner',
|
||||
agentType: 'codex' as const,
|
||||
@@ -218,7 +218,7 @@ describe('agent completion coordinator', () => {
|
||||
getSettings: () => null,
|
||||
inspectProcess: vi.fn(async () => processResult(foregroundProcess)),
|
||||
dispatchCompletion,
|
||||
onProcessExitCompletion,
|
||||
onCompletionStatusRepair,
|
||||
isLive: () => true,
|
||||
shouldPollProcessCadence: () => false
|
||||
})
|
||||
@@ -228,9 +228,13 @@ describe('agent completion coordinator', () => {
|
||||
foregroundProcess = null
|
||||
await vi.advanceTimersByTimeAsync(6_000)
|
||||
|
||||
expect(onProcessExitCompletion).toHaveBeenCalledWith({
|
||||
agent: 'codex',
|
||||
processName: 'codex'
|
||||
expect(onCompletionStatusRepair).toHaveBeenCalledWith({
|
||||
source: 'process-exit',
|
||||
title: 'codex',
|
||||
agent: {
|
||||
agent: 'codex',
|
||||
processName: 'codex'
|
||||
}
|
||||
})
|
||||
expect(dispatchCompletion).toHaveBeenCalledWith('codex', {
|
||||
source: 'process-exit',
|
||||
@@ -338,6 +342,92 @@ describe('agent completion coordinator', () => {
|
||||
expect(dispatchCompletion).toHaveBeenCalledWith('codex done')
|
||||
})
|
||||
|
||||
it('repairs a working status row from a title completion signal', () => {
|
||||
const dispatchCompletion = vi.fn()
|
||||
const onCompletionStatusRepair = vi.fn(() => ({
|
||||
state: 'done' as const,
|
||||
prompt: 'Fix title-only fatal states',
|
||||
agentType: 'codex' as const,
|
||||
lastAssistantMessage: 'Codex stopped before emitting Stop.'
|
||||
}))
|
||||
const coordinator = createAgentCompletionCoordinator({
|
||||
paneKey: 'tab-1:leaf-1',
|
||||
getPtyId: () => 'pty-1',
|
||||
getSettings: () => null,
|
||||
inspectProcess: vi.fn(),
|
||||
dispatchCompletion,
|
||||
onCompletionStatusRepair,
|
||||
isLive: () => true
|
||||
})
|
||||
|
||||
coordinator.observeTitle('Codex working')
|
||||
coordinator.observeTitle('Codex done')
|
||||
|
||||
expect(onCompletionStatusRepair).toHaveBeenCalledWith({
|
||||
source: 'title',
|
||||
title: 'Codex done',
|
||||
agentType: 'codex'
|
||||
})
|
||||
expect(dispatchCompletion).toHaveBeenCalledWith('Codex done', {
|
||||
source: 'title',
|
||||
quietedHookDone: false,
|
||||
agentStatus: {
|
||||
state: 'done',
|
||||
prompt: 'Fix title-only fatal states',
|
||||
agentType: 'codex',
|
||||
lastAssistantMessage: 'Codex stopped before emitting Stop.'
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('repairs process-exit status even when title completion already deduped notification', async () => {
|
||||
let foregroundProcess: string | null = 'codex'
|
||||
const dispatchCompletion = vi.fn()
|
||||
const onCompletionStatusRepair = vi.fn((signal) =>
|
||||
signal.source === 'process-exit'
|
||||
? {
|
||||
state: 'done' as const,
|
||||
prompt: 'Repair after dedupe',
|
||||
agentType: 'codex' as const
|
||||
}
|
||||
: null
|
||||
)
|
||||
const coordinator = createAgentCompletionCoordinator({
|
||||
paneKey: 'tab-1:leaf-1',
|
||||
getPtyId: () => 'pty-1',
|
||||
getSettings: () => null,
|
||||
inspectProcess: vi.fn(async () => processResult(foregroundProcess)),
|
||||
dispatchCompletion,
|
||||
onCompletionStatusRepair,
|
||||
isLive: () => true
|
||||
})
|
||||
|
||||
coordinator.startProcessTracking()
|
||||
vi.advanceTimersByTime(2_000)
|
||||
await flushAsyncTicks()
|
||||
coordinator.observeTitle('⠋ codex')
|
||||
coordinator.observeTitle('codex done')
|
||||
|
||||
foregroundProcess = null
|
||||
vi.advanceTimersByTime(750)
|
||||
await flushAsyncTicks()
|
||||
vi.advanceTimersByTime(750)
|
||||
await flushAsyncTicks()
|
||||
|
||||
expect(dispatchCompletion).toHaveBeenCalledTimes(1)
|
||||
expect(dispatchCompletion).toHaveBeenCalledWith('codex done')
|
||||
expect(onCompletionStatusRepair).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
source: 'process-exit',
|
||||
title: 'codex',
|
||||
agent: {
|
||||
agent: 'codex',
|
||||
processName: 'codex'
|
||||
}
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('does not dispatch a cwd title after an explicit agent working title if the shell owns the pane', async () => {
|
||||
const dispatchCompletion = vi.fn()
|
||||
const coordinator = createAgentCompletionCoordinator({
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
/* oxlint-disable max-lines */
|
||||
import { detectAgentStatusFromTitle, type AgentStatus } from '../../../../shared/agent-detection'
|
||||
import type { ParsedAgentStatusPayload } from '../../../../shared/agent-status-types'
|
||||
import type { AgentType, ParsedAgentStatusPayload } from '../../../../shared/agent-status-types'
|
||||
import {
|
||||
isRecognizedAgentType,
|
||||
recognizeAgentProcess,
|
||||
@@ -227,6 +227,14 @@ export function createAgentCompletionCoordinator(
|
||||
return null
|
||||
}
|
||||
|
||||
function resolveTitleCompletionAgentType(title: string): AgentType | undefined {
|
||||
const titleAgentIdentity = titleCompletionAgentIdentity(title)
|
||||
if (titleAgentIdentity) {
|
||||
return titleAgentIdentity
|
||||
}
|
||||
return lastForegroundAgent?.agent
|
||||
}
|
||||
|
||||
function completionIdentityAlreadyNotified(
|
||||
completionIdentity: LastCompletionIdentity | null | undefined
|
||||
): boolean {
|
||||
@@ -247,6 +255,31 @@ export function createAgentCompletionCoordinator(
|
||||
)
|
||||
}
|
||||
|
||||
function repairCompletionStatus(
|
||||
source: CompletionSource,
|
||||
title: string,
|
||||
optionsOverride: {
|
||||
processExitAgent?: RecognizedAgentProcess
|
||||
}
|
||||
): AgentCompletionStatusSnapshot | null | undefined {
|
||||
if (source === 'process-exit' && optionsOverride.processExitAgent) {
|
||||
return options.onCompletionStatusRepair?.({
|
||||
source,
|
||||
title,
|
||||
agent: optionsOverride.processExitAgent
|
||||
})
|
||||
}
|
||||
if (source !== 'title') {
|
||||
return undefined
|
||||
}
|
||||
const agentType = resolveTitleCompletionAgentType(title)
|
||||
return options.onCompletionStatusRepair?.({
|
||||
source,
|
||||
title,
|
||||
...(agentType ? { agentType } : {})
|
||||
})
|
||||
}
|
||||
|
||||
function dispatchCompletion(
|
||||
source: CompletionSource,
|
||||
title: string,
|
||||
@@ -257,6 +290,7 @@ export function createAgentCompletionCoordinator(
|
||||
processExitAgent?: RecognizedAgentProcess
|
||||
} = {}
|
||||
): void {
|
||||
const repairedAgentStatus = repairCompletionStatus(source, title, optionsOverride)
|
||||
if (source !== 'hook' && pendingHookDoneTimer !== null) {
|
||||
return
|
||||
}
|
||||
@@ -282,11 +316,7 @@ export function createAgentCompletionCoordinator(
|
||||
if (optionsOverride.completionIdentity) {
|
||||
lastCompletionIdentityByPaneKey.set(options.paneKey, optionsOverride.completionIdentity)
|
||||
}
|
||||
const agentStatus =
|
||||
optionsOverride.agentStatus ??
|
||||
(source === 'process-exit' && optionsOverride.processExitAgent
|
||||
? (options.onProcessExitCompletion?.(optionsOverride.processExitAgent) ?? undefined)
|
||||
: undefined)
|
||||
const agentStatus = optionsOverride.agentStatus ?? repairedAgentStatus
|
||||
if (optionsOverride.quietedHookDone === true || agentStatus) {
|
||||
options.dispatchCompletion(title, {
|
||||
source,
|
||||
|
||||
@@ -32,6 +32,87 @@ describe('Codex error output status detector', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('waits for the full line when the stream-disconnect error splits after the colon', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
|
||||
expect(detector.observe('■ stream disconnected before completion:')).toBe(false)
|
||||
expect(onStreamError).not.toHaveBeenCalled()
|
||||
|
||||
expect(
|
||||
detector.observe(
|
||||
' error sending request for url (http://openclaw:2455/backend-api/codex/responses)\r\n'
|
||||
)
|
||||
).toBe(true)
|
||||
|
||||
expect(onStreamError).toHaveBeenCalledTimes(1)
|
||||
expect(onStreamError).toHaveBeenCalledWith(
|
||||
'stream disconnected before completion: error sending request for url (http://openclaw:2455/backend-api/codex/responses)'
|
||||
)
|
||||
})
|
||||
|
||||
it('detects a stream-disconnect error in the middle of a large PTY chunk', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
const chunk =
|
||||
`${'a'.repeat(8_000)}\r\n` +
|
||||
'■ stream disconnected before completion: context window exceeded\r\n' +
|
||||
`${'b'.repeat(8_000)}\r\n`
|
||||
|
||||
expect(detector.observe(chunk)).toBe(true)
|
||||
|
||||
expect(onStreamError).toHaveBeenCalledWith(
|
||||
'stream disconnected before completion: context window exceeded'
|
||||
)
|
||||
})
|
||||
|
||||
it('does not complete transient Codex retry notices', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
|
||||
expect(
|
||||
detector.observe(
|
||||
'stream error: stream disconnected before completion: temporary network failure; retrying 1/5 in 217ms\r\n'
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
detector.observe(
|
||||
'stream disconnected before completion: temporary network failure; retrying 2/5 in 431ms\r\n'
|
||||
)
|
||||
).toBe(false)
|
||||
|
||||
expect(onStreamError).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('keeps the lowercase Codex fatal marker invariant', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
|
||||
expect(
|
||||
detector.observe('Stream disconnected before completion: status panel repaint\r\n')
|
||||
).toBe(false)
|
||||
|
||||
expect(onStreamError).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not complete quoted or grepped stream-disconnect text', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
|
||||
expect(
|
||||
detector.observe(
|
||||
'log.txt: stream disconnected before completion: old error from an earlier turn\r\n'
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
detector.observe(
|
||||
'"stream disconnected before completion: old error from an earlier turn"\r\n'
|
||||
)
|
||||
).toBe(false)
|
||||
|
||||
expect(onStreamError).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('strips terminal control sequences before reporting the error message', () => {
|
||||
const onStreamError = vi.fn()
|
||||
const detector = createCodexErrorOutputStatusDetector({ onStreamError })
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
type CodexErrorOutputStatusDetector = {
|
||||
observe: (data: string) => boolean
|
||||
reset: () => void
|
||||
}
|
||||
|
||||
const ESC = String.fromCharCode(0x1b)
|
||||
@@ -12,10 +13,10 @@ const INCOMPLETE_ANSI_ESCAPE_RE = new RegExp(
|
||||
`${ESC}(?:\\[[0-?]*[ -/]*|\\][^${BEL}${ESC}]*|\\S?)?$`,
|
||||
'g'
|
||||
)
|
||||
const RECENT_RAW_TEXT_LIMIT = 300
|
||||
const STATUS_SCAN_TEXT_LIMIT = 4096
|
||||
const CODEX_STREAM_DISCONNECTED_MARKER = 'stream disconnected before completion:'
|
||||
const CODEX_STREAM_DISCONNECTED_RE = /\bstream disconnected before completion:[^\r\n]*/
|
||||
const CODEX_STREAM_DISCONNECTED_CARRY_LENGTH = CODEX_STREAM_DISCONNECTED_MARKER.length - 1
|
||||
const MAX_PENDING_STREAM_ERROR_LINE_LENGTH = 8_000
|
||||
const RETRY_NOTICE_RE = /;\s*retrying\b/
|
||||
|
||||
function terminalControlMayAffectText(data: string): boolean {
|
||||
for (let index = 0; index < data.length; index += 1) {
|
||||
@@ -48,91 +49,138 @@ function stripTerminalControl(data: string): string {
|
||||
return output
|
||||
}
|
||||
|
||||
function appendRecentRawText(previousRawText: string, data: string): string {
|
||||
if (data.length >= RECENT_RAW_TEXT_LIMIT) {
|
||||
return data.slice(-RECENT_RAW_TEXT_LIMIT)
|
||||
function appendMarkerCarry(carry: string, data: string): string {
|
||||
if (data.length >= CODEX_STREAM_DISCONNECTED_CARRY_LENGTH) {
|
||||
return data.slice(-CODEX_STREAM_DISCONNECTED_CARRY_LENGTH)
|
||||
}
|
||||
return (previousRawText + data).slice(-RECENT_RAW_TEXT_LIMIT)
|
||||
return (carry + data).slice(-CODEX_STREAM_DISCONNECTED_CARRY_LENGTH)
|
||||
}
|
||||
|
||||
function buildStatusScanRawText(prefix: string, data: string): string {
|
||||
const boundedPrefix =
|
||||
prefix.length > RECENT_RAW_TEXT_LIMIT + 1 ? prefix.slice(-(RECENT_RAW_TEXT_LIMIT + 1)) : prefix
|
||||
const dataBudget = STATUS_SCAN_TEXT_LIMIT - boundedPrefix.length
|
||||
|
||||
if (dataBudget <= 0) {
|
||||
return boundedPrefix.slice(-STATUS_SCAN_TEXT_LIMIT)
|
||||
}
|
||||
if (data.length <= dataBudget) {
|
||||
return boundedPrefix + data
|
||||
}
|
||||
|
||||
const headBudget = Math.max(0, Math.floor((dataBudget - 1) / 2))
|
||||
const tailBudget = Math.max(0, dataBudget - headBudget - 1)
|
||||
const head = headBudget > 0 ? data.slice(0, headBudget) : ''
|
||||
const tail = tailBudget > 0 ? data.slice(-tailBudget) : ''
|
||||
// Why: PTY chunks can be large pastes or hidden-output restores; stream-error
|
||||
// detection only needs chunk-boundary context plus the current edge windows.
|
||||
return `${boundedPrefix}${head}\n${tail}`
|
||||
function updatePendingLine(pendingLine: string, data: string): string {
|
||||
return (pendingLine + data).slice(0, MAX_PENDING_STREAM_ERROR_LINE_LENGTH)
|
||||
}
|
||||
|
||||
function rawTextMayContainCodexStreamError(rawText: string): boolean {
|
||||
// Why: Codex emits this error in lowercase; avoid allocating a lowercased
|
||||
// copy for every ordinary PTY chunk on the no-match hot path.
|
||||
return rawText.includes(CODEX_STREAM_DISCONNECTED_MARKER)
|
||||
function findLineEnd(value: string, start: number): number {
|
||||
const carriageReturnIndex = value.indexOf('\r', start)
|
||||
const newlineIndex = value.indexOf('\n', start)
|
||||
if (carriageReturnIndex === -1) {
|
||||
return newlineIndex
|
||||
}
|
||||
if (newlineIndex === -1) {
|
||||
return carriageReturnIndex
|
||||
}
|
||||
return Math.min(carriageReturnIndex, newlineIndex)
|
||||
}
|
||||
|
||||
function findOverlappingCodexStreamError(
|
||||
scanText: string,
|
||||
previousTextLength: number
|
||||
): RegExpMatchArray | null {
|
||||
const re = new RegExp(CODEX_STREAM_DISCONNECTED_RE.source, 'g')
|
||||
for (const match of scanText.matchAll(re)) {
|
||||
const start = match.index ?? 0
|
||||
if (start + match[0].length > previousTextLength) {
|
||||
return match
|
||||
function findLineStart(value: string, markerIndex: number): number {
|
||||
const previousCarriageReturn = value.lastIndexOf('\r', markerIndex)
|
||||
const previousNewline = value.lastIndexOf('\n', markerIndex)
|
||||
return Math.max(previousCarriageReturn, previousNewline) + 1
|
||||
}
|
||||
|
||||
function isLikelyCodexFatalLinePrefix(prefix: string): boolean {
|
||||
const trimmed = prefix.trim()
|
||||
// Why: transient retry notices prefix the marker with words like
|
||||
// "stream error:"; Codex's fatal TUI cell only has whitespace/glyph chrome.
|
||||
return trimmed === '' || !/[A-Za-z0-9:/"'`|\\-]/.test(trimmed)
|
||||
}
|
||||
|
||||
function normalizeStreamErrorLine(line: string): string | null {
|
||||
const strippedLine = stripTerminalControl(line)
|
||||
const markerIndex = strippedLine.indexOf(CODEX_STREAM_DISCONNECTED_MARKER)
|
||||
if (markerIndex === -1) {
|
||||
return null
|
||||
}
|
||||
if (!isLikelyCodexFatalLinePrefix(strippedLine.slice(0, markerIndex))) {
|
||||
return null
|
||||
}
|
||||
const message = strippedLine.slice(markerIndex).replace(/\s+/g, ' ').trim()
|
||||
if (!message || RETRY_NOTICE_RE.test(message)) {
|
||||
return null
|
||||
}
|
||||
return message
|
||||
}
|
||||
|
||||
function findCompletedStreamErrorLine(rawText: string): {
|
||||
message: string | null
|
||||
pendingLine: string | null
|
||||
} {
|
||||
let searchStart = 0
|
||||
while (searchStart < rawText.length) {
|
||||
const markerIndex = rawText.indexOf(CODEX_STREAM_DISCONNECTED_MARKER, searchStart)
|
||||
if (markerIndex === -1) {
|
||||
return { message: null, pendingLine: null }
|
||||
}
|
||||
const lineStart = findLineStart(rawText, markerIndex)
|
||||
const lineEnd = findLineEnd(rawText, markerIndex)
|
||||
if (lineEnd === -1) {
|
||||
const pendingLine = rawText.slice(lineStart)
|
||||
const message = normalizeStreamErrorLine(pendingLine)
|
||||
return message
|
||||
? { message: null, pendingLine: pendingLine.slice(0, MAX_PENDING_STREAM_ERROR_LINE_LENGTH) }
|
||||
: { message: null, pendingLine: null }
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
function extractCodexStreamErrorMessage(scanText: string, match: RegExpMatchArray): string {
|
||||
const start = match.index ?? 0
|
||||
const lineEndCandidates = [scanText.indexOf('\r', start), scanText.indexOf('\n', start)].filter(
|
||||
(index) => index >= 0
|
||||
)
|
||||
const lineEnd = lineEndCandidates.length > 0 ? Math.min(...lineEndCandidates) : scanText.length
|
||||
return scanText.slice(start, lineEnd).replace(/\s+/g, ' ').trim()
|
||||
const message = normalizeStreamErrorLine(rawText.slice(lineStart, lineEnd))
|
||||
if (message) {
|
||||
return { message, pendingLine: null }
|
||||
}
|
||||
searchStart = markerIndex + CODEX_STREAM_DISCONNECTED_MARKER.length
|
||||
}
|
||||
return { message: null, pendingLine: null }
|
||||
}
|
||||
|
||||
export function createCodexErrorOutputStatusDetector(args: {
|
||||
onStreamError: (message: string) => void
|
||||
}): CodexErrorOutputStatusDetector {
|
||||
let recentRawText = ''
|
||||
let markerCarry = ''
|
||||
let pendingLine: string | null = null
|
||||
|
||||
const reset = (): void => {
|
||||
markerCarry = ''
|
||||
pendingLine = null
|
||||
}
|
||||
|
||||
return {
|
||||
observe(data: string): boolean {
|
||||
const previousRawText = recentRawText
|
||||
recentRawText = appendRecentRawText(previousRawText, data)
|
||||
const scanRawText = buildStatusScanRawText(previousRawText, data)
|
||||
if (pendingLine !== null) {
|
||||
pendingLine = updatePendingLine(pendingLine, data)
|
||||
const lineEnd = findLineEnd(pendingLine, 0)
|
||||
if (lineEnd === -1 && pendingLine.length < MAX_PENDING_STREAM_ERROR_LINE_LENGTH) {
|
||||
markerCarry = appendMarkerCarry(markerCarry, data)
|
||||
return false
|
||||
}
|
||||
const message = normalizeStreamErrorLine(
|
||||
lineEnd === -1 ? pendingLine : pendingLine.slice(0, lineEnd)
|
||||
)
|
||||
pendingLine = null
|
||||
markerCarry = appendMarkerCarry(markerCarry, data)
|
||||
if (!message) {
|
||||
return false
|
||||
}
|
||||
args.onStreamError(message)
|
||||
return true
|
||||
}
|
||||
|
||||
if (!rawTextMayContainCodexStreamError(scanRawText)) {
|
||||
const seam = markerCarry + data.slice(0, CODEX_STREAM_DISCONNECTED_CARRY_LENGTH)
|
||||
if (
|
||||
!data.includes(CODEX_STREAM_DISCONNECTED_MARKER) &&
|
||||
!seam.includes(CODEX_STREAM_DISCONNECTED_MARKER)
|
||||
) {
|
||||
markerCarry = appendMarkerCarry(markerCarry, data)
|
||||
return false
|
||||
}
|
||||
|
||||
const scanText = stripTerminalControl(scanRawText)
|
||||
const previousTextLength = previousRawText ? stripTerminalControl(previousRawText).length : 0
|
||||
const match = findOverlappingCodexStreamError(scanText, previousTextLength)
|
||||
if (!match) {
|
||||
const rawText = data.includes(CODEX_STREAM_DISCONNECTED_MARKER) ? data : markerCarry + data
|
||||
const result = findCompletedStreamErrorLine(rawText)
|
||||
pendingLine = result.pendingLine
|
||||
markerCarry = appendMarkerCarry(markerCarry, data)
|
||||
if (!result.message) {
|
||||
return false
|
||||
}
|
||||
|
||||
const message = extractCodexStreamErrorMessage(scanText, match)
|
||||
if (!message) {
|
||||
return false
|
||||
}
|
||||
args.onStreamError(message)
|
||||
args.onStreamError(result.message)
|
||||
return true
|
||||
}
|
||||
},
|
||||
reset
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1537,6 +1537,121 @@ describe('connectPanePty', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('does not complete a Codex row from transient stream-disconnect retry output', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
|
||||
const transport = createMockTransport()
|
||||
transport.connect.mockImplementation(async ({ callbacks }: { callbacks: ConnectCallbacks }) => {
|
||||
capturedDataCallback.current = callbacks.onData ?? null
|
||||
return 'pty-codex'
|
||||
})
|
||||
transportFactoryQueue.push(transport)
|
||||
|
||||
const pane = createPane(1)
|
||||
const paneKey = makePaneKey('tab-1', LEAF_1)
|
||||
mockStoreState.agentStatusByPaneKey[paneKey] = {
|
||||
paneKey,
|
||||
state: 'working',
|
||||
prompt: 'Keep generating',
|
||||
agentType: 'codex',
|
||||
updatedAt: Date.now(),
|
||||
stateStartedAt: Date.now(),
|
||||
stateHistory: []
|
||||
}
|
||||
|
||||
connectPanePty(pane as never, createManager(1) as never, createDeps() as never)
|
||||
await flushAsyncTicks()
|
||||
expect(capturedDataCallback.current).not.toBeNull()
|
||||
|
||||
capturedDataCallback.current?.(
|
||||
'stream error: stream disconnected before completion: temporary network failure; retrying 1/5 in 217ms\r\n'
|
||||
)
|
||||
|
||||
expect(mockStoreState.setAgentStatus).not.toHaveBeenCalled()
|
||||
expect(mockStoreState.agentStatusByPaneKey[paneKey]).toMatchObject({
|
||||
state: 'working',
|
||||
prompt: 'Keep generating',
|
||||
agentType: 'codex'
|
||||
})
|
||||
})
|
||||
|
||||
it('does not complete a later Codex turn from a repainted stream-disconnect line', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
|
||||
const transport = createMockTransport()
|
||||
transport.connect.mockImplementation(async ({ callbacks }: { callbacks: ConnectCallbacks }) => {
|
||||
capturedDataCallback.current = callbacks.onData ?? null
|
||||
return 'pty-codex'
|
||||
})
|
||||
transportFactoryQueue.push(transport)
|
||||
|
||||
const pane = createPane(1)
|
||||
const paneKey = makePaneKey('tab-1', LEAF_1)
|
||||
mockStoreState.agentStatusByPaneKey[paneKey] = {
|
||||
paneKey,
|
||||
state: 'working',
|
||||
prompt: 'First turn',
|
||||
agentType: 'codex',
|
||||
updatedAt: Date.now(),
|
||||
stateStartedAt: Date.now(),
|
||||
stateHistory: []
|
||||
}
|
||||
|
||||
connectPanePty(pane as never, createManager(1) as never, createDeps() as never)
|
||||
await flushAsyncTicks()
|
||||
expect(capturedDataCallback.current).not.toBeNull()
|
||||
|
||||
const fatalLine =
|
||||
'■ stream disconnected before completion: error sending request for url (http://openclaw:2455/backend-api/codex/responses)\r\n'
|
||||
capturedDataCallback.current?.(fatalLine)
|
||||
expect(mockStoreState.agentStatusByPaneKey[paneKey]).toMatchObject({
|
||||
state: 'done',
|
||||
prompt: 'First turn',
|
||||
agentType: 'codex'
|
||||
})
|
||||
|
||||
mockStoreState.setAgentStatus.mockClear()
|
||||
mockStoreState.agentStatusByPaneKey[paneKey] = {
|
||||
paneKey,
|
||||
state: 'working',
|
||||
prompt: 'Second turn',
|
||||
agentType: 'codex',
|
||||
updatedAt: Date.now(),
|
||||
stateStartedAt: Date.now(),
|
||||
stateHistory: []
|
||||
}
|
||||
|
||||
capturedDataCallback.current?.(fatalLine)
|
||||
|
||||
expect(mockStoreState.setAgentStatus).not.toHaveBeenCalled()
|
||||
expect(mockStoreState.agentStatusByPaneKey[paneKey]).toMatchObject({
|
||||
state: 'working',
|
||||
prompt: 'Second turn',
|
||||
agentType: 'codex'
|
||||
})
|
||||
})
|
||||
|
||||
it('does not run stream-disconnect completion for idle panes with no working Codex row', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
|
||||
const transport = createMockTransport()
|
||||
transport.connect.mockImplementation(async ({ callbacks }: { callbacks: ConnectCallbacks }) => {
|
||||
capturedDataCallback.current = callbacks.onData ?? null
|
||||
return 'pty-codex'
|
||||
})
|
||||
transportFactoryQueue.push(transport)
|
||||
|
||||
connectPanePty(createPane(1) as never, createManager(1) as never, createDeps() as never)
|
||||
await flushAsyncTicks()
|
||||
expect(capturedDataCallback.current).not.toBeNull()
|
||||
|
||||
capturedDataCallback.current?.(
|
||||
'■ stream disconnected before completion: error sending request for url (http://openclaw:2455/backend-api/codex/responses)\r\n'
|
||||
)
|
||||
|
||||
expect(mockStoreState.setAgentStatus).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('delivers terminal-paste startup commands through xterm before submitting', async () => {
|
||||
const pendingTimeouts: (() => void)[] = []
|
||||
const originalSetTimeout = globalThis.setTimeout
|
||||
@@ -10268,6 +10383,116 @@ describe('connectPanePty', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('repairs a working Codex row from a title completion before notification dedupe', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const transport = createMockTransport('pty-codex')
|
||||
transportFactoryQueue.push(transport)
|
||||
|
||||
vi.useFakeTimers()
|
||||
const paneKey = makePaneKey('tab-1', LEAF_1)
|
||||
mockStoreState.agentStatusByPaneKey[paneKey] = {
|
||||
state: 'working',
|
||||
prompt: 'Handle fatal Codex errors',
|
||||
updatedAt: Date.now(),
|
||||
stateStartedAt: Date.now(),
|
||||
agentType: 'codex',
|
||||
paneKey,
|
||||
terminalTitle: 'Codex working',
|
||||
stateHistory: [],
|
||||
lastAssistantMessage: 'Usage limit reached.'
|
||||
}
|
||||
const pane = createPane(1)
|
||||
const manager = createManager(1)
|
||||
const deps = createDeps()
|
||||
|
||||
connectPanePty(pane as never, manager as never, deps as never)
|
||||
|
||||
const titleHandler = createdTransportOptions[0]?.onTitleChange as
|
||||
| ((title: string, rawTitle: string) => void)
|
||||
| undefined
|
||||
if (!titleHandler) {
|
||||
throw new Error('Expected onTitleChange to be registered')
|
||||
}
|
||||
|
||||
titleHandler('Codex working', 'Codex working')
|
||||
titleHandler('Codex done', 'Codex done')
|
||||
|
||||
expect(mockStoreState.agentStatusByPaneKey[paneKey]).toMatchObject({
|
||||
state: 'done',
|
||||
prompt: 'Handle fatal Codex errors',
|
||||
agentType: 'codex',
|
||||
lastAssistantMessage: 'Usage limit reached.'
|
||||
})
|
||||
vi.advanceTimersByTime(AGENT_TASK_COMPLETE_NOTIFICATION_GRACE_MS)
|
||||
|
||||
expect(deps.dispatchNotification).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
source: 'agent-task-complete',
|
||||
terminalTitle: 'Codex done',
|
||||
paneKey,
|
||||
agentStatusSnapshot: expect.objectContaining({
|
||||
state: 'done',
|
||||
prompt: 'Handle fatal Codex errors',
|
||||
agentType: 'codex',
|
||||
lastAssistantMessage: 'Usage limit reached.'
|
||||
})
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('repairs a working Codex row from the real process-exit backstop path', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const transport = createMockTransport('pty-codex')
|
||||
transportFactoryQueue.push(transport)
|
||||
|
||||
vi.useFakeTimers()
|
||||
vi.spyOn(Math, 'random').mockReturnValue(0.5)
|
||||
let foregroundProcess: string | null = 'codex'
|
||||
const api = (
|
||||
globalThis as unknown as {
|
||||
window: {
|
||||
api: {
|
||||
pty: {
|
||||
getForegroundProcess: ReturnType<typeof vi.fn>
|
||||
hasChildProcesses: ReturnType<typeof vi.fn>
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
).window.api
|
||||
api.pty.getForegroundProcess.mockImplementation(async () => foregroundProcess)
|
||||
api.pty.hasChildProcesses.mockImplementation(async () => foregroundProcess !== null)
|
||||
const paneKey = makePaneKey('tab-1', LEAF_1)
|
||||
mockStoreState.agentStatusByPaneKey[paneKey] = {
|
||||
state: 'working',
|
||||
prompt: 'Handle vanished Codex process',
|
||||
updatedAt: Date.now(),
|
||||
stateStartedAt: Date.now(),
|
||||
agentType: 'codex',
|
||||
paneKey,
|
||||
terminalTitle: 'Codex working',
|
||||
stateHistory: [],
|
||||
lastAssistantMessage: 'Connection failed.'
|
||||
}
|
||||
const pane = createPane(1)
|
||||
const manager = createManager(1)
|
||||
const deps = createDeps()
|
||||
|
||||
connectPanePty(pane as never, manager as never, deps as never)
|
||||
await flushAsyncTicks()
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
|
||||
foregroundProcess = null
|
||||
await vi.advanceTimersByTimeAsync(1_500)
|
||||
|
||||
expect(mockStoreState.agentStatusByPaneKey[paneKey]).toMatchObject({
|
||||
state: 'done',
|
||||
prompt: 'Handle vanished Codex process',
|
||||
agentType: 'codex',
|
||||
lastAssistantMessage: 'Connection failed.'
|
||||
})
|
||||
})
|
||||
|
||||
it('does not dispatch generic spinner completions when process inspection finds no agent', async () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const transport = createMockTransport('pty-shell')
|
||||
|
||||
@@ -104,7 +104,10 @@ import {
|
||||
shouldSuppressCodexAutoApprovalSyntheticTitle,
|
||||
shouldSuppressCodexAutoApprovalStatus
|
||||
} from './codex-auto-approval-notification-suppression'
|
||||
import type { AgentCompletionStatusSnapshot } from './agent-completion-coordinator-types'
|
||||
import type {
|
||||
AgentCompletionStatusRepairSignal,
|
||||
AgentCompletionStatusSnapshot
|
||||
} from './agent-completion-coordinator-types'
|
||||
import {
|
||||
markTerminalBracketedPasteInterrupted,
|
||||
observeTerminalBracketedPasteModeOutput
|
||||
@@ -164,10 +167,7 @@ import {
|
||||
beginAgentStartupDeliveryAttempt,
|
||||
releaseAgentStartupDeliveryAttempt
|
||||
} from '@/lib/agent-startup-delayed-delivery'
|
||||
import {
|
||||
isExpectedAgentProcess,
|
||||
type RecognizedAgentProcess
|
||||
} from '../../../../shared/agent-process-recognition'
|
||||
import { isExpectedAgentProcess } from '../../../../shared/agent-process-recognition'
|
||||
|
||||
const pendingSpawnByPaneKey = new Map<string, Promise<string | null>>()
|
||||
const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
|
||||
@@ -1477,21 +1477,24 @@ export function connectPanePty(
|
||||
delete pane.container.dataset.ptyId
|
||||
}
|
||||
|
||||
const completeWorkingStatusFromProcessExit = (
|
||||
agent: RecognizedAgentProcess
|
||||
const completeWorkingStatusFromCompletionSignal = (
|
||||
signal: AgentCompletionStatusRepairSignal
|
||||
): AgentCompletionStatusSnapshot | null => {
|
||||
const currentState = useAppStore.getState()
|
||||
const currentEntry = currentState.agentStatusByPaneKey[cacheKey]
|
||||
if (!currentEntry || currentEntry.state !== 'working') {
|
||||
return null
|
||||
}
|
||||
const ownerAgent = getAuthoritativePaneAgent() ?? agent.agent
|
||||
const signalAgent = signal.source === 'process-exit' ? signal.agent.agent : signal.agentType
|
||||
const ownerAgent = getAuthoritativePaneAgent() ?? signalAgent ?? currentEntry.agentType
|
||||
const agentType = resolveCompatibleAgentTypeForOwner(
|
||||
currentEntry.agentType ?? agent.agent,
|
||||
currentEntry.agentType ?? signalAgent,
|
||||
ownerAgent
|
||||
)
|
||||
const processAgentType = resolveCompatibleAgentTypeForOwner(agent.agent, ownerAgent)
|
||||
if (!agentType || agentType !== processAgentType) {
|
||||
const signalAgentType = signalAgent
|
||||
? resolveCompatibleAgentTypeForOwner(signalAgent, ownerAgent)
|
||||
: undefined
|
||||
if (!agentType || (signalAgentType && agentType !== signalAgentType)) {
|
||||
return null
|
||||
}
|
||||
const statusPayload = {
|
||||
@@ -1505,8 +1508,8 @@ export function connectPanePty(
|
||||
const currentTitle =
|
||||
currentState.runtimePaneTitlesByTabId?.[deps.tabId]?.[pane.id] ?? currentEntry.terminalTitle
|
||||
const statusTitle = resolveAgentStatusTerminalTitle(statusPayload, currentTitle)
|
||||
// Why: the foreground-process backstop can prove the TUI disappeared even
|
||||
// when the agent missed its final hook; keep the explicit row in sync.
|
||||
// Why: title/process completion can prove a turn ended even when the agent
|
||||
// missed its final hook; keep the explicit row in sync before dedupe gates.
|
||||
currentState.setAgentStatus(cacheKey, statusPayload, statusTitle)
|
||||
const storedStatus = useAppStore.getState().agentStatusByPaneKey[cacheKey]
|
||||
return typeof storedStatus?.stateStartedAt === 'number'
|
||||
@@ -1521,7 +1524,7 @@ export function connectPanePty(
|
||||
inspectProcess: inspectRuntimeTerminalProcess,
|
||||
dispatchCompletion: (title, meta) =>
|
||||
scheduleAgentTaskCompleteNotification(title, {
|
||||
allowDoneDetailAfterGrace: meta?.quietedHookDone,
|
||||
allowDoneDetailAfterGrace: meta?.quietedHookDone === true || Boolean(meta?.agentStatus),
|
||||
...(meta?.agentStatus ? { agentStatusSnapshot: meta.agentStatus } : {})
|
||||
}),
|
||||
dispatchAttention: (title, meta) =>
|
||||
@@ -1530,7 +1533,7 @@ export function connectPanePty(
|
||||
}),
|
||||
shouldPollProcessCadence: () =>
|
||||
isAgentTaskCompleteTrackingEnabled() && deps.isVisibleRef.current,
|
||||
onProcessExitCompletion: completeWorkingStatusFromProcessExit,
|
||||
onCompletionStatusRepair: completeWorkingStatusFromCompletionSignal,
|
||||
isLive: () => {
|
||||
if (disposed) {
|
||||
return false
|
||||
@@ -1839,9 +1842,22 @@ export function connectPanePty(
|
||||
// Stop hook Orca normally uses to clear an explicit working row.
|
||||
currentState.setAgentStatus(cacheKey, statusPayload, statusTitle)
|
||||
}
|
||||
const shouldObserveCodexStreamErrorOutput = (): boolean => {
|
||||
const currentEntry = useAppStore.getState().agentStatusByPaneKey[cacheKey]
|
||||
if (!currentEntry || currentEntry.state !== 'working') {
|
||||
return false
|
||||
}
|
||||
const agentType = resolveCompatibleAgentTypeForOwner(
|
||||
currentEntry.agentType ?? getAuthoritativePaneAgent(),
|
||||
getAuthoritativePaneAgent()
|
||||
)
|
||||
return agentType === 'codex'
|
||||
}
|
||||
const codexErrorOutputStatusDetector = createCodexErrorOutputStatusDetector({
|
||||
onStreamError: completeCodexOutputStreamErrorStatus
|
||||
})
|
||||
let codexStreamErrorOutputDetectionArmed = false
|
||||
let codexStreamErrorOutputFastPathConsumed = false
|
||||
const observeTerminalGitHubPRLink = createTerminalGitHubPRLinkDetector()
|
||||
const reportPanePtyVisibility = (ptyId: string | null | undefined, visible: boolean): void => {
|
||||
if (!ptyId || isRemoteRuntimePtyId(ptyId)) {
|
||||
@@ -1858,6 +1874,11 @@ export function connectPanePty(
|
||||
if (activePanePtyBinding && activePanePtyBinding !== ptyId) {
|
||||
reportPanePtyVisibility(activePanePtyBinding, false)
|
||||
}
|
||||
if (activePanePtyBinding !== ptyId) {
|
||||
codexStreamErrorOutputDetectionArmed = false
|
||||
codexStreamErrorOutputFastPathConsumed = false
|
||||
codexErrorOutputStatusDetector.reset()
|
||||
}
|
||||
setPanePtyFitBinding(ptyId)
|
||||
activePanePtyBinding = ptyId
|
||||
reportPanePtyVisibility(ptyId, deps.isVisibleRef.current)
|
||||
@@ -4476,7 +4497,17 @@ export function connectPanePty(
|
||||
for (const link of observeTerminalGitHubPRLink(data)) {
|
||||
useAppStore.getState().observeTerminalGitHubPullRequestLink(deps.worktreeId, link)
|
||||
}
|
||||
codexErrorOutputStatusDetector.observe(data)
|
||||
if (shouldObserveCodexStreamErrorOutput() && !codexStreamErrorOutputFastPathConsumed) {
|
||||
codexStreamErrorOutputDetectionArmed = true
|
||||
if (codexErrorOutputStatusDetector.observe(data)) {
|
||||
codexStreamErrorOutputDetectionArmed = false
|
||||
codexStreamErrorOutputFastPathConsumed = true
|
||||
codexErrorOutputStatusDetector.reset()
|
||||
}
|
||||
} else if (codexStreamErrorOutputDetectionArmed) {
|
||||
codexStreamErrorOutputDetectionArmed = false
|
||||
codexErrorOutputStatusDetector.reset()
|
||||
}
|
||||
commandCodeOutputStatusDetector.observe(data)
|
||||
commandLifecycle.handlePtyData(data)
|
||||
// Why: split-pane layouts have multiple visible-but-inactive panes whose
|
||||
|
||||
Reference in New Issue
Block a user