Restore run.ts from main
This commit is contained in:
Danny Kim
+391
-49
@@ -1,28 +1,40 @@
|
||||
// Runs one ordered agent profile. Credential decryption and reviewer posting
|
||||
// live in separate processes so neither privileged secret reaches this runner.
|
||||
import { chmod, readFile, writeFile } from "node:fs/promises";
|
||||
import { chmod, chown, readFile, writeFile } from "node:fs/promises";
|
||||
import process from "node:process";
|
||||
import { createInterface } from "node:readline";
|
||||
import type { Readable } from "node:stream";
|
||||
import { text as readText } from "node:stream/consumers";
|
||||
import { setTimeout as sleep } from "node:timers/promises";
|
||||
|
||||
import { codexAuthentication, persistCodexAuth, stageCodexAuth } from "./auth.ts";
|
||||
import { runAgent } from "./agents.ts";
|
||||
import type { AgentAnswer } from "./agents.ts";
|
||||
import { output } from "./command.ts";
|
||||
import { killProfileProcesses, readProfileFile } from "./isolation.ts";
|
||||
import { isObject } from "./json.ts";
|
||||
import { runProfiles } from "./profiles.ts";
|
||||
import type { ProfileFailure } from "./profiles.ts";
|
||||
import {
|
||||
API,
|
||||
env,
|
||||
EVENT,
|
||||
gitea,
|
||||
INDEX,
|
||||
makeProfileFile,
|
||||
MANIFEST,
|
||||
PROCESS_ENV,
|
||||
REPO,
|
||||
runAgentCommand,
|
||||
WORKSPACE,
|
||||
} from "./run-context.ts";
|
||||
codexAuthentication,
|
||||
isClaudeAuthFailure,
|
||||
isCodexAuthFailure,
|
||||
persistCodexAuth,
|
||||
retryInvalidToken,
|
||||
stageCodexAuth,
|
||||
} from "./auth.ts";
|
||||
import type { ClaudeFailureFields, CodexAuthentication } from "./auth.ts";
|
||||
import { chunks, endWith, output, writeChunk } from "./command.ts";
|
||||
import type { CommandStatus } from "./command.ts";
|
||||
import { agentCommand, killProfileProcesses, readProfileFile } from "./isolation.ts";
|
||||
import { isObject } from "./json.ts";
|
||||
import {
|
||||
collectOpenCodeOutput,
|
||||
openCodeArgs,
|
||||
openCodeError,
|
||||
parseOpenCodeEvents,
|
||||
} from "./opencode.ts";
|
||||
import {
|
||||
AgentAuthError,
|
||||
agentEnvironment,
|
||||
runProfiles,
|
||||
sanitizedEnvironment,
|
||||
withProfile,
|
||||
} from "./profiles.ts";
|
||||
import type { ProfileFailure } from "./profiles.ts";
|
||||
import { readRuntimeManifest } from "./runtime.ts";
|
||||
import type { RuntimeOutcome, RuntimeProfile } from "./runtime.ts";
|
||||
|
||||
interface AttemptResult {
|
||||
@@ -32,16 +44,61 @@ interface AttemptResult {
|
||||
profile: Pick<RuntimeProfile, "type" | "model">;
|
||||
}
|
||||
|
||||
interface RenderedPrompt {
|
||||
prompt: string;
|
||||
reviewPath: string;
|
||||
anchorsPath: string;
|
||||
// Claude Code's stream-json `result` event, in its own field names.
|
||||
type ClaudeResult = ClaudeFailureFields &
|
||||
Partial<Record<"is_error", boolean>> & {
|
||||
subtype: string;
|
||||
};
|
||||
|
||||
const env = (name: string): string => {
|
||||
const value = process.env[name];
|
||||
if (value === undefined) {
|
||||
throw new Error(`${name} is not set`);
|
||||
}
|
||||
return value;
|
||||
};
|
||||
|
||||
function definedEnvironment(): Record<string, string> {
|
||||
const result: Record<string, string> = {};
|
||||
for (const [name, value] of Object.entries(process.env)) {
|
||||
if (value !== undefined) {
|
||||
result[name] = value;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
const API = env("GITEA_API_URL");
|
||||
const REPO = env("GITEA_REPOSITORY");
|
||||
const INDEX = env("ISSUE_INDEX");
|
||||
const EVENT = env("EVENT_NAME");
|
||||
const AUTHOR_TOKEN = env("GITEA_TOKEN");
|
||||
const WORKSPACE = env("GITHUB_WORKSPACE");
|
||||
const MANIFEST = await readRuntimeManifest(env("BOT_AGENTS_MANIFEST"));
|
||||
const PROCESS_ENV = sanitizedEnvironment(definedEnvironment());
|
||||
const RULES_PATH = "repos/commons/code-rules/raw/README.md";
|
||||
const CODEX_AUTH = "/root/.codex/auth.json";
|
||||
const SKILLS_INSTALL = "skills add obra/superpowers";
|
||||
|
||||
async function gitea(path: string, body?: unknown): Promise<Response> {
|
||||
const response = await fetch(`${API}/${path}`, {
|
||||
method: body === undefined ? "GET" : "POST",
|
||||
headers: {
|
||||
Authorization: `token ${AUTHOR_TOKEN}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: body === undefined ? null : JSON.stringify(body),
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new Error(`${path}: ${response.status} ${await response.text()}`);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
async function postAuthorComment(body: string): Promise<void> {
|
||||
await gitea(`repos/${REPO}/issues/${INDEX}/comments`, { body });
|
||||
}
|
||||
|
||||
async function runRoot(command: string, args: string[]): Promise<void> {
|
||||
const { success, code } = await output(command, { args, env: PROCESS_ENV, clearEnv: true });
|
||||
if (!success) {
|
||||
@@ -63,6 +120,22 @@ async function readProfileToken(profile: RuntimeProfile): Promise<string | undef
|
||||
return token;
|
||||
}
|
||||
|
||||
async function runAgentCommand(
|
||||
profile: RuntimeProfile,
|
||||
command: string,
|
||||
args: string[],
|
||||
token?: string,
|
||||
): Promise<void> {
|
||||
const { success, code } = await agentCommand(profile, command, args, {
|
||||
cwd: WORKSPACE,
|
||||
env: agentEnvironment(PROCESS_ENV, profile, token),
|
||||
clearEnv: true,
|
||||
}).output();
|
||||
if (!success) {
|
||||
throw new Error(`${command} ${args[0]} exited with ${code}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function prepareProfile(profile: RuntimeProfile): Promise<void> {
|
||||
await runAgentCommand(profile, "git", [
|
||||
"config",
|
||||
@@ -88,17 +161,18 @@ function formatComment(comment: unknown): string {
|
||||
return `## ${String(login)} at ${String(createdAt)}\n\n${String(body)}\n`;
|
||||
}
|
||||
|
||||
async function promptValues(
|
||||
reviewPath: string,
|
||||
anchorsPath: string,
|
||||
): Promise<Record<string, string>> {
|
||||
async function renderPrompt(
|
||||
profile: RuntimeProfile,
|
||||
): Promise<{ prompt: string; reviewPath: string; anchorsPath: string }> {
|
||||
const reviewPath = await makeProfileFile(profile, "review.md");
|
||||
const anchorsPath = await makeProfileFile(profile, "anchors.json");
|
||||
const comments: unknown = await (
|
||||
await gitea(`repos/${REPO}/issues/${INDEX}/comments?limit=100`)
|
||||
).json();
|
||||
if (!Array.isArray(comments)) {
|
||||
throw new TypeError("comments: unexpected Gitea response");
|
||||
}
|
||||
return {
|
||||
const values: Record<string, string> = {
|
||||
COMMENT: env("COMMENT"),
|
||||
ISSUE_COMMENTS: comments.map((comment: unknown) => formatComment(comment)).join("\n"),
|
||||
CODE_RULES: await (await gitea(RULES_PATH)).text(),
|
||||
@@ -109,12 +183,6 @@ async function promptValues(
|
||||
REVIEW_PATH: reviewPath,
|
||||
ANCHORS_PATH: anchorsPath,
|
||||
};
|
||||
}
|
||||
|
||||
async function renderPrompt(profile: RuntimeProfile): Promise<RenderedPrompt> {
|
||||
const reviewPath = await makeProfileFile(profile, "review.md");
|
||||
const anchorsPath = await makeProfileFile(profile, "anchors.json");
|
||||
const values = await promptValues(reviewPath, anchorsPath);
|
||||
const template = await readFile(new URL("../prompt.md", import.meta.url), "utf8");
|
||||
return {
|
||||
prompt: template.replaceAll(
|
||||
@@ -126,45 +194,319 @@ async function renderPrompt(profile: RuntimeProfile): Promise<RenderedPrompt> {
|
||||
};
|
||||
}
|
||||
|
||||
async function runClaude(profile: RuntimeProfile, token: string, prompt: string): Promise<string> {
|
||||
const claude = agentCommand(
|
||||
profile,
|
||||
"claude",
|
||||
[
|
||||
"--print",
|
||||
"--dangerously-skip-permissions",
|
||||
"--model",
|
||||
profile.model,
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
"--verbose",
|
||||
prompt,
|
||||
],
|
||||
{
|
||||
cwd: WORKSPACE,
|
||||
env: {
|
||||
...agentEnvironment(PROCESS_ENV, profile, token),
|
||||
IS_SANDBOX: "1",
|
||||
},
|
||||
clearEnv: true,
|
||||
stdout: "piped",
|
||||
stderr: "piped",
|
||||
},
|
||||
).spawn();
|
||||
let result: ClaudeResult | undefined;
|
||||
let error = "";
|
||||
const copyError = (async (): Promise<void> => {
|
||||
const decoder = new TextDecoder();
|
||||
for await (const chunk of chunks(claude.stderr)) {
|
||||
const part = decoder.decode(chunk, { stream: true });
|
||||
error += part;
|
||||
console.error(part);
|
||||
}
|
||||
error += decoder.decode();
|
||||
})();
|
||||
for await (const line of createInterface({ input: claude.stdout, crlfDelay: Infinity })) {
|
||||
if (line !== "") {
|
||||
const event: unknown = JSON.parse(line);
|
||||
if (isObject(event)) {
|
||||
const content = isObject(event.message) ? event.message.content : undefined;
|
||||
for (const part of Array.isArray(content) ? content : []) {
|
||||
if (isObject(part)) {
|
||||
const text = part.thinking ?? part.text ?? part.name;
|
||||
if (typeof text === "string" && text !== "") {
|
||||
console.log(text);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (event.type === "result") {
|
||||
const { is_error: isError, result: answer, subtype, api_error_status: apiStatus } = event;
|
||||
result = {
|
||||
...(typeof isError === "boolean" && { is_error: isError }),
|
||||
...(typeof answer === "string" && { result: answer }),
|
||||
subtype: String(subtype),
|
||||
...((typeof apiStatus === "number" || apiStatus === null) && {
|
||||
api_error_status: apiStatus,
|
||||
}),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
const [status] = await Promise.all([claude.status, copyError]);
|
||||
if (result?.is_error === true) {
|
||||
const failure = new Error(result.result ?? `claude ended with ${result.subtype}`);
|
||||
if (isClaudeAuthFailure(result)) {
|
||||
throw new AgentAuthError("claude sign-in failed", { cause: failure });
|
||||
}
|
||||
throw failure;
|
||||
}
|
||||
if (result?.result === undefined) {
|
||||
throw new Error(error.trim() || `claude ended with ${result?.subtype ?? "no result"}`);
|
||||
}
|
||||
if (!status.success) {
|
||||
throw new Error(error.trim() || `claude exited with ${status.code}`);
|
||||
}
|
||||
return result.result;
|
||||
}
|
||||
|
||||
// Polls every 100 ms until the login prints something, failing if it exits first.
|
||||
async function pollForInstructions(
|
||||
shownText: () => string,
|
||||
status: Promise<CommandStatus>,
|
||||
): Promise<void> {
|
||||
if (shownText().trim() !== "") {
|
||||
return;
|
||||
}
|
||||
const exited = await Promise.race([status.then(() => true), sleep(100, false)]);
|
||||
if (exited) {
|
||||
throw new AgentAuthError("AI bot login produced no instructions");
|
||||
}
|
||||
await pollForInstructions(shownText, status);
|
||||
}
|
||||
|
||||
async function codexDeviceLogin(
|
||||
profile: RuntimeProfile,
|
||||
environment: Record<string, string>,
|
||||
): Promise<void> {
|
||||
const login = agentCommand(profile, "codex", ["login", "--device-auth"], {
|
||||
cwd: WORKSPACE,
|
||||
env: environment,
|
||||
clearEnv: true,
|
||||
stdout: "piped",
|
||||
stderr: "piped",
|
||||
}).spawn();
|
||||
let shown = "";
|
||||
const collect = async (stream: Readable): Promise<void> => {
|
||||
const decoder = new TextDecoder();
|
||||
for await (const chunk of chunks(stream)) {
|
||||
shown += decoder.decode(chunk, { stream: true });
|
||||
}
|
||||
shown += decoder.decode();
|
||||
};
|
||||
const drained = Promise.all([collect(login.stdout), collect(login.stderr)]);
|
||||
const status = login.status;
|
||||
await sleep(3000);
|
||||
await pollForInstructions(() => shown, status);
|
||||
await postAuthorComment(withProfile(shown, profile));
|
||||
const [result] = await Promise.all([status, drained]);
|
||||
if (!result.success) {
|
||||
throw new AgentAuthError(`codex login exited with ${result.code}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function codexApiLogin(
|
||||
profile: RuntimeProfile,
|
||||
environment: Record<string, string>,
|
||||
token: string,
|
||||
): Promise<void> {
|
||||
const login = agentCommand(profile, "codex", ["login", "--with-api-key"], {
|
||||
cwd: WORKSPACE,
|
||||
env: environment,
|
||||
clearEnv: true,
|
||||
stdin: "piped",
|
||||
stderr: "piped",
|
||||
}).spawn();
|
||||
await endWith(login.stdin, `${token}\n`);
|
||||
const [status, error] = await Promise.all([login.status, readText(login.stderr)]);
|
||||
if (!status.success) {
|
||||
throw new Error(error.trim() || `codex login exited with ${status.code}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function makeProfileFile(profile: RuntimeProfile, name: string): Promise<string> {
|
||||
const path = `${profile.home}/${name}`;
|
||||
await writeFile(path, "", { mode: 0o600 });
|
||||
await chmod(path, 0o600);
|
||||
await chown(path, profile.uid, profile.gid);
|
||||
return path;
|
||||
}
|
||||
|
||||
async function runCodex(
|
||||
profile: RuntimeProfile,
|
||||
prompt: string,
|
||||
authentication: CodexAuthentication,
|
||||
): Promise<string> {
|
||||
const token = authentication.kind === "api-key" ? authentication.token : undefined;
|
||||
const environment = agentEnvironment(PROCESS_ENV, profile, token);
|
||||
if (authentication.kind === "api-key") {
|
||||
await codexApiLogin(profile, environment, authentication.token);
|
||||
} else {
|
||||
const loggedIn = await agentCommand(profile, "codex", ["login", "status"], {
|
||||
cwd: WORKSPACE,
|
||||
env: environment,
|
||||
clearEnv: true,
|
||||
}).output();
|
||||
if (!loggedIn.success) {
|
||||
await codexDeviceLogin(profile, environment);
|
||||
}
|
||||
}
|
||||
const outputPath = await makeProfileFile(profile, "result.md");
|
||||
const attempt = async (): Promise<{ status: CommandStatus; error: string }> => {
|
||||
const codex = agentCommand(
|
||||
profile,
|
||||
"codex",
|
||||
[
|
||||
"exec",
|
||||
"--model",
|
||||
profile.model,
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--output-last-message",
|
||||
outputPath,
|
||||
prompt,
|
||||
],
|
||||
{
|
||||
cwd: WORKSPACE,
|
||||
env: environment,
|
||||
clearEnv: true,
|
||||
stdout: "inherit",
|
||||
stderr: "piped",
|
||||
},
|
||||
).spawn();
|
||||
const decoder = new TextDecoder();
|
||||
let error = "";
|
||||
for await (const chunk of chunks(codex.stderr)) {
|
||||
await writeChunk(process.stderr, chunk);
|
||||
error += decoder.decode(chunk, { stream: true });
|
||||
}
|
||||
error += decoder.decode();
|
||||
// Codex echoes the prompt, whose comments may quote auth errors.
|
||||
return { status: await codex.status, error: error.replaceAll(prompt, "") };
|
||||
};
|
||||
const result =
|
||||
authentication.kind === "api-key"
|
||||
? await attempt()
|
||||
: await retryInvalidToken(attempt, async () => {
|
||||
await runAgentCommand(profile, "codex", ["logout"]);
|
||||
await codexDeviceLogin(profile, environment);
|
||||
});
|
||||
if (!result.status.success) {
|
||||
const failure = new Error(result.error.trim() || `codex exited with ${result.status.code}`);
|
||||
if (isCodexAuthFailure(result.error)) {
|
||||
throw new AgentAuthError("codex sign-in failed", { cause: failure });
|
||||
}
|
||||
throw failure;
|
||||
}
|
||||
return outputPath;
|
||||
}
|
||||
|
||||
async function runOpenCode(
|
||||
profile: RuntimeProfile,
|
||||
token: string,
|
||||
prompt: string,
|
||||
): Promise<string> {
|
||||
const child = agentCommand(profile, "opencode", openCodeArgs(profile.model, prompt), {
|
||||
cwd: WORKSPACE,
|
||||
env: agentEnvironment(PROCESS_ENV, profile, token),
|
||||
clearEnv: true,
|
||||
stdout: "piped",
|
||||
stderr: "piped",
|
||||
}).spawn();
|
||||
const collected = collectOpenCodeOutput(
|
||||
chunks(child.stdout),
|
||||
chunks(child.stderr),
|
||||
async (chunk) => {
|
||||
await writeChunk(process.stdout, chunk);
|
||||
},
|
||||
async (chunk) => {
|
||||
await writeChunk(process.stderr, chunk);
|
||||
},
|
||||
);
|
||||
const [status, streams] = await Promise.all([child.status, collected]);
|
||||
const error = streams.stderr.trim();
|
||||
if (!status.success) {
|
||||
throw (
|
||||
openCodeError(streams.stdout) ?? new Error(error || `opencode exited with ${status.code}`)
|
||||
);
|
||||
}
|
||||
return parseOpenCodeEvents(streams.stdout);
|
||||
}
|
||||
|
||||
async function runProfile(profile: RuntimeProfile): Promise<AttemptResult> {
|
||||
const token = await readProfileToken(profile);
|
||||
const authentication = profile.type === "codex" ? codexAuthentication(token) : null;
|
||||
let stagedCodexAuth: string | null = null;
|
||||
let answer: AgentAnswer = {};
|
||||
let rendered: RenderedPrompt | undefined;
|
||||
let body: string | undefined;
|
||||
let resultPath: string | null = null;
|
||||
let reviewPath: string | undefined;
|
||||
let anchorsPath: string | undefined;
|
||||
try {
|
||||
if (authentication?.kind === "persistent") {
|
||||
stagedCodexAuth = await stageCodexAuth(CODEX_AUTH, profile.home, profile.uid, profile.gid);
|
||||
}
|
||||
await prepareProfile(profile);
|
||||
rendered = await renderPrompt(profile);
|
||||
answer = await runAgent(profile, token, rendered.prompt, authentication);
|
||||
const rendered = await renderPrompt(profile);
|
||||
reviewPath = rendered.reviewPath;
|
||||
anchorsPath = rendered.anchorsPath;
|
||||
if (profile.type === "claude") {
|
||||
body = await runClaude(profile, token ?? "", rendered.prompt);
|
||||
} else if (profile.type === "codex") {
|
||||
resultPath = await runCodex(
|
||||
profile,
|
||||
rendered.prompt,
|
||||
authentication ?? codexAuthentication(token),
|
||||
);
|
||||
} else {
|
||||
body = await runOpenCode(profile, token ?? "", rendered.prompt);
|
||||
}
|
||||
} finally {
|
||||
await killProfileProcesses(profile.uid);
|
||||
if (stagedCodexAuth !== null) {
|
||||
await persistCodexAuth(stagedCodexAuth, CODEX_AUTH);
|
||||
}
|
||||
}
|
||||
const body =
|
||||
answer.resultPath === undefined
|
||||
? answer.body
|
||||
: await readProfileFile(profile, answer.resultPath);
|
||||
if (body === undefined || rendered === undefined) {
|
||||
if (resultPath !== null) {
|
||||
body = await readProfileFile(profile, resultPath);
|
||||
}
|
||||
if (body === undefined || reviewPath === undefined || anchorsPath === undefined) {
|
||||
throw new Error("agent attempt produced no result");
|
||||
}
|
||||
const { reviewPath, anchorsPath } = rendered;
|
||||
return { body, reviewPath, anchorsPath, profile: { type: profile.type, model: profile.model } };
|
||||
return {
|
||||
body,
|
||||
reviewPath,
|
||||
anchorsPath,
|
||||
profile: { type: profile.type, model: profile.model },
|
||||
};
|
||||
}
|
||||
|
||||
async function writeOutcome(outcome: RuntimeOutcome): Promise<void> {
|
||||
await writeFile(MANIFEST.outcomeFile, JSON.stringify(outcome), { mode: 0o600 });
|
||||
await writeFile(MANIFEST.outcomeFile, JSON.stringify(outcome), {
|
||||
mode: 0o600,
|
||||
});
|
||||
await chmod(MANIFEST.outcomeFile, 0o600);
|
||||
}
|
||||
|
||||
const failures: ProfileFailure[] = [];
|
||||
try {
|
||||
await configureGitAuthor();
|
||||
await writeOutcome({ ok: true, ...(await runProfiles(MANIFEST.profiles, runProfile, failures)) });
|
||||
await writeOutcome({
|
||||
ok: true,
|
||||
...(await runProfiles(MANIFEST.profiles, runProfile, failures)),
|
||||
});
|
||||
} catch (error) {
|
||||
console.error(error);
|
||||
await writeOutcome({ ok: false, failures });
|
||||
|
||||
Reference in New Issue
Block a user