fix: 统一 session send 续跑入口
This commit is contained in:
AgentRun Codex
+159
-2
@@ -6,8 +6,8 @@ import { openAgentRunStoreFromEnv } from "./store.js";
|
||||
import { AgentRunError, errorToJson } from "../common/errors.js";
|
||||
import { asRecord, validateBackendProfile, validateCreateCommand, validateCreateQueueTask, validateCreateRun, validateQueueTaskState, validateSessionListState } from "../common/validation.js";
|
||||
import { isBackendProfile } from "../common/backend-profiles.js";
|
||||
import type { ApiErrorBody, ApiOkBody, JsonRecord, JsonValue, QueueTaskRecord, RunEvent } from "../common/types.js";
|
||||
import { createKubernetesRunnerJob } from "./kubernetes-runner-job.js";
|
||||
import type { ApiErrorBody, ApiOkBody, CommandRecord, JsonRecord, JsonValue, QueueTaskRecord, RunEvent, RunRecord, SessionRecord } from "../common/types.js";
|
||||
import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js";
|
||||
import { dispatchQueueTask, refreshQueueTaskFromCore } from "./queue-dispatch.js";
|
||||
import { buildRunResult } from "./result.js";
|
||||
import { runnerJobStatusSummary } from "./runner-job-status.js";
|
||||
@@ -35,6 +35,21 @@ function sessionPvcOptionsForRequest(serverDefaults: { kubectlHandler?: import("
|
||||
return pvcOptions(runnerJobDefaults);
|
||||
}
|
||||
|
||||
function runnerJobDefaultsForRequest(defaults: ManagerServerOptions["runnerJobDefaults"], sourceCommit: string): RunnerJobDefaults {
|
||||
const namespace = defaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01";
|
||||
return {
|
||||
namespace,
|
||||
managerUrl: defaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`,
|
||||
image: defaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "",
|
||||
sourceCommit,
|
||||
...optionalStringRecord("envIdentity", defaults?.envIdentity ?? process.env.AGENTRUN_ENV_IDENTITY),
|
||||
...optionalStringRecord("artifactCatalogFile", defaults?.artifactCatalogFile ?? process.env.AGENTRUN_ARTIFACT_CATALOG_FILE),
|
||||
serviceAccountName: defaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT ?? "agentrun-v01-runner",
|
||||
...(defaults?.kubectlCommand ? { kubectlCommand: defaults.kubectlCommand } : {}),
|
||||
...(defaults?.unideskSshEndpointEnv ? { unideskSshEndpointEnv: defaults.unideskSshEndpointEnv } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
export interface ManagerServerOptions {
|
||||
store?: AgentRunStore;
|
||||
port?: number;
|
||||
@@ -341,6 +356,10 @@ async function route({ method, url, body, store, sourceCommit, authSummary, runn
|
||||
if (runId) input.runId = runId;
|
||||
return await store.listSessionOutput(sessionOutputMatch[1] ?? "", input) as unknown as JsonValue;
|
||||
}
|
||||
const sessionSendMatch = path.match(/^\/api\/v1\/sessions\/([^/]+)\/send$/u);
|
||||
if (method === "POST" && sessionSendMatch) {
|
||||
return await sendToSession({ store, sessionId: sessionSendMatch[1] ?? "", body, sourceCommit, runnerJobDefaults }) as JsonValue;
|
||||
}
|
||||
const sessionReadMatch = path.match(/^\/api\/v1\/sessions\/([^/]+)\/read$/u);
|
||||
if (method === "POST" && sessionReadMatch) {
|
||||
const record = body === null ? {} : asRecord(body, "read");
|
||||
@@ -606,6 +625,144 @@ async function applyNamedAipodSpec(name: string, body: unknown, dir?: string): P
|
||||
return await applyAipodSpec(spec, dir) as JsonValue;
|
||||
}
|
||||
|
||||
async function sendToSession(input: { store: AgentRunStore; sessionId: string; body: unknown; sourceCommit: string; runnerJobDefaults?: ManagerServerOptions["runnerJobDefaults"] }): Promise<JsonRecord> {
|
||||
const record = asRecord(input.body ?? {}, "sessionSend");
|
||||
const dryRun = record.dryRun === true;
|
||||
const existing = await input.store.getSession(input.sessionId);
|
||||
const active = existing ? await activeReceivableCommand(input.store, existing) : null;
|
||||
const payload = sessionSendPayload(record);
|
||||
const commandIdempotencyKey = optionalString(record.commandIdempotencyKey) ?? optionalString(record.idempotencyKey);
|
||||
if (active) {
|
||||
const commandBody: JsonRecord = { type: "steer", payload };
|
||||
if (commandIdempotencyKey) commandBody.idempotencyKey = commandIdempotencyKey;
|
||||
const request = { method: "POST", path: `/api/v1/runs/${active.run.id}/commands`, commandType: "steer", payloadBytes: jsonByteLength(payload), valuesPrinted: false };
|
||||
if (dryRun) return sessionSendPlan(input.sessionId, "steer", active, request, null);
|
||||
const command = await input.store.createCommand(active.run.id, validateCreateCommand(commandBody));
|
||||
return sessionSendResponse({ sessionId: input.sessionId, decision: "steer", run: active.run, command, runnerJob: null, activeBefore: active, dryRun: false });
|
||||
}
|
||||
|
||||
const runRecord = asRecord(record.run ?? record.runBase ?? null, "sessionSend.run");
|
||||
const runBody = sessionSendRunBody(input.sessionId, runRecord);
|
||||
const commandBody: JsonRecord = { type: "turn", payload };
|
||||
if (commandIdempotencyKey) commandBody.idempotencyKey = commandIdempotencyKey;
|
||||
const runnerJobBody = record.runnerJob === undefined || record.runnerJob === null ? {} : asRecord(record.runnerJob, "sessionSend.runnerJob");
|
||||
const createRunnerJob = record.createRunnerJob !== false;
|
||||
const request = {
|
||||
method: "POST",
|
||||
path: `/api/v1/sessions/${input.sessionId}/send`,
|
||||
commandType: "turn",
|
||||
runBytes: jsonByteLength(runBody),
|
||||
payloadBytes: jsonByteLength(payload),
|
||||
createRunnerJob,
|
||||
runnerJobBytes: createRunnerJob ? jsonByteLength(runnerJobBody) : 0,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
if (dryRun) return sessionSendPlan(input.sessionId, "turn", active, request, runBody);
|
||||
const run = await input.store.createRun(validateCreateRun(runBody));
|
||||
const command = await input.store.createCommand(run.id, validateCreateCommand(commandBody));
|
||||
let runnerJob: JsonValue = null;
|
||||
if (createRunnerJob) {
|
||||
runnerJob = await createKubernetesRunnerJob({
|
||||
store: input.store,
|
||||
runId: run.id,
|
||||
input: { ...runnerJobBody, commandId: command.id } as never,
|
||||
defaults: runnerJobDefaultsForRequest(input.runnerJobDefaults, input.sourceCommit),
|
||||
}) as unknown as JsonValue;
|
||||
}
|
||||
return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run, command, runnerJob, activeBefore: active, dryRun: false });
|
||||
}
|
||||
|
||||
async function activeReceivableCommand(store: AgentRunStore, session: SessionRecord): Promise<{ run: RunRecord; command: CommandRecord; reason: string; leaseExpired: boolean } | null> {
|
||||
if (!session.activeRunId || !session.activeCommandId) return null;
|
||||
const [run, command] = await Promise.all([store.getRun(session.activeRunId), store.getCommand(session.activeCommandId)]);
|
||||
if (run.sessionRef?.sessionId !== session.sessionId || command.runId !== run.id) return null;
|
||||
if (runIsTerminal(run) || commandIsTerminal(command)) return null;
|
||||
if (command.type !== "turn") return null;
|
||||
const leaseExpired = run.leaseExpiresAt ? Date.parse(run.leaseExpiresAt) <= Date.now() : false;
|
||||
if (leaseExpired) return null;
|
||||
if (command.state !== "acknowledged") return null;
|
||||
if (run.status !== "claimed" && run.status !== "running") return null;
|
||||
return { run, command, reason: "active-turn-running", leaseExpired };
|
||||
}
|
||||
|
||||
function sessionSendPayload(record: JsonRecord): JsonRecord {
|
||||
const payload = asJsonRecord(record.payload);
|
||||
if (payload) return payload;
|
||||
const prompt = optionalString(record.prompt) ?? optionalString(record.message) ?? optionalString(record.text);
|
||||
if (!prompt) throw new AgentRunError("schema-invalid", "sessions send requires payload or non-empty prompt/message/text", { httpStatus: 400 });
|
||||
return { prompt };
|
||||
}
|
||||
|
||||
function sessionSendRunBody(sessionId: string, runRecord: JsonRecord): JsonRecord {
|
||||
const sessionRef = asJsonRecord(runRecord.sessionRef) ?? {};
|
||||
const metadata = asJsonRecord(sessionRef.metadata) ?? {};
|
||||
return { ...runRecord, sessionRef: { ...sessionRef, sessionId, metadata } };
|
||||
}
|
||||
|
||||
function sessionSendPlan(sessionId: string, decision: "steer" | "turn", active: Awaited<ReturnType<typeof activeReceivableCommand>>, request: JsonRecord, runBody: JsonRecord | null): JsonRecord {
|
||||
return {
|
||||
action: "session-send-plan",
|
||||
dryRun: true,
|
||||
mutation: false,
|
||||
sessionId,
|
||||
decision,
|
||||
internalCommandType: decision,
|
||||
activeBefore: active ? activeBeforeSummary(active) : null,
|
||||
request,
|
||||
...(runBody ? { run: { bodyBytes: jsonByteLength(runBody), sessionRef: summarizeSendSessionRef(runBody), valuesPrinted: false } } : {}),
|
||||
next: { confirm: `./scripts/agentrun sessions send ${sessionId} --prompt-stdin`, note: "Remove --dry-run to perform the mutation. Manager will decide internal steer vs turn from durable session state." },
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function sessionSendResponse(input: { sessionId: string; decision: "steer" | "turn"; run: RunRecord; command: CommandRecord; runnerJob: JsonValue; activeBefore: Awaited<ReturnType<typeof activeReceivableCommand>>; dryRun: false }): JsonRecord {
|
||||
return {
|
||||
action: "session-send",
|
||||
dryRun: input.dryRun,
|
||||
mutation: true,
|
||||
sessionId: input.sessionId,
|
||||
decision: input.decision,
|
||||
internalCommandType: input.command.type,
|
||||
run: input.run as unknown as JsonRecord,
|
||||
command: input.command as unknown as JsonRecord,
|
||||
runnerJob: input.runnerJob,
|
||||
activeBefore: input.activeBefore ? activeBeforeSummary(input.activeBefore) : null,
|
||||
pollCommands: {
|
||||
show: `./scripts/agentrun sessions show ${input.sessionId} --reader-id cli`,
|
||||
trace: `./scripts/agentrun sessions trace ${input.sessionId} --after-seq 0 --limit 100`,
|
||||
output: `./scripts/agentrun sessions output ${input.sessionId} --after-seq 0 --limit 100`,
|
||||
read: `./scripts/agentrun sessions read ${input.sessionId} --reader-id cli`,
|
||||
cancel: `./scripts/agentrun sessions cancel ${input.sessionId}`,
|
||||
},
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function activeBeforeSummary(active: NonNullable<Awaited<ReturnType<typeof activeReceivableCommand>>>): JsonRecord {
|
||||
return { runId: active.run.id, commandId: active.command.id, commandState: active.command.state, runStatus: active.run.status, leaseExpiresAt: active.run.leaseExpiresAt, leaseExpired: active.leaseExpired, reason: active.reason, valuesPrinted: false };
|
||||
}
|
||||
|
||||
function summarizeSendSessionRef(runBody: JsonRecord): JsonRecord {
|
||||
const ref = asJsonRecord(runBody.sessionRef) ?? {};
|
||||
return { sessionId: optionalString(ref.sessionId), conversationId: optionalString(ref.conversationId), threadId: optionalString(ref.threadId), metadataKeys: Object.keys(asJsonRecord(ref.metadata) ?? {}).sort(), valuesPrinted: false };
|
||||
}
|
||||
|
||||
function runIsTerminal(run: RunRecord): boolean {
|
||||
return run.status === "completed" || run.status === "failed" || run.status === "blocked" || run.status === "cancelled";
|
||||
}
|
||||
|
||||
function commandIsTerminal(command: CommandRecord): boolean {
|
||||
return command.state === "completed" || command.state === "failed" || command.state === "cancelled";
|
||||
}
|
||||
|
||||
function optionalString(value: JsonValue | undefined): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
||||
}
|
||||
|
||||
function jsonByteLength(value: unknown): number {
|
||||
return Buffer.byteLength(JSON.stringify(value ?? null), "utf8");
|
||||
}
|
||||
|
||||
function integerQuery(url: URL, key: string, fallback: number): number {
|
||||
const value = Number(url.searchParams.get(key));
|
||||
return Number.isInteger(value) && value >= 0 ? value : fallback;
|
||||
|
||||
Reference in New Issue
Block a user