diff --git a/src/common/types.ts b/src/common/types.ts index 731be30..ec0999e 100644 --- a/src/common/types.ts +++ b/src/common/types.ts @@ -271,12 +271,12 @@ export interface CreateCommandInput extends JsonRecord { } export interface RunnerDispatchRequest extends JsonRecord { - kind: "kubernetes-job"; + kind: "host-process" | "kubernetes-job"; input: JsonRecord; } export type RunnerDispatchIntentState = "pending" | "dispatching" | "retry" | "dispatched" | "failed" | "cancelled"; -export type RunnerDispatchOutcome = "kubernetes-job" | "active-runner-reused"; +export type RunnerDispatchOutcome = "host-process" | "kubernetes-job" | "active-runner-reused"; export interface RunnerDispatchCompletion { dispatchOutcome: RunnerDispatchOutcome; @@ -288,7 +288,7 @@ export interface RunnerDispatchIntentRecord extends JsonRecord { id: string; runId: string; commandId: string; - kind: "kubernetes-job"; + kind: "host-process" | "kubernetes-job"; input: JsonRecord; inputHash: string; state: RunnerDispatchIntentState; diff --git a/src/common/validation.ts b/src/common/validation.ts index fde2ed3..7b6cf06 100644 --- a/src/common/validation.ts +++ b/src/common/validation.ts @@ -401,9 +401,10 @@ export function validateCreateCommand(input: unknown): CreateCommandInput { if (record.dispatch !== undefined && record.dispatch !== null) { if (type !== "turn") throw new AgentRunError("schema-invalid", "dispatch is only supported for turn commands", { httpStatus: 400 }); const dispatchRecord = asRecord(record.dispatch, "command.dispatch"); - if (requiredString(dispatchRecord, "kind") !== "kubernetes-job") throw new AgentRunError("schema-invalid", "command.dispatch.kind must be kubernetes-job", { httpStatus: 400 }); + const kind = requiredString(dispatchRecord, "kind"); + if (kind !== "host-process" && kind !== "kubernetes-job") throw new AgentRunError("schema-invalid", "command.dispatch.kind must be host-process or kubernetes-job", { httpStatus: 400 }); const dispatchInput = validateDurableDispatchInput(asRecord(dispatchRecord.input ?? {}, "command.dispatch.input")); - dispatch = { kind: "kubernetes-job", input: dispatchInput }; + dispatch = { kind, input: dispatchInput }; } return { type, payload, ...(idempotencyKey ? { idempotencyKey } : {}), ...(dispatch ? { dispatch } : {}) }; } diff --git a/src/mgr/host-process-runner.ts b/src/mgr/host-process-runner.ts new file mode 100644 index 0000000..b8f9317 --- /dev/null +++ b/src/mgr/host-process-runner.ts @@ -0,0 +1,140 @@ +// SPEC: PJ2026-010306 嵌入式AgentBench +// SPEC_VERSION: draft-2026-07-24-p1-l0-l1-arm2d +import { closeSync, existsSync, mkdirSync, openSync, writeFileSync } from "node:fs"; +import { spawn } from "node:child_process"; +import path from "node:path"; +import type { JsonRecord } from "../common/types.js"; +import { AgentRunError } from "../common/errors.js"; +import { isLeaseExpired, isTerminalCommandState, isTerminalRunStatus, type AgentRunStore } from "./store.js"; +import { throwIfAborted } from "./abortable-child.js"; + +export interface HostProcessRunnerDefaults { + managerUrl: string; + sourceCommit: string; + command: string; + entrypoint: string; + cwd: string; + stateDir: string; + logDir: string; + apiKeyFile?: string; + runnerIdleTimeoutMs?: number; + backendRetryMaxAttempts?: number; + backendRetryInitialBackoffMs?: number; + backendRetryMaxBackoffMs?: number; + env?: NodeJS.ProcessEnv; +} + +export async function createHostProcessRunner(options: { + store: AgentRunStore; + runId: string; + input: JsonRecord; + defaults: HostProcessRunnerDefaults; + signal?: AbortSignal; +}): Promise { + throwIfAborted(options.signal, "host-process runner dispatch attempt"); + const commandId = requiredString(options.input.commandId, "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 }); + if (!existsSync(options.defaults.cwd)) throw new AgentRunError("infra-failed", `host runner cwd does not exist: ${options.defaults.cwd}`, { httpStatus: 503 }); + if (!existsSync(options.defaults.entrypoint)) throw new AgentRunError("infra-failed", `host runner entrypoint does not exist: ${options.defaults.entrypoint}`, { httpStatus: 503 }); + if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? "schema-invalid", `run ${run.id} is already terminal: ${run.status}`, { httpStatus: 409 }); + if (isTerminalCommandState(command.state) || command.state !== "pending") throw new AgentRunError(command.state === "cancelled" ? "cancelled" : "schema-invalid", `command ${commandId} is not pending: ${command.state}`, { httpStatus: 409 }); + if (run.claimedBy && !isLeaseExpired(run.leaseExpiresAt)) { + await options.store.appendEvent(run.id, "backend_status", { + phase: "runner-process-skipped-active-runner", + commandId, + runnerId: run.claimedBy, + leaseExpiresAt: run.leaseExpiresAt, + reason: "fresh-active-runner-lease", + valuesPrinted: false, + }); + return { action: "reuse-active-runner", mutation: false, runId: run.id, commandId, runnerId: run.claimedBy, leaseExpiresAt: run.leaseExpiresAt, reason: "fresh-active-runner-lease", valuesPrinted: false }; + } + + const runnerId = optionalString(options.input.runnerId) ?? `runner_host_${process.pid}_${Date.now().toString(36)}`; + const attemptId = optionalString(options.input.attemptId) ?? `attempt_${Date.now().toString(36)}`; + const safeRunId = filePart(run.id); + const safeRunnerId = filePart(runnerId); + mkdirSync(options.defaults.stateDir, { recursive: true }); + mkdirSync(options.defaults.logDir, { recursive: true }); + const logPath = path.join(options.defaults.logDir, `${safeRunId}-${safeRunnerId}.log`); + const statePath = path.join(options.defaults.stateDir, `${safeRunId}-${safeRunnerId}.json`); + const logFd = openSync(logPath, "a"); + const env: NodeJS.ProcessEnv = { + ...process.env, + ...options.defaults.env, + ...transientEnvironment(options.input.transientEnv), + AGENTRUN_MGR_URL: options.defaults.managerUrl, + AGENTRUN_RUN_ID: run.id, + AGENTRUN_COMMAND_ID: commandId, + AGENTRUN_RUNNER_ID: runnerId, + AGENTRUN_ATTEMPT_ID: attemptId, + AGENTRUN_SOURCE_COMMIT: options.defaults.sourceCommit, + AGENTRUN_LOG_PATH: logPath, + ...(options.defaults.apiKeyFile ? { AGENTRUN_API_KEY_FILE: options.defaults.apiKeyFile } : {}), + ...(options.defaults.runnerIdleTimeoutMs ? { AGENTRUN_RUNNER_IDLE_TIMEOUT_MS: String(options.defaults.runnerIdleTimeoutMs) } : {}), + ...(options.defaults.backendRetryMaxAttempts ? { AGENTRUN_BACKEND_RETRY_MAX_ATTEMPTS: String(options.defaults.backendRetryMaxAttempts) } : {}), + ...(options.defaults.backendRetryInitialBackoffMs ? { AGENTRUN_BACKEND_RETRY_INITIAL_BACKOFF_MS: String(options.defaults.backendRetryInitialBackoffMs) } : {}), + ...(options.defaults.backendRetryMaxBackoffMs ? { AGENTRUN_BACKEND_RETRY_MAX_BACKOFF_MS: String(options.defaults.backendRetryMaxBackoffMs) } : {}), + }; + const child = spawn(options.defaults.command, [options.defaults.entrypoint], { + cwd: options.defaults.cwd, + env, + detached: true, + stdio: ["ignore", logFd, logFd], + }); + try { + await childStarted(child); + } finally { + closeSync(logFd); + } + child.unref(); + const state = { placement: "host-process", pid: child.pid ?? null, runId: run.id, commandId, runnerId, attemptId, state: "running", startedAt: new Date().toISOString(), logPath, valuesPrinted: false }; + writeFileSync(statePath, `${JSON.stringify(state, null, 2)}\n`, { encoding: "utf8", mode: 0o600 }); + child.once("exit", (code, signal) => { + writeFileSync(statePath, `${JSON.stringify({ ...state, state: "exited", exitCode: code, signal, exitedAt: new Date().toISOString() }, null, 2)}\n`, { encoding: "utf8", mode: 0o600 }); + }); + await options.store.appendEvent(run.id, "backend_status", { + phase: "host-process-runner-started", + commandId, + runnerId, + attemptId, + pid: child.pid ?? null, + logPath, + statePath, + valuesPrinted: false, + }); + return { action: "create-host-process", mutation: true, runId: run.id, commandId, runnerId, attemptId, pid: child.pid ?? null, logPath, statePath, valuesPrinted: false }; +} + +function childStarted(child: ReturnType): Promise { + return new Promise((resolve, reject) => { + child.once("spawn", resolve); + child.once("error", reject); + }); +} + +function transientEnvironment(value: unknown): NodeJS.ProcessEnv { + if (!Array.isArray(value)) return {}; + return Object.fromEntries(value.flatMap((item) => { + if (!item || typeof item !== "object" || Array.isArray(item)) return []; + const record = item as JsonRecord; + return typeof record.name === "string" && typeof record.value === "string" ? [[record.name, record.value]] : []; + })); +} + +function requiredString(value: unknown, field: string): string { + const result = optionalString(value); + if (!result) throw new AgentRunError("schema-invalid", `${field} is required`, { httpStatus: 400 }); + return result; +} + +function optionalString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function filePart(value: string): string { + return value.replace(/[^a-zA-Z0-9._-]/gu, "_").slice(0, 120); +} diff --git a/src/mgr/runner-dispatch-intent.ts b/src/mgr/runner-dispatch-intent.ts index b1c0e18..068b0f0 100644 --- a/src/mgr/runner-dispatch-intent.ts +++ b/src/mgr/runner-dispatch-intent.ts @@ -18,7 +18,7 @@ export function newRunnerDispatchIntent(command: CommandRecord, request: RunnerD id, runId: command.runId, commandId: command.id, - kind: "kubernetes-job", + kind: request.kind, input, inputHash: stableHash(input), state: "pending", diff --git a/src/mgr/runner-dispatcher.ts b/src/mgr/runner-dispatcher.ts index 7c215ef..b127a2e 100644 --- a/src/mgr/runner-dispatcher.ts +++ b/src/mgr/runner-dispatcher.ts @@ -1,13 +1,17 @@ +// SPEC: PJ2026-010306 嵌入式AgentBench +// SPEC_VERSION: draft-2026-07-24-p1-l0-l1-arm2d import type { JsonRecord, RunnerDispatchCompletion, RunnerDispatchIntentRecord } from "../common/types.js"; import { AgentRunError } from "../common/errors.js"; import { redactJson, redactText } from "../common/redaction.js"; import { nowIso } from "../common/validation.js"; import { runnerDispatchFailureKind, runnerDispatchFailureReason, type AgentRunStore } from "./store.js"; import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; +import { createHostProcessRunner, type HostProcessRunnerDefaults } from "./host-process-runner.js"; import { childProcessKillGraceMs } from "./abortable-child.js"; export interface RunnerDispatcherOptions { enabled: boolean; + placement: "host-process" | "kubernetes-job"; intervalMs: number; batchSize: number; leaseMs: number; @@ -20,6 +24,7 @@ export interface RunnerDispatcherOptions { export async function dispatchRunnerIntentsOnce(input: { store: AgentRunStore; defaults: RunnerJobDefaults; + hostProcessDefaults?: HostProcessRunnerDefaults; options: RunnerDispatcherOptions; signal?: AbortSignal; }): Promise { @@ -38,7 +43,13 @@ export async function dispatchRunnerIntentsOnce(input: { if (!intent) break; claimedCount++; try { - const result = await withAttemptDeadline(intent, input.options.attemptTimeoutMs, input.signal, async (signal) => await createKubernetesRunnerJob({ store: input.store, runId: intent.runId, input: intent.input as never, defaults: input.defaults, signal })); + const result = await withAttemptDeadline(intent, input.options.attemptTimeoutMs, input.signal, async (signal) => { + if (intent.kind === "host-process") { + if (!input.hostProcessDefaults) throw new AgentRunError("infra-failed", "host-process runner defaults are not configured", { httpStatus: 503 }); + return await createHostProcessRunner({ store: input.store, runId: intent.runId, input: intent.input, defaults: input.hostProcessDefaults, signal }); + } + return await createKubernetesRunnerJob({ store: input.store, runId: intent.runId, input: intent.input as never, defaults: input.defaults, signal }); + }); const completion = completionFromResult(result); await input.store.completeRunnerDispatchIntent(intent, completion, { phase: "runner-dispatch-completed", commandId: intent.commandId, dispatchIntentId: intent.id, plannedRunnerJobId: intent.runnerJobId, runnerJobId: completion.actualRunnerJobId, activeRunnerId: completion.activeRunnerId, dispatchOutcome: completion.dispatchOutcome, attemptCount: intent.attemptCount, adopted: result.adopted === true, valuesPrinted: false }); dispatchedCount++; @@ -76,6 +87,7 @@ export async function dispatchRunnerIntentsOnce(input: { export function startRunnerDispatcher(input: { store: AgentRunStore; defaults: RunnerJobDefaults; + hostProcessDefaults?: HostProcessRunnerDefaults; options: RunnerDispatcherOptions; onError?: (error: unknown) => void; }): () => void { @@ -106,7 +118,8 @@ function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): Js failureKind: runnerDispatchFailureKind(error instanceof AgentRunError ? error.failureKind : null), reason: runnerDispatchFailureReason(error instanceof AgentRunError ? details.reason : null), message, - kubernetesObservation: details, + dispatchObservation: details, + ...(intent.kind === "kubernetes-job" ? { kubernetesObservation: details } : { hostProcessObservation: details }), dispatchIntentId: intent.id, runId: intent.runId, commandId: intent.commandId, @@ -150,6 +163,11 @@ function completionFromResult(result: JsonRecord): RunnerDispatchCompletion { if (!activeRunnerId) throw new AgentRunError("infra-failed", "active runner reuse did not return runnerId", { httpStatus: 502 }); return { dispatchOutcome: "active-runner-reused", actualRunnerJobId: null, activeRunnerId }; } + if (result.action === "create-host-process") { + const activeRunnerId = stringValue(result.runnerId); + if (!activeRunnerId) throw new AgentRunError("infra-failed", "host-process runner dispatch did not return runnerId", { httpStatus: 502 }); + return { dispatchOutcome: "host-process", actualRunnerJobId: null, activeRunnerId }; + } if (result.action !== "create-kubernetes-job") throw new AgentRunError("infra-failed", "runner dispatch returned an unsupported outcome", { httpStatus: 502 }); const actualRunnerJobId = stringValue(result.runnerJobId); if (!actualRunnerJobId) throw new AgentRunError("infra-failed", "Kubernetes runner dispatch did not return runnerJobId", { httpStatus: 502 }); diff --git a/src/mgr/server.ts b/src/mgr/server.ts index aea7d1e..39901b5 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -1,3 +1,5 @@ +// SPEC: PJ2026-010306 嵌入式AgentBench +// SPEC_VERSION: draft-2026-07-24-p1-l0-l1-arm2d import type { Server } from "node:http"; import { createServer } from "node:http"; import type { AddressInfo } from "node:net"; @@ -27,6 +29,7 @@ import { aipodSpecFromInput, applyAipodSpec, deleteAipodSpec, listAipodSpecs, re import { staticWorkReadyCapabilitySummary } from "../common/work-ready.js"; import { agentRunBusinessTraceId, agentRunDiagnosticOtelTraceContext, emitAgentRunOtelSpan, type AgentRunDiagnosticOtelTraceContext } from "../common/otel-trace.js"; import { recordRunnerBootObservation } from "./runner-boot-observation.js"; +import type { HostProcessRunnerDefaults } from "./host-process-runner.js"; function pvcOptions(defaults: { kubectlCommand?: string } | undefined): SessionPvcOptions { return defaults?.kubectlCommand ? { kubectlCommand: defaults.kubectlCommand } : {}; @@ -84,17 +87,27 @@ function runnerJobDefaultsForRequest(defaults: ManagerServerOptions["runnerJobDe }; } -function assertRunnerRuntimeConfigured(defaults: ManagerServerOptions["runnerJobDefaults"]): void { - const required = [ +function assertRunnerRuntimeConfigured(defaults: ManagerServerOptions["runnerJobDefaults"], dispatcher: RunnerDispatcherOptions, hostDefaults: HostProcessRunnerDefaults | undefined): void { + if (!dispatcher.enabled) return; + const shared = [ ["AGENTRUN_RUNTIME_NAMESPACE", defaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE], ["AGENTRUN_INTERNAL_MGR_URL", defaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL], + ["AGENTRUN_LANE", defaults?.lane ?? process.env.AGENTRUN_LANE], + ] as const; + const placementRequired = dispatcher.placement === "host-process" ? [ + ["AGENTRUN_HOST_RUNNER_COMMAND", hostDefaults?.command], + ["AGENTRUN_HOST_RUNNER_ENTRYPOINT", hostDefaults?.entrypoint], + ["AGENTRUN_HOST_RUNNER_CWD", hostDefaults?.cwd], + ["AGENTRUN_HOST_RUNNER_STATE_DIR", hostDefaults?.stateDir], + ["AGENTRUN_HOST_RUNNER_LOG_DIR", hostDefaults?.logDir], + ] as const : [ ["AGENTRUN_RUNNER_IMAGE", defaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE], ["AGENTRUN_RUNNER_SERVICE_ACCOUNT", defaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT], ["AGENTRUN_RUNNER_JOB_NAME_PREFIX", defaults?.jobNamePrefix ?? process.env.AGENTRUN_RUNNER_JOB_NAME_PREFIX], ["AGENTRUN_RUNNER_API_KEY_SECRET_NAME", defaults?.runnerApiKeySecretRef?.name ?? process.env.AGENTRUN_RUNNER_API_KEY_SECRET_NAME], ["AGENTRUN_RUNNER_API_KEY_SECRET_KEY", defaults?.runnerApiKeySecretRef?.key ?? process.env.AGENTRUN_RUNNER_API_KEY_SECRET_KEY], - ["AGENTRUN_LANE", defaults?.lane ?? process.env.AGENTRUN_LANE], ] as const; + const required = [...shared, ...placementRequired]; const missing = required.filter(([, value]) => typeof value !== "string" || value.trim().length === 0).map(([name]) => name); if (missing.length > 0) { throw new AgentRunError("infra-failed", "durable runner dispatcher runtime configuration is incomplete", { httpStatus: 503, details: { missing, valuesPrinted: false } }); @@ -191,8 +204,10 @@ function runnerReconcilerLogSummary(result: JsonRecord): JsonRecord { function runnerDispatcherOptionsForRuntime(defaults: ManagerServerOptions["runnerDispatcherOptions"] | undefined): RunnerDispatcherOptions { const enabled = defaults?.enabled ?? optionalBooleanEnv("AGENTRUN_RUNNER_DISPATCHER_ENABLED", process.env.AGENTRUN_RUNNER_DISPATCHER_ENABLED, false); + const placement = defaults?.placement ?? runnerPlacement(process.env.AGENTRUN_RUNNER_PLACEMENT ?? "kubernetes-job"); const options: RunnerDispatcherOptions = { enabled, + placement, intervalMs: runtimePositiveInteger(defaults?.intervalMs, "AGENTRUN_RUNNER_DISPATCHER_INTERVAL_MS", enabled), batchSize: runtimePositiveInteger(defaults?.batchSize, "AGENTRUN_RUNNER_DISPATCHER_BATCH_SIZE", enabled), leaseMs: runtimePositiveInteger(defaults?.leaseMs, "AGENTRUN_RUNNER_DISPATCHER_LEASE_MS", enabled), @@ -205,6 +220,40 @@ function runnerDispatcherOptionsForRuntime(defaults: ManagerServerOptions["runne return options; } +function hostProcessRunnerDefaultsForRuntime(defaults: ManagerServerOptions["hostProcessRunnerDefaults"], runnerDefaults: RunnerJobDefaults): HostProcessRunnerDefaults | undefined { + const command = defaults?.command ?? process.env.AGENTRUN_HOST_RUNNER_COMMAND; + const entrypoint = defaults?.entrypoint ?? process.env.AGENTRUN_HOST_RUNNER_ENTRYPOINT; + const cwd = defaults?.cwd ?? process.env.AGENTRUN_HOST_RUNNER_CWD; + const stateDir = defaults?.stateDir ?? process.env.AGENTRUN_HOST_RUNNER_STATE_DIR; + const logDir = defaults?.logDir ?? process.env.AGENTRUN_HOST_RUNNER_LOG_DIR; + if (!command || !entrypoint || !cwd || !stateDir || !logDir) return undefined; + const apiKeyFile = defaults?.apiKeyFile ?? process.env.AGENTRUN_API_KEY_FILE; + const runnerIdleTimeoutMs = defaults?.runnerIdleTimeoutMs ?? runnerDefaults.runnerIdleTimeoutMs; + const backendRetryMaxAttempts = defaults?.backendRetryMaxAttempts ?? runnerDefaults.backendRetryMaxAttempts; + const backendRetryInitialBackoffMs = defaults?.backendRetryInitialBackoffMs ?? runnerDefaults.backendRetryInitialBackoffMs; + const backendRetryMaxBackoffMs = defaults?.backendRetryMaxBackoffMs ?? runnerDefaults.backendRetryMaxBackoffMs; + return { + managerUrl: defaults?.managerUrl ?? runnerDefaults.managerUrl, + sourceCommit: defaults?.sourceCommit ?? runnerDefaults.sourceCommit, + command, + entrypoint, + cwd, + stateDir, + logDir, + ...(apiKeyFile ? { apiKeyFile } : {}), + ...(runnerIdleTimeoutMs ? { runnerIdleTimeoutMs } : {}), + ...(backendRetryMaxAttempts ? { backendRetryMaxAttempts } : {}), + ...(backendRetryInitialBackoffMs ? { backendRetryInitialBackoffMs } : {}), + ...(backendRetryMaxBackoffMs ? { backendRetryMaxBackoffMs } : {}), + ...(defaults?.env ? { env: defaults.env } : {}), + }; +} + +function runnerPlacement(value: string): "host-process" | "kubernetes-job" { + if (value === "host-process" || value === "kubernetes-job") return value; + throw new AgentRunError("schema-invalid", "AGENTRUN_RUNNER_PLACEMENT must be host-process or kubernetes-job", { httpStatus: 500 }); +} + function kafkaOutboxRelayOptionsForRuntime(defaults: ManagerServerOptions["kafkaOutboxRelayOptions"] | undefined): KafkaOutboxRelayOptions { const enabled = defaults?.enabled ?? envFlag(process.env.AGENTRUN_KAFKA_ENABLED); const config = defaults?.config ?? agentRunKafkaConfig(process.env); @@ -256,7 +305,8 @@ export interface ManagerServerOptions { providerProfileOptions?: { namespace?: string; kubectlCommand?: string; fetchImpl?: typeof fetch; modelCatalogTimeoutMs?: number }; toolCredentialOptions?: { namespace?: string; kubectlCommand?: string }; runnerReconcilerOptions?: { enabled?: boolean; namespace?: string; kubectlCommand?: string; intervalMs?: number; batchSize?: number; startupRecovery?: RunnerStartupRecoveryOptions }; - runnerDispatcherOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; maxAttempts?: number; retryBackoffMs?: number; attemptTimeoutMs?: number; owner?: string }; + runnerDispatcherOptions?: { enabled?: boolean; placement?: "host-process" | "kubernetes-job"; intervalMs?: number; batchSize?: number; leaseMs?: number; maxAttempts?: number; retryBackoffMs?: number; attemptTimeoutMs?: number; owner?: string }; + hostProcessRunnerDefaults?: Partial & Pick; kafkaOutboxRelayOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; retryBackoffMs?: number; owner?: string; config?: AgentRunKafkaConfig }; aipodSpecDir?: string; auth?: ManagerAuthConfig; @@ -289,8 +339,9 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit); const runnerReconcilerOptions = runnerReconcilerOptionsForRuntime(options.runnerReconcilerOptions, resolvedRunnerJobDefaults); const runnerDispatcherOptions = runnerDispatcherOptionsForRuntime(options.runnerDispatcherOptions); + const hostProcessRunnerDefaults = hostProcessRunnerDefaultsForRuntime(options.hostProcessRunnerDefaults, resolvedRunnerJobDefaults); const kafkaOutboxRelayOptions = kafkaOutboxRelayOptionsForRuntime(options.kafkaOutboxRelayOptions); - assertRunnerRuntimeConfigured(runnerJobDefaults); + assertRunnerRuntimeConfigured(runnerJobDefaults, runnerDispatcherOptions, hostProcessRunnerDefaults); const aipodSpecDir = options.aipodSpecDir ?? process.env.AGENTRUN_AIPOD_SPEC_DIR; const auth = options.auth ?? managerServerAuthConfigFromEnv(); const authSummary = managerAuthSummary(auth); @@ -306,7 +357,7 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr onResult: (result) => console.info(JSON.stringify(runnerReconcilerLogSummary(result))), onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })), }) : null; - const stopRunnerDispatcher = runnerDispatcherOptions.enabled ? startRunnerDispatcher({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions, onError: (error) => console.warn(JSON.stringify({ code: "runner-dispatcher-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; + const stopRunnerDispatcher = runnerDispatcherOptions.enabled ? startRunnerDispatcher({ store, defaults: resolvedRunnerJobDefaults, ...(hostProcessRunnerDefaults ? { hostProcessDefaults: hostProcessRunnerDefaults } : {}), options: runnerDispatcherOptions, onError: (error) => console.warn(JSON.stringify({ code: "runner-dispatcher-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; const stopKafkaOutboxRelay = kafkaOutboxRelayOptions.enabled ? startKafkaEventOutboxRelay({ store, options: kafkaOutboxRelayOptions, onError: (error) => console.warn(JSON.stringify({ code: "kafka-outbox-relay-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; const server = createServer(async (req, res) => { const traceId = `trc_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`; @@ -327,7 +378,7 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr if (diagnosticTraceContext !== null && !isSupportedDiagnosticReadPath(url.pathname)) { throw new AgentRunError("schema-invalid", "diagnostic trace context is not supported for this GET route", { httpStatus: 400, details: { reason: "diagnostic-trace-route-unsupported", route: url.pathname, valuesPrinted: false } }); } - const data = await route({ method, url, body: await readBody(req), signal: requestAbort.signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, ...(runnerJobDefaults ? { runnerJobDefaults } : {}), ...(sessionPvcDefaults ? { sessionPvcDefaults } : {}), ...(providerProfileDefaults ? { providerProfileDefaults } : {}), ...(toolCredentialDefaults ? { toolCredentialDefaults } : {}), ...(aipodSpecDir ? { aipodSpecDir } : {}) }); + const data = await route({ method, url, body: await readBody(req), signal: requestAbort.signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, ...(hostProcessRunnerDefaults ? { hostProcessRunnerDefaults } : {}), ...(runnerJobDefaults ? { runnerJobDefaults } : {}), ...(sessionPvcDefaults ? { sessionPvcDefaults } : {}), ...(providerProfileDefaults ? { providerProfileDefaults } : {}), ...(toolCredentialDefaults ? { toolCredentialDefaults } : {}), ...(aipodSpecDir ? { aipodSpecDir } : {}) }); writeJson(res, 200, { ok: true, data, traceId }); } catch (error) { const agentError = normalizeError(error); @@ -684,7 +735,7 @@ function commandResultOtelAttributes(result: JsonValue): JsonRecord { }; } -async function route({ method, url, body, signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerJobDefaults, sessionPvcDefaults, providerProfileDefaults, toolCredentialDefaults, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, aipodSpecDir }: { method: string; url: URL; body: unknown; signal: AbortSignal; store: AgentRunStore; sourceCommit: string; authSummary?: JsonRecord; diagnosticTraceContext: AgentRunDiagnosticOtelTraceContext | null; runnerJobDefaults?: NonNullable; sessionPvcDefaults?: NonNullable; providerProfileDefaults?: NonNullable; toolCredentialDefaults?: NonNullable; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise { +async function route({ method, url, body, signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerJobDefaults, hostProcessRunnerDefaults, sessionPvcDefaults, providerProfileDefaults, toolCredentialDefaults, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, aipodSpecDir }: { method: string; url: URL; body: unknown; signal: AbortSignal; store: AgentRunStore; sourceCommit: string; authSummary?: JsonRecord; diagnosticTraceContext: AgentRunDiagnosticOtelTraceContext | null; runnerJobDefaults?: NonNullable; hostProcessRunnerDefaults?: HostProcessRunnerDefaults; sessionPvcDefaults?: NonNullable; providerProfileDefaults?: NonNullable; toolCredentialDefaults?: NonNullable; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise { const path = url.pathname; if (method === "GET" && (path === "/health" || path === "/health/live" || path === "/health/readiness")) { const [database, runnerDispatch, kafkaEventOutbox] = await Promise.all([store.health(), store.runnerDispatchStatus(), store.kafkaEventOutboxStatus()]); @@ -1108,7 +1159,7 @@ async function route({ method, url, body, signal, store, sourceCommit, authSumma }) as JsonValue; } if (method === "POST" && path === "/api/v1/reconciler/runner-dispatch") { - return await dispatchRunnerIntentsOnce({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions }) as JsonValue; + return await dispatchRunnerIntentsOnce({ store, defaults: resolvedRunnerJobDefaults, ...(hostProcessRunnerDefaults ? { hostProcessDefaults: hostProcessRunnerDefaults } : {}), options: runnerDispatcherOptions }) as JsonValue; } if (method === "POST" && path === "/api/v1/reconciler/kafka-event-outbox") { return await relayKafkaEventOutboxOnce({ store, options: kafkaOutboxRelayOptions }) as JsonValue; @@ -1279,7 +1330,7 @@ async function sendToSession(input: { store: AgentRunStore; sessionId: string; b const requestedRun = validateCreateRun(runBody); if (existing) assertSessionBoundary(existing, requestedRun); const commandInput = validateCreateCommand(commandBody); - if (createRunnerJob) commandInput.dispatch = { kind: "kubernetes-job", input: pendingAdmission?.intent?.input ?? runnerJobBody }; + if (createRunnerJob) commandInput.dispatch = { kind: input.runnerDispatcherOptions.placement, input: pendingAdmission?.intent?.input ?? runnerJobBody }; const admitted = await input.store.admitSessionTurn({ sessionId: input.sessionId, run: pendingAdmission ? frozenRunContract(pendingAdmission.run) : requestedRun, command: commandInput }); const { run, command } = admitted; if (admitted.disposition === "created" && !admitted.runReused) void emitAgentRunOtelSpan("run_created", run, process.env, { startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", backendProfile: run.backendProfile, providerId: run.providerId } }); diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index 92fe2bc..eb2299d 100644 --- a/src/runner/run-once.ts +++ b/src/runner/run-once.ts @@ -1,3 +1,5 @@ +// SPEC: PJ2026-010306 嵌入式AgentBench +// SPEC_VERSION: draft-2026-07-24-p1-l0-l1-arm2d import { appendFile, mkdir, writeFile } from "node:fs/promises"; import path from "node:path"; import { RunnerManagerApi, failureKindFromError, terminalStatusForFailure, errorMessage } from "./manager-api.js"; @@ -102,9 +104,7 @@ export async function runOnce(options: RunnerOnceOptions): Promise { const failureKind = failureKindFromError(error); const message = errorMessage(error); const details = failureDetailsFromError(error); - await api.appendEvent(options.runId, { type: "error", payload: { failureKind, message, phase: "runner-image-work-ready-smoke", attemptId, runnerId: runner.id, ...(details ? { details } : {}) } }); - const finalRun = await api.reportStatus(options.runId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, failureMessage: message }) as RunRecord; - return { runner, terminalStatus: finalRun.terminalStatus, failureKind, run: finalRun, commandsProcessed: 0, commandResults: [], stopped: "image-work-ready-smoke-failed" } as JsonRecord; + await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-capability-warning", checkScope: "runner-image-work-ready-smoke", blocking: false, failureKind, message, attemptId, runnerId: runner.id, authoritativeFact: "continue-runner-core-execution", ...(details ? { details } : {}), valuesPrinted: false } }); } let claimed: RunRecord; try { @@ -160,8 +160,13 @@ export async function runOnce(options: RunnerOnceOptions): Promise { initialPrompt = materialized.initialPrompt; await api.appendEvent(options.runId, { type: "backend_status", payload: { ...materialized.event, commandId: command.id, attemptId, runnerId: runner.id } }); if (requiresBundledWorkReadyTools(claimed)) { - const bundleWorkReady = await smokeBundledWorkReadyCapabilities(resourceEnv ?? options.env ?? process.env); - await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-bundle-work-ready-smoke", commandId: command.id, attemptId, runnerId: runner.id, ...bundleWorkReady } }); + try { + const bundleWorkReady = await smokeBundledWorkReadyCapabilities(resourceEnv ?? options.env ?? process.env); + await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-bundle-work-ready-smoke", commandId: command.id, attemptId, runnerId: runner.id, ...bundleWorkReady } }); + } catch (error) { + const failureKind = failureKindFromError(error); + await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-capability-warning", checkScope: "runner-bundle-work-ready-smoke", blocking: false, failureKind, message: errorMessage(error), commandId: command.id, attemptId, runnerId: runner.id, authoritativeFact: "continue-materialized-resource-execution", details: failureDetailsFromError(error), valuesPrinted: false } }); + } } } } catch (error) { diff --git a/src/selftest/cases/21-runner-durable-dispatch.ts b/src/selftest/cases/21-runner-durable-dispatch.ts index 04bc111..bef988e 100644 --- a/src/selftest/cases/21-runner-durable-dispatch.ts +++ b/src/selftest/cases/21-runner-durable-dispatch.ts @@ -118,7 +118,7 @@ function defaults(kubectlCommand: string): RunnerJobDefaults { } function dispatcherOptions(maxAttempts: number): RunnerDispatcherOptions { - return { enabled: true, intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "selftest-dispatcher" }; + return { enabled: true, placement: "kubernetes-job", intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "selftest-dispatcher" }; } async function assertActiveRunnerReuseHasNoDanglingJob(kubectlCommand: string): Promise { diff --git a/src/selftest/cases/21a-runner-host-process.ts b/src/selftest/cases/21a-runner-host-process.ts new file mode 100644 index 0000000..8a31d66 --- /dev/null +++ b/src/selftest/cases/21a-runner-host-process.ts @@ -0,0 +1,81 @@ +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import type { CreateRunInput } from "../../common/types.js"; +import { validateCreateCommand } from "../../common/validation.js"; +import { dispatchRunnerIntentsOnce, type RunnerDispatcherOptions } from "../../mgr/runner-dispatcher.js"; +import { MemoryAgentRunStore } from "../../mgr/store.js"; +import type { RunnerJobDefaults } from "../../mgr/kubernetes-runner-job.js"; +import type { SelfTestCase } from "../harness.js"; + +const selfTest: SelfTestCase = async (context) => { + const store = new MemoryAgentRunStore(); + const run = store.createRun(runInput()); + const command = store.createCommand(run.id, validateCreateCommand({ + type: "turn", + payload: { prompt: "host process dispatch" }, + dispatch: { kind: "host-process", input: {} }, + })); + const result = await dispatchRunnerIntentsOnce({ + store, + defaults: kubernetesDefaults(), + hostProcessDefaults: { + managerUrl: "http://127.0.0.1:1", + sourceCommit: "selftest", + command: process.execPath, + entrypoint: path.join(context.root, "src/selftest/fixtures/fake-host-runner.mjs"), + cwd: context.root, + stateDir: path.join(context.tmp, "host-runner-state"), + logDir: path.join(context.tmp, "host-runner-logs"), + }, + options: dispatcherOptions(), + }); + assert.equal(result.dispatchedCount, 1); + assert.equal(result.failedCount, 0); + const intent = store.getRunnerDispatchIntent(command.id); + assert.equal(intent?.kind, "host-process"); + assert.equal(intent?.dispatchOutcome, "host-process"); + assert.match(intent?.activeRunnerId ?? "", /^runner_[a-f0-9]{24}$/u); + assert.equal(intent?.actualRunnerJobId, null); + assert.equal(store.listRunnerJobs(run.id, command.id).length, 0); + const started = store.listEvents(run.id, 0, 100).find((event) => event.payload.phase === "host-process-runner-started"); + assert.equal(typeof started?.payload.pid, "number"); + const statePath = String(started?.payload.statePath ?? ""); + const logPath = String(started?.payload.logPath ?? ""); + await waitFor(async () => (JSON.parse(await readFile(statePath, "utf8")) as { state?: string }).state === "exited", 2_000); + assert.match(await readFile(logPath, "utf8"), /"placement":"host-process"/u); + return { name: "runner-host-process", tests: ["host-process intent dispatches without Kubernetes", "host runner leaves PID, state, and log evidence"] }; +}; + +function dispatcherOptions(): RunnerDispatcherOptions { + return { enabled: true, placement: "host-process", intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts: 1, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "host-process-selftest" }; +} + +function kubernetesDefaults(): RunnerJobDefaults { + return { namespace: "unused", managerUrl: "http://127.0.0.1:1", runnerApiKeySecretRef: { name: "unused", key: "unused" }, image: "unused", sourceCommit: "selftest", kubectlCommand: "/must-not-be-called" }; +} + +function runInput(): CreateRunInput { + return { + tenantId: "hwlab", + projectId: "pikasTech/HWLAB", + workspaceRef: { kind: "opaque", path: "." }, + sessionRef: null, + resourceBundleRef: null, + providerId: "gpt.pika", + backendProfile: "codex", + executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 30_000, network: "default", secretScope: { providerCredentials: [], toolCredentials: [] } }, + traceSink: null, + }; +} + +async function waitFor(predicate: () => Promise, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error("timed out waiting for host runner exit evidence"); +} + +export default selfTest; diff --git a/src/selftest/cases/66-session-turn-admission.ts b/src/selftest/cases/66-session-turn-admission.ts index 3bcc06b..9d7143a 100644 --- a/src/selftest/cases/66-session-turn-admission.ts +++ b/src/selftest/cases/66-session-turn-admission.ts @@ -662,7 +662,7 @@ function defaults(kubectlCommand: string, withRetention = false): RunnerJobDefau } function dispatcherOptions(maxAttempts: number): RunnerDispatcherOptions { - return { enabled: true, intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "session-admission-selftest" }; + return { enabled: true, placement: "kubernetes-job", intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "session-admission-selftest" }; } class FailingSessionCommandStore extends MemoryAgentRunStore { diff --git a/src/selftest/fixtures/fake-host-runner.mjs b/src/selftest/fixtures/fake-host-runner.mjs new file mode 100644 index 0000000..4c60351 --- /dev/null +++ b/src/selftest/fixtures/fake-host-runner.mjs @@ -0,0 +1,9 @@ +await new Promise((resolve) => setTimeout(resolve, 20)); +console.log(JSON.stringify({ + ok: true, + placement: "host-process", + runId: process.env.AGENTRUN_RUN_ID ?? null, + commandId: process.env.AGENTRUN_COMMAND_ID ?? null, + runnerId: process.env.AGENTRUN_RUNNER_ID ?? null, + valuesPrinted: false, +}));