diff --git a/.changeset/breezy-poems-listen.md b/.changeset/breezy-poems-listen.md new file mode 100644 index 0000000..042c808 --- /dev/null +++ b/.changeset/breezy-poems-listen.md @@ -0,0 +1,5 @@ +--- +'@arcasilesgroup/ai-shell': patch +--- + +Fix "Something went wrong" after `Your script:`. The first generation stream was consumed twice (script + explanation readers), which killed the process with an unhandled "Cannot iterate over a consumed stream" rejection, and any mid-stream provider error left the read promise unsettled so the CLI hung on a spinner with no message. The explanation now always comes from its own request, stream errors reject with a readable `KnownError`, and the keypress listener is removed when a stream ends. diff --git a/justfile b/justfile index cd94e14..e09c1b8 100644 --- a/justfile +++ b/justfile @@ -11,13 +11,17 @@ build: lint: npm run lint -# No unit suite yet; the smoke recipe is the contract test that exists: -# the built binary must report its version and resolve its defaults. It runs -# against a throwaway HOME so it never reads a developer's real key and never -# depends on one existing. +# The stream check is the offline regression gate for the double-consumed +# SSE stream bug ("Something went wrong", 2026-08-28): it runs the real +# completion helpers against a fake local SSE server, no key or network. +# The smoke recipe is the contract test that exists: the built binary must +# report its version and resolve its defaults. It runs against a throwaway +# HOME so it never reads a developer's real key and never depends on one +# existing. test: #!/usr/bin/env bash set -euo pipefail + node scripts/stream-consume.check.mjs tmp="$(mktemp -d)" printf 'API_KEY=sk-smoketest\n' > "$tmp/.ai-shell" HOME="$tmp" node dist/cli.mjs --version @@ -34,7 +38,7 @@ security: counts: @echo 'RAN lint=1 # npm run lint (prettier + eslint) checked the whole tree' - @echo 'RAN tests=2 # the two smoke assertions in the `test` recipe above' + @echo 'RAN tests=4 # the stream-consume check plus the three smoke assertions in the `test` recipe above' check: wired build lint test security counts diff --git a/scripts/stream-consume.check.mjs b/scripts/stream-consume.check.mjs new file mode 100644 index 0000000..2cbab89 --- /dev/null +++ b/scripts/stream-consume.check.mjs @@ -0,0 +1,152 @@ +// Contract check for the "Something went wrong" crash (ai-debug, 2026-08-28). +// +// Fails when: +// 1. getScriptAndInfo returns a second reader (readInfo) over the same +// single-consumption SDK stream -> the process dies with an unhandled +// rejection "Cannot iterate over a consumed stream". +// 2. readData swallows a mid-stream SSE error and its promise never +// settles (the "hangs forever, then says nothing" symptom). +// +// Run: node scripts/stream-consume.check.mjs (offline, deterministic) +import http from 'node:http'; +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const jiti = require('jiti')(process.cwd() + '/entry.js', { + interopDefault: true, +}); +const { getScriptAndInfo } = jiti('./src/helpers/completion.ts'); + +// The check runs with non-TTY stdin; readData's setRawMode(true) would throw +// for the wrong reason there. Real runs have a TTY. Neutralise it here. +if (!process.stdin.isTTY) { + process.stdin.setRawMode = () => process.stdin; +} + +// Watchdog: the pre-fix code hangs on `await readScript` under some stdin +// setups; never let the check itself hang. +setTimeout(() => { + console.error('FAIL: overall timeout (a read promise never settled)'); + process.exit(1); +}, 25000).unref(); + +let requests = 0; +const sse = (res, events) => { + res.writeHead(200, { + 'content-type': 'text/event-stream', + 'transfer-encoding': 'chunked', + }); + for (const e of events) res.write(`data: ${JSON.stringify(e)}\n\n`); + res.write('data: [DONE]\n\n'); + res.end(); +}; +const contentChunk = (content) => ({ + id: 'c1', + object: 'chat.completion.chunk', + created: 0, + model: 'fake', + choices: [{ index: 0, delta: { content }, finish_reason: null }], +}); + +const server = http.createServer((req, res) => { + requests++; + const path = new URL(req.url, 'http://x').pathname; + req.resume(); + req.on('end', () => { + if (path.endsWith('/err/v1/chat/completions')) { + // Valid SSE response carrying an error payload (what providers do when + // the model call fails after the stream has opened). + res.writeHead(200, { 'content-type': 'text/event-stream' }); + res.write(`data: ${JSON.stringify({ error: { message: 'boom' } })}\n\n`); + res.end(); + return; + } + sse(res, [ + contentChunk('```bash\n'), + contentChunk('echo hello\n'), + contentChunk('```'), + ]); + }); +}); +await new Promise((r) => server.listen(0, '127.0.0.1', r)); +const base = `http://127.0.0.1:${server.address().port}`; + +const unhandled = []; +process.on('unhandledRejection', (e) => unhandled.push(e)); + +const withTimeout = (p, ms, label) => + Promise.race([ + p.then( + (v) => ({ ok: true, v }), + (e) => ({ ok: false, e }) + ), + new Promise((r) => setTimeout(() => r({ timeout: true, label }), ms)), + ]); + +const args = { + prompt: 'say hello', + key: 'sk-fake', + model: 'fake', + apiEndpoint: `${base}/v1`, +}; + +// --- Case 1: the script stream must be consumed exactly once, no ghosts --- +const readers = await getScriptAndInfo(args); +const script = await readers.readScript(() => {}); + +const ghost = + typeof readers.readInfo === 'function' + ? await withTimeout( + Promise.race([ + readers + .readInfo(() => {}) + .then((v) => `resolved:${JSON.stringify(v)}`), + new Promise((r) => setTimeout(() => r('HUNG'), 2000)), + ]), + 2500, + 'readInfo' + ) + : { ok: true, v: 'absent' }; + +// Give any rejection from the shared stream a tick to surface. +await new Promise((r) => setTimeout(r, 100)); +if (ghost.timeout) { + console.error('FAIL: readInfo hung instead of finishing'); +} +if (unhandled.length) { + console.error( + 'FAIL: unhandled rejection(s):', + unhandled.map((e) => e?.message ?? String(e)).join(' | ') + ); +} + +// --- Case 2: a mid-stream error must reject the read promise, not hang --- +const errReaders = await getScriptAndInfo({ + ...args, + apiEndpoint: `${base}/err/v1`, +}); +const errResult = await withTimeout( + errReaders.readScript(() => {}), + 2000, + 'readScript-on-error-stream' +); +if (errResult.timeout) { + console.error( + `FAIL: ${errResult.label} never settled (hang on mid-stream error)` + ); +} + +server.close(); +const pass = + script.trim() === 'echo hello' && + !unhandled.length && + !ghost.timeout && + ghost.v === 'absent' && + !errResult.timeout && + errResult.ok === false; +console.log( + pass + ? 'PASS: stream consumed once, errors surface, no unhandled rejections' + : 'FAIL (see above)' +); +process.exit(pass ? 0 : 1); diff --git a/src/commands/chat.ts b/src/commands/chat.ts index bff828e..01a7cf9 100644 --- a/src/commands/chat.ts +++ b/src/commands/chat.ts @@ -92,5 +92,5 @@ async function getResponse({ apiEndpoint, }); - return { readResponse: readData(stream) }; + return { readResponse: readData(stream, apiEndpoint) }; } diff --git a/src/helpers/completion.ts b/src/helpers/completion.ts index d4abf4d..0c552e2 100644 --- a/src/helpers/completion.ts +++ b/src/helpers/completion.ts @@ -47,9 +47,11 @@ export async function getScriptAndInfo({ model, apiEndpoint, }); + // The SDK stream can be iterated exactly once. The explanation always + // comes from a second request (see explainInSecondRequest), so this only + // hands out one reader. return { - readScript: readData(stream, ...shellCodeExclusions), - readInfo: readData(stream, ...shellCodeExclusions), + readScript: readData(stream, apiEndpoint, ...shellCodeExclusions), }; } @@ -82,50 +84,57 @@ export async function generateCompletion({ return completion; } catch (err) { - if (err instanceof APIConnectionError) { - throw new KnownError( - `Error connecting to ${apiEndpoint}. Is the endpoint reachable and are you connected to the internet?\n${err.message}` - ); - } + throw normalizeCompletionError(err, apiEndpoint); + } +} - if (err instanceof APIError) { - const messageString = err.error - ? JSON.stringify(err.error, null, 2) - : err.message; - if (err.status === 429) { - throw new KnownError( - dedent` - Request failed with status 429 (rate limit or quota exceeded). This is usually due to an incorrect billing setup or excessive quota usage at your provider. - - Check your API key and plan at the provider configured in API_ENDPOINT. - - Full message from the API: - ` + - '\n\n' + - messageString + - '\n' - ); - } - const authHint = - err.status === 401 - ? '\n' + - 'Your API key does not match this endpoint. The key may belong to a different provider.\n' + - `Current endpoint: ${apiEndpoint}\n` + - 'Fix it with `ai config` (provider setup) or set API_ENDPOINT in ~/.ai-shell.' - : ''; - throw new KnownError( +// Maps a failure — request-time or mid-stream — into a KnownError with an +// actionable message. Shared by generateCompletion and readData so every +// caller of the stream sees the same error contract. +function normalizeCompletionError(err: unknown, apiEndpoint: string): unknown { + if (err instanceof APIConnectionError) { + return new KnownError( + `Error connecting to ${apiEndpoint}. Is the endpoint reachable and are you connected to the internet?\n${err.message}` + ); + } + + if (err instanceof APIError) { + const messageString = err.error + ? JSON.stringify(err.error, null, 2) + : err.message; + if (err.status === 429) { + return new KnownError( dedent` - Request to the API failed with status ${err.status}: + Request failed with status 429 (rate limit or quota exceeded). This is usually due to an incorrect billing setup or excessive quota usage at your provider. + + Check your API key and plan at the provider configured in API_ENDPOINT. + + Full message from the API: ` + '\n\n' + messageString + - authHint + '\n' ); } - - throw err instanceof Error ? new KnownError(err.message) : err; + const authHint = + err.status === 401 + ? '\n' + + 'Your API key does not match this endpoint. The key may belong to a different provider.\n' + + `Current endpoint: ${apiEndpoint}\n` + + 'Fix it with `ai config` (provider setup) or set API_ENDPOINT in ~/.ai-shell.' + : ''; + return new KnownError( + dedent` + Request to the API failed with status ${err.status}: + ` + + '\n\n' + + messageString + + authHint + + '\n' + ); } + + return err instanceof Error ? new KnownError(err.message) : err; } // Minimal non-streaming completion used by `ai config test` and the wizard. @@ -178,7 +187,7 @@ export async function getExplanation({ model, apiEndpoint, }); - return { readExplanation: readData(stream) }; + return { readExplanation: readData(stream, apiEndpoint) }; } export async function getRevision({ @@ -203,17 +212,18 @@ export async function getRevision({ apiEndpoint, }); return { - readScript: readData(stream, ...shellCodeExclusions), + readScript: readData(stream, apiEndpoint, ...shellCodeExclusions), }; } export const readData = ( chunkStream: AsyncIterable, + apiEndpoint: string, ...excluded: (RegExp | string | undefined)[] ) => (writer: (data: string) => void): Promise => { - const { promise, resolve } = Promise.withResolvers(); + const { promise, resolve, reject } = Promise.withResolvers(); (async () => { let stopTextStream = false; @@ -228,41 +238,54 @@ export const readData = input: process.stdin, }); - process.stdin.setRawMode(true); - - process.stdin.on('keypress', (_key, keyData) => { - if (stopTextStreamKeys.includes(keyData.name)) { + const onKeyPress = (_key: unknown, keyData: { name?: string }) => { + if (stopTextStreamKeys.includes(keyData?.name ?? '')) { stopTextStream = true; } - }); + }; - for await (const chunk of chunkStream) { - if (stopTextStream) { - break; - } - const content = chunk.choices[0]?.delta?.content ?? ''; - - if (!dataStart) { - buffer += content; - if (buffer.match(excludedPrefix ?? '')) { - dataStart = true; - buffer = ''; - // The delta that completes the opening fence is not part of the - // output. With no marker to wait for, write it through. - if (excludedPrefix) continue; + process.stdin.setRawMode(true); + process.stdin.on('keypress', onKeyPress); + + try { + for await (const chunk of chunkStream) { + if (stopTextStream) { + break; + } + const content = chunk.choices[0]?.delta?.content ?? ''; + + if (!dataStart) { + buffer += content; + if (buffer.match(excludedPrefix ?? '')) { + dataStart = true; + buffer = ''; + // The delta that completes the opening fence is not part of the + // output. With no marker to wait for, write it through. + if (excludedPrefix) continue; + } } - } - if (dataStart && content) { - const contentWithoutExcluded = stripRegexPatterns(content, excluded); + if (dataStart && content) { + const contentWithoutExcluded = stripRegexPatterns( + content, + excluded + ); - data += contentWithoutExcluded; - writer(contentWithoutExcluded); + data += contentWithoutExcluded; + writer(contentWithoutExcluded); + } } + resolve(data); + } catch (err) { + // A mid-stream failure (provider error event, dropped connection) + // must reach the caller. Swallowing it here left the promise unsettled + // forever: the UI hung on a spinner and clack's global + // unhandled-rejection handler printed "Something went wrong". + reject(normalizeCompletionError(err, apiEndpoint)); + } finally { + process.stdin.removeListener('keypress', onKeyPress); + rl.close(); } - - rl.close(); - resolve(data); })(); return promise; diff --git a/src/prompt.ts b/src/prompt.ts index 080b284..64c6241 100644 --- a/src/prompt.ts +++ b/src/prompt.ts @@ -116,7 +116,7 @@ export async function prompt({ const thePrompt = usePrompt || (await getPrompt()); const spin = p.spinner(); spin.start(i18n.t(`Loading...`)); - const { readInfo, readScript } = await getScriptAndInfo({ + const { readScript } = await getScriptAndInfo({ prompt: thePrompt, key, model, @@ -130,21 +130,18 @@ export async function prompt({ console.log(dim('•')); if (!skipCommandExplanation) { spin.start(i18n.t(`Getting explanation...`)); - const info = await readInfo(process.stdout.write.bind(process.stdout)); - if (!info) { - const { readExplanation } = await getExplanation({ - script, - key, - model, - apiEndpoint, - }); - spin.stop(`${i18n.t('Explanation')}:`); - console.log(''); - await readExplanation(process.stdout.write.bind(process.stdout)); - console.log(''); - console.log(''); - console.log(dim('•')); - } + const { readExplanation } = await getExplanation({ + script, + key, + model, + apiEndpoint, + }); + spin.stop(`${i18n.t('Explanation')}:`); + console.log(''); + await readExplanation(process.stdout.write.bind(process.stdout)); + console.log(''); + console.log(''); + console.log(dim('•')); } await runOrReviseFlow(script, key, model, apiEndpoint, silentMode);