feat: 打通 v0.1 runner job 正式路径
This commit is contained in:
Codex
@@ -0,0 +1,137 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactJson, redactText } from "../common/redaction.js";
|
||||
import type { AgentRunStore } from "./store.js";
|
||||
import type { JsonRecord } from "../common/types.js";
|
||||
import { renderRunnerJobManifest } from "../runner/k8s-job.js";
|
||||
|
||||
export interface RunnerJobDefaults {
|
||||
namespace: string;
|
||||
managerUrl: string;
|
||||
image: string;
|
||||
sourceCommit: string;
|
||||
serviceAccountName?: string;
|
||||
kubectlCommand?: string;
|
||||
}
|
||||
|
||||
export interface CreateRunnerJobInput extends JsonRecord {
|
||||
commandId: string;
|
||||
managerUrl?: string;
|
||||
image?: string;
|
||||
namespace?: string;
|
||||
attemptId?: string;
|
||||
runnerId?: string;
|
||||
sourceCommit?: string;
|
||||
serviceAccountName?: string;
|
||||
}
|
||||
|
||||
export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults }): Promise<JsonRecord> {
|
||||
const commandId = stringField(options.input, "commandId");
|
||||
const run = await options.store.getRun(options.runId);
|
||||
const command = await options.store.getCommand(commandId);
|
||||
if (command.runId !== run.id) throw new AgentRunError("schema-invalid", `command ${commandId} does not belong to run ${run.id}`, { httpStatus: 400 });
|
||||
if (command.type !== "turn") throw new AgentRunError("schema-invalid", `command ${commandId} is not a turn command`, { httpStatus: 400 });
|
||||
|
||||
const image = optionalString(options.input.image) ?? options.defaults.image;
|
||||
if (!image) throw new AgentRunError("schema-invalid", "runner job image is required; set --image or AGENTRUN_RUNNER_IMAGE", { httpStatus: 400 });
|
||||
const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace;
|
||||
const managerUrl = optionalString(options.input.managerUrl) ?? options.defaults.managerUrl;
|
||||
const sourceCommit = optionalString(options.input.sourceCommit) ?? options.defaults.sourceCommit;
|
||||
const serviceAccountName = optionalString(options.input.serviceAccountName) ?? options.defaults.serviceAccountName;
|
||||
|
||||
const renderOptions = {
|
||||
run,
|
||||
commandId,
|
||||
managerUrl,
|
||||
image,
|
||||
namespace,
|
||||
sourceCommit,
|
||||
...(serviceAccountName ? { serviceAccountName } : {}),
|
||||
};
|
||||
const attemptId = optionalString(options.input.attemptId);
|
||||
const runnerId = optionalString(options.input.runnerId);
|
||||
const render = renderRunnerJobManifest({ ...renderOptions, ...(attemptId ? { attemptId } : {}), ...(runnerId ? { runnerId } : {}) });
|
||||
const created = await kubectlCreate(render.manifest, options.defaults.kubectlCommand ?? "kubectl");
|
||||
return {
|
||||
action: "create-kubernetes-job",
|
||||
mutation: true,
|
||||
jobIdentity: {
|
||||
kind: "Job",
|
||||
namespace: render.namespace,
|
||||
name: render.jobName,
|
||||
serviceAccountName: render.serviceAccountName,
|
||||
uid: objectPath(created, ["metadata", "uid"]),
|
||||
},
|
||||
runner: {
|
||||
runId: run.id,
|
||||
commandId,
|
||||
attemptId: render.attemptId,
|
||||
runnerId: render.runnerId,
|
||||
backendProfile: run.backendProfile,
|
||||
managerUrl,
|
||||
sourceCommit,
|
||||
placement: "kubernetes-job",
|
||||
logPath: `kubectl -n ${render.namespace} logs job/${render.jobName}`,
|
||||
},
|
||||
secretRefs: render.secretRefs.map((item) => ({ profile: item.profile, name: item.secretRef.name, namespace: item.secretRef.namespace ?? render.namespace, keys: item.secretRef.keys ?? [], mountPath: item.mountPath, valuesPrinted: false })),
|
||||
pollCommands: {
|
||||
run: `bun scripts/agentrun-cli.ts runs show ${run.id} --manager-url ${managerUrl}`,
|
||||
command: `bun scripts/agentrun-cli.ts commands show ${commandId} --run-id ${run.id} --manager-url ${managerUrl}`,
|
||||
events: `bun scripts/agentrun-cli.ts runs events ${run.id} --manager-url ${managerUrl} --after-seq 0 --limit 100`,
|
||||
},
|
||||
warnings: render.warnings,
|
||||
kubernetes: {
|
||||
created: true,
|
||||
valuesPrinted: false,
|
||||
apiVersion: objectPath(created, ["apiVersion"]),
|
||||
kind: objectPath(created, ["kind"]),
|
||||
resourceVersion: objectPath(created, ["metadata", "resourceVersion"]),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function kubectlCreate(manifest: JsonRecord, kubectlCommand: string): Promise<JsonRecord> {
|
||||
const child = spawn(kubectlCommand, ["create", "-f", "-", "-o", "json"], { stdio: ["pipe", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
|
||||
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
|
||||
child.stdin.end(`${JSON.stringify(manifest)}\n`);
|
||||
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
|
||||
child.on("error", reject);
|
||||
child.on("close", (code, signal) => resolve({ code, signal }));
|
||||
}).catch((error: unknown) => {
|
||||
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
|
||||
});
|
||||
if (result.code !== 0) {
|
||||
throw new AgentRunError("infra-failed", `kubectl create runner job failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(stderr.slice(-4000)), stdout: redactText(stdout.slice(-2000)), signal: result.signal }) });
|
||||
}
|
||||
try {
|
||||
const parsed = JSON.parse(stdout) as unknown;
|
||||
if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return redactJson(parsed as JsonRecord);
|
||||
} catch (error) {
|
||||
throw new AgentRunError("infra-failed", `kubectl returned invalid JSON: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { stdoutPreview: redactText(stdout.slice(0, 1000)) } });
|
||||
}
|
||||
throw new AgentRunError("infra-failed", "kubectl returned non-object JSON", { httpStatus: 502 });
|
||||
}
|
||||
|
||||
function stringField(record: JsonRecord, key: string): string {
|
||||
const value = record[key];
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${key} is required`, { httpStatus: 400 });
|
||||
return value.trim();
|
||||
}
|
||||
|
||||
function optionalString(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
function objectPath(value: unknown, path: string[]): string | null {
|
||||
let current: unknown = value;
|
||||
for (const key of path) {
|
||||
if (typeof current !== "object" || current === null || Array.isArray(current)) return null;
|
||||
current = (current as Record<string, unknown>)[key];
|
||||
}
|
||||
return typeof current === "string" ? current : null;
|
||||
}
|
||||
@@ -6,7 +6,7 @@ import { redactJson } from "../common/redaction.js";
|
||||
import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, RunEvent, RunnerRecord, RunRecord, RunStatus, TerminalStatus } from "../common/types.js";
|
||||
import { newId, nowIso, stableHash } from "../common/validation.js";
|
||||
import type { AgentRunStore, StoreHealth } from "./store.js";
|
||||
import { statusFromTerminal } from "./store.js";
|
||||
import { commandStateFromTerminal, statusFromTerminal } from "./store.js";
|
||||
|
||||
interface PostgresStoreOptions {
|
||||
connectionString: string;
|
||||
@@ -308,6 +308,19 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return commandFromRow(row);
|
||||
}
|
||||
|
||||
async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): Promise<CommandRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
|
||||
const row = existing.rows[0];
|
||||
if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
|
||||
const command = commandFromRow(row);
|
||||
const state = commandStateFromTerminal(result.terminalStatus);
|
||||
const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, nowIso()]);
|
||||
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
|
||||
return commandFromRow(updated.rows[0]);
|
||||
});
|
||||
}
|
||||
|
||||
async appendEvent(runId: string, type: EventType, payload: JsonRecord): Promise<RunEvent> {
|
||||
return this.withTransaction(async (client) => {
|
||||
await this.requireRunForUpdate(client, runId);
|
||||
|
||||
+34
-2
@@ -6,12 +6,20 @@ import { openAgentRunStoreFromEnv } from "./store.js";
|
||||
import { AgentRunError, errorToJson } from "../common/errors.js";
|
||||
import { asRecord, validateCreateCommand, validateCreateRun } from "../common/validation.js";
|
||||
import type { ApiErrorBody, ApiOkBody, JsonRecord, JsonValue, RunEvent } from "../common/types.js";
|
||||
import { createKubernetesRunnerJob } from "./kubernetes-runner-job.js";
|
||||
|
||||
export interface ManagerServerOptions {
|
||||
store?: AgentRunStore;
|
||||
port?: number;
|
||||
host?: string;
|
||||
sourceCommit?: string;
|
||||
runnerJobDefaults?: {
|
||||
namespace?: string;
|
||||
managerUrl?: string;
|
||||
image?: string;
|
||||
serviceAccountName?: string;
|
||||
kubectlCommand?: string;
|
||||
};
|
||||
}
|
||||
|
||||
export interface StartedManagerServer {
|
||||
@@ -23,12 +31,13 @@ export interface StartedManagerServer {
|
||||
export async function startManagerServer(options: ManagerServerOptions = {}): Promise<StartedManagerServer> {
|
||||
const store = options.store ?? await openAgentRunStoreFromEnv();
|
||||
const sourceCommit = options.sourceCommit ?? process.env.AGENTRUN_SOURCE_COMMIT ?? "unknown";
|
||||
const runnerJobDefaults = options.runnerJobDefaults;
|
||||
const server = createServer(async (req, res) => {
|
||||
const traceId = `trc_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`;
|
||||
try {
|
||||
const method = req.method ?? "GET";
|
||||
const url = new URL(req.url ?? "/", "http://agentrun.local");
|
||||
const data = await route({ method, url, body: await readBody(req), store, sourceCommit });
|
||||
const data = await route({ method, url, body: await readBody(req), store, sourceCommit, ...(runnerJobDefaults ? { runnerJobDefaults } : {}) });
|
||||
writeJson(res, 200, { ok: true, data, traceId });
|
||||
} catch (error) {
|
||||
const agentError = normalizeError(error);
|
||||
@@ -49,7 +58,7 @@ async function readBody(req: import("node:http").IncomingMessage): Promise<unkno
|
||||
return JSON.parse(text) as unknown;
|
||||
}
|
||||
|
||||
async function route({ method, url, body, store, sourceCommit }: { method: string; url: URL; body: unknown; store: AgentRunStore; sourceCommit: string }): Promise<JsonValue> {
|
||||
async function route({ method, url, body, store, sourceCommit, runnerJobDefaults }: { method: string; url: URL; body: unknown; store: AgentRunStore; sourceCommit: string; runnerJobDefaults?: NonNullable<ManagerServerOptions["runnerJobDefaults"]> }): Promise<JsonValue> {
|
||||
const path = url.pathname;
|
||||
if (method === "GET" && (path === "/health" || path === "/health/live" || path === "/health/readiness")) {
|
||||
const database = await store.health();
|
||||
@@ -69,6 +78,23 @@ async function route({ method, url, body, store, sourceCommit }: { method: strin
|
||||
const commandCreateMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/commands$/u);
|
||||
if (method === "POST" && commandCreateMatch) return await store.createCommand(commandCreateMatch[1] ?? "", validateCreateCommand(body)) as unknown as JsonValue;
|
||||
if (method === "GET" && commandCreateMatch) return { items: await store.listCommands(commandCreateMatch[1] ?? "", integerQuery(url, "afterSeq", 0), integerQuery(url, "limit", 20)) as unknown as JsonValue };
|
||||
const runnerJobMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/runner-jobs$/u);
|
||||
if (method === "POST" && runnerJobMatch) {
|
||||
const namespace = runnerJobDefaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01";
|
||||
return await createKubernetesRunnerJob({
|
||||
store,
|
||||
runId: runnerJobMatch[1] ?? "",
|
||||
input: asRecord(body ?? {}, "runnerJob") as never,
|
||||
defaults: {
|
||||
namespace,
|
||||
managerUrl: runnerJobDefaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`,
|
||||
image: runnerJobDefaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "",
|
||||
sourceCommit,
|
||||
serviceAccountName: runnerJobDefaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT ?? "agentrun-v01-runner",
|
||||
...(runnerJobDefaults?.kubectlCommand ? { kubectlCommand: runnerJobDefaults.kubectlCommand } : {}),
|
||||
},
|
||||
}) as unknown as JsonValue;
|
||||
}
|
||||
const commandShowMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/commands\/([^/]+)$/u);
|
||||
if (method === "GET" && commandShowMatch) return await store.getCommand(commandShowMatch[2] ?? "") as unknown as JsonValue;
|
||||
if (method === "POST" && path === "/api/v1/runners/register") return await store.registerRunner(asRecord(body ?? {}, "runner")) as unknown as JsonValue;
|
||||
@@ -100,6 +126,12 @@ async function route({ method, url, body, store, sourceCommit }: { method: strin
|
||||
}
|
||||
const ackMatch = path.match(/^\/api\/v1\/commands\/([^/]+)\/ack$/u);
|
||||
if (method === "POST" && ackMatch) return await store.ackCommand(ackMatch[1] ?? "") as unknown as JsonValue;
|
||||
const commandStatusMatch = path.match(/^\/api\/v1\/commands\/([^/]+)\/status$/u);
|
||||
if (method === "PATCH" && commandStatusMatch) {
|
||||
const record = asRecord(body, "commandStatus");
|
||||
const terminalStatus = record.terminalStatus === "completed" || record.terminalStatus === "failed" || record.terminalStatus === "blocked" || record.terminalStatus === "cancelled" ? record.terminalStatus : "failed";
|
||||
return await store.finishCommand(commandStatusMatch[1] ?? "", { terminalStatus, failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null, failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null }) as unknown as JsonValue;
|
||||
}
|
||||
throw new AgentRunError("schema-invalid", `unsupported route ${method} ${path}`, { httpStatus: 404 });
|
||||
}
|
||||
|
||||
|
||||
@@ -28,6 +28,7 @@ export interface AgentRunStore {
|
||||
claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
|
||||
heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
|
||||
ackCommand(commandId: string): MaybePromise<CommandRecord>;
|
||||
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): MaybePromise<CommandRecord>;
|
||||
appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): MaybePromise<RunEvent>;
|
||||
finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): MaybePromise<RunRecord>;
|
||||
backends(): MaybePromise<JsonRecord[]>;
|
||||
@@ -132,6 +133,14 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
return next;
|
||||
}
|
||||
|
||||
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): CommandRecord {
|
||||
const command = this.getCommand(commandId);
|
||||
const next = { ...command, state: commandStateFromTerminal(result.terminalStatus), updatedAt: nowIso() };
|
||||
this.commands.set(commandId, next);
|
||||
this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: next.state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
|
||||
return next;
|
||||
}
|
||||
|
||||
appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): RunEvent {
|
||||
this.getRun(runId);
|
||||
const events = this.eventsByRun.get(runId) ?? [];
|
||||
@@ -166,3 +175,9 @@ export function statusFromTerminal(terminalStatus: TerminalStatus): RunRecord["s
|
||||
if (terminalStatus === "blocked") return "blocked";
|
||||
return "failed";
|
||||
}
|
||||
|
||||
export function commandStateFromTerminal(terminalStatus: TerminalStatus): CommandRecord["state"] {
|
||||
if (terminalStatus === "completed") return "completed";
|
||||
if (terminalStatus === "cancelled") return "cancelled";
|
||||
return "failed";
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user