mirror of
https://github.com/thedotmack/claude-mem.git
synced 2026-09-20 04:23:02 +08:00
fix(observer): requeue the batch when the child reports a transport failure (#3998)
When the spawned CLI cannot reach the provider it returns its own error string and exits 0. That used to classify as prose and confirm (drop) the claimed batch. Detect the CLI's transport report, reset the batch to pending, and keep the session alive on a transport abort so the work survives — the complete fix for #3752. Rebased onto main: provider_switch stays on the preserve list, and transport errors are not appended to conversation history. Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Alex Newman <thedotmack@users.noreply.github.com>
This commit is contained in:
Alex Newman
@@ -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 `<envelope> 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) || /<skip_summary\b/i.test(raw)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 401/403 is a credential problem, not a transport one. Leave it to the auth
|
||||
// detector so the user still gets the /login remediation instead of a silent
|
||||
// retry against a provider that will keep refusing.
|
||||
if (isAuthFailureObserverOutput(raw)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const text = raw.toLowerCase().replace(/\s+/g, ' ').trim();
|
||||
|
||||
// Every pattern below is anchored, without exception. The child hands its
|
||||
// failure back as the WHOLE response, so an error shape at the start is what
|
||||
// separates it from an observer narrating a past incident — and a narrative
|
||||
// is exactly what an unanchored token search catches. "The observer noted
|
||||
// that fetch failed during the previous deploy" is a completed no-op batch,
|
||||
// not a dead network.
|
||||
//
|
||||
// Anchoring can under-detect, if a future CLI prefixes its error with a
|
||||
// timestamp or a log level. That is the safe direction to be wrong in: a
|
||||
// missed detection is the behaviour that already exists today, while a false
|
||||
// positive requeues the batch AND pauses the generator, which is a retry loop.
|
||||
return (
|
||||
// "Fetch error …", "Network error …", "Connection error …" — the noun is
|
||||
// itself the condition, so no separate diagnosis is needed. The envelope
|
||||
// must still be REPORTING an error rather than naming one: an error report
|
||||
// ends there or introduces its detail with punctuation, while prose
|
||||
// continues into a sentence. "Connection error handling was reviewed" is a
|
||||
// completed observation, and anchoring alone does not catch it — the
|
||||
// narrative starts with the token.
|
||||
envelopeReportsAFailure(ENVELOPE_IS_A_REPORT.fetchNetworkConnection, text, false) ||
|
||||
// "API Error", "HTTP error" and "Request error" are envelopes, not
|
||||
// diagnoses: they front a dead socket and a refused request identically,
|
||||
// and the two want opposite handling. Requeuing a 400 or an unknown model
|
||||
// is a loop — it will fail the same way on every retry — so the envelope
|
||||
// only counts once the message also names a network condition. A genuine
|
||||
// 5xx still arrives, through the status patterns below.
|
||||
// The report test applies here too. Without it, "API error handling for
|
||||
// connection resets was reviewed" satisfies both halves — it opens with the
|
||||
// envelope and mentions a condition — while being ordinary prose.
|
||||
envelopeReportsAFailure(ENVELOPE_IS_A_REPORT.apiHttpRequest, text, true) ||
|
||||
// "Error: socket hang up" — but not "Error: no such file or directory".
|
||||
envelopeReportsAFailure(ENVELOPE_IS_A_REPORT.bareError, text, true) ||
|
||||
// Node's own shapes: "connect ECONNREFUSED 127.0.0.1:443".
|
||||
/^(?:connect|getaddrinfo|read|write|socket)\b.*\b(?:econnrefused|econnreset|etimedout|enotfound|enetunreach|ehostunreach|epipe|econnaborted|eai_again|eproto)\b/.test(text) ||
|
||||
// The bare code, or the bare phrase, as the entire message.
|
||||
/^(?:econnrefused|econnreset|etimedout|enotfound|enetunreach|ehostunreach|epipe|econnaborted|eai_again|eproto)\b/.test(text) ||
|
||||
/^(?:fetch failed|socket hang up|connectionrefused)\b/.test(text) ||
|
||||
// A 5xx reported as the response itself.
|
||||
/^(?:api|http|request)\s*(?:error\s*)?:?\s*5\d{2}\b/.test(text) ||
|
||||
/^request failed with\s+5\d{2}\b/.test(text) ||
|
||||
/^status\s*[:=]?\s*5\d{2}\b/.test(text) ||
|
||||
/^5\d{2}\s+(?:internal server error|bad gateway|service unavailable|gateway timeout)\b/.test(text)
|
||||
);
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
isAuthFailureObserverOutput,
|
||||
isContextOverflowObserverOutput,
|
||||
isQuotaLimitedObserverOutput,
|
||||
isTransportFailureObserverOutput,
|
||||
previewOutput,
|
||||
} from '../../../sdk/output-classifier.js';
|
||||
import { updateCursorContextForProject } from '../../integrations/CursorHooksInstaller.js';
|
||||
@@ -303,7 +304,8 @@ export async function processAgentResponse(
|
||||
!!text &&
|
||||
(isContextOverflowObserverOutput(text) ||
|
||||
isQuotaLimitedObserverOutput(text) ||
|
||||
isAuthFailureObserverOutput(text));
|
||||
isAuthFailureObserverOutput(text) ||
|
||||
isTransportFailureObserverOutput(text));
|
||||
|
||||
if (text && !isRejectionProse) {
|
||||
session.conversationHistory.push({ role: 'assistant', content: text });
|
||||
@@ -373,6 +375,30 @@ export async function processAgentResponse(
|
||||
return;
|
||||
}
|
||||
|
||||
// A response that is the child's OWN transport/API failure is not the
|
||||
// observer declining to say anything — it is the observer never having
|
||||
// run. Preserve the batch like the quota and auth cases above; confirming
|
||||
// it here would turn a transient network fault into permanent data loss
|
||||
// (#3752).
|
||||
if (isTransportFailureObserverOutput(text)) {
|
||||
session.consecutiveInvalidOutputs = 0;
|
||||
|
||||
await sessionManager.resetProcessingToPending(session.sessionDbId);
|
||||
session.abortReason = 'transport:observer_text';
|
||||
try {
|
||||
session.abortController.abort();
|
||||
} catch {
|
||||
// best-effort; AbortController.abort() should not throw in normal use.
|
||||
}
|
||||
worker?.broadcastProcessingStatus?.();
|
||||
logger.error('PARSER', `${agentName} could not reach the provider; queued batch preserved for retry`, {
|
||||
sessionId: session.sessionDbId,
|
||||
outputClass: 'transport',
|
||||
preview: previewOutput(text),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Classify the non-XML output so a dropped batch is visible, not silent.
|
||||
// Ordinary idle/prose is a claimed no-op batch: confirm it and do not build
|
||||
// any respawn debt from repeated skip acknowledgements.
|
||||
|
||||
@@ -53,7 +53,12 @@ export async function handleGeneratorExit(
|
||||
// in-RAM buffer (SessionManager.removeSessionImmediate -> 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),
|
||||
|
||||
@@ -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('<observation><title>Not logged in</title></observation>')).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(
|
||||
'<observation><title>fetch failed on cold start</title></observation>'
|
||||
)).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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user