diff --git a/plugins/opencode/scripts/lib/opencode-server.mjs b/plugins/opencode/scripts/lib/opencode-server.mjs index fa4dece..786f492 100644 --- a/plugins/opencode/scripts/lib/opencode-server.mjs +++ b/plugins/opencode/scripts/lib/opencode-server.mjs @@ -341,6 +341,13 @@ export class OpencodeServerClient { }); } + rejectQuestion(requestID, options = {}) { + // The reject route takes no request body. + return this.request("POST", `/question/${encodePathSegment(requestID)}/reject`, { + signal: options.signal + }); + } + health(options = {}) { return this.request("GET", "/global/health", { signal: options.signal }); } diff --git a/plugins/opencode/scripts/lib/opencode.mjs b/plugins/opencode/scripts/lib/opencode.mjs index 659c24e..a43b0de 100644 --- a/plugins/opencode/scripts/lib/opencode.mjs +++ b/plugins/opencode/scripts/lib/opencode.mjs @@ -377,6 +377,10 @@ function createTurnCaptureState(sessionID, options = {}) { finalMessage: "", structuredOutput: null, reasoningSummary: [], + // Latest snapshot of every streamed part, keyed by part id. Insertion + // order is preserved across snapshot replacements, which keeps text + // assembly stable while parts stream in (issue #28). + parts: new Map(), touchedFiles: new Set(), commandExecutions: [], error: null, @@ -400,17 +404,21 @@ function labelForSession(state, sessionID) { } function trackSession(state, event) { - const sessionID = extractSessionId(event); + // Real session.created/session.updated events carry the Session object in + // `properties.info` (id, parentID, title, agent); other events only carry + // `properties.sessionID`. + const info = event?.properties?.info ?? event?.info ?? event?.session ?? null; + const sessionID = extractSessionId(event) ?? info?.id ?? null; if (!sessionID) { return; } - const parentID = extractParentSessionId(event); + const parentID = info?.parentID ?? extractParentSessionId(event); if (sessionID === state.sessionID || state.sessionIDs.has(parentID)) { state.sessionIDs.add(sessionID); const label = - event?.session?.title ?? - event?.session?.agent ?? + info?.title ?? + info?.agent ?? event?.agent ?? event?.properties?.title ?? event?.properties?.agent ?? @@ -501,6 +509,160 @@ function applyMessageParts(state, parts, sessionID) { } } +function partIsAssembledText(part) { + return String(part?.type ?? "") === "text" && typeof part?.text === "string" && !part.synthetic && !part.ignored; +} + +// Rebuild the main-session final message from streamed text-part snapshots. +// Parts belong to messages, so assemble the text of one target message: the +// turn's assistant message when known, otherwise the message of the newest +// streamed text part (its parts still arrive in order). +function rebuildMainSessionText(state, hintMessageID = null) { + const textParts = []; + for (const part of state.parts.values()) { + if ((part.sessionID ?? state.sessionID) !== state.sessionID || !partIsAssembledText(part)) { + continue; + } + textParts.push(part); + } + if (textParts.length === 0) { + return; + } + + const candidates = [state.messageID, hintMessageID, textParts[textParts.length - 1].messageID]; + const target = candidates.find((id) => id && textParts.some((part) => part.messageID === id)); + const text = textParts + .filter((part) => (part.messageID ?? target) === target) + .map((part) => part.text) + .join(""); + if (text) { + state.finalMessage = text; + } +} + +// One streamed part snapshot (message.part.updated => properties.part). The +// event carries the part's full current state, so replace by part id. +function applyPartSnapshot(state, part) { + if (!part || typeof part !== "object" || !part.id) { + return; + } + const stored = { ...(state.parts.get(part.id) ?? {}), ...part }; + state.parts.set(part.id, stored); + + const sessionID = stored.sessionID ?? state.sessionID; + if (sessionID && !state.sessionIDs.has(sessionID)) { + return; + } + const subagentLabel = labelForSession(state, sessionID); + const type = String(stored.type ?? ""); + const completed = Boolean(stored.time?.end) || stored.state?.status === "completed"; + + if (!subagentLabel) { + const structured = extractStructuredOutput([stored]); + if (structured !== undefined) { + state.structuredOutput = structured; + } + } + + if (partIsAssembledText(stored)) { + if (!subagentLabel) { + rebuildMainSessionText(state, stored.messageID ?? null); + if (completed && stored.text) { + emitLogEvent(state.onProgress, { + message: `Assistant message captured: ${shorten(stored.text, 96)}`, + stderrMessage: null, + phase: "finalizing", + logTitle: "Assistant message", + logBody: stored.text + }); + } + } else if (completed && stored.text) { + emitLogEvent(state.onProgress, { + message: `Subagent ${subagentLabel}: ${shorten(stored.text, 96)}`, + stderrMessage: null, + logTitle: `Subagent ${subagentLabel} message`, + logBody: stored.text + }); + } + return; + } + + if (type === "reasoning" && completed && stored.text) { + const reasoning = extractReasoningFromParts([stored]); + mergeReasoning(state, reasoning); + if (reasoning.length > 0) { + emitLogEvent(state.onProgress, { + message: subagentLabel + ? `Subagent ${subagentLabel} reasoning: ${shorten(reasoning[0], 96)}` + : `Reasoning summary captured: ${shorten(reasoning[0], 96)}`, + stderrMessage: null, + logTitle: subagentLabel ? `Subagent ${subagentLabel} reasoning summary` : "Reasoning summary", + logBody: reasoning.map((section) => `- ${section}`).join("\n") + }); + } + } +} + +// Incremental text (message.part.delta => {sessionID, messageID, partID, +// field, delta}). Deltas only ever extend a part we already saw a snapshot +// for; the next snapshot is authoritative and replaces the accumulated value. +function applyPartDelta(state, props) { + const partID = typeof props?.partID === "string" ? props.partID : null; + const field = typeof props?.field === "string" ? props.field : null; + const delta = typeof props?.delta === "string" ? props.delta : null; + if (!partID || !field || delta == null) { + return; + } + const existing = state.parts.get(partID); + if (!existing) { + return; + } + existing[field] = (typeof existing[field] === "string" ? existing[field] : "") + delta; + if (partIsAssembledText(existing) && (existing.sessionID ?? state.sessionID) === state.sessionID) { + rebuildMainSessionText(state, existing.messageID ?? null); + } +} + +function extractQuestionRequestId(event) { + // The question request id lives at properties.id (que_...); the envelope's + // own id is the evt_... event id and must not be used. + const id = event?.properties?.id; + return typeof id === "string" && id ? id : null; +} + +function describeQuestions(event) { + const questions = Array.isArray(event?.properties?.questions) ? event.properties.questions : []; + const summary = questions + .map((question) => question?.header ?? question?.question) + .filter((value) => typeof value === "string" && value) + .slice(0, 2) + .join("; "); + return summary || null; +} + +// OpenCode's question tool blocks its session on a deferred reply. Headless +// runs have nobody to answer, so reject immediately — the model receives the +// rejection and must proceed autonomously — instead of stalling the turn +// until the outer timeout (issue #28 / review M-02). +async function respondToQuestion(client, state, event) { + const requestID = extractQuestionRequestId(event); + if (!requestID) { + return; + } + const described = describeQuestions(event); + emitProgress( + state.onProgress, + `Rejecting OpenCode question ${requestID}${described ? ` (${shorten(described, 96)})` : ""}: headless runs cannot answer interactive questions.`, + "running" + ); + try { + await client.rejectQuestion(requestID); + } catch (error) { + state.error = error; + emitProgress(state.onProgress, `OpenCode question rejection failed: ${error.message}`, "failed"); + } +} + function describePermissionRequest(event) { const category = (typeof event?.permission === "string" ? event.permission : null) ?? @@ -581,14 +743,46 @@ async function applyOpenCodeEvent(client, state, event, meta = {}) { return; } + if (type === "session.created" || type === "session.updated") { + // Child registration already happened in trackSession above. + return; + } + if (type === "permission.asked" || type === "permission.v2.asked") { await respondToPermission(client, state, event, sessionID ?? state.sessionID); return; } + if (type === "question.asked" || type === "question.v2.asked") { + await respondToQuestion(client, state, event); + return; + } + if (type === "message.updated") { - state.messageID = event?.message?.id ?? event?.info?.id ?? event?.id ?? state.messageID; - applyMessageParts(state, extractMessageParts(event), sessionID ?? state.sessionID); + // Metadata only in the real contract: properties.info carries the Message. + // Parts arrive separately via message.part.updated / message.part.delta. + const info = event?.properties?.info ?? event?.info ?? event?.message ?? null; + const infoSessionID = info?.sessionID ?? sessionID ?? state.sessionID; + if (info?.role === "assistant" && infoSessionID === state.sessionID) { + if (typeof info.id === "string" && info.id) { + state.messageID = info.id; + rebuildMainSessionText(state); + } + // json_schema output also lands on the assistant message itself. + if (info.structured !== undefined) { + state.structuredOutput = info.structured; + } + } + return; + } + + if (type === "message.part.updated") { + applyPartSnapshot(state, event?.properties?.part ?? event?.part ?? null); + return; + } + + if (type === "message.part.delta") { + applyPartDelta(state, event?.properties ?? null); return; } diff --git a/tests/event-contract.test.mjs b/tests/event-contract.test.mjs new file mode 100644 index 0000000..7c1e0dc --- /dev/null +++ b/tests/event-contract.test.mjs @@ -0,0 +1,239 @@ +import net from "node:net"; +import path from "node:path"; +import { spawn } from "node:child_process"; +import test from "node:test"; +import assert from "node:assert/strict"; +import { fileURLToPath } from "node:url"; +import { createRequire } from "node:module"; + +import { installFakeOpencode } from "./fake-opencode-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; + +// Pinned from a real `opencode serve` /doc (see contract.version). The fake +// fixture must emit events that satisfy this contract so the suite cannot +// drift back to invented event shapes (issue #28 / review H-03). +const require = createRequire(import.meta.url); +const contract = require("./opencode-event-contract.json"); + +async function canListenLocalhost() { + return new Promise((resolve) => { + const server = net.createServer(); + let settled = false; + function finish(value) { + if (settled) { + return; + } + settled = true; + resolve(value); + } + server.once("error", () => finish(false)); + server.listen(0, "127.0.0.1", () => { + server.close(() => finish(true)); + }); + }); +} + +const LOCAL_LISTEN_AVAILABLE = await canListenLocalhost(); +const LOCAL_LISTEN_SKIP = LOCAL_LISTEN_AVAILABLE ? false : "local 127.0.0.1 listen is unavailable in this sandbox"; + +function startFixtureServer(binDir, env = {}) { + const child = spawn("node", [path.join(binDir, "opencode"), "serve", "--hostname", "127.0.0.1", "--port", "0"], { + env: { + ...process.env, + FAKE_OPENCODE_STATE_PATH: path.join(binDir, "fake-opencode-state.json"), + // The raw-HTTP contract run talks to the fixture without credentials. + OPENCODE_SERVER_PASSWORD: "", + ...env + }, + windowsHide: true + }); + + const url = new Promise((resolve, reject) => { + let output = ""; + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + output += chunk; + const match = output.match(/listening on (http:\/\/[^\s]+)/); + if (match) { + resolve(match[1]); + } + }); + child.once("error", reject); + child.once("exit", () => reject(new Error("fixture server exited before listening"))); + setTimeout(() => reject(new Error("fixture server did not start")), 5000).unref(); + }); + + return { child, url }; +} + +function collectSseEvents(baseUrl, onEvent) { + const controller = new AbortController(); + const done = (async () => { + const response = await fetch(`${baseUrl}/event`, { + headers: { accept: "text/event-stream" }, + signal: controller.signal + }); + assert.equal(response.status, 200); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + for (;;) { + const { done: finished, value } = await reader.read(); + if (finished) { + break; + } + buffer += decoder.decode(value, { stream: true }); + for (;;) { + const blockEnd = buffer.indexOf("\n\n"); + if (blockEnd === -1) { + break; + } + const block = buffer.slice(0, blockEnd); + buffer = buffer.slice(blockEnd + 2); + const data = block + .split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trimStart()) + .join("\n"); + if (data) { + await onEvent(JSON.parse(data)); + } + } + } + })().catch((error) => { + if (!controller.signal.aborted) { + throw error; + } + }); + return { stop: () => controller.abort(), done }; +} + +function validatePinnedEvent(event, errors) { + const spec = contract.events[event?.type]; + if (!spec) { + errors.push(`unpinned event type emitted by fixture: ${event?.type}`); + return; + } + + for (const key of spec.required) { + if (!(key in event)) { + errors.push(`${event.type}: missing envelope key "${key}"`); + } + } + if (typeof event.id !== "string" || !event.id.startsWith("evt_")) { + errors.push(`${event.type}: envelope id must match ^evt_ (got ${event.id})`); + } + const properties = event.properties ?? {}; + for (const key of spec.properties.required) { + if (!(key in properties)) { + errors.push(`${event.type}: missing properties key "${key}"`); + } + } + for (const key of Object.keys(properties)) { + if (!spec.properties.keys.includes(key)) { + errors.push(`${event.type}: unexpected properties key "${key}" (additionalProperties: false upstream)`); + } + } + + const objectErrors = (objectName, value, label) => { + for (const key of contract.objects[objectName].required) { + if (!value || !(key in value)) { + errors.push(`${event.type}: ${label} missing required "${key}" (${objectName})`); + } + } + }; + + if (event.type === "session.created" || event.type === "session.updated") { + objectErrors("Session", properties.info, "properties.info"); + } + if (event.type === "message.updated") { + const role = properties.info?.role; + objectErrors(role === "user" ? "UserMessage" : "AssistantMessage", properties.info, "properties.info"); + } + if (event.type === "message.part.updated") { + const type = properties.part?.type; + const objectName = type === "reasoning" ? "ReasoningPart" : type === "tool" ? "ToolPart" : "TextPart"; + objectErrors(objectName, properties.part, "properties.part"); + } +} + +test("fake fixture events satisfy the pinned OpenCode event contract", { skip: LOCAL_LISTEN_SKIP }, async () => { + const binDir = makeTempDir(); + installFakeOpencode(binDir); + const { child, url } = startFixtureServer(binDir, { + FAKE_OPENCODE_ASK_QUESTION: "1", + FAKE_OPENCODE_SUBAGENT: "1", + FAKE_OPENCODE_STREAM_DELTAS: "1" + }); + + const events = []; + const errors = []; + let sawIdle; + const idle = new Promise((resolve) => { + sawIdle = resolve; + }); + let stream = null; + + try { + const baseUrl = await url; + stream = collectSseEvents(baseUrl, async (event) => { + events.push(event); + validatePinnedEvent(event, errors); + // Answer the interactive flows the way the real client must: deny the + // gated permission, reject the question. + if (event.type === "permission.asked") { + await fetch(`${baseUrl}/session/${event.properties.sessionID}/permissions/${event.properties.id}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ response: "reject" }) + }); + } + if (event.type === "question.asked") { + await fetch(`${baseUrl}/question/${event.properties.id}/reject`, { method: "POST" }); + } + if (event.type === "session.idle") { + sawIdle(); + } + }); + + const session = await fetch(`${baseUrl}/session`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ agent: "build" }) + }).then((response) => response.json()); + assert.ok(session.id, "fixture session created"); + + const message = fetch(`${baseUrl}/session/${session.id}/message`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ agent: "build", parts: [{ type: "text", text: "contract run" }] }) + }).catch(() => null); + + await Promise.race([ + idle, + new Promise((_, reject) => setTimeout(() => reject(new Error("session.idle never arrived")), 8000).unref()) + ]); + await message; + + assert.deepEqual(errors, [], errors.join("\n")); + const seenTypes = new Set(events.map((event) => event.type)); + for (const expected of [ + "session.created", + "message.updated", + "session.next.step.started", + "file.edited", + "permission.asked", + "question.asked", + "message.part.updated", + "message.part.delta", + "session.idle" + ]) { + assert.ok(seenTypes.has(expected), `expected fixture to emit ${expected}; saw ${[...seenTypes].join(", ")}`); + } + assert.equal(contract.version, "1.17.15"); + } finally { + stream?.stop(); + await stream?.done.catch(() => {}); + child.kill(); + } +}); diff --git a/tests/fake-opencode-fixture.mjs b/tests/fake-opencode-fixture.mjs index 38294bf..f13fd94 100644 --- a/tests/fake-opencode-fixture.mjs +++ b/tests/fake-opencode-fixture.mjs @@ -38,6 +38,7 @@ const path = require("node:path"); const STATE_PATH = process.env.FAKE_OPENCODE_STATE_PATH || ${JSON.stringify(statePath)}; const clients = new Set(); const pendingPermissions = new Map(); +const pendingQuestions = new Map(); function loadState() { if (!fs.existsSync(STATE_PATH)) { @@ -86,13 +87,47 @@ function sendJson(res, value, status = 200) { res.end(JSON.stringify(value)); } -function emit(event) { - const line = "data: " + JSON.stringify(event) + "\\n\\n"; +// Real event envelope (pinned in tests/opencode-event-contract.json): +// { id: "evt_...", type, properties }. +let nextEventId = 1; +function emit(type, properties) { + const line = "data: " + JSON.stringify({ id: "evt_" + nextEventId++, type, properties }) + "\\n\\n"; for (const client of clients) { client.write(line); } } +function sessionInfo(session, parentID) { + return { + id: session.id, + slug: "fake-" + session.id, + projectID: "prj_fake", + directory: session.directory || process.cwd(), + title: session.title || "", + version: "1.17.15", + time: { created: Date.now(), updated: Date.now() }, + ...(session.agent ? { agent: session.agent } : {}), + ...(parentID ? { parentID } : {}) + }; +} + +function assistantInfo(sessionID, messageID, agent) { + return { + id: messageID, + sessionID, + role: "assistant", + time: { created: Date.now() }, + parentID: messageID + "_user", + modelID: "fake-model", + providerID: "fake", + mode: "normal", + agent: agent || "build", + path: { cwd: process.cwd(), root: process.cwd() }, + cost: 0, + tokens: { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } } + }; +} + function textFromMessage(body) { return (body.parts || []) .map((part) => typeof part.text === "string" ? part.text : "") @@ -148,6 +183,7 @@ function structuredOutputParts(body) { { type: "tool", tool: "StructuredOutput", + callID: "call_structured_1", state: { status: "completed", input: exampleForSchema(body.format.schema, "result") @@ -168,6 +204,16 @@ async function waitForPermission(permissionID) { }); } +async function waitForQuestion(requestID) { + return new Promise((resolve) => { + const timeout = setTimeout(() => resolve(false), 2000); + pendingQuestions.set(requestID, () => { + clearTimeout(timeout); + resolve(true); + }); + }); +} + async function handleMessage(req, res, sessionID) { const body = await readJson(req); const state = loadState(); @@ -183,34 +229,94 @@ async function handleMessage(req, res, sessionID) { state.lastMessage = { sessionID, messageID, body, prompt }; saveState(state); - emit({ type: "session.next.step.started", sessionID }); - if (session.agent === "build" || body.agent === "build") { + const agent = session.agent || body.agent || "build"; + const info = assistantInfo(sessionID, messageID, agent); + emit("message.updated", { sessionID, info }); + emit("session.next.step.started", { + timestamp: Date.now(), + sessionID, + assistantMessageID: messageID, + agent, + model: { providerID: "fake", modelID: "fake-model" } + }); + + if (agent === "build") { // Workspace edits are covered by the stock build agent's wildcard allow // and never produce a permission round-trip. - emit({ type: "file.edited", sessionID, path: "generated.txt" }); - // A guard category (e.g. external_directory) reaches "ask". The companion - // must deny it; only an (incorrect) approval lets the gated edit proceed. + emit("file.edited", { file: "generated.txt" }); + // A guard category (external_directory) reaches "ask". The companion must + // deny it; only an (incorrect) approval lets the gated edit proceed. const permissionID = "perm_" + messageID; - emit({ - type: "permission.asked", + emit("permission.asked", { + id: permissionID, sessionID, - permissionID, - permission: { id: permissionID, tool: "edit" }, - patterns: ["/outside/workspace/secret.txt"] + permission: "external_directory", + patterns: ["/outside/workspace/secret.txt"], + metadata: {}, + always: ["/outside/workspace/secret.txt"], + tool: { messageID, callID: "call_edit_1" } }); const reply = await waitForPermission(permissionID); if (reply && reply.response !== "reject") { - emit({ type: "file.edited", sessionID, path: "/outside/workspace/secret.txt" }); + emit("file.edited", { file: "/outside/workspace/secret.txt" }); } } + if (process.env.FAKE_OPENCODE_ASK_QUESTION === "1") { + // The question tool blocks the session on a deferred reply; a headless + // client must reject it via POST /question/{requestID}/reject. + const questionID = "que_" + messageID; + emit("question.asked", { + id: questionID, + sessionID, + questions: [ + { + question: "Which approach should I take?", + header: "Approach", + options: [{ label: "Option A" }, { label: "Option B" }] + } + ] + }); + await waitForQuestion(questionID); + } + + if (process.env.FAKE_OPENCODE_SUBAGENT === "1") { + // A child session announces itself via session.created with parentID in + // properties.info; its output must not pollute the parent final message. + const childID = sessionID + "_child"; + const childMessageID = "msg_child_" + messageID; + emit("session.created", { + sessionID: childID, + info: sessionInfo({ id: childID, directory: process.cwd(), title: "Subtask: explore", agent: "explore" }, sessionID) + }); + emit("message.updated", { sessionID: childID, info: assistantInfo(childID, childMessageID, "explore") }); + emit("message.part.updated", { + sessionID: childID, + part: { + id: "prt_" + childMessageID, + sessionID: childID, + messageID: childMessageID, + type: "text", + text: "Child exploration output.", + time: { start: Date.now(), end: Date.now() } + }, + time: Date.now() + }); + } + const finalText = prompt.includes("follow up") ? "Resumed the prior OpenCode run.\\nFollow-up prompt accepted." : "Handled the requested task.\\nTask prompt accepted."; - const parts = structuredOutputParts(body) || [{ type: "text", text: finalText }]; + const parts = (structuredOutputParts(body) || [{ type: "text", text: finalText }]).map((part, index) => ({ + id: "prt_" + messageID + "_" + index, + sessionID, + messageID, + ...(part.type === "text" ? { time: { start: Date.now(), end: Date.now() } } : {}), + ...part + })); const failMode = process.env.FAKE_OPENCODE_MESSAGE_FAIL; if (failMode === "empty-recovery" || failMode === "snapshot-fails-empty-recovery") { - emit({ type: "session.idle", sessionID }); + emit("session.idle", { sessionID }); res.destroy(); return; } @@ -221,12 +327,33 @@ async function handleMessage(req, res, sessionID) { const finalState = loadState(); finalState.lastResponseParts = parts; finalState.responses = finalState.responses || []; - finalState.responses.push({ sessionID, info: { id: messageID, role: "assistant", sessionID }, parts }); + finalState.responses.push({ sessionID, info, parts }); saveState(finalState); + + if (process.env.FAKE_OPENCODE_STREAM_DELTAS === "1" && parts[0].type === "text") { + // Stream the text purely as an initial empty snapshot plus deltas, then + // drop the POST and withhold recovery data: the client must assemble the + // final message from the delta stream alone. + const full = parts[0].text; + emit("message.part.updated", { + sessionID, + part: Object.assign({}, parts[0], { text: "", time: { start: Date.now() } }), + time: Date.now() + }); + emit("message.part.delta", { sessionID, messageID, partID: parts[0].id, field: "text", delta: full.slice(0, 8) }); + emit("message.part.delta", { sessionID, messageID, partID: parts[0].id, field: "text", delta: full.slice(8) }); + const cleanState = loadState(); + cleanState.responses = (cleanState.responses || []).filter((entry) => entry.info.id !== messageID); + saveState(cleanState); + emit("session.idle", { sessionID }); + res.destroy(); + return; + } + // Issue #2 regression hooks. "transport": deliver the turn over the event // stream but drop the /message HTTP response mid-flight (like undici timing - // out the held-open POST). "recover": additionally withhold message.updated so - // the client must re-fetch the finished message via GET /session/:id/message. + // out the held-open POST). "recover": additionally withhold the part events + // so the client must re-fetch the finished message via GET /session/:id/message. // "delayed-events" drops the POST before completion events arrive, matching // the real failure ordering seen in issue #2 review. "mismatched-recover" // gives recovery a stale event-derived message id, then expects fallback to @@ -234,26 +361,30 @@ async function handleMessage(req, res, sessionID) { if (failMode === "delayed-events") { res.destroy(); setTimeout(() => { - emit({ type: "message.updated", sessionID, message: { id: messageID, parts } }); - emit({ type: "session.idle", sessionID }); + for (const part of parts) { + emit("message.part.updated", { sessionID, part, time: Date.now() }); + } + emit("session.idle", { sessionID }); }, 25); return; } if (failMode === "mismatched-recover") { - emit({ type: "message.updated", sessionID, message: { id: messageID + "_event_only" } }); - emit({ type: "session.idle", sessionID }); + emit("message.updated", { sessionID, info: assistantInfo(sessionID, messageID + "_event_only", agent) }); + emit("session.idle", { sessionID }); res.destroy(); return; } if (failMode !== "recover") { - emit({ type: "message.updated", sessionID, message: { id: messageID, parts } }); + for (const part of parts) { + emit("message.part.updated", { sessionID, part, time: Date.now() }); + } } - emit({ type: "session.idle", sessionID }); + emit("session.idle", { sessionID }); if (failMode === "transport" || failMode === "recover") { res.destroy(); return; } - sendJson(res, { info: { id: messageID, sessionID }, parts }); + sendJson(res, { info, parts }); } function handleSessionListCli() { @@ -422,7 +553,7 @@ const server = http.createServer(async (req, res) => { state.sessions.unshift(session); state.lastCreateSession = body; saveState(state); - emit({ type: "session.created", sessionID: session.id, session }); + emit("session.created", { sessionID: session.id, info: sessionInfo(session, body.parentID) }); sendJson(res, session); return; } @@ -457,6 +588,20 @@ const server = http.createServer(async (req, res) => { return; } + const questionRejectMatch = url.pathname.match(/^\\/question\\/([^/]+)\\/reject$/); + if (req.method === "POST" && questionRejectMatch) { + const requestID = decodeURIComponent(questionRejectMatch[1]); + const state = loadState(); + state.questionRejections = state.questionRejections || []; + state.questionRejections.push({ requestID }); + saveState(state); + const resolve = pendingQuestions.get(requestID); + pendingQuestions.delete(requestID); + resolve?.(); + sendJson(res, true); + return; + } + const permissionMatch = url.pathname.match(/^\\/session\\/([^/]+)\\/permissions\\/([^/]+)$/); if (req.method === "POST" && permissionMatch) { const body = await readJson(req); diff --git a/tests/opencode-event-contract.json b/tests/opencode-event-contract.json new file mode 100644 index 0000000..93b7c0c --- /dev/null +++ b/tests/opencode-event-contract.json @@ -0,0 +1,364 @@ +{ + "source": "opencode /doc", + "version": "1.17.15", + "events": { + "session.created": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID", + "info" + ], + "keys": [ + "sessionID", + "info" + ] + } + }, + "session.updated": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID", + "info" + ], + "keys": [ + "sessionID", + "info" + ] + } + }, + "session.idle": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID" + ], + "keys": [ + "sessionID" + ] + } + }, + "session.error": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [], + "keys": [ + "sessionID", + "error" + ] + } + }, + "message.updated": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID", + "info" + ], + "keys": [ + "sessionID", + "info" + ] + } + }, + "message.part.updated": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID", + "part", + "time" + ], + "keys": [ + "sessionID", + "part", + "time" + ] + } + }, + "message.part.delta": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "sessionID", + "messageID", + "partID", + "field", + "delta" + ], + "keys": [ + "sessionID", + "messageID", + "partID", + "field", + "delta" + ] + } + }, + "file.edited": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "file" + ], + "keys": [ + "file" + ] + } + }, + "permission.asked": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "id", + "sessionID", + "permission", + "patterns", + "metadata", + "always" + ], + "keys": [ + "id", + "sessionID", + "permission", + "patterns", + "metadata", + "always", + "tool" + ] + } + }, + "question.asked": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "id", + "sessionID", + "questions" + ], + "keys": [ + "id", + "sessionID", + "questions", + "tool" + ] + } + }, + "session.next.step.started": { + "required": [ + "id", + "type", + "properties" + ], + "properties": { + "required": [ + "timestamp", + "sessionID", + "assistantMessageID", + "agent", + "model" + ], + "keys": [ + "timestamp", + "sessionID", + "assistantMessageID", + "agent", + "model", + "snapshot" + ] + } + } + }, + "objects": { + "TextPart": { + "required": [ + "id", + "sessionID", + "messageID", + "type", + "text" + ], + "keys": [ + "id", + "sessionID", + "messageID", + "type", + "text", + "synthetic", + "ignored", + "time", + "metadata" + ] + }, + "ReasoningPart": { + "required": [ + "id", + "sessionID", + "messageID", + "type", + "text", + "time" + ], + "keys": [ + "id", + "sessionID", + "messageID", + "type", + "text", + "metadata", + "time" + ] + }, + "ToolPart": { + "required": [ + "id", + "sessionID", + "messageID", + "type", + "callID", + "tool", + "state" + ], + "keys": [ + "id", + "sessionID", + "messageID", + "type", + "callID", + "tool", + "state", + "metadata" + ] + }, + "AssistantMessage": { + "required": [ + "id", + "sessionID", + "role", + "time", + "parentID", + "modelID", + "providerID", + "mode", + "agent", + "path", + "cost", + "tokens" + ], + "keys": [ + "id", + "sessionID", + "role", + "time", + "error", + "parentID", + "modelID", + "providerID", + "mode", + "agent", + "path", + "summary", + "cost", + "tokens", + "structured", + "variant", + "finish" + ] + }, + "UserMessage": { + "required": [ + "id", + "sessionID", + "role", + "time", + "agent", + "model" + ], + "keys": [ + "id", + "sessionID", + "role", + "time", + "format", + "summary", + "agent", + "model", + "system", + "tools" + ] + }, + "Session": { + "required": [ + "id", + "slug", + "projectID", + "directory", + "title", + "version", + "time" + ], + "keys": [ + "id", + "slug", + "projectID", + "workspaceID", + "directory", + "path", + "parentID", + "summary", + "cost", + "tokens", + "share", + "title", + "agent", + "model", + "version", + "metadata", + "time", + "permission", + "revert" + ] + } + } +} diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index d7a6f4d..f3a9606 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -1036,6 +1036,80 @@ test("task recovery falls back to the newest assistant message when event messag } }); +test("task rejects headless question asks instead of stalling (issue #28)", { skip: LOCAL_LISTEN_SKIP }, () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeOpencode(binDir); + initGitRepo(repo); + const env = buildTestEnv(binDir, { FAKE_OPENCODE_ASK_QUESTION: "1" }); + + try { + const result = run("node", [SCRIPT, "task", "--json", "task that provokes a question"], { + cwd: repo, + env + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + assert.equal(payload.status, 0); + assert.match(payload.rawOutput, /Handled the requested task/); + + const fakeState = readFakeState(binDir); + assert.equal(fakeState.questionRejections.length, 1); + assert.match(fakeState.questionRejections[0].requestID, /^que_/); + } finally { + cleanupServer(repo, env); + } +}); + +test("task assembles the final message from part deltas without transport or recovery (issue #28)", { skip: LOCAL_LISTEN_SKIP }, () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeOpencode(binDir); + initGitRepo(repo); + // The POST response is dropped and the message list withheld, so the final + // text can only come from the message.part.updated + message.part.delta + // stream — the exact channel the pre-#28 client ignored. + const env = buildTestEnv(binDir, { FAKE_OPENCODE_STREAM_DELTAS: "1" }); + + try { + const result = run("node", [SCRIPT, "task", "--json", "stream this answer"], { + cwd: repo, + env + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + assert.equal(payload.status, 0); + assert.match(payload.rawOutput, /^Handled the requested task/); + } finally { + cleanupServer(repo, env); + } +}); + +test("subagent session output does not pollute the main final message (issue #28)", { skip: LOCAL_LISTEN_SKIP }, () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeOpencode(binDir); + initGitRepo(repo); + const env = buildTestEnv(binDir, { FAKE_OPENCODE_SUBAGENT: "1" }); + + try { + const result = run("node", [SCRIPT, "task", "--json", "task with a subagent"], { + cwd: repo, + env + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + assert.equal(payload.status, 0); + assert.match(payload.rawOutput, /Handled the requested task/); + assert.doesNotMatch(payload.rawOutput, /Child exploration output/); + } finally { + cleanupServer(repo, env); + } +}); + test("task fails when completion has no recoverable current-turn message (issue #2)", { skip: LOCAL_LISTEN_SKIP }, () => { const repo = makeTempDir(); const binDir = makeTempDir();