From 57fe9eafa1f9a07661641aea01aebb925fff89b3 Mon Sep 17 00:00:00 2001 From: StanPonomarev Date: Sat, 18 Jul 2026 17:24:00 +0200 Subject: [PATCH] fix: handle long-running executor prompts --- app/package-lock.json | 12 +- app/package.json | 3 +- app/src/adapters/opencode/runner.ts | 69 +++++---- app/src/adapters/opencode/transport.ts | 146 ++++++++++++++++++ .../controller/handlers/workers.ts | 14 +- app/src/application/controller/main.ts | 4 +- app/src/application/execution/main.ts | 22 ++- app/src/application/execution/worker.ts | 32 ++-- app/src/core/contracts.ts | 13 ++ app/src/tests/adapters/opencode.test.ts | 49 ++++++ 10 files changed, 304 insertions(+), 60 deletions(-) create mode 100644 app/src/adapters/opencode/transport.ts create mode 100644 app/src/tests/adapters/opencode.test.ts diff --git a/app/package-lock.json b/app/package-lock.json index bb7240a..0b982dc 100644 --- a/app/package-lock.json +++ b/app/package-lock.json @@ -8,7 +8,8 @@ "name": "gitea-agent-server", "version": "1.0.0", "dependencies": { - "@opencode-ai/sdk": "1.17.18" + "@opencode-ai/sdk": "1.17.18", + "undici": "8.7.0" }, "devDependencies": { "@biomejs/biome": "2.5.3", @@ -277,6 +278,15 @@ "node": ">=14.17" } }, + "node_modules/undici": { + "version": "8.7.0", + "resolved": "https://registry.npmjs.org/undici/-/undici-8.7.0.tgz", + "integrity": "sha512-N7iQtfyLhIMOFgQubvmLV26svHpO0bqKnAiWotTQCVKCmWrcGbBotPuW1x+xwYZ2VHdSTVUfPQQnlEt1/LouTQ==", + "license": "MIT", + "engines": { + "node": ">=22.19.0" + } + }, "node_modules/undici-types": { "version": "7.16.0", "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.16.0.tgz", diff --git a/app/package.json b/app/package.json index e007553..0213f8b 100644 --- a/app/package.json +++ b/app/package.json @@ -19,7 +19,8 @@ "node": ">=24.13" }, "dependencies": { - "@opencode-ai/sdk": "1.17.18" + "@opencode-ai/sdk": "1.17.18", + "undici": "8.7.0" }, "devDependencies": { "@biomejs/biome": "2.5.3", diff --git a/app/src/adapters/opencode/runner.ts b/app/src/adapters/opencode/runner.ts index 0f79d64..4110d06 100644 --- a/app/src/adapters/opencode/runner.ts +++ b/app/src/adapters/opencode/runner.ts @@ -4,6 +4,7 @@ import { createOpencode, createOpencodeClient, } from "@opencode-ai/sdk/v2"; +import { promptWithBudget, withRequestContext } from "./transport.js"; const startupAttempts = 10; const promptTimeout = 20 * 60_000; @@ -44,7 +45,7 @@ export class OpenCodeRunner { this.startupAbort = controller; this.starting = starting; try { - await starting; + await withRequestContext("OpenCode startup failed", () => starting); } finally { if (this.starting === starting) this.starting = undefined; if (this.startupAbort === controller) this.startupAbort = undefined; @@ -85,15 +86,20 @@ export class OpenCodeRunner { title: string, signal?: AbortSignal, ): Promise { - if (!this.client) throw new Error("OpenCode is not started"); + const client = this.client; + if (!client) throw new Error("OpenCode is not started"); const requestSignal = this.operationSignal(signal); - const result = await this.client.session.create( - { - directory: this.workspace, - title, - agent, - }, - requestSignal ? { signal: requestSignal } : undefined, + const result = await withRequestContext( + `OpenCode ${agent} session creation failed`, + () => + client.session.create( + { + directory: this.workspace, + title, + agent, + }, + requestSignal ? { signal: requestSignal } : undefined, + ), ); if (!result.data) throw new Error("OpenCode returned no session data"); return result.data.id; @@ -105,15 +111,20 @@ export class OpenCodeRunner { title: string, signal?: AbortSignal, ): Promise { - if (!this.client) throw new Error("OpenCode is not started"); + const client = this.client; + if (!client) throw new Error("OpenCode is not started"); if (!existingId) return this.createSession(agent, title, signal); const requestSignal = this.operationSignal(signal); - const result = await this.client.session.get( - { sessionID: existingId, directory: this.workspace }, - requestSignal - ? { signal: requestSignal, throwOnError: false } - : { throwOnError: false }, + const result = await withRequestContext( + `OpenCode ${agent} session lookup failed`, + () => + client.session.get( + { sessionID: existingId, directory: this.workspace }, + requestSignal + ? { signal: requestSignal, throwOnError: false } + : { throwOnError: false }, + ), ); if (!result.data) { if (result.response.status === 404) @@ -142,23 +153,19 @@ export class OpenCodeRunner { schema: Record, signal?: AbortSignal, ): Promise { - if (!this.client) throw new Error("OpenCode is not started"); + const client = this.client; + if (!client) throw new Error("OpenCode is not started"); const callerSignal = this.operationSignal(signal); - const timeoutSignal = AbortSignal.timeout(promptTimeout); - const promptSignal = callerSignal - ? AbortSignal.any([callerSignal, timeoutSignal]) - : timeoutSignal; - const request = this.client.session.prompt( - { - sessionID, - directory: this.workspace, - agent, - parts: [{ type: "text", text }], - format: { type: "json_schema", schema, retryCount: 2 }, - }, - { signal: promptSignal }, - ); - const result = await request; + const result = await promptWithBudget({ + client, + sessionID, + directory: this.workspace, + agent, + text, + schema, + timeout: promptTimeout, + ...(callerSignal ? { signal: callerSignal } : {}), + }); if (!result.data) throw new Error("OpenCode returned no prompt data"); const info = result.data.info as AssistantMessage; if (info.error) { diff --git a/app/src/adapters/opencode/transport.ts b/app/src/adapters/opencode/transport.ts new file mode 100644 index 0000000..2aa71db --- /dev/null +++ b/app/src/adapters/opencode/transport.ts @@ -0,0 +1,146 @@ +import type { createOpencodeClient } from "@opencode-ai/sdk/v2"; +import { Agent, type Dispatcher } from "undici"; +import { log } from "../../core/contracts.js"; + +type OpenCodeClient = ReturnType; + +export async function withRequestContext( + message: string, + operation: () => Promise, +): Promise { + const startedAt = Date.now(); + try { + return await operation(); + } catch (error) { + throw new Error(`${message} after ${Date.now() - startedAt}ms`, { + cause: error, + }); + } +} + +export async function withPromptTransport( + operation: (customFetch: typeof fetch) => Promise, +): Promise { + const dispatcher = new Agent({ + connectTimeout: 30_000, + headersTimeout: 0, + bodyTimeout: 0, + }); + try { + return await operation((input, init) => + fetch(input, { + ...init, + dispatcher, + } as RequestInit & { dispatcher: Dispatcher }), + ); + } finally { + await dispatcher.close(); + } +} + +export async function withWorkBudget(input: { + timeout: number; + signal?: AbortSignal; + work: (signal: AbortSignal) => Promise; + stop: () => Promise; + finish: () => Promise; +}): Promise { + const controller = new AbortController(); + const signal = input.signal + ? AbortSignal.any([input.signal, controller.signal]) + : controller.signal; + const work = input.work(signal); + let timer: NodeJS.Timeout | undefined; + try { + const outcome = await Promise.race([ + work.then((result) => ({ kind: "done" as const, result })), + new Promise<{ kind: "budget" }>((resolve) => { + timer = setTimeout( + () => resolve({ kind: "budget" }), + input.timeout, + ); + timer.unref(); + }), + ]); + if (outcome.kind === "done") return outcome.result; + try { + await input.stop(); + } finally { + controller.abort(new Error("OpenCode work budget reached")); + await work.catch(() => undefined); + } + input.signal?.throwIfAborted(); + return await input.finish(); + } finally { + if (timer) clearTimeout(timer); + } +} + +export function promptWithBudget(input: { + client: OpenCodeClient; + sessionID: string; + directory: string; + agent: string; + text: string; + schema: Record; + timeout: number; + signal?: AbortSignal; +}) { + const prompt = (text: string, signal?: AbortSignal) => + withPromptTransport((customFetch) => + withRequestContext(`OpenCode ${input.agent} prompt failed`, () => + input.client.session.prompt( + { + sessionID: input.sessionID, + directory: input.directory, + agent: input.agent, + parts: [{ type: "text", text }], + format: { + type: "json_schema", + schema: input.schema, + retryCount: 2, + }, + }, + { + fetch: customFetch, + ...(signal ? { signal } : {}), + }, + ), + ), + ); + return withWorkBudget({ + timeout: input.timeout, + ...(input.signal ? { signal: input.signal } : {}), + work: (signal) => prompt(input.text, signal), + stop: async () => { + console.log( + log( + "info", + "OpenCode work budget reached; requesting final response", + { agent: input.agent }, + ), + ); + await withRequestContext( + `OpenCode ${input.agent} work-budget stop failed`, + async () => { + const stopped = await input.client.session.abort( + { + sessionID: input.sessionID, + directory: input.directory, + }, + input.signal ? { signal: input.signal } : undefined, + ); + if (!stopped.data) + throw new Error( + "OpenCode did not confirm session stop", + ); + }, + ); + }, + finish: () => + prompt( + "The work budget has been reached. Do not use tools or perform further investigation. Using the persisted work so far, return the best valid structured response now.", + input.signal, + ), + }); +} diff --git a/app/src/application/controller/handlers/workers.ts b/app/src/application/controller/handlers/workers.ts index d98b2ea..4af6c64 100644 --- a/app/src/application/controller/handlers/workers.ts +++ b/app/src/application/controller/handlers/workers.ts @@ -1,6 +1,7 @@ import { rm } from "node:fs/promises"; import type { AgentStore } from "../../../adapters/database/store.js"; import type { ActorPolicy } from "../../../core/config.js"; +import { log } from "../../../core/contracts.js"; import { parseCommentPayload, parseLabelPayload, @@ -160,19 +161,6 @@ export async function pumpOutbox( } } -export function log( - level: string, - message: string, - fields: Record, -): string { - return JSON.stringify({ - level, - message, - ...fields, - time: new Date().toISOString(), - }); -} - function verifyIdentity( repository: WebhookRepository, id: number, diff --git a/app/src/application/controller/main.ts b/app/src/application/controller/main.ts index 9284aac..6913de5 100644 --- a/app/src/application/controller/main.ts +++ b/app/src/application/controller/main.ts @@ -9,9 +9,9 @@ import { repositoryParts, validateServerUrl, } from "../../core/config.js"; -import { formatError, requireEnv } from "../../core/contracts.js"; +import { formatError, log, requireEnv } from "../../core/contracts.js"; import { type PublicationContext, safeFailure } from "../publication/status.js"; -import { log, pumpDeliveries, pumpOutbox } from "./handlers/workers.js"; +import { pumpDeliveries, pumpOutbox } from "./handlers/workers.js"; import { handleHttp, observeHttpRequest } from "./server.js"; async function main(): Promise { diff --git a/app/src/application/execution/main.ts b/app/src/application/execution/main.ts index 4d647e0..58ad34d 100644 --- a/app/src/application/execution/main.ts +++ b/app/src/application/execution/main.ts @@ -6,7 +6,7 @@ import { repositoryParts, validateServerUrl, } from "../../core/config.js"; -import { formatError, requireEnv } from "../../core/contracts.js"; +import { formatError, log, requireEnv } from "../../core/contracts.js"; import { executeJob } from "./worker.js"; const leaseMs = 30_000; @@ -28,6 +28,12 @@ async function main(): Promise { const worker = `executor-${randomUUID()}`; process.env.GITEA_READ_TOKEN = readToken; process.env.GITEA_SERVER_URL = serverUrl; + console.log( + log("info", "Executor ready", { + worker, + repository: `${repository.owner}/${repository.repo}`, + }), + ); try { while (!shutdown.signal.aborted) { const job = store.leaseJob(worker, leaseMs); @@ -35,6 +41,14 @@ async function main(): Promise { await delay(500, shutdown.signal); continue; } + console.log( + log("info", "Execution started", { + jobId: job.id, + mode: job.mode, + issueNumber: job.issueNumber, + attempt: job.attempts, + }), + ); await executeJob({ store, job, @@ -50,6 +64,8 @@ async function main(): Promise { } catch (error) { if (!shutdown.signal.aborted) throw error; } finally { + if (shutdown.signal.aborted) + console.log(log("info", "Executor stopped", { worker })); store.close(); } } @@ -70,6 +86,8 @@ function delay(milliseconds: number, signal: AbortSignal): Promise { } main().catch((error) => { - console.error(formatError(error)); + console.error( + log("error", "Executor failed", { error: formatError(error) }), + ); process.exitCode = 1; }); diff --git a/app/src/application/execution/worker.ts b/app/src/application/execution/worker.ts index 651c8e9..6280b6e 100644 --- a/app/src/application/execution/worker.ts +++ b/app/src/application/execution/worker.ts @@ -6,6 +6,7 @@ import { GiteaClient } from "../../adapters/gitea/client/client.js"; import { findAcceptedPlan } from "../../adapters/gitea/issues.js"; import { formatError, + log, protocolVersion, type Result, } from "../../core/contracts.js"; @@ -117,6 +118,7 @@ async function executePlan( }, }); input.store.finishExecution(input.job.id, input.worker, output.result); + logCompletion(input, output.result); } async function executeImplementation( @@ -162,6 +164,20 @@ async function executeImplementation( }, }); input.store.finishExecution(input.job.id, input.worker, output.result); + logCompletion(input, output.result); +} + +function logCompletion( + input: Parameters[0], + result: Result, +): void { + console.log( + log("info", "Execution completed", { + jobId: input.job.id, + mode: input.job.mode, + status: result.status, + }), + ); } async function handleFailure( @@ -173,11 +189,9 @@ async function handleFailure( input.shutdown.aborted || !input.store.ownsLease(input.job.id, input.worker) ) { - console.error( - JSON.stringify({ - level: "info", + console.log( + log("info", "Execution interrupted; lease will be recovered", { jobId: input.job.id, - message: "Execution interrupted; lease will be recovered", }), ); return; @@ -194,18 +208,16 @@ async function handleFailure( input.store.finishExecution(input.job.id, input.worker, result); } catch (finishError) { console.error( - JSON.stringify({ - level: "error", + log("error", "Execution finalization failed", { jobId: input.job.id, - message: formatError(finishError), + error: formatError(finishError), }), ); } console.error( - JSON.stringify({ - level: "error", + log("error", "Execution failed", { jobId: input.job.id, - message: formatError(error), + error: formatError(error), }), ); } diff --git a/app/src/core/contracts.ts b/app/src/core/contracts.ts index 4bc0672..19ee2f2 100644 --- a/app/src/core/contracts.ts +++ b/app/src/core/contracts.ts @@ -7,6 +7,19 @@ export const generatedLabel = "agent:generated"; export const blockedLabel = "agent:blocked"; export const planReadyLabel = "agent:plan-ready"; +export function log( + level: string, + message: string, + fields: Record, +): string { + return JSON.stringify({ + level, + message, + ...fields, + time: new Date().toISOString(), + }); +} + export type Mode = "plan" | "implement"; export interface PlanData { diff --git a/app/src/tests/adapters/opencode.test.ts b/app/src/tests/adapters/opencode.test.ts new file mode 100644 index 0000000..29918aa --- /dev/null +++ b/app/src/tests/adapters/opencode.test.ts @@ -0,0 +1,49 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { withWorkBudget } from "../../adapters/opencode/transport.js"; + +test("returns completed OpenCode work without finalization", async () => { + let stopped = false; + let finalized = false; + const result = await withWorkBudget({ + timeout: 1_000, + work: async () => "complete", + stop: async () => { + stopped = true; + }, + finish: async () => { + finalized = true; + return "final"; + }, + }); + assert.equal(result, "complete"); + assert.equal(stopped, false); + assert.equal(finalized, false); +}); + +test("stops over-budget OpenCode work before requesting a final response", async () => { + const events: string[] = []; + const result = await withWorkBudget({ + timeout: 1, + work: (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener( + "abort", + () => { + events.push("settled"); + reject(signal.reason); + }, + { once: true }, + ); + }), + stop: async () => { + events.push("stopped"); + }, + finish: async () => { + events.push("finalized"); + return "final"; + }, + }); + assert.equal(result, "final"); + assert.deepEqual(events, ["stopped", "settled", "finalized"]); +});