diff --git a/bin/agy-proxy.mjs b/bin/agy-proxy.mjs index d49e95f81..da7af89bf 100755 --- a/bin/agy-proxy.mjs +++ b/bin/agy-proxy.mjs @@ -464,7 +464,7 @@ const server = http.createServer((req, res) => { stderrBuffer += chunk.toString(); }); - child.on('close', code => { + child.on('close', (code, signal) => { if (responseEnded) return; // Fallback retry if resuming failed without output @@ -480,6 +480,28 @@ const server = http.createServer((req, res) => { processJsonLine(lineBuffer.trim()); } + // Failure is judged by the child's exit: a non-zero code or + // a killing signal means the run failed, however much content + // was already produced (#678). Partial output must not mask + // a failure — it used to be reported as a successful 'stop' + // carrying the truncated text (or an empty 200 when nothing + // was produced at all), and stderrBuffer was never read. + if (code !== 0) { + const stderrLine = stderrBuffer.trim().split('\n').pop(); + const fallback = signal + ? `agy process terminated by signal ${signal}` + : `agy process exited with code ${code}`; + const detail = (stderrLine || fallback).trim(); + const errorBody = { error: { message: `agy execution failed: ${detail}` } }; + // The 200 + text/event-stream headers are already on the + // wire, so frame the error as an SSE event and close the + // stream with [DONE] (mirrors muse-proxy #670). + res.write(`data: ${JSON.stringify(errorBody)}\n\n`); + res.write('data: [DONE]\n\n'); + res.end(); + return; + } + const finishChunk = { id: reqId, object: 'chat.completion.chunk', @@ -530,7 +552,7 @@ const server = http.createServer((req, res) => { child.stdout.on('data', chunk => { stdoutBuffer += chunk.toString(); }); child.stderr.on('data', chunk => { stderrBuffer += chunk.toString(); }); - child.on('close', code => { + child.on('close', (code, signal) => { if (res.writableEnded) return; let responseText = ''; @@ -550,8 +572,13 @@ const server = http.createServer((req, res) => { recordConversationId(detectedConvId); } - if (code !== 0 && !responseText) { - if (allowRetry && targetConvId) { + // Failure is judged by the child's exit, not by whether it + // produced output (#678): a run that fails after a partial + // response must not be reported as a successful 'stop' + // carrying the truncated text. The resume retry stays + // gated on empty output, as before. + if (code !== 0) { + if (!responseText && allowRetry && targetConvId) { console.log(`Resuming conversation ${targetConvId} failed (code ${code}), retrying as fresh conversation...`); executeJson(null, formatPrompt(body.messages), false); return; @@ -561,6 +588,7 @@ const server = http.createServer((req, res) => { res.end(JSON.stringify({ error: 'agy process exited with non-zero code', code, + ...(signal ? { signal } : {}), stderr: stderrBuffer.trim() })); return; diff --git a/bin/agy-proxy.test.mjs b/bin/agy-proxy.test.mjs new file mode 100644 index 000000000..546c332bc --- /dev/null +++ b/bin/agy-proxy.test.mjs @@ -0,0 +1,201 @@ +// Tests for agy-proxy's failure detection (#678). +// Run: node --test bin/agy-proxy.test.mjs +// +// The proxy spawns one `agy` child per request, with the prompt delivered +// as the `-p` argument. These tests put a stub `agy` at +// $HOME/.local/bin/agy (the proxy's binary resolution prefers it) whose +// behaviour is keyed on markers in the prompt: a run that exited non-zero +// used to be reported as a successful 'stop' — in stream mode always +// (the close handler never checked the code), and in non-stream mode +// whenever the run had produced partial output. +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import fs from 'node:fs'; +import net from 'node:net'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { fileURLToPath } from 'node:url'; + +const binDir = path.dirname(fileURLToPath(import.meta.url)); + +const STUB = `#!/bin/sh +prompt="" +fmt="json" +conv="" +prev="" +for a in "$@"; do + if [ "$prev" = "-p" ]; then prompt="$a"; fi + if [ "$prev" = "--conversation" ]; then conv="$a"; fi + if [ "$a" = "stream-json" ]; then fmt="stream"; fi + prev="$a" +done +case "$prompt" in + *RETRYME*) + if [ -n "$conv" ]; then + echo "stub agy: resume failed" >&2 + exit 3 + fi ;; + *FAILME*) + echo "stub agy: auth expired" >&2 + exit 3 ;; + *FAILPARTIAL*) + if [ "$fmt" = "stream" ]; then + printf '%s\\n' '{"event":"step_update","step_update":{"text_delta":"partial answer"}}' + else + printf '%s\\n' '{"response":"partial answer"}' + fi + echo "stub agy: run failed midway" >&2 + exit 3 ;; + *SIGKILLME*) + kill -9 $$ ;; +esac +if [ "$fmt" = "stream" ]; then + printf '%s\\n' '{"event":"step_update","step_update":{"text_delta":"Hello world"}}' + printf '%s\\n' '{"event":"result","result":{"usage":{"input_tokens":100,"output_tokens":3,"total_tokens":103}}}' +else + printf '%s\\n' '{"response":"Hello world","usage":{"input_tokens":100,"output_tokens":3,"total_tokens":103}}' +fi +exit 0 +`; + +const freePort = () => new Promise((resolve, reject) => { + const srv = net.createServer(); + srv.once('error', reject); + srv.listen(0, '127.0.0.1', () => { + const { port } = srv.address(); + srv.close(() => resolve(port)); + }); +}); + +const startProxy = async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'agy-proxy-test-')); + const home = path.join(tmp, 'home'); + fs.mkdirSync(path.join(home, '.local', 'bin'), { recursive: true }); + fs.writeFileSync(path.join(home, '.local', 'bin', 'agy'), STUB, { mode: 0o755 }); + const port = await freePort(); + const child = spawn(process.execPath, [path.join(binDir, 'agy-proxy.mjs')], { + env: { + ...process.env, + HOME: home, + XDG_STATE_HOME: path.join(tmp, 'state'), + AGY_PROXY_PORT: String(port) + }, + stdio: ['ignore', 'pipe', 'pipe'] + }); + let log = ''; + child.stdout.on('data', d => { log += d; }); + child.stderr.on('data', d => { log += d; }); + const deadline = Date.now() + 10000; + for (;;) { + try { + const res = await fetch(`http://127.0.0.1:${port}/health`); + if (res.ok) break; + } catch { /* not up yet */ } + if (Date.now() > deadline) throw new Error(`proxy did not start:\n${log}`); + await new Promise(r => setTimeout(r, 50)); + } + return { port, stop: () => child.kill('SIGKILL') }; +}; + +const post = (port, body, headers = {}) => fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...headers }, + body: JSON.stringify(body) +}); + +const parseSse = (sse) => { + const dataLines = sse.split('\n').filter(l => l.startsWith('data: ')); + assert.equal(dataLines[dataLines.length - 1], 'data: [DONE]', 'stream must end with [DONE]'); + return dataLines.slice(0, -1).map(l => JSON.parse(l.slice('data: '.length))); +}; + +const assertStreamFailure = async (port, marker, detail) => { + const messages = [{ role: 'user', content: `please ${marker} now` }]; + const streamRes = await post(port, { messages, stream: true }); + assert.equal(streamRes.status, 200); + assert.match(streamRes.headers.get('content-type'), /text\/event-stream/); + const events = parseSse(await streamRes.text()); + const errorEvent = events.find(e => e.error); + assert.ok(errorEvent, `${marker}: stream must carry an error event`); + assert.match(errorEvent.error.message, detail, `${marker}: stream detail must name the cause, got: ${errorEvent.error.message}`); + const stopChunk = events.find(e => e.choices && e.choices[0].finish_reason === 'stop'); + assert.equal(stopChunk, undefined, `${marker}: no chunk may claim finish_reason 'stop'`); +}; + +test('successful run is unchanged in both modes', async (t) => { + const { port, stop } = await startProxy(); + t.after(stop); + const messages = [{ role: 'user', content: 'hi' }]; + + const plainRes = await post(port, { messages }); + assert.equal(plainRes.status, 200); + const plain = JSON.parse(await plainRes.text()); + assert.equal(plain.choices[0].message.content, 'Hello world'); + assert.equal(plain.choices[0].finish_reason, 'stop'); + + const streamRes = await post(port, { messages, stream: true }); + assert.equal(streamRes.status, 200); + const sse = await streamRes.text(); + assert.match(sse, /Hello world/); + assert.match(sse, /"finish_reason":"stop"/); + assert.ok(sse.trimEnd().endsWith('data: [DONE]')); +}); + +test('execution failure with no output: 502 non-stream, SSE error + [DONE] stream (#678)', async (t) => { + const { port, stop } = await startProxy(); + t.after(stop); + const messages = [{ role: 'user', content: 'please FAILME now' }]; + + const plainRes = await post(port, { messages }); + assert.equal(plainRes.status, 502, 'non-stream failure must not be a 200 stop'); + const plainBody = JSON.parse(await plainRes.text()); + assert.match(plainBody.stderr, /stub agy: auth expired/); + + await assertStreamFailure(port, 'FAILME', /agy execution failed: stub agy: auth expired/); +}); + +test('partial output then failure: masked success becomes an error in both modes (#678)', async (t) => { + const { port, stop } = await startProxy(); + t.after(stop); + const messages = [{ role: 'user', content: 'please FAILPARTIAL now' }]; + + const plainRes = await post(port, { messages }); + assert.equal(plainRes.status, 502, 'non-stream partial failure must not be a 200 stop'); + const plainBody = JSON.parse(await plainRes.text()); + assert.match(plainBody.stderr, /stub agy: run failed midway/); + + await assertStreamFailure(port, 'FAILPARTIAL', /agy execution failed: stub agy: run failed midway/); +}); + +test('signal-killed run: stream fallback names the signal, non-stream is a 502 (#678)', async (t) => { + const { port, stop } = await startProxy(); + t.after(stop); + const messages = [{ role: 'user', content: 'please SIGKILLME now' }]; + + const plainRes = await post(port, { messages }); + assert.equal(plainRes.status, 502); + const plainBody = JSON.parse(await plainRes.text()); + assert.equal(plainBody.signal, 'SIGKILL'); + + await assertStreamFailure(port, 'SIGKILLME', /agy execution failed: agy process terminated by signal SIGKILL/); +}); + +test('failed resume still retries once as a fresh conversation (#678)', async (t) => { + const { port, stop } = await startProxy(); + t.after(stop); + const messages = [{ role: 'user', content: 'please RETRYME now' }]; + const headers = { 'x-conversation-id': 'conv-123' }; + + const plainRes = await post(port, { messages }, headers); + assert.equal(plainRes.status, 200); + const plain = JSON.parse(await plainRes.text()); + assert.equal(plain.choices[0].message.content, 'Hello world'); + assert.equal(plain.choices[0].finish_reason, 'stop'); + + const streamRes = await post(port, { messages, stream: true }, headers); + assert.equal(streamRes.status, 200); + const sse = await streamRes.text(); + assert.match(sse, /Hello world/); + assert.match(sse, /"finish_reason":"stop"/); +});