ci-agent-runner/app/src/application/execution/worker.ts

import { mkdir, rm } from "node:fs/promises";
import { join } from "node:path";
import { checkoutTrustedRevision } from "../../adapters/git/repository/checkout.js";
import { checkoutPullRequestRevision } from "../../adapters/git/repository/pull.js";
import { GiteaClient } from "../../adapters/gitea/client/client.js";
import {
    formatError,
    log,
    protocolVersion,
    type Result,
} from "../../core/contracts.js";
import {
    type JobExecutionInput,
    runCheckedOutJob,
    runPullReviewJob,
} from "./jobs/execute.js";

const leaseMs = 30_000;

export async function executeJob(input: JobExecutionInput): Promise<void> {
    const controller = new AbortController();
    const signal = AbortSignal.any([input.shutdown, controller.signal]);
    const monitor = setInterval(() => {
        if (input.store.isCancelRequested(input.job.id))
            controller.abort(new Error("Agent job was cancelled"));
        else if (!input.store.heartbeat(input.job.id, input.worker, leaseMs))
            controller.abort(new Error("Agent job lease was lost"));
    }, 5_000);
    monitor.unref();
    try {
        const stableWorkspace = join(
            input.workspaceRoot,
            String(input.job.repositoryId),
            String(input.job.issueNumber),
        );
        const workspace =
            input.job.attempts === 1
                ? stableWorkspace
                : `${stableWorkspace}-recovery-${input.job.id}-${input.job.attempts}`;
        await rm(workspace, { recursive: true, force: true });
        await mkdir(workspace, { recursive: true, mode: 0o700 });
        input.store.setWorkspace(input.job.id, input.worker, workspace);
        const client = new GiteaClient(
            input.serverUrl,
            input.readToken,
            input.repository.owner,
            input.repository.repo,
            signal,
        );
        const repository = await client.getRepository();
        if (repository.id !== input.job.repositoryId)
            throw new Error("Configured repository identity changed");
        if (input.job.kind === "pull-review") {
            const pull = await client.getPullRequest(input.job.targetNumber);
            if (pull.state !== "open" || pull.base.repo_id !== repository.id)
                throw new Error(
                    "Review requires an open pull request in this repository",
                );
            await checkoutPullRequestRevision({
                workspace,
                serverUrl: input.serverUrl,
                baseRepository: repository.full_name,
                pullRequestNumber: pull.number,
                headSha: pull.head.sha,
                mergeBaseSha: pull.merge_base,
                readToken: input.readToken,
                signal,
            });
            const output = await runPullReviewJob(
                input,
                client,
                workspace,
                pull,
                signal,
            );
            finish(input, output.result);
            return;
        }
        const base = await client.getBranch(repository.default_branch);
        if (!base)
            throw new Error(
                `Default branch ${repository.default_branch} was not found`,
            );
        await checkoutTrustedRevision({
            workspace,
            serverUrl: input.serverUrl,
            repository: `${input.repository.owner}/${input.repository.repo}`,
            sha: base.commit.id,
            readToken: input.readToken,
            signal,
        });
        const output = await runCheckedOutJob(input, client, workspace, signal);
        finish(input, output.result);
    } catch (error) {
        await handleFailure(input, signal, error);
    } finally {
        clearInterval(monitor);
    }
}

function finish(input: JobExecutionInput, result: Result): void {
    input.store.finishExecution(input.job.id, input.worker, result);
    logCompletion(input, result);
}

function logCompletion(input: JobExecutionInput, result: Result): void {
    console.log(
        log("info", "Execution completed", {
            jobId: input.job.id,
            mode: input.job.mode,
            status: result.status,
        }),
    );
}

async function handleFailure(
    input: JobExecutionInput,
    signal: AbortSignal,
    error: unknown,
): Promise<void> {
    if (
        input.shutdown.aborted ||
        !input.store.ownsLease(input.job.id, input.worker)
    ) {
        console.log(
            log("info", "Execution interrupted; lease will be recovered", {
                jobId: input.job.id,
            }),
        );
        return;
    }
    const result: Result = {
        version: protocolVersion,
        mode: input.job.mode,
        status: "failed",
        message: signal.aborted
            ? "Agent job was cancelled or interrupted"
            : formatError(error),
    };
    try {
        input.store.finishExecution(input.job.id, input.worker, result);
    } catch (finishError) {
        console.error(
            log("error", "Execution finalization failed", {
                jobId: input.job.id,
                error: formatError(finishError),
            }),
        );
    }
    console.error(
        log("error", "Execution failed", {
            jobId: input.job.id,
            error: formatError(error),
        }),
    );
}