diff --git a/src/sdk/output-classifier.ts b/src/sdk/output-classifier.ts index 4c0bd8882..5f8f0d432 100644 --- a/src/sdk/output-classifier.ts +++ b/src/sdk/output-classifier.ts @@ -167,3 +167,179 @@ export function isAuthFailureObserverOutput(raw: unknown): boolean { /\/login\b.{0,40}\b(?:to\s+authenticate|again|to\s+continue|and\s+retry|reauthenticate|credentials|provider|claude)\b/.test(text) ); } + +/** + * Detect the spawned CLI's own transport/API failure returned as the response + * body. + * + * When the child cannot reach the provider it does not crash — it prints its + * error and exits 0, so the text arrives here looking like any other non-XML + * output. Without a class of its own it falls to the generic prose branch, + * which confirms (drops) the claimed batch: a transient network fault becomes + * permanent data loss (#3752). + * + * Kept deliberately tight. A false positive requeues work and pauses the + * generator, so the patterns anchor on the shapes the CLI actually emits — + * the error as the whole response — rather than on any mention of a network + * word, which an observer narrating a connectivity bug would trip. + */ +/** + * Conditions that make a failure a *transport* failure — the request never got + * an answer — as opposed to one the server understood and refused. Only the + * former is worth requeuing: a refusal fails identically on every retry, so + * treating it as transport turns one bad batch into an endless one. + */ +const NETWORK_CONDITION = + /\b(?:connect|connection|network|socket|dns|proxy|tls|ssl|certificate|unreachable|econnrefused|econnreset|etimedout|enotfound|enetunreach|ehostunreach|epipe|econnaborted|eai_again|eproto)\b|\bfetch failed\b|\bsocket hang up\b/; + +/** + * An ` error` prefix only counts when the response is *reporting* the + * error rather than talking about one. A report either stops at the envelope or + * introduces its detail with punctuation; prose runs straight on into a + * sentence ("Connection error handling was reviewed"). Punctuation is a + * necessary condition, not a sufficient one — see + * selfDescribingEnvelopeReportsAFailure for the half it does not carry. + * + * A trailing full stop is deliberately NOT a delimiter here. "Network error." + * therefore goes undetected, which is the safe direction: a miss leaves today's + * behaviour in place, while a false positive requeues the batch and pauses the + * generator — a retry loop over work that already completed. + */ +const ENVELOPE_IS_A_REPORT = { + fetchNetworkConnection: /^(?:fetch|network|connection)\s*error\b\s*(?::|-|–|—|$)\s*/, + apiHttpRequest: /^(?:api|http|request)\s*error\b\s*(?::|-|–|—|$)\s*/, + bareError: /^error\s*:\s*/, +}; + +/** + * A verb of being is what separates a sentence *about* an error from the error + * itself. What a CLI puts after its envelope is a fragment naming a condition — + * "upstream closed", "peer reset the stream", "ECONNRESET". What an observer + * puts there predicates about one — "recovery is already covered by the retry + * wrapper". + */ +const DETAIL_IS_A_CLAUSE = /\b(?:is|are|was|were|be|been|being|has|have|had)\b/; + +/** + * The words that only a failure uses. Deliberately narrower than + * NETWORK_CONDITION, whose generic nouns — connect, connection, network, + * socket, proxy — are exactly what prose about networking code is full of, so + * they cannot re-admit a clause this detector has already judged to be prose. + */ +const CONCRETE_FAILURE = + /\b(?:econnrefused|econnreset|etimedout|enotfound|enetunreach|ehostunreach|epipe|econnaborted|eai_again|eproto|unreachable|refused|timed out)\b|\bfetch failed\b|\bsocket hang up\b|\breset by peer\b/; + +/** + * Whether an envelope at the head of `text` is *reporting* a failure. + * + * Every envelope family goes through this, because the hole is the same in all + * of them: punctuation says the response is shaped like a report, and nothing + * says it is one. `Network error: recovery is already covered by the retry + * wrapper` and `API Error: connection handling was reviewed` are both completed + * observations that satisfy the shape. + * + * What separates them from a real failure is grammar. A CLI's detail is a + * fragment naming a condition — "upstream closed", "TLS handshake failed", + * "ECONNRESET". An observer's is a clause predicating about one — "recovery + * **is** already covered". A clause is admitted only when it names a concrete + * failure, which is why CONCRETE_FAILURE excludes the generic nouns the prose + * itself is built from — `connection`, `socket`, and the bare noun `timeout`, + * which a sentence about raising one uses exactly as often as a failure does. + * + * `requireCondition` is the difference between the families: `api|http|request` + * is an envelope that fronts a dead socket and a refused request identically, + * so it needs a network condition before it counts at all, while `fetch|network + * |connection` names the condition itself and needs none. A genuine 5xx behind + * an API envelope does not depend on this path — the status patterns below + * catch it whatever its prose looks like. + * + * This errs towards missing: "Network error: the endpoint is behind a firewall" + * is a real failure it rejects. That is the direction the whole detector errs + * in, and the cheap one — a miss leaves today's behaviour in place, while a + * false positive requeues completed work and pauses the generator. + */ +function envelopeReportsAFailure( + envelope: RegExp, + text: string, + requireCondition: boolean +): boolean { + const match = envelope.exec(text); + if (match === null) { + return false; + } + + const detail = text.slice(match[0].length).trim(); + + if (requireCondition && !NETWORK_CONDITION.test(detail)) { + return false; + } + + if (detail === '' || !DETAIL_IS_A_CLAUSE.test(detail)) { + return true; + } + + return CONCRETE_FAILURE.test(detail); +} + +export function isTransportFailureObserverOutput(raw: unknown): boolean { + if (typeof raw !== 'string' || raw.trim() === '') { + return false; + } + + if (/<(observation|summary)\b/i.test(raw) || / buffer.dispose), // wiping the very queue/conversationHistory the switch is meant to preserve. const abortCategory = (reason ?? '').split(':')[0]; - if (abortCategory === 'quota' || abortCategory === 'auth' || abortCategory === 'overflow' || abortCategory === 'provider_switch') { + // Every category listed here has ALREADY called resetProcessingToPending + // (except provider_switch, which parks a live buffer for a provider change). + // Falling through to finalizeSession would remove the session and undo that + // preservation — the second half of #3752. + const PRESERVES_CLAIMED_WORK = ['quota', 'auth', 'overflow', 'provider_switch', 'transport']; + if (PRESERVES_CLAIMED_WORK.includes(abortCategory)) { logger.warn('SESSION', `Generator paused for ${abortCategory}; preserving buffered work`, { sessionId: sessionDbId, pendingCount: sessionManager.getMessageBuffer().getPendingCount(sessionDbId), diff --git a/tests/sdk/output-classifier.test.ts b/tests/sdk/output-classifier.test.ts index be55c92b1..3ac588bf8 100644 --- a/tests/sdk/output-classifier.test.ts +++ b/tests/sdk/output-classifier.test.ts @@ -4,6 +4,7 @@ import { isAuthFailureObserverOutput, isContextOverflowObserverOutput, isQuotaLimitedObserverOutput, + isTransportFailureObserverOutput, previewOutput, } from '../../src/sdk/output-classifier.js'; @@ -242,3 +243,208 @@ describe('isAuthFailureObserverOutput recognises the CLI signed-out wording (#36 expect(isAuthFailureObserverOutput('Not logged in')).toBe(false); }); }); + +describe('isTransportFailureObserverOutput (#3752)', () => { + it('classifies the CLI error the child returns instead of crashing', () => { + // Verbatim from the issue report. + expect(isTransportFailureObserverOutput( + 'API Error: Connection refused - a firewall or proxy may be blocking it (ConnectionRefused)' + )).toBe(true); + }); + + it('classifies the common transport surfaces', () => { + expect(isTransportFailureObserverOutput('connect ECONNREFUSED 127.0.0.1:443')).toBe(true); + expect(isTransportFailureObserverOutput('getaddrinfo ENOTFOUND api.anthropic.com')).toBe(true); + expect(isTransportFailureObserverOutput('fetch failed')).toBe(true); + expect(isTransportFailureObserverOutput('Error: socket hang up ECONNRESET')).toBe(true); + expect(isTransportFailureObserverOutput('API Error: 503 Service Unavailable')).toBe(true); + expect(isTransportFailureObserverOutput('Request failed with 502')).toBe(true); + }); + + // A false positive requeues the batch AND pauses the generator, so the + // detector has to stay off observer narrative that merely discusses network + // failure — exactly the kind of thing this project's own observations say. + it('does not classify observer prose that merely talks about connectivity', () => { + expect(isTransportFailureObserverOutput( + 'Traced the flake to a proxy that drops idle sockets; the retry now handles the connection reset.' + )).toBe(false); + expect(isTransportFailureObserverOutput('No observations to record.')).toBe(false); + expect(isTransportFailureObserverOutput('')).toBe(false); + expect(isTransportFailureObserverOutput(null)).toBe(false); + }); + + it('does not steal XML output', () => { + expect(isTransportFailureObserverOutput( + 'fetch failed on cold start' + )).toBe(false); + }); + + // 401/403 must keep reaching the auth branch, which gives the user a /login + // remediation instead of retrying against a provider that will keep refusing. + it('leaves authentication failures to the auth detector', () => { + expect(isTransportFailureObserverOutput('API Error: 401 Unauthorized')).toBe(false); + expect(isTransportFailureObserverOutput('Authentication failed. Please run /login to authenticate.')).toBe(false); + }); +}); + + +describe('isTransportFailureObserverOutput rejects narrative (review on #3752)', () => { + // The three cases the reviewer reproduced. Each contains a transport term + // that an unanchored search matched, and each is an ordinary completed + // observation — confirming it is correct, requeueing it is a retry loop. + const NARRATIVES = [ + 'The observer noted that fetch failed during the previous deploy and the rollback restored service.', + 'The observer documented ECONNRESET on the idle pool and the fix that followed.', + 'The observer recorded that the upstream returned HTTP 503 for four minutes.', + ]; + + for (const prose of NARRATIVES) { + it(`does not classify: ${prose.slice(0, 44)}…`, () => { + expect(isTransportFailureObserverOutput(prose)).toBe(false); + }); + } + + it('still classifies the same terms when they lead the response', () => { + expect(isTransportFailureObserverOutput('fetch failed')).toBe(true); + expect(isTransportFailureObserverOutput('ECONNRESET')).toBe(true); + expect(isTransportFailureObserverOutput('HTTP 503 Service Unavailable')).toBe(true); + expect(isTransportFailureObserverOutput('503 Service Unavailable')).toBe(true); + expect(isTransportFailureObserverOutput('socket hang up')).toBe(true); + }); + + it('does not classify a narrative that merely opens with the word error', () => { + expect(isTransportFailureObserverOutput( + 'Error handling in the fetch layer was reviewed; no changes were needed.' + )).toBe(false); + }); +}); + +describe('isTransportFailureObserverOutput separates envelope from diagnosis (review on #3752)', () => { + // `API Error:` is what the CLI prefixes to everything it failed at. It fronts + // a dead socket and a rejected request the same way, and the two want + // opposite handling: the transport branch resets the claimed batch to pending + // and aborts the session for a later retry, which for a request the server + // understood and refused is a loop that never terminates. + const PERSISTENT = [ + 'API Error: 400 Bad Request', + 'API Error: invalid model', + 'API Error: model not found: claude-nonexistent', + 'HTTP error: 404 Not Found', + 'Request error: unsupported parameter "max_tokens"', + 'API Error: 422 Unprocessable Entity', + ]; + + for (const output of PERSISTENT) { + it(`does not requeue a persistent failure: ${output}`, () => { + expect(isTransportFailureObserverOutput(output)).toBe(false); + }); + } + + // The same envelope, now naming a condition that means the request never got + // an answer. These must still requeue. + const TRANSPORT = [ + 'API Error: Connection refused - a firewall or proxy may be blocking it (ConnectionRefused)', + 'API Error: fetch failed', + 'API Error: socket hang up', + 'HTTP error: ETIMEDOUT', + 'Request error: getaddrinfo ENOTFOUND api.anthropic.com', + 'API Error: TLS handshake failed', + 'API Error: 503 Service Unavailable', + 'Request error: 502', + ]; + + for (const output of TRANSPORT) { + it(`still requeues: ${output}`, () => { + expect(isTransportFailureObserverOutput(output)).toBe(true); + }); + } + + // The nouns that ARE the condition need no separate diagnosis — but they do + // have to be reporting an error rather than naming one. + it('keeps the self-describing envelopes free of a condition requirement', () => { + expect(isTransportFailureObserverOutput('Fetch error: upstream closed')).toBe(true); + expect(isTransportFailureObserverOutput('Network error')).toBe(true); + expect(isTransportFailureObserverOutput('Connection error - ECONNRESET')).toBe(true); + expect(isTransportFailureObserverOutput('Connection error: peer reset the stream')).toBe(true); + }); + + // Anchoring does not save the self-describing envelopes: this prose STARTS + // with the token. Reported on the first round of this branch, and the whole + // reason the report test exists. + const ENVELOPE_PROSE = [ + 'Connection error handling was reviewed; no changes were needed.', + 'Network error recovery is already covered by the retry wrapper.', + 'Fetch error paths were consolidated into one helper.', + 'API error handling for connection resets was reviewed.', + 'Request error messages now include the socket address.', + ]; + + for (const prose of ENVELOPE_PROSE) { + it(`does not requeue completed prose: ${prose.slice(0, 40)}…`, () => { + expect(isTransportFailureObserverOutput(prose)).toBe(false); + }); + } + + // Punctuation does not make an envelope a report — the narrative can be + // punctuated too. Reported on this branch after the rebase; the unpunctuated + // list above was the whole test, so a colon walked straight through it. + const PUNCTUATED_ENVELOPE_PROSE = [ + 'Network error: recovery is already covered by the retry wrapper', + 'Connection error: handling was reviewed, and no changes were needed', + 'Fetch error: paths were consolidated into one helper', + 'Network error: connection handling is being refactored this sprint', + ]; + + for (const prose of PUNCTUATED_ENVELOPE_PROSE) { + it(`does not requeue punctuated prose: ${prose.slice(0, 40)}…`, () => { + expect(isTransportFailureObserverOutput(prose)).toBe(false); + }); + } + + // The API and bare-`Error:` families had the same hole, and a network + // condition did not close it: the generic noun the prose is *about* is the + // same noun the condition list is made of. + const PUNCTUATED_ENVELOPE_PROSE_WITH_CONDITION = [ + 'API Error: connection handling was reviewed', + 'HTTP error: the socket timeout has been raised to 30s', + 'Request error: proxy support was added to the client', + 'Error: connection pooling is now handled by the driver', + ]; + + for (const prose of PUNCTUATED_ENVELOPE_PROSE_WITH_CONDITION) { + it(`does not requeue an explained condition: ${prose.slice(0, 40)}…`, () => { + expect(isTransportFailureObserverOutput(prose)).toBe(false); + }); + } + + // The 5xx path is independent of all of this, which is what makes the clause + // test affordable on the API family: a real server failure is still caught + // however its prose reads. + it('still reports a 5xx behind an API envelope, clause or not', () => { + expect( + isTransportFailureObserverOutput('API Error: 503 Service Unavailable, the gateway is down') + ).toBe(true); + expect(isTransportFailureObserverOutput('Request error: 502')).toBe(true); + }); + + // …and the clause test must not swallow a real failure that happens to be + // written as one. A concrete failure word re-admits it; a generic network + // noun deliberately does not, since that is what the prose above is made of. + it('still reports a clause that names a concrete failure', () => { + expect( + isTransportFailureObserverOutput('Connection error: the connection was reset by peer') + ).toBe(true); + expect( + isTransportFailureObserverOutput('Network error: the provider is unreachable') + ).toBe(true); + expect( + isTransportFailureObserverOutput('Fetch error: the request has timed out') + ).toBe(true); + }); + + // A 4xx must not slip through on the strength of an unrelated number. + it('does not treat a 4xx as retryable because a 5xx-shaped number appears later', () => { + expect(isTransportFailureObserverOutput('API Error: 400 Bad Request')).toBe(false); + expect(isTransportFailureObserverOutput('API Error: 429 Too Many Requests')).toBe(false); + }); +}); diff --git a/tests/worker/agents/response-processor.test.ts b/tests/worker/agents/response-processor.test.ts index 22b02aeed..5480f45db 100644 --- a/tests/worker/agents/response-processor.test.ts +++ b/tests/worker/agents/response-processor.test.ts @@ -591,8 +591,134 @@ describe('ResponseProcessor', () => { expect(session.earliestPendingTimestamp).toBeNull(); expect(mockStoreObservations).not.toHaveBeenCalled(); }); + + // #3752: when the spawned CLI cannot reach the provider it returns its own + // error string rather than crashing, so the text lands in this same branch. + // Confirming it drops the claimed batch permanently — the reporter lost 882 + // observations over six days that way. + it('requeues the claimed batch when the response is the child transport failure', async () => { + const confirmClaimedMessages = mock(() => Promise.resolve(0)); + const resetProcessingToPending = mock(() => Promise.resolve(0)); + mockSessionManager = { + getMessageIterator: async function* () { yield* []; }, + getPendingMessageStore: () => ({ confirmProcessed: mock(() => {}) }), + confirmClaimedMessages, + resetProcessingToPending, + } as unknown as SessionManager; + + const session = createMockSession(); + const responseText = + 'API Error: Connection refused - a firewall or proxy may be blocking it (ConnectionRefused)'; + + await processAgentResponse( + responseText, + session, + mockDbManager, + mockSessionManager, + mockWorker, + 100, + null, + 'TestAgent' + ); + + expect(resetProcessingToPending).toHaveBeenCalledWith(1); + // The whole point: the batch must NOT be confirmed away. + expect(confirmClaimedMessages).not.toHaveBeenCalled(); + expect(mockStoreObservations).not.toHaveBeenCalled(); + }); + + it('pauses the generator with a preserving abort reason on transport failure', async () => { + mockSessionManager = { + getMessageIterator: async function* () { yield* []; }, + getPendingMessageStore: () => ({ confirmProcessed: mock(() => {}) }), + confirmClaimedMessages: mock(() => Promise.resolve(0)), + resetProcessingToPending: mock(() => Promise.resolve(0)), + } as unknown as SessionManager; + + const session = createMockSession(); + + await processAgentResponse( + 'getaddrinfo ENOTFOUND api.anthropic.com', + session, + mockDbManager, + mockSessionManager, + mockWorker, + 100, + null, + 'TestAgent' + ); + + // handleGeneratorExit keys off the category before the colon; 'transport' + // is what keeps it from finalizing the session and undoing the requeue. + expect(session.abortReason).toBe('transport:observer_text'); + expect(session.abortController.signal.aborted).toBe(true); + expect(logger.error).toHaveBeenCalledWith( + 'PARSER', + expect.stringMatching(/could not reach the provider/), + expect.objectContaining({ sessionId: 1, outputClass: 'transport' }) + ); + }); + + it('still confirms ordinary prose so low-signal batches do not loop', async () => { + const confirmClaimedMessages = mock(() => Promise.resolve(0)); + const resetProcessingToPending = mock(() => Promise.resolve(0)); + mockSessionManager = { + getMessageIterator: async function* () { yield* []; }, + getPendingMessageStore: () => ({ confirmProcessed: mock(() => {}) }), + confirmClaimedMessages, + resetProcessingToPending, + } as unknown as SessionManager; + + const session = createMockSession(); + + await processAgentResponse( + 'Traced the flake to a proxy that drops idle sockets; the retry now handles the connection reset.', + session, + mockDbManager, + mockSessionManager, + mockWorker, + 100, + null, + 'TestAgent' + ); + + expect(confirmClaimedMessages).toHaveBeenCalledWith(1); + expect(resetProcessingToPending).not.toHaveBeenCalled(); + }); + + // The same guarantee one punctuation mark over: a completed observation + // that opens with an envelope AND a colon must still be confirmed, or the + // session pauses and retries work that already finished. + it('still confirms a punctuated envelope narrative', async () => { + const confirmClaimedMessages = mock(() => Promise.resolve(0)); + const resetProcessingToPending = mock(() => Promise.resolve(0)); + mockSessionManager = { + getMessageIterator: async function* () { yield* []; }, + getPendingMessageStore: () => ({ confirmProcessed: mock(() => {}) }), + confirmClaimedMessages, + resetProcessingToPending, + } as unknown as SessionManager; + + const session = createMockSession(); + + await processAgentResponse( + 'Network error: recovery is already covered by the retry wrapper', + session, + mockDbManager, + mockSessionManager, + mockWorker, + 100, + null, + 'TestAgent' + ); + + expect(confirmClaimedMessages).toHaveBeenCalledWith(1); + expect(resetProcessingToPending).not.toHaveBeenCalled(); + expect(session.abortController.signal.aborted).toBe(false); + }); }); + describe('context-window overflow recovery (#3800)', () => { function overflowSessionManager() { const resetProcessingToPending = mock(() => Promise.resolve(1)); diff --git a/tests/worker/generator-exit-preserves-work.test.ts b/tests/worker/generator-exit-preserves-work.test.ts new file mode 100644 index 000000000..8927b1d24 --- /dev/null +++ b/tests/worker/generator-exit-preserves-work.test.ts @@ -0,0 +1,58 @@ +import { describe, it, expect, mock } from 'bun:test'; + +import { handleGeneratorExit } from '../../src/services/worker/session/GeneratorExitHandler.js'; +import type { ActiveSession } from '../../src/services/worker-types.js'; +import type { SessionManager } from '../../src/services/worker/SessionManager.js'; +import type { SessionCompletionHandler } from '../../src/services/worker/session/SessionCompletionHandler.js'; + +/** + * #3752, second half. + * + * Resetting the claimed batch back to pending only survives if the generator + * exit that follows does NOT finalize the session — finalizeSession removes it + * and the preservation is undone. `quota` and `auth` were already on that list; + * `transport` had to join them, and nothing else in the codebase would have + * caught its absence. + */ +function buildSession(abortReason: string): ActiveSession { + return { + sessionDbId: 7, + abortReason, + abortController: new AbortController(), + generatorPromise: Promise.resolve(), + currentProvider: 'claude', + } as unknown as ActiveSession; +} + +function buildDeps() { + const finalizeSession = mock(() => Promise.resolve()); + const removeSessionImmediate = mock(() => {}); + const sessionManager = { + getMessageBuffer: () => ({ getPendingCount: () => 3 }), + removeSessionImmediate, + } as unknown as SessionManager; + const completionHandler = { finalizeSession } as unknown as SessionCompletionHandler; + return { deps: { sessionManager, completionHandler }, finalizeSession, removeSessionImmediate }; +} + +describe('handleGeneratorExit — reasons that preserve claimed work', () => { + for (const reason of ['transport:observer_text', 'quota:observer_text', 'auth:observer_text', 'overflow:observer_text', 'provider_switch']) { + it(`leaves the session alive for ${reason}`, async () => { + const { deps, finalizeSession, removeSessionImmediate } = buildDeps(); + + await handleGeneratorExit(buildSession(reason), reason, deps); + + expect(finalizeSession).not.toHaveBeenCalled(); + expect(removeSessionImmediate).not.toHaveBeenCalled(); + }); + } + + it('still finalizes on an ordinary idle exit', async () => { + const { deps, finalizeSession, removeSessionImmediate } = buildDeps(); + + await handleGeneratorExit(buildSession('idle'), 'idle', deps); + + expect(finalizeSession).toHaveBeenCalledWith(7); + expect(removeSessionImmediate).toHaveBeenCalledWith(7); + }); +});