fix: handle long-running executor prompts

2026-07-18 17:24:00 +02:00
parent a05975dec5
commit 57fe9eafa1
10 changed files with 304 additions and 60 deletions
@@ -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",
@@ -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",
@@ -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<string> {
-        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<string> {
-        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<string, unknown>,
        signal?: AbortSignal,
    ): Promise<unknown> {
-        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) {
@@ -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<typeof createOpencodeClient>;
+
+export async function withRequestContext<T>(
+    message: string,
+    operation: () => Promise<T>,
+): Promise<T> {
+    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<T>(
+    operation: (customFetch: typeof fetch) => Promise<T>,
+): Promise<T> {
+    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<T>(input: {
+    timeout: number;
+    signal?: AbortSignal;
+    work: (signal: AbortSignal) => Promise<T>;
+    stop: () => Promise<void>;
+    finish: () => Promise<T>;
+}): Promise<T> {
+    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<string, unknown>;
+    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,
+            ),
+    });
+}
@@ -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, unknown>,
-): string {
-    return JSON.stringify({
-        level,
-        message,
-        ...fields,
-        time: new Date().toISOString(),
-    });
-}
-
 function verifyIdentity(
    repository: WebhookRepository,
    id: number,
@@ -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<void> {
@@ -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<void> {
    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<void> {
                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<void> {
    } 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<void> {
 }

 main().catch((error) => {
-    console.error(formatError(error));
+    console.error(
+        log("error", "Executor failed", { error: formatError(error) }),
+    );
    process.exitCode = 1;
 });
@@ -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<typeof executeJob>[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),
        }),
    );
 }
@@ -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, unknown>,
+): string {
+    return JSON.stringify({
+        level,
+        message,
+        ...fields,
+        time: new Date().toISOString(),
+    });
+}
+
 export type Mode = "plan" | "implement";

 export interface PlanData {
@@ -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<string>((_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"]);
+});