From db9016a8ba4ccd8a92614ef2776e0cd86e03a07f Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 12:26:13 -0600 Subject: [PATCH 1/7] fix(opencode): persist question requests in SQLite Question requests (que_*) lived only in process memory, so any opencode restart under a parked question orphaned the coordinator's durable pointer: every answer submit then 404ed (~44% of answerQuestion attempts in prod over 30 days). Requests now persist in a question_request table; replies to requests whose waiter died with the process complete the interrupted tool part with the real answers and re-enter the assistant loop. Co-Authored-By: Claude Fable 5 --- packages/core/schema.json | 283 +++++++++++++++--- packages/core/src/database/migration.gen.ts | 1 + .../20260820182352_question_request.ts | 21 ++ packages/core/src/database/schema.gen.ts | 11 + packages/core/src/session/sql.ts | 17 ++ packages/opencode/src/question/format.ts | 9 + packages/opencode/src/question/index.ts | 112 +++++-- packages/opencode/src/question/resume.ts | 50 ++++ .../instance/httpapi/handlers/question.ts | 16 +- packages/opencode/src/session/prompt.ts | 6 + packages/opencode/src/tool/question.ts | 7 +- 11 files changed, 457 insertions(+), 76 deletions(-) create mode 100644 packages/core/src/database/migration/20260820182352_question_request.ts create mode 100644 packages/opencode/src/question/format.ts create mode 100644 packages/opencode/src/question/resume.ts diff --git a/packages/core/schema.json b/packages/core/schema.json index d0eeeebd5c41..e5e0e0ac9199 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,10 @@ { "version": "7", "dialect": "sqlite", - "id": "f14a9b18-8207-487e-a3d3-227e629ba9ad", - "prevIds": ["169a0f0f-d58f-479f-b024-fa1c7b9a09db"], + "id": "1e809201-35a1-4c99-8ba8-0ce36b675bcf", + "prevIds": [ + "f14a9b18-8207-487e-a3d3-227e629ba9ad" + ], "ddl": [ { "name": "workspace", @@ -56,6 +58,10 @@ "name": "part", "entityType": "tables" }, + { + "name": "question_request", + "entityType": "tables" + }, { "name": "session_context_epoch", "entityType": "tables" @@ -880,6 +886,56 @@ "entityType": "columns", "table": "part" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "question_request" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "question_request" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "question_request" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "question_request" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "question_request" + }, { "type": "text", "notNull": false, @@ -1481,9 +1537,13 @@ "table": "session_share" }, { - "columns": ["project_id"], + "columns": [ + "project_id" + ], "tableTo": "project", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1492,9 +1552,13 @@ "table": "workspace" }, { - "columns": ["active_account_id"], + "columns": [ + "active_account_id" + ], "tableTo": "account", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "SET NULL", "nameExplicit": false, @@ -1503,9 +1567,13 @@ "table": "account_state" }, { - "columns": ["aggregate_id"], + "columns": [ + "aggregate_id" + ], "tableTo": "event_sequence", - "columnsTo": ["aggregate_id"], + "columnsTo": [ + "aggregate_id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1514,9 +1582,13 @@ "table": "event" }, { - "columns": ["project_id"], + "columns": [ + "project_id" + ], "tableTo": "project", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1525,9 +1597,13 @@ "table": "permission" }, { - "columns": ["project_id"], + "columns": [ + "project_id" + ], "tableTo": "project", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1536,9 +1612,13 @@ "table": "project_directory" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1547,9 +1627,13 @@ "table": "message" }, { - "columns": ["message_id"], + "columns": [ + "message_id" + ], "tableTo": "message", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1558,9 +1642,28 @@ "table": "part" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], + "tableTo": "session", + "columnsTo": [ + "id" + ], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_question_request_session_id_session_id_fk", + "entityType": "fks", + "table": "question_request" + }, + { + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1569,9 +1672,13 @@ "table": "session_context_epoch" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1580,9 +1687,13 @@ "table": "session_input" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1591,9 +1702,13 @@ "table": "session_message" }, { - "columns": ["project_id"], + "columns": [ + "project_id" + ], "tableTo": "project", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1602,9 +1717,13 @@ "table": "session" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1613,9 +1732,13 @@ "table": "todo" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "tableTo": "session", - "columnsTo": ["id"], + "columnsTo": [ + "id" + ], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1624,133 +1747,183 @@ "table": "session_share" }, { - "columns": ["email", "url"], + "columns": [ + "email", + "url" + ], "nameExplicit": false, "name": "control_account_pk", "entityType": "pks", "table": "control_account" }, { - "columns": ["project_id", "directory"], + "columns": [ + "project_id", + "directory" + ], "nameExplicit": false, "name": "project_directory_pk", "entityType": "pks", "table": "project_directory" }, { - "columns": ["session_id", "position"], + "columns": [ + "session_id", + "position" + ], "nameExplicit": false, "name": "todo_pk", "entityType": "pks", "table": "todo" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "workspace_pk", "table": "workspace", "entityType": "pks" }, { - "columns": ["name"], + "columns": [ + "name" + ], "nameExplicit": false, "name": "data_migration_pk", "table": "data_migration", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "account_state_pk", "table": "account_state", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "account_pk", "table": "account", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "credential_pk", "table": "credential", "entityType": "pks" }, { - "columns": ["aggregate_id"], + "columns": [ + "aggregate_id" + ], "nameExplicit": false, "name": "event_sequence_pk", "table": "event_sequence", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "event_pk", "table": "event", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "permission_pk", "table": "permission", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "project_pk", "table": "project", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "message_pk", "table": "message", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "part_pk", "table": "part", "entityType": "pks" }, { - "columns": ["session_id"], + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "question_request_pk", + "table": "question_request", + "entityType": "pks" + }, + { + "columns": [ + "session_id" + ], "nameExplicit": false, "name": "session_context_epoch_pk", "table": "session_context_epoch", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "session_input_pk", "table": "session_input", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "session_message_pk", "table": "session_message", "entityType": "pks" }, { - "columns": ["id"], + "columns": [ + "id" + ], "nameExplicit": false, "name": "session_pk", "table": "session", "entityType": "pks" }, { - "columns": ["session_id"], + "columns": [ + "session_id" + ], "nameExplicit": false, "name": "session_share_pk", "table": "session_share", @@ -1872,6 +2045,20 @@ "entityType": "indexes", "table": "part" }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "question_request_session_idx", + "entityType": "indexes", + "table": "question_request" + }, { "columns": [ { @@ -2068,4 +2255,4 @@ } ], "renames": [] -} +} \ No newline at end of file diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index e6ea4eaa1477..031bd3ddd30b 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -40,5 +40,6 @@ export const migrations = ( import("./migration/20260622142730_simplify_session_context_epoch"), import("./migration/20260622170816_reset_v2_session_state"), import("./migration/20260622202450_simplify_session_input"), + import("./migration/20260820182352_question_request"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260820182352_question_request.ts b/packages/core/src/database/migration/20260820182352_question_request.ts new file mode 100644 index 000000000000..2b8731fe2a8e --- /dev/null +++ b/packages/core/src/database/migration/20260820182352_question_request.ts @@ -0,0 +1,21 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260820182352_question_request", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(` + CREATE TABLE \`question_request\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text NOT NULL, + \`time_created\` integer NOT NULL, + \`time_updated\` integer NOT NULL, + \`data\` text NOT NULL, + CONSTRAINT \`fk_question_request_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE + ); + `) + yield* tx.run(`CREATE INDEX \`question_request_session_idx\` ON \`question_request\` (\`session_id\`);`) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index ed60fde6c55f..7657be6fcc4a 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -145,6 +145,16 @@ export default { CONSTRAINT \`fk_part_message_id_message_id_fk\` FOREIGN KEY (\`message_id\`) REFERENCES \`message\`(\`id\`) ON DELETE CASCADE ); `) + yield* tx.run(` + CREATE TABLE \`question_request\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text NOT NULL, + \`time_created\` integer NOT NULL, + \`time_updated\` integer NOT NULL, + \`data\` text NOT NULL, + CONSTRAINT \`fk_question_request_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE + ); + `) yield* tx.run(` CREATE TABLE \`session_context_epoch\` ( \`session_id\` text PRIMARY KEY, @@ -246,6 +256,7 @@ export default { ) yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`) yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`) + yield* tx.run(`CREATE INDEX \`question_request_session_idx\` ON \`question_request\` (\`session_id\`);`) yield* tx.run( `CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`, ) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 264a1d2cca0a..7937e70f0c51 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -14,6 +14,7 @@ import { Timestamps } from "../database/schema.sql" import type { SystemContext } from "../system-context/index" import { AgentV2 } from "../agent" import type { Revert } from "@opencode-ai/schema/revert" +import { QuestionV1 } from "@opencode-ai/schema/question-v1" type SessionMessageData = Omit<(typeof SessionMessage.Message)["Encoded"], "type" | "id"> type V1MessageData = Omit @@ -116,6 +117,22 @@ export const TodoTable = sqliteTable( ], ) +type QuestionRequestData = Omit + +export const QuestionRequestTable = sqliteTable( + "question_request", + { + id: text().$type().primaryKey(), + session_id: text() + .$type() + .notNull() + .references(() => SessionTable.id, { onDelete: "cascade" }), + ...Timestamps, + data: text({ mode: "json" }).notNull().$type(), + }, + (table) => [index("question_request_session_idx").on(table.session_id)], +) + export const SessionMessageTable = sqliteTable( "session_message", { diff --git a/packages/opencode/src/question/format.ts b/packages/opencode/src/question/format.ts new file mode 100644 index 000000000000..a4fc24bdc71d --- /dev/null +++ b/packages/opencode/src/question/format.ts @@ -0,0 +1,9 @@ +export function formatAnswerOutput(input: { + questions: ReadonlyArray<{ question: string }> + answers: ReadonlyArray> +}) { + const formatted = input.questions + .map((q, i) => `"${q.question}"="${input.answers[i]?.length ? input.answers[i].join(", ") : "Unanswered"}"`) + .join(", ") + return `User has answered your questions: ${formatted}. You can now continue with the user's answers in mind.` +} diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 8afc141072d5..01d2ff8bcb8e 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -1,5 +1,8 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Deferred, Effect, Layer, Schema, Context } from "effect" +import { asc, eq } from "drizzle-orm" +import { Database } from "@opencode-ai/core/database/database" +import { QuestionRequestTable, SessionTable } from "@opencode-ai/core/session/sql" import { InstanceState } from "@/effect/instance-state" import { SessionID } from "@/session/schema" import { QuestionID } from "./schema" @@ -43,6 +46,16 @@ interface State { pending: Map } +export type ReplyOutcome = + | { readonly outcome: "resolved" } + | { readonly outcome: "orphaned"; readonly request: Request } + +type QuestionRequestRow = typeof QuestionRequestTable.$inferSelect + +function rowToRequest(row: QuestionRequestRow): Request { + return { ...row.data, id: row.id, sessionID: row.session_id } +} + // Service export interface Interface { @@ -54,8 +67,9 @@ export interface Interface { readonly reply: (input: { requestID: QuestionID answers: ReadonlyArray - }) => Effect.Effect + }) => Effect.Effect readonly reject: (requestID: QuestionID) => Effect.Effect + readonly rejectAllForSession: (sessionID: SessionID) => Effect.Effect readonly list: () => Effect.Effect> } @@ -65,12 +79,16 @@ const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2Bridge.Service + const { db } = yield* Database.Service const state = yield* InstanceState.make( Effect.fn("Question.state")(function* () { const state = { pending: new Map(), } + // Rows in question_request intentionally survive this finalizer so a + // restarted process can still resolve replies to questions it no + // longer has waiters for. yield* Effect.addFinalizer(() => Effect.gen(function* () { for (const item of state.pending.values()) { @@ -100,6 +118,15 @@ const layer = Layer.effect( questions: input.questions, tool: input.tool, } + yield* db + .insert(QuestionRequestTable) + .values({ + id, + session_id: input.sessionID, + data: { questions: input.questions, tool: input.tool }, + }) + .run() + .pipe(Effect.orDie) pending.set(id, { info, deferred }) yield* events.publish(Event.Asked, info) @@ -115,47 +142,90 @@ const layer = Layer.effect( requestID: QuestionID answers: ReadonlyArray }) { - const pending = (yield* InstanceState.get(state)).pending - const existing = pending.get(input.requestID) - if (!existing) { + const row = yield* db + .select() + .from(QuestionRequestTable) + .where(eq(QuestionRequestTable.id, input.requestID)) + .get() + .pipe(Effect.orDie) + if (!row) { yield* Effect.logWarning("reply for unknown request", { requestID: input.requestID }) return yield* new NotFoundError({ requestID: input.requestID }) } - pending.delete(input.requestID) + yield* db.delete(QuestionRequestTable).where(eq(QuestionRequestTable.id, row.id)).run().pipe(Effect.orDie) + const request = rowToRequest(row) yield* Effect.logInfo("replied", { requestID: input.requestID, answers: input.answers }) yield* events.publish(Event.Replied, { - sessionID: existing.info.sessionID, - requestID: existing.info.id, + sessionID: request.sessionID, + requestID: request.id, answers: input.answers.map((a) => [...a]), }) + const pending = (yield* InstanceState.get(state)).pending + const existing = pending.get(input.requestID) + if (!existing) { + yield* Effect.logInfo("reply for orphaned request", { requestID: input.requestID }) + return { outcome: "orphaned", request } as const + } + pending.delete(input.requestID) yield* Deferred.succeed(existing.deferred, input.answers) + return { outcome: "resolved" } as const }) - const reject = Effect.fn("Question.reject")(function* (requestID: QuestionID) { + const rejectRow = Effect.fn("Question.rejectRow")(function* (row: QuestionRequestRow) { + yield* db.delete(QuestionRequestTable).where(eq(QuestionRequestTable.id, row.id)).run().pipe(Effect.orDie) + yield* Effect.logInfo("rejected", { requestID: row.id }) + yield* events.publish(Event.Rejected, { + sessionID: row.session_id, + requestID: row.id, + }) const pending = (yield* InstanceState.get(state)).pending - const existing = pending.get(requestID) - if (!existing) { + const existing = pending.get(row.id) + if (!existing) return + pending.delete(row.id) + yield* Deferred.fail(existing.deferred, new RejectedError()) + }) + + const reject = Effect.fn("Question.reject")(function* (requestID: QuestionID) { + const row = yield* db + .select() + .from(QuestionRequestTable) + .where(eq(QuestionRequestTable.id, requestID)) + .get() + .pipe(Effect.orDie) + if (!row) { yield* Effect.logWarning("reject for unknown request", { requestID }) return yield* new NotFoundError({ requestID }) } - pending.delete(requestID) - yield* Effect.logInfo("rejected", { requestID }) - yield* events.publish(Event.Rejected, { - sessionID: existing.info.sessionID, - requestID: existing.info.id, - }) - yield* Deferred.fail(existing.deferred, new RejectedError()) + yield* rejectRow(row) + }) + + const rejectAllForSession = Effect.fn("Question.rejectAllForSession")(function* (sessionID: SessionID) { + const rows = yield* db + .select() + .from(QuestionRequestTable) + .where(eq(QuestionRequestTable.session_id, sessionID)) + .all() + .pipe(Effect.orDie) + yield* Effect.forEach(rows, rejectRow, { discard: true }) }) const list = Effect.fn("Question.list")(function* () { - const pending = (yield* InstanceState.get(state)).pending - return Array.from(pending.values(), (x) => x.info) + const ctx = yield* InstanceState.context + const rows = yield* db + .select({ request: QuestionRequestTable }) + .from(QuestionRequestTable) + .innerJoin(SessionTable, eq(QuestionRequestTable.session_id, SessionTable.id)) + .where(eq(SessionTable.project_id, ctx.project.id)) + .orderBy(asc(QuestionRequestTable.id)) + .all() + .pipe(Effect.orDie) + return rows.map((x) => rowToRequest(x.request)) }) - return Service.of({ ask, reply, reject, list }) + return Service.of({ ask, reply, reject, rejectAllForSession, list }) }), ) -export const node = LayerNode.make({ service: Service, layer: layer, deps: [EventV2Bridge.node] }) +export const node = LayerNode.make({ service: Service, layer: layer, deps: [EventV2Bridge.node, Database.node] }) export * as Question from "." diff --git a/packages/opencode/src/question/resume.ts b/packages/opencode/src/question/resume.ts new file mode 100644 index 000000000000..239cc3c67eaf --- /dev/null +++ b/packages/opencode/src/question/resume.ts @@ -0,0 +1,50 @@ +import { Effect, Option } from "effect" +import { Session } from "@/session/session" +import { SessionPrompt } from "@/session/prompt" +import { SessionStatus } from "@/session/status" +import { formatAnswerOutput } from "./format" +import type { Answer, Request } from "." + +/** + * Continues a session after a reply arrived for a question whose in-process + * waiter died with a restart. The persisted request lets us complete the + * interrupted tool call with the real answers and re-enter the assistant loop + * as if the answer had resolved normally. + */ +export const resumeOrphanedReply = Effect.fn("Question.resumeOrphanedReply")(function* (input: { + request: Request + answers: ReadonlyArray +}) { + const sessions = yield* Session.Service + const prompts = yield* SessionPrompt.Service + const status = yield* SessionStatus.Service + + const tool = input.request.tool + if (tool) { + const message = yield* sessions.findMessage(input.request.sessionID, (msg) => msg.info.id === tool.messageID) + const part = Option.isSome(message) + ? message.value.parts.find((candidate) => candidate.type === "tool" && candidate.callID === tool.callID) + : undefined + if (part && part.type === "tool" && part.state.status !== "completed") { + const started = "time" in part.state && part.state.time ? part.state.time.start : Date.now() + yield* sessions.updatePart({ + ...part, + state: { + status: "completed", + input: part.state.input, + output: formatAnswerOutput({ questions: input.request.questions, answers: input.answers }), + title: `Asked ${input.request.questions.length} question${input.request.questions.length > 1 ? "s" : ""}`, + metadata: { answers: input.answers.map((answer) => [...answer]) }, + time: { start: started, end: Date.now() }, + }, + }) + } + } + + const current = yield* status.get(input.request.sessionID) + if (current.type !== "idle") { + yield* Effect.logInfo("skipping question resume, session busy", { sessionID: input.request.sessionID }) + return + } + yield* prompts.loop({ sessionID: input.request.sessionID }) +}) diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts index e9ed8cc89154..69792a6db22d 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts @@ -1,6 +1,7 @@ import { Question } from "@/question" +import { resumeOrphanedReply } from "@/question/resume" import { QuestionID } from "@/question/schema" -import { Effect } from "effect" +import { Effect, Scope } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { InstanceHttpApi } from "../api" import { QuestionNotFoundError } from "../errors" @@ -8,6 +9,9 @@ import { QuestionNotFoundError } from "../errors" export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question", (handlers) => Effect.gen(function* () { const svc = yield* Question.Service + // Build-time scope outlives any single request, so a resume forked into it + // keeps running after the reply endpoint has already responded. + const scope = yield* Scope.Scope const list = Effect.fn("QuestionHttpApi.list")(function* () { return yield* svc.list() @@ -17,7 +21,7 @@ export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question" params: { requestID: QuestionID } payload: Question.Reply }) { - yield* svc + const result = yield* svc .reply({ requestID: ctx.params.requestID, answers: ctx.payload.answers, @@ -32,6 +36,14 @@ export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question" ), ), ) + if (result.outcome === "orphaned") { + yield* resumeOrphanedReply({ request: result.request, answers: ctx.payload.answers }).pipe( + Effect.catchCause((cause) => + Effect.logError("question resume failed", { requestID: ctx.params.requestID, cause }), + ), + Effect.forkIn(scope), + ) + } return true }) diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 839924fd0a4c..3222a1d0d3cc 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -33,6 +33,7 @@ import { NamedError } from "@opencode-ai/core/util/error" import { SessionProcessor } from "./processor" import { Tool } from "@/tool/tool" import { Permission } from "@/permission" +import { Question } from "@/question" import { SessionStatus } from "./status" import { LLM } from "./llm" import { Shell } from "@opencode-ai/core/shell" @@ -141,6 +142,7 @@ const layer = Layer.effect( const flags = yield* RuntimeFlags.Service const database = yield* Database.Service const { db } = database + const questions = yield* Question.Service const ops = Effect.fn("SessionPrompt.ops")(function* () { return { cancel: (sessionID: SessionID) => cancel(sessionID), @@ -152,6 +154,9 @@ const layer = Layer.effect( const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) { yield* Effect.logInfo("cancel", { "session.id": sessionID }) yield* state.cancel(sessionID) + // Stop must clear persisted question rows so a docked ask can't outlive + // the turn it belonged to; process shutdown deliberately does not. + yield* questions.rejectAllForSession(sessionID) }) const resolvePromptParts = Effect.fn("SessionPrompt.resolvePromptParts")(function* (template: string) { @@ -1644,6 +1649,7 @@ export const node = LayerNode.make({ EventV2Bridge.node, RuntimeFlags.node, Database.node, + Question.node, ], }) diff --git a/packages/opencode/src/tool/question.ts b/packages/opencode/src/tool/question.ts index 51f1e71e28fe..13f898c6fff3 100644 --- a/packages/opencode/src/tool/question.ts +++ b/packages/opencode/src/tool/question.ts @@ -1,6 +1,7 @@ import { Effect, Schema } from "effect" import * as Tool from "./tool" import { Question } from "../question" +import { formatAnswerOutput } from "../question/format" import DESCRIPTION from "./question.txt" export const Parameters = Schema.Struct({ @@ -27,13 +28,9 @@ export const QuestionTool = Tool.define `"${q.question}"="${answers[i]?.length ? answers[i].join(", ") : "Unanswered"}"`) - .join(", ") - return { title: `Asked ${params.questions.length} question${params.questions.length > 1 ? "s" : ""}`, - output: `User has answered your questions: ${formatted}. You can now continue with the user's answers in mind.`, + output: formatAnswerOutput({ questions: params.questions, answers }), metadata: { answers, }, From 6fe1e97210063e1deb383daf44c4818427badbaa Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 12:40:21 -0600 Subject: [PATCH 2/7] test(opencode): cover question persistence across instance reload Seed real project/session rows in question tests (the new question_request table enforces its session FK), assert reply outcomes, and add coverage for rejectAllForSession plus the reload-survival + orphaned-reply heal path. Co-Authored-By: Claude Fable 5 --- packages/opencode/config.json | 3 + packages/opencode/test/fixture/fixture.ts | 29 +++++++ .../opencode/test/question/question.test.ts | 81 ++++++++++++++++++- packages/opencode/test/tool/question.test.ts | 6 +- 4 files changed, 114 insertions(+), 5 deletions(-) create mode 100644 packages/opencode/config.json diff --git a/packages/opencode/config.json b/packages/opencode/config.json new file mode 100644 index 000000000000..c3eb6a56fa87 --- /dev/null +++ b/packages/opencode/config.json @@ -0,0 +1,3 @@ +{ + "$schema": "https://opencode.ai/config.json" +} \ No newline at end of file diff --git a/packages/opencode/test/fixture/fixture.ts b/packages/opencode/test/fixture/fixture.ts index 53e3142cdfb3..e20af805c984 100644 --- a/packages/opencode/test/fixture/fixture.ts +++ b/packages/opencode/test/fixture/fixture.ts @@ -11,7 +11,13 @@ import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" import type { Config } from "@/config/config" import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Database } from "@opencode-ai/core/database/database" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionTable } from "@opencode-ai/core/session/sql" import { InstanceRef } from "../../src/effect/instance-ref" +import { InstanceState } from "../../src/effect/instance-state" +import type { SessionID } from "../../src/session/schema" import { InstanceBootstrap } from "../../src/project/bootstrap-service" import type { InstanceContext } from "../../src/project/instance-context" import { InstanceRuntime } from "../../src/project/instance-runtime" @@ -23,6 +29,29 @@ export const testInstanceStoreLayer = LayerNode.compile(InstanceStore.node, [ [InstanceStore.bootstrapNode, noopBootstrap], ]) +/** Persisted question requests reference real session rows; tests that ask questions seed one here. */ +export const seedSession = Effect.fn("TestFixture.seedSession")(function* (sessionID: SessionID) { + const { db } = yield* Database.Service + const ctx = yield* InstanceState.context + yield* db + .insert(ProjectTable) + .values({ id: ctx.project.id, worktree: AbsolutePath.make(ctx.worktree), sandboxes: [] }) + .onConflictDoNothing() + .run() + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: ctx.project.id, + slug: sessionID, + directory: ctx.directory, + title: "test session", + version: "test", + }) + .onConflictDoNothing() + .run() +}) + export async function provideTestInstance(input: { directory: string init?: Effect.Effect diff --git a/packages/opencode/test/question/question.test.ts b/packages/opencode/test/question/question.test.ts index 9adb1e0c9ff7..760f6148efe1 100644 --- a/packages/opencode/test/question/question.test.ts +++ b/packages/opencode/test/question/question.test.ts @@ -1,17 +1,20 @@ import { afterEach, expect } from "bun:test" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Cause, Effect, Exit, Fiber, Layer, Queue } from "effect" +import { Database } from "@opencode-ai/core/database/database" import { Question } from "../../src/question" import { InstanceRef } from "../../src/effect/instance-ref" import { InstanceStore } from "../../src/project/instance-store" import { QuestionID } from "../../src/question/schema" -import { disposeAllInstances, provideInstance, testInstanceStoreLayer, tmpdirScoped } from "../fixture/fixture" +import { disposeAllInstances, provideInstance, seedSession, testInstanceStoreLayer, tmpdirScoped } from "../fixture/fixture" import { SessionID } from "../../src/session/schema" import { testEffect } from "../lib/effect" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { EventV2Bridge } from "../../src/event-v2-bridge" -const questionLayer = LayerNode.compile(LayerNode.group([Question.node, EventV2Bridge.node, CrossSpawnSpawner.node])) +const questionLayer = LayerNode.compile( + LayerNode.group([Question.node, EventV2Bridge.node, CrossSpawnSpawner.node, Database.node]), +) const it = testEffect(questionLayer) const lifecycle = testEffect(Layer.mergeAll(questionLayer, testInstanceStoreLayer)) @@ -20,6 +23,7 @@ const askEffect = Effect.fn("QuestionTest.ask")(function* (input: { questions: ReadonlyArray tool?: Question.Tool }) { + yield* seedSession(input.sessionID) const question = yield* Question.Service return yield* question.ask(input) }) @@ -31,7 +35,7 @@ const replyEffect = Effect.fn("QuestionTest.reply")(function* (input: { answers: ReadonlyArray }) { const question = yield* Question.Service - yield* question.reply(input) + return yield* question.reply(input) }) const rejectEffect = Effect.fn("QuestionTest.reject")(function* (id: QuestionID) { @@ -144,11 +148,12 @@ it.instance( const pending = yield* waitForPending(1) const requestID = pending[0].id - yield* replyEffect({ + const outcome = yield* replyEffect({ requestID, answers: [["Option 1"]], }) + expect(outcome.outcome).toBe("resolved") expect(yield* Fiber.join(fiber)).toEqual([["Option 1"]]) }), { git: true }, @@ -457,3 +462,71 @@ lifecycle.live("pending question rejects on instance reload", () => if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(Question.RejectedError) }), ) + +it.instance( + "rejectAllForSession - clears rows and fails the waiter", + () => + Effect.gen(function* () { + const fiber = yield* askEffect({ + sessionID: SessionID.make("ses_cancel"), + questions: [ + { + question: "Cancel me?", + header: "Cancel", + options: [{ label: "Yes", description: "Yes" }], + }, + ], + }).pipe(Effect.forkScoped) + + expect(yield* waitForPending(1)).toHaveLength(1) + yield* Question.Service.use((svc) => svc.rejectAllForSession(SessionID.make("ses_cancel"))) + + expect((yield* Fiber.await(fiber))._tag).toBe("Failure") + expect(yield* listEffect).toHaveLength(0) + }), + { git: true }, +) + +lifecycle.live("request survives instance reload and reply heals it", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped({ git: true }) + const questions = [ + { + question: "Survive me?", + header: "Survive", + options: [{ label: "Yes", description: "Yes" }], + }, + ] + const fiber = yield* askEffect({ + sessionID: SessionID.make("ses_survive"), + questions, + }).pipe(provideInstance(dir), Effect.forkScoped) + + const pending = yield* waitForPending(1).pipe(provideInstance(dir)) + yield* InstanceStore.Service.use((store) => store.reload({ directory: dir })) + yield* Fiber.await(fiber) + + const survived = yield* listEffect.pipe(provideInstance(dir)) + expect(survived).toHaveLength(1) + expect(survived[0].id).toBe(pending[0].id) + expect(survived[0].sessionID).toBe(SessionID.make("ses_survive")) + expect(survived[0].questions).toEqual(questions) + + const replied = yield* Effect.gen(function* () { + const events = yield* EventV2Bridge.Service + const seen = yield* Queue.unbounded() + const off = yield* events.listen((event) => { + if (event.type === Question.Event.Replied.type) Queue.offerUnsafe(seen, event.data) + return Effect.void + }) + yield* Effect.addFinalizer(() => off) + const outcome = yield* replyEffect({ requestID: pending[0].id, answers: [["Yes"]] }) + expect(outcome.outcome).toBe("orphaned") + return yield* Queue.take(seen).pipe(Effect.timeout("2 seconds")) + }).pipe(provideInstance(dir)) + expect(replied).toMatchObject({ requestID: pending[0].id, answers: [["Yes"]] }) + + const after = yield* listEffect.pipe(provideInstance(dir)) + expect(after).toHaveLength(0) + }), +) diff --git a/packages/opencode/test/tool/question.test.ts b/packages/opencode/test/tool/question.test.ts index d8c6df77e8a2..c10bf3f556a4 100644 --- a/packages/opencode/test/tool/question.test.ts +++ b/packages/opencode/test/tool/question.test.ts @@ -1,11 +1,13 @@ import { describe, expect } from "bun:test" import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Database } from "@opencode-ai/core/database/database" import { Effect, Fiber, Queue } from "effect" import { QuestionTool } from "../../src/tool/question" import { Question } from "../../src/question" import { SessionID, MessageID } from "../../src/session/schema" import { Agent } from "../../src/agent/agent" import { Truncate } from "@/tool/truncate" +import { seedSession } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { EventV2Bridge } from "../../src/event-v2-bridge" @@ -21,7 +23,7 @@ const ctx = { } const it = testEffect( - LayerNode.compile(LayerNode.group([Question.node, EventV2Bridge.node, Truncate.node, Agent.node])), + LayerNode.compile(LayerNode.group([Question.node, EventV2Bridge.node, Truncate.node, Agent.node, Database.node])), ) const pending = Effect.fn("QuestionToolTest.pending")(function* (question: Question.Interface) { @@ -44,6 +46,7 @@ const pending = Effect.fn("QuestionToolTest.pending")(function* (question: Quest describe("tool.question", () => { it.instance("should successfully execute with valid question parameters", () => Effect.gen(function* () { + yield* seedSession(ctx.sessionID) const question = yield* Question.Service const toolInfo = yield* QuestionTool const tool = yield* toolInfo.init() @@ -70,6 +73,7 @@ describe("tool.question", () => { it.instance("should now pass with a header longer than 12 but less than 30 chars", () => Effect.gen(function* () { + yield* seedSession(ctx.sessionID) const question = yield* Question.Service const toolInfo = yield* QuestionTool const tool = yield* toolInfo.init() From f52b2d6dd8d70aa800419530cc8f9cf9c0855fe6 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 12:52:13 -0600 Subject: [PATCH 3/7] chore(opencode): drop stray config.json test artifact Co-Authored-By: Claude Fable 5 --- packages/opencode/config.json | 3 --- 1 file changed, 3 deletions(-) delete mode 100644 packages/opencode/config.json diff --git a/packages/opencode/config.json b/packages/opencode/config.json deleted file mode 100644 index c3eb6a56fa87..000000000000 --- a/packages/opencode/config.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "$schema": "https://opencode.ai/config.json" -} \ No newline at end of file From ec701971dbbbea826f88985dab71ff46784baa44 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 14:04:12 -0600 Subject: [PATCH 4/7] refactor(opencode): claim question rows atomically via delete-returning Concurrent replies (or reply racing reject) could both read the same row and double-publish before either delete landed; deleting with returning() makes the row delete the single claim, restoring the pre-persistence double-submit semantics. Also trims review comments to one line. Co-Authored-By: Claude Fable 5 --- bun.lock | 36 ----------------- packages/opencode/src/question/index.ts | 40 +++++++++---------- packages/opencode/src/question/resume.ts | 7 +--- .../instance/httpapi/handlers/question.ts | 3 +- packages/opencode/src/session/prompt.ts | 3 +- 5 files changed, 21 insertions(+), 68 deletions(-) diff --git a/bun.lock b/bun.lock index 907c04c31d2f..26b83f6e90d9 100644 --- a/bun.lock +++ b/bun.lock @@ -1437,8 +1437,6 @@ "@babel/types": ["@babel/types@7.29.7", "", { "dependencies": { "@babel/helper-string-parser": "^7.29.7", "@babel/helper-validator-identifier": "^7.29.7" } }, "sha512-4zBIxpPzowiZpusoFkyGVwakdRJUyuH5PxQ/PrqghfdFWWasvnCdPfQXHrenDai+gyLARulZjZowCOj6fjT4pA=="], - "@bcoe/v8-coverage": ["@bcoe/v8-coverage@1.0.2", "", {}, "sha512-6zABk/ECA/QYSCQ1NGiVwwbQerUCZ+TQbp64Q3AgmfNvurHH0j8TtXa1qbShXA6qqkpAj4V5W8pP6mLe1mcMqA=="], - "@bufbuild/protobuf": ["@bufbuild/protobuf@2.12.0", "", {}, "sha512-B/XlCaFIP8LOwzo+bz5uFzATYokcwCKQcghqnlfwSmM5eX/qTkvDBnDPs+gXtX/RyjxJ4DRikECcPJbyALA8FA=="], "@bufbuild/protoplugin": ["@bufbuild/protoplugin@2.12.0", "", { "dependencies": { "@bufbuild/protobuf": "2.12.0", "@typescript/vfs": "^1.6.2", "typescript": "5.4.5" } }, "sha512-ORlDITp8AFUXzIhLRoMCG+ud+D3MPKWb5HQXBoskMMnjeyEjE1H1qLonVNPyOr8lkx3xSfYUo8a0dvOZJVAzow=="], @@ -2991,8 +2989,6 @@ "@vitejs/plugin-react": ["@vitejs/plugin-react@4.7.0", "", { "dependencies": { "@babel/core": "^7.28.0", "@babel/plugin-transform-react-jsx-self": "^7.27.1", "@babel/plugin-transform-react-jsx-source": "^7.27.1", "@rolldown/pluginutils": "1.0.0-beta.27", "@types/babel__core": "^7.20.5", "react-refresh": "^0.17.0" }, "peerDependencies": { "vite": "^4.2.0 || ^5.0.0 || ^6.0.0 || ^7.0.0" } }, "sha512-gUu9hwfWvvEDBBmgtAowQCojwZmJ5mcLn3aufeCsitijs3+f2NsrPtlAWIR6OPiqljl96GVCUbLe0HyqIpVaoA=="], - "@vitest/coverage-v8": ["@vitest/coverage-v8@4.1.8", "", { "dependencies": { "@bcoe/v8-coverage": "^1.0.2", "@vitest/utils": "4.1.8", "ast-v8-to-istanbul": "^1.0.0", "istanbul-lib-coverage": "^3.2.2", "istanbul-lib-report": "^3.0.1", "istanbul-reports": "^3.2.0", "magicast": "^0.5.2", "obug": "^2.1.1", "std-env": "^4.0.0-rc.1", "tinyrainbow": "^3.1.0" }, "peerDependencies": { "@vitest/browser": "4.1.8", "vitest": "4.1.8" }, "optionalPeers": ["@vitest/browser"] }, "sha512-lt3kovsyHwYe00wq4D1ti0Z974fWj4NLp6siqiyEufUpyFwK9Yhi7rBhac9JL5aA0zoMrJqc4vYPZRUnI7l7nw=="], - "@vitest/expect": ["@vitest/expect@3.2.4", "", { "dependencies": { "@types/chai": "^5.2.2", "@vitest/spy": "3.2.4", "@vitest/utils": "3.2.4", "chai": "^5.2.0", "tinyrainbow": "^2.0.0" } }, "sha512-Io0yyORnB6sikFlt8QW5K7slY4OjqNX9jmJQ02QDda8lyM6B5oNgVWoSoKPac8/kgnCUzuHQKrSLtu/uOqqrig=="], "@vitest/mocker": ["@vitest/mocker@4.1.7", "", { "dependencies": { "@vitest/spy": "4.1.7", "estree-walker": "^3.0.3", "magic-string": "^0.30.21" }, "peerDependencies": { "msw": "^2.4.9", "vite": "^6.0.0 || ^7.0.0 || ^8.0.0" }, "optionalPeers": ["msw", "vite"] }, "sha512-vY7nuamKgfvpA1Koa3oYIw/k7D6kZnpGyNMZW8loow2bsBYla1TFdqTaXncWdRn4pgwNs+90RhnXhJScDwQeJA=="], @@ -3113,8 +3109,6 @@ "ast-types": ["ast-types@0.16.1", "", { "dependencies": { "tslib": "^2.0.1" } }, "sha512-6t10qk83GOG8p0vKmaCr8eiilZwO171AvbROMtvvNiwrTly62t+7XkA8RdIIVbpMhCASAsxgAzdRSwh6nw/5Dg=="], - "ast-v8-to-istanbul": ["ast-v8-to-istanbul@1.0.3", "", { "dependencies": { "@jridgewell/trace-mapping": "^0.3.31", "estree-walker": "^3.0.3", "js-tokens": "^10.0.0" } }, "sha512-jCMQ6ZylLPudp0CDfBmQBZUsrh1/8psbmu9ibeVWKuHWD0YrH9YABwlKu5kVEFoT0GCQQW9Z/SxfuEbbkGQCRg=="], - "astral-regex": ["astral-regex@2.0.0", "", {}, "sha512-Z7tMw1ytTXt5jqMcOP+OQteU1VuNK9Y02uuJtKQ1Sv69jXQKKg5cibLwGJow8yzZP+eAc18EmLGPal0bp36rvQ=="], "astring": ["astring@1.9.0", "", { "bin": { "astring": "bin/astring" } }, "sha512-LElXdjswlqjWrPpJFg1Fx4wpkOCxj1TDHlSV4PlaRxHGWko024xICaa97ZkMfs6DRKlCguiAI+rbXv5GWwXIkg=="], @@ -3599,8 +3593,6 @@ "encodeurl": ["encodeurl@2.0.0", "", {}, "sha512-Q0n9HRi4m6JuGIV1eFlmvJB7ZEVxu93IrMyiMsGC0lrMJMWzRgx6WGquyfQgZVb31vhGgXnfmPNNXmxnOkRBrg=="], - "encoding": ["encoding@0.1.13", "", { "dependencies": { "iconv-lite": "^0.6.2" } }, "sha512-ETBauow1T35Y/WZMkio9jiM0Z5xjHHmJ4XmjZOq1l/dXz3lr2sRn87nJy20RupqSh1F2m3HHPSp8ShIPQJrJ3A=="], - "end-of-stream": ["end-of-stream@1.4.5", "", { "dependencies": { "once": "^1.4.0" } }, "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg=="], "engine.io-client": ["engine.io-client@6.6.5", "", { "dependencies": { "@socket.io/component-emitter": "~3.1.0", "debug": "~4.4.1", "engine.io-parser": "~5.2.1", "ws": "~8.20.1", "xmlhttprequest-ssl": "~2.1.1" } }, "sha512-QCwxUDULPlXv8F6tqMMKx5dNkTe6OaBYRMPYeXKBlyOoKvAmE0ac6pW7fFhSscJ/5SI7666/U/B+MElbsrJlIg=="], @@ -4121,12 +4113,6 @@ "isomorphic-ws": ["isomorphic-ws@5.0.0", "", { "peerDependencies": { "ws": "*" } }, "sha512-muId7Zzn9ywDsyXgTIafTry2sV3nySZeUDe6YedVd1Hvuuep5AsIlqK+XefWpYTyJG5e503F2xIuT2lcU6rCSw=="], - "istanbul-lib-coverage": ["istanbul-lib-coverage@3.2.2", "", {}, "sha512-O8dpsF+r0WV/8MNRKfnmrtCWhuKjxrq2w+jpzBL5UZKTi2LeVWnWOmWRxFlesJONmc+wLAGvKQZEOanko0LFTg=="], - - "istanbul-lib-report": ["istanbul-lib-report@3.0.1", "", { "dependencies": { "istanbul-lib-coverage": "^3.0.0", "make-dir": "^4.0.0", "supports-color": "^7.1.0" } }, "sha512-GCfE1mtsHGOELCU8e/Z7YWzpmybrx/+dSTfLrvY8qRmaY6zXTKWn6WQIjaAFw069icm6GVMNkgu0NzI4iPZUNw=="], - - "istanbul-reports": ["istanbul-reports@3.2.0", "", { "dependencies": { "html-escaper": "^2.0.0", "istanbul-lib-report": "^3.0.0" } }, "sha512-HGYWWS/ehqTV3xN10i23tkPkpH46MLCIMFNCaaKNavAXTF1RkqxawEPtnjnGZ6XKSInBKkiOA5BKS+aZiY3AvA=="], - "iterate-iterator": ["iterate-iterator@1.0.2", "", {}, "sha512-t91HubM4ZDQ70M9wqp+pcNpu8OyJ9UAtXntT/Bcsvp5tZMnz9vRa+IunKXeI8AnfZMTv0jNuVEmGeLSMjVvfPw=="], "iterate-value": ["iterate-value@1.0.2", "", { "dependencies": { "es-get-iterator": "^1.0.2", "iterate-iterator": "^1.0.1" } }, "sha512-A6fMAio4D2ot2r/TYzr4yUWrmwNdsN5xL7+HUiyACE4DXm+q8HtPcnFTp+NnW3k4N05tZ7FVYFFb2CR13NxyHQ=="], @@ -4295,8 +4281,6 @@ "magicast": ["magicast@0.3.5", "", { "dependencies": { "@babel/parser": "^7.25.4", "@babel/types": "^7.25.4", "source-map-js": "^1.2.0" } }, "sha512-L0WhttDl+2BOsybvEOLK7fW3UA0OQ0IQ2d6Zl2x/a6vVRs3bAY0ECOSHHeL5jD+SbOpOCUEi0y1DgHEn9Qn1AQ=="], - "make-dir": ["make-dir@4.0.0", "", { "dependencies": { "semver": "^7.5.3" } }, "sha512-hXdUTZYIVOt1Ex//jAQi+wTZZpUpwBj/0QsOzqegb3rGMMeJiSEu5xLHnYfBrRV4RH2+OCSOO95Is/7x1WJ4bw=="], - "make-fetch-happen": ["make-fetch-happen@15.0.6", "", { "dependencies": { "@gar/promise-retry": "^1.0.0", "@npmcli/agent": "^4.0.0", "@npmcli/redact": "^4.0.0", "cacache": "^20.0.1", "http-cache-semantics": "^4.1.1", "minipass": "^7.0.2", "minipass-fetch": "^5.0.0", "minipass-flush": "^1.0.5", "minipass-pipeline": "^1.2.4", "negotiator": "^1.0.0", "proc-log": "^6.0.0", "ssri": "^13.0.0" } }, "sha512-Je0fLJ0F5atA7F+eIlLzk+Wkcl57JDf4kf+EW8xiP5E31xOQxkIxTbgf1Oi1Lw9tRI9UEMRdI5Vz2xTzoNU1Jw=="], "markdown-extensions": ["markdown-extensions@2.0.0", "", {}, "sha512-o5vL7aDWatOTX8LzaS1WMoaoxIiLRQJuIKKe2wAw6IeULDHaqbiqiggmx+pKvZDb1Sj+pE46Sn1T7lCqfFtg1Q=="], @@ -6089,10 +6073,6 @@ "@solidjs/start/vite": ["vite@7.1.10", "", { "dependencies": { "esbuild": "^0.25.0", "fdir": "^6.5.0", "picomatch": "^4.0.3", "postcss": "^8.5.6", "rollup": "^4.43.0", "tinyglobby": "^0.2.15" }, "optionalDependencies": { "fsevents": "~2.3.3" }, "peerDependencies": { "@types/node": "^20.19.0 || >=22.12.0", "jiti": ">=1.21.0", "less": "^4.0.0", "lightningcss": "^1.21.0", "sass": "^1.70.0", "sass-embedded": "^1.70.0", "stylus": ">=0.54.8", "sugarss": "^5.0.0", "terser": "^5.16.0", "tsx": "^4.8.1", "yaml": "^2.4.2" }, "optionalPeers": ["@types/node", "jiti", "less", "lightningcss", "sass", "sass-embedded", "stylus", "sugarss", "terser", "tsx", "yaml"], "bin": { "vite": "bin/vite.js" } }, "sha512-CmuvUBzVJ/e3HGxhg6cYk88NGgTnBoOo7ogtfJJ0fefUWAxN/WDSUa50o+oVBxuIhO8FoEZW0j2eW7sfjs5EtA=="], - "@standard-community/standard-json/effect": ["effect@4.0.0-beta.74", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.8.0", "find-my-way-ts": "^0.1.6", "ini": "^7.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^2.0.1", "multipasta": "^0.2.7", "toml": "^4.1.1", "uuid": "^14.0.0", "yaml": "^2.9.0" } }, "sha512-Yx+Kh12U+i2FmjwEfKs+ePFmpMd43RPD1oGqc/VraSS9bYzvF0Ff3PojwEFEVEewp8xc92Uxu28gTspU4qyvHA=="], - - "@standard-community/standard-openapi/effect": ["effect@4.0.0-beta.74", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.8.0", "find-my-way-ts": "^0.1.6", "ini": "^7.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^2.0.1", "multipasta": "^0.2.7", "toml": "^4.1.1", "uuid": "^14.0.0", "yaml": "^2.9.0" } }, "sha512-Yx+Kh12U+i2FmjwEfKs+ePFmpMd43RPD1oGqc/VraSS9bYzvF0Ff3PojwEFEVEewp8xc92Uxu28gTspU4qyvHA=="], - "@storybook/csf-plugin/unplugin": ["unplugin@2.3.11", "", { "dependencies": { "@jridgewell/remapping": "^2.3.5", "acorn": "^8.15.0", "picomatch": "^4.0.3", "webpack-virtual-modules": "^0.6.2" } }, "sha512-5uKD0nqiYVzlmCRs01Fhs2BdkEgBS3SAVP6ndrBsuK42iC2+JHyxM05Rm9G8+5mkmRtzMZGY8Ct5+mliZxU/Ww=="], "@tailwindcss/oxide/detect-libc": ["detect-libc@2.1.2", "", {}, "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ=="], @@ -6121,10 +6101,6 @@ "@types/plist/xmlbuilder": ["xmlbuilder@15.1.1", "", {}, "sha512-yMqGBqtXyeN1e3TGYvgNgDVZ3j84W4cwkOXQswghol6APgZWaff9lnbvN7MHYJOiXsvGPXtjTYJEiC9J2wv9Eg=="], - "@vitest/coverage-v8/@vitest/utils": ["@vitest/utils@4.1.8", "", { "dependencies": { "@vitest/pretty-format": "4.1.8", "convert-source-map": "^2.0.0", "tinyrainbow": "^3.1.0" } }, "sha512-uOJamYALNhfJ6iolExyQM40yIQwDqYnkKtQ5VCiSe17E33H0aQ/u+1GlRuz4LZBk6Mm3sg90G9hEbmEt37C1Zg=="], - - "@vitest/coverage-v8/magicast": ["magicast@0.5.3", "", { "dependencies": { "@babel/parser": "^7.29.3", "@babel/types": "^7.29.0", "source-map-js": "^1.2.1" } }, "sha512-pVKE4UdSQ7DvHzivsCIFx2BJn1mHG6KsyrFcaxFx6tONdneEuThrDx0Cj3AMg58KyN4pzYT+LHOotxDQDjNvkw=="], - "@vitest/expect/@vitest/utils": ["@vitest/utils@3.2.4", "", { "dependencies": { "@vitest/pretty-format": "3.2.4", "loupe": "^3.1.4", "tinyrainbow": "^2.0.0" } }, "sha512-fB2V0JFrQSMsCo9HiSq3Ezpdv4iYaXRG1Sx8edX3MwxfyNn83mKiGzOcH+Fkxt4MHxr3y42fQi1oeAInqgX2QA=="], "@vitest/expect/tinyrainbow": ["tinyrainbow@2.0.0", "", {}, "sha512-op4nsTR47R6p0vMUUoYl/a+ljLFVtlfaXkLQmqfLR1qHma1h/ysYk4hEXZ880bf2CYgTskvTa/e196Vd5dDQXw=="], @@ -6167,8 +6143,6 @@ "archiver-utils/is-stream": ["is-stream@2.0.1", "", {}, "sha512-hFoiJiTl63nn+kstHGBtewWSKnQLpyb155KHheA1l39uvtO9nWIop1p3udqPcUd/xbF1VLMO4n7OI6p7RbngDg=="], - "ast-v8-to-istanbul/js-tokens": ["js-tokens@10.0.0", "", {}, "sha512-lM/UBzQmfJRo9ABXbPWemivdCW8V2G8FHaHdypQaIy523snUjog0W71ayWXTjiR+ixeMyVHN2XcpnTd/liPg/Q=="], - "astro/@astrojs/internal-helpers": ["@astrojs/internal-helpers@0.6.1", "", {}, "sha512-l5Pqf6uZu31aG+3Lv8nl/3s4DbUzdlxTWDof4pEpto6GUJNhhCbelVi9dEyurOVyqaelwmS9oSyOWOENSfgo9A=="], "astro/acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="], @@ -6259,8 +6233,6 @@ "electron-winstaller/fs-extra": ["fs-extra@7.0.1", "", { "dependencies": { "graceful-fs": "^4.1.2", "jsonfile": "^4.0.0", "universalify": "^0.1.0" } }, "sha512-YJDaCJZEnBmcbw13fvdAM9AwNOJwOzrE4pqMqBq5nFiEqXUqHwlK4B+3pUw6JNvfSPtX05xFHtYy/1ni01eGCw=="], - "encoding/iconv-lite": ["iconv-lite@0.6.3", "", { "dependencies": { "safer-buffer": ">= 2.1.2 < 3.0.0" } }, "sha512-4fCk79wshMdzMp2rH06qWrJE4iolqLhCUH+OiuIgU++RB0+94NlDL81atO7GX55uUKueo0txHNtvEyI6D7WdMw=="], - "engine.io-client/ws": ["ws@8.20.1", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-It4dO0K5v//JtTXuPkfEOaI3uUN87iYPnqo/ZzqCoG3g8uhA66QUMs/SrM0YK7/NAu+r4LMh/9dq2A7k+rHs+w=="], "esast-util-from-js/acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="], @@ -6311,8 +6283,6 @@ "iconv-corefoundation/node-addon-api": ["node-addon-api@1.7.2", "", {}, "sha512-ibPK3iA+vaY1eEjESkQkM0BbCqFOaZMiXRTtdB0u7b4djtY6JnsjvPdUHVMg6xQt3B8fpTTWHI9A+ADjM9frzg=="], - "istanbul-reports/html-escaper": ["html-escaper@2.0.2", "", {}, "sha512-H2iMtd0I4Mt5eYiapRdIDjp+XzelXQ0tFE4JS7YFwFevXXMmOp9myNrUvCg0D6ws8iqkRPBfKHgbwig1SmlLfg=="], - "js-beautify/glob": ["glob@10.5.0", "", { "dependencies": { "foreground-child": "^3.1.0", "jackspeak": "^3.1.2", "minimatch": "^9.0.4", "minipass": "^7.1.2", "package-json-from-dist": "^1.0.0", "path-scurry": "^1.11.1" }, "bin": { "glob": "dist/esm/bin.mjs" } }, "sha512-DfXN8DfhJ7NH3Oe7cFmu3NCu1wKbkReJ8TorzSAFbSKrlNaQSKfIzqYqVY8zlbs2NLBbWpRiU52GX2PbaBVNkg=="], "js-beautify/nopt": ["nopt@7.2.1", "", { "dependencies": { "abbrev": "^2.0.0" }, "bin": { "nopt": "bin/nopt.js" } }, "sha512-taM24ViiimT/XntxbPyJQzCG+p4EKOpgD3mxFwW38mGjVUrfERQOeY4EDHjdnptttfHuHQXFx+lTP08Q+mLa/w=="], @@ -6911,18 +6881,12 @@ "@solidjs/start/shiki/@shikijs/types": ["@shikijs/types@1.29.2", "", { "dependencies": { "@shikijs/vscode-textmate": "^10.0.1", "@types/hast": "^3.0.4" } }, "sha512-VJjK0eIijTZf0QSTODEXCqinjBn0joAHQ+aPSBzrv4O2d/QSbsMw+ZeSRx03kV34Hy7NzUvV/7NqfYGRLrASmw=="], - "@standard-community/standard-json/effect/@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="], - - "@standard-community/standard-openapi/effect/@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="], - "@storybook/csf-plugin/unplugin/acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="], "@storybook/csf-plugin/unplugin/webpack-virtual-modules": ["webpack-virtual-modules@0.6.2", "", {}, "sha512-66/V2i5hQanC51vBQKPH4aI8NMAcBW59FVBs+rC7eGHupMyfn34q7rZIE+ETlJ+XTevqfUhVVBgSUNSW2flEUQ=="], "@tailwindcss/oxide-wasm32-wasi/@napi-rs/wasm-runtime/@tybys/wasm-util": ["@tybys/wasm-util@0.10.2", "", { "dependencies": { "tslib": "^2.4.0" } }, "sha512-RoBvJ2X0wuKlWFIjrwffGw1IqZHKQqzIchKaadZZfnNpsAYp2mM0h36JtPCjNDAHGgYez/15uMBpfGwchhiMgg=="], - "@vitest/coverage-v8/@vitest/utils/@vitest/pretty-format": ["@vitest/pretty-format@4.1.8", "", { "dependencies": { "tinyrainbow": "^3.1.0" } }, "sha512-9GasEBxpZ1VYIpqHf/0+YGg121uSNwCKOJqIrTwWP/TB7DmFCiaBpNl3aPZzoLWfWkuqhbH8vJIVobZkvdo2cA=="], - "@vitest/expect/@vitest/utils/@vitest/pretty-format": ["@vitest/pretty-format@3.2.4", "", { "dependencies": { "tinyrainbow": "^2.0.0" } }, "sha512-IVNZik8IVRJRTr9fxlitMKeJeXFFFN0JaB9PHPGQ8NKQbGpfjlTx9zO4RefN8gp7eqjNy8nyK3NZmBzOPeIxtA=="], "accepts/mime-types/mime-db": ["mime-db@1.52.0", "", {}, "sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg=="], diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 01d2ff8bcb8e..5c7291773fa3 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -86,9 +86,7 @@ const layer = Layer.effect( pending: new Map(), } - // Rows in question_request intentionally survive this finalizer so a - // restarted process can still resolve replies to questions it no - // longer has waiters for. + // Rows intentionally survive shutdown so a restarted process can still resolve replies. yield* Effect.addFinalizer(() => Effect.gen(function* () { for (const item of state.pending.values()) { @@ -138,21 +136,25 @@ const layer = Layer.effect( ) }) + // Deleting the row is the claim: whichever concurrent reply/reject wins the delete owns the request. + const claim = Effect.fn("Question.claim")(function* (requestID: QuestionID) { + return yield* db + .delete(QuestionRequestTable) + .where(eq(QuestionRequestTable.id, requestID)) + .returning() + .get() + .pipe(Effect.orDie) + }) + const reply = Effect.fn("Question.reply")(function* (input: { requestID: QuestionID answers: ReadonlyArray }) { - const row = yield* db - .select() - .from(QuestionRequestTable) - .where(eq(QuestionRequestTable.id, input.requestID)) - .get() - .pipe(Effect.orDie) + const row = yield* claim(input.requestID) if (!row) { yield* Effect.logWarning("reply for unknown request", { requestID: input.requestID }) return yield* new NotFoundError({ requestID: input.requestID }) } - yield* db.delete(QuestionRequestTable).where(eq(QuestionRequestTable.id, row.id)).run().pipe(Effect.orDie) const request = rowToRequest(row) yield* Effect.logInfo("replied", { requestID: input.requestID, answers: input.answers }) yield* events.publish(Event.Replied, { @@ -171,8 +173,7 @@ const layer = Layer.effect( return { outcome: "resolved" } as const }) - const rejectRow = Effect.fn("Question.rejectRow")(function* (row: QuestionRequestRow) { - yield* db.delete(QuestionRequestTable).where(eq(QuestionRequestTable.id, row.id)).run().pipe(Effect.orDie) + const rejectClaimed = Effect.fn("Question.rejectClaimed")(function* (row: QuestionRequestRow) { yield* Effect.logInfo("rejected", { requestID: row.id }) yield* events.publish(Event.Rejected, { sessionID: row.session_id, @@ -186,27 +187,22 @@ const layer = Layer.effect( }) const reject = Effect.fn("Question.reject")(function* (requestID: QuestionID) { - const row = yield* db - .select() - .from(QuestionRequestTable) - .where(eq(QuestionRequestTable.id, requestID)) - .get() - .pipe(Effect.orDie) + const row = yield* claim(requestID) if (!row) { yield* Effect.logWarning("reject for unknown request", { requestID }) return yield* new NotFoundError({ requestID }) } - yield* rejectRow(row) + yield* rejectClaimed(row) }) const rejectAllForSession = Effect.fn("Question.rejectAllForSession")(function* (sessionID: SessionID) { const rows = yield* db - .select() - .from(QuestionRequestTable) + .delete(QuestionRequestTable) .where(eq(QuestionRequestTable.session_id, sessionID)) + .returning() .all() .pipe(Effect.orDie) - yield* Effect.forEach(rows, rejectRow, { discard: true }) + yield* Effect.forEach(rows, rejectClaimed, { discard: true }) }) const list = Effect.fn("Question.list")(function* () { diff --git a/packages/opencode/src/question/resume.ts b/packages/opencode/src/question/resume.ts index 239cc3c67eaf..11a1b983e2eb 100644 --- a/packages/opencode/src/question/resume.ts +++ b/packages/opencode/src/question/resume.ts @@ -5,12 +5,7 @@ import { SessionStatus } from "@/session/status" import { formatAnswerOutput } from "./format" import type { Answer, Request } from "." -/** - * Continues a session after a reply arrived for a question whose in-process - * waiter died with a restart. The persisted request lets us complete the - * interrupted tool call with the real answers and re-enter the assistant loop - * as if the answer had resolved normally. - */ +/** Completes a restart-orphaned question's tool call with the real answers and re-enters the assistant loop. */ export const resumeOrphanedReply = Effect.fn("Question.resumeOrphanedReply")(function* (input: { request: Request answers: ReadonlyArray diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts index 69792a6db22d..b6c0d4e0086d 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts @@ -9,8 +9,7 @@ import { QuestionNotFoundError } from "../errors" export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question", (handlers) => Effect.gen(function* () { const svc = yield* Question.Service - // Build-time scope outlives any single request, so a resume forked into it - // keeps running after the reply endpoint has already responded. + // Build-time scope: resumes forked into it outlive the reply request. const scope = yield* Scope.Scope const list = Effect.fn("QuestionHttpApi.list")(function* () { diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 3222a1d0d3cc..69dca1211a83 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -154,8 +154,7 @@ const layer = Layer.effect( const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) { yield* Effect.logInfo("cancel", { "session.id": sessionID }) yield* state.cancel(sessionID) - // Stop must clear persisted question rows so a docked ask can't outlive - // the turn it belonged to; process shutdown deliberately does not. + // Stop clears persisted question rows; process shutdown deliberately does not. yield* questions.rejectAllForSession(sessionID) }) From 2ce39913f26266e9f82a1cb1eede036d046852f5 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 15:30:13 -0600 Subject: [PATCH 5/7] fix(opencode): harden question claim scoping and lost-waiter replies Claiming a question row now requires the row's session to belong to the instance's project (same scoping as list), so co-located instances sharing the machine-global DB cannot consume each other's live requests. A reply whose Deferred.succeed loses to the instance-disposal finalizer now falls through to the orphaned outcome instead of reporting resolved for a waiter that was already failed. Co-Authored-By: Claude Fable 5 --- packages/opencode/src/question/index.ts | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 5c7291773fa3..97669392cc20 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -1,6 +1,6 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Deferred, Effect, Layer, Schema, Context } from "effect" -import { asc, eq } from "drizzle-orm" +import { and, asc, eq, inArray } from "drizzle-orm" import { Database } from "@opencode-ai/core/database/database" import { QuestionRequestTable, SessionTable } from "@opencode-ai/core/session/sql" import { InstanceState } from "@/effect/instance-state" @@ -137,10 +137,21 @@ const layer = Layer.effect( }) // Deleting the row is the claim: whichever concurrent reply/reject wins the delete owns the request. + // Scoped to this instance's project (like list) so co-located instances sharing the global DB + // cannot consume each other's live requests. const claim = Effect.fn("Question.claim")(function* (requestID: QuestionID) { + const ctx = yield* InstanceState.context return yield* db .delete(QuestionRequestTable) - .where(eq(QuestionRequestTable.id, requestID)) + .where( + and( + eq(QuestionRequestTable.id, requestID), + inArray( + QuestionRequestTable.session_id, + db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.project_id, ctx.project.id)), + ), + ), + ) .returning() .get() .pipe(Effect.orDie) @@ -169,7 +180,12 @@ const layer = Layer.effect( return { outcome: "orphaned", request } as const } pending.delete(input.requestID) - yield* Deferred.succeed(existing.deferred, input.answers) + const delivered = yield* Deferred.succeed(existing.deferred, input.answers) + if (!delivered) { + // Instance disposal already failed this waiter; the answers still need the orphaned heal path. + yield* Effect.logInfo("reply raced instance disposal, treating as orphaned", { requestID: input.requestID }) + return { outcome: "orphaned", request } as const + } return { outcome: "resolved" } as const }) From ba777bf9d4b9ebf38745d3bc0eaa47c335306812 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 15:30:13 -0600 Subject: [PATCH 6/7] test(opencode): cover the orphaned question resume path Three prompt-harness tests for resumeOrphanedReply: the heal completes the dangling tool part with the shared answer format and re-enters the loop, busy sessions complete the part but skip the loop, and a missing tool part is tolerated. Co-Authored-By: Claude Fable 5 --- packages/opencode/test/session/prompt.test.ts | 97 ++++++++++++++++++- 1 file changed, 96 insertions(+), 1 deletion(-) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 86ba20e98561..818035e4275c 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -6,7 +6,7 @@ import { SessionProjector } from "@opencode-ai/core/session/projector" import { eq } from "drizzle-orm" import { EventV2Bridge } from "@/event-v2-bridge" import { expect } from "bun:test" -import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect" +import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Option } from "effect" import path from "path" import { fileURLToPath } from "url" import { NamedError } from "@opencode-ai/core/util/error" @@ -24,6 +24,9 @@ import { Git } from "../../src/git" import { Image } from "../../src/image/image" import { Question } from "../../src/question" +import { formatAnswerOutput } from "../../src/question/format" +import { resumeOrphanedReply } from "../../src/question/resume" +import { QuestionID } from "../../src/question/schema" import { Todo } from "../../src/session/todo" import { Session } from "@/session/session" import { SessionMessageTable } from "@opencode-ai/core/session/sql" @@ -2401,3 +2404,95 @@ noLLMServer.instance( }), 30_000, ) + +// Orphaned question resume + +const questionRequest = (input: { sessionID: SessionID; messageID: MessageID; callID: string }): Question.Request => ({ + id: QuestionID.ascending(), + sessionID: input.sessionID, + questions: [ + { + question: "Resume me?", + header: "Resume", + options: [{ label: "Yes", description: "Yes" }], + }, + ], + tool: { messageID: input.messageID, callID: input.callID }, +}) + +const seedDanglingQuestion = Effect.fn("test.seedDanglingQuestion")(function* (input: { + sessionID: SessionID + messageID: MessageID + callID: string +}) { + const sessions = yield* Session.Service + yield* sessions.updatePart({ + id: PartID.ascending(), + messageID: input.messageID, + sessionID: input.sessionID, + type: "tool", + callID: input.callID, + tool: "question", + state: { status: "running", input: {}, time: { start: 1 } }, + }) +}) + +const questionToolPart = Effect.fn("test.questionToolPart")(function* (sessionID: SessionID, messageID: MessageID) { + const sessions = yield* Session.Service + const message = yield* sessions.findMessage(sessionID, (msg) => msg.info.id === messageID) + return completedTool(Option.isSome(message) ? [...message.value.parts] : []) +}) + +it.instance("orphaned reply heals the dangling question part and re-enters the loop", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Pinned" }) + const seeded = yield* seed(chat.id) + yield* seedDanglingQuestion({ sessionID: chat.id, messageID: seeded.assistant.id, callID: "que-call" }) + yield* llm.text("resumed") + + const request = questionRequest({ sessionID: chat.id, messageID: seeded.assistant.id, callID: "que-call" }) + yield* resumeOrphanedReply({ request, answers: [["Yes"]] }) + + const part = yield* questionToolPart(chat.id, seeded.assistant.id) + expect(part?.state.output).toBe(formatAnswerOutput({ questions: request.questions, answers: [["Yes"]] })) + expect(part?.state.title).toBe("Asked 1 question") + expect(part?.state.metadata).toEqual({ answers: [["Yes"]] }) + expect(yield* llm.hits).toHaveLength(1) + }), +) + +it.instance("orphaned reply resume completes the part but skips the loop while the session is busy", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const chat = yield* sessions.create({ title: "Pinned" }) + const seeded = yield* seed(chat.id) + yield* seedDanglingQuestion({ sessionID: chat.id, messageID: seeded.assistant.id, callID: "que-call" }) + yield* status.set(chat.id, { type: "busy" }) + + const request = questionRequest({ sessionID: chat.id, messageID: seeded.assistant.id, callID: "que-call" }) + yield* resumeOrphanedReply({ request, answers: [["Yes"]] }) + + const part = yield* questionToolPart(chat.id, seeded.assistant.id) + expect(part?.state.output).toBe(formatAnswerOutput({ questions: request.questions, answers: [["Yes"]] })) + expect(yield* llm.hits).toHaveLength(0) + }), +) + +it.instance("orphaned reply resume tolerates a missing tool part and still re-enters the loop", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Pinned" }) + const seeded = yield* seed(chat.id) + yield* llm.text("resumed") + + const request = questionRequest({ sessionID: chat.id, messageID: seeded.assistant.id, callID: "no-such-call" }) + yield* resumeOrphanedReply({ request, answers: [["Yes"]] }) + + expect(yield* llm.hits).toHaveLength(1) + }), +) From c02b392ba6834c64bb9079e1f21dbeba4d9e745b Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 20 Aug 2026 15:30:13 -0600 Subject: [PATCH 7/7] ci: gate core schema migrations in fork-ci packages/core's database-migration tests (snapshot parity --check plus the migration corpus) now run in the fork-ci test job; nothing gated the schema artifacts before this. Co-Authored-By: Claude Fable 5 --- .github/workflows/fork-ci.yml | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/.github/workflows/fork-ci.yml b/.github/workflows/fork-ci.yml index 2b61bd69f139..7df5c90ef690 100644 --- a/.github/workflows/fork-ci.yml +++ b/.github/workflows/fork-ci.yml @@ -1,7 +1,8 @@ # Fork CI. The inherited upstream workflows (test.yml, typecheck.yml, ...) all # target blacksmith-* self-hosted runners this fork does not have, so they never # run. This is the fork's own gate on GitHub-hosted runners, scoped to -# packages/opencode — where every fork patch lands. +# packages/opencode — where every fork patch lands — plus packages/core's +# schema migration gates. name: fork-ci on: @@ -56,3 +57,7 @@ jobs: - name: HttpApi exerciser gates working-directory: packages/opencode run: bun run test:httpapi + + - name: Core schema migration gates + working-directory: packages/core + run: bun test test/database-migration.test.ts