mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 由AgentRun选择原生runner placement

Browse Source
This commit is contained in:
root
2026-07-24 17:19:02 +02:00
parent f4e9862573
commit 5eeeff5b35
3 changed files with 94 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+30 -6
View File
@@ -1,4 +1,6 @@
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; draft-2026-06-25-p0-session-warm-runner-contract; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
// SPEC: PJ2026-010306 嵌入式AgentBench
// SPEC_VERSION: draft-2026-07-24-p2-l0-l1-arm2d
// Responsibility: AgentRun command admission and control adapter; Kafka owns event/result projection.
import { randomUUID } from "node:crypto";
@@ -546,20 +548,42 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI
}
const dispatchParams = paramsWithAgentRunSessionThread(attemptParams, profileThreadId);
const runInput = buildAgentRunCreateRunInput({ params: dispatchParams, env, traceId, backendProfile, sessionId, toolCapabilities });
const dispatchInput = buildAgentRunDurableDispatchInput({ env, traceId, backendProfile });
const commandInput = buildAgentRunCommandInput({ params: dispatchParams, traceId, backendProfile, sessionId });
const runCreateStartedAt = Date.now();
run = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs", method: "POST", body: runInput, timeoutMs, env, traceStore, traceId, backendProfile, stage: "run-create" });
const admission = await agentRunDispatchJson({
fetchImpl,
managerUrl,
path: `/api/v1/sessions/${encodeURIComponent(sessionId)}/send`,
method: "POST",
body: {
payload: commandInput.payload,
commandIdempotencyKey: commandInput.idempotencyKey,
run: runInput,
runnerJob: dispatchInput,
createRunnerJob: true,
},
timeoutMs,
env,
traceStore,
traceId,
backendProfile,
stage: "session-turn-admission",
});
if (firstNonEmpty(admission?.decision, admission?.internalCommandType, admission?.command?.type) !== "turn") {
throw adapterError("agentrun_session_send_delivery_invalid", "AgentRun fresh session admission did not return a turn delivery.");
}
run = admission?.run;
command = admission?.command;
const runId = requiredString(run?.id, "run.id");
void emitCodeAgentOtelSpan("agentrun_run_create", traceId, env, { startTimeMs: runCreateStartedAt, attributes: { runId, backendProfile, managerUrl } });
traceStore.append(traceId, agentRunTraceEvent({
type: "backend", status: "running",
label: "agentrun:run:created",
message: "AgentRun run " + runId + " created through internal k3s Service DNS.",
message: "AgentRun manager 已原子接纳本次 session、run、command 与 runner dispatch。",
runId, backendProfile, waitingFor: "agentrun-command-create", valuesPrinted: false,
}, backendProfile));
const dispatchInput = buildAgentRunDurableDispatchInput({ env, traceId, backendProfile });
const commandInput = buildAgentRunCommandInput({ params: dispatchParams, traceId, backendProfile, sessionId, dispatchInput });
const commandCreateStartedAt = Date.now();
command = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/commands", method: "POST", body: commandInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, stage: "command-create" });
const commandCreateStartedAt = runCreateStartedAt;
const commandId = requiredString(command?.id, "command.id");
const dispatchIntent = requireAgentRunDurableDispatchIntent(command?.dispatchIntent);
void emitCodeAgentOtelSpan("agentrun_command_create", traceId, env, { startTimeMs: commandCreateStartedAt, attributes: { runId, commandId, backendProfile, managerUrl } });
internal/cloud/code-agent-agentrun-durable-admission.test.ts
+36 -34
View File
@@ -54,25 +54,25 @@ test("fresh gpt.pika turn returns running only after durable dispatch intent is
assert.equal(result.agentRun.durableDispatch, true);
assert.equal(result.agentRun.status, "dispatch-intent-pending");
const runCall = calls.find((call) => call.path === "/api/v1/runs");
assert.ok(runCall);
assert.deepEqual(runCall.body.workspaceRef, { kind: "opaque", path: "." });
const commandCall = calls.find((call) => call.path.endsWith("/commands"));
assert.ok(commandCall);
assert.equal(commandCall.body.idempotencyKey, traceId);
assert.equal(commandCall.body.payload.traceId, traceId);
assert.equal(commandCall.body.payload.sessionId, "ses_agentrun_gpt_pika_durable");
assert.equal(commandCall.body.payload.hwlabSessionId, "ses_gpt_pika_durable");
assert.equal(commandCall.body.payload.conversationId, "cnv_gpt_pika_durable");
assert.equal(commandCall.body.dispatch.kind, "kubernetes-job");
assert.equal(Object.hasOwn(commandCall.body.dispatch.input, "commandId"), false);
assert.equal(commandCall.body.dispatch.input.idempotencyKey, traceId);
assert.equal(Object.hasOwn(commandCall.body.dispatch.input, "namespace"), false);
assert.equal(Object.hasOwn(commandCall.body.dispatch.input, "image"), false);
assert.equal(Object.hasOwn(commandCall.body.dispatch.input, "serviceAccountName"), false);
assert.equal(commandCall.body.dispatch.input.transientEnv.every((entry) => entry.sensitive === false), true);
assert.equal(commandCall.body.dispatch.input.transientEnv.every((entry) => !entry.name.startsWith("AGENTRUN_") && !entry.name.startsWith("KUBERNETES_")), true);
assert.equal(JSON.stringify(commandCall.body).includes("hwl_live_must_not_be_persisted"), false);
const admissionCall = calls.find((call) => call.path === "/api/v1/sessions/ses_agentrun_gpt_pika_durable/send");
assert.ok(admissionCall);
assert.deepEqual(admissionCall.body.run.workspaceRef, { kind: "opaque", path: "." });
assert.equal(admissionCall.body.commandIdempotencyKey, traceId);
assert.equal(admissionCall.body.payload.traceId, traceId);
assert.equal(admissionCall.body.payload.sessionId, "ses_agentrun_gpt_pika_durable");
assert.equal(admissionCall.body.payload.hwlabSessionId, "ses_gpt_pika_durable");
assert.equal(admissionCall.body.payload.conversationId, "cnv_gpt_pika_durable");
assert.equal(Object.hasOwn(admissionCall.body, "dispatch"), false);
assert.equal(Object.hasOwn(admissionCall.body.runnerJob, "commandId"), false);
assert.equal(admissionCall.body.runnerJob.idempotencyKey, traceId);
assert.equal(Object.hasOwn(admissionCall.body.runnerJob, "namespace"), false);
assert.equal(Object.hasOwn(admissionCall.body.runnerJob, "image"), false);
assert.equal(Object.hasOwn(admissionCall.body.runnerJob, "serviceAccountName"), false);
assert.equal(admissionCall.body.runnerJob.transientEnv.every((entry) => entry.sensitive === false), true);
assert.equal(admissionCall.body.runnerJob.transientEnv.every((entry) => !entry.name.startsWith("AGENTRUN_") && !entry.name.startsWith("KUBERNETES_")), true);
assert.equal(JSON.stringify(admissionCall.body).includes("hwl_live_must_not_be_persisted"), false);
assert.equal(calls.some((call) => call.path === "/api/v1/runs"), false);
assert.equal(calls.some((call) => call.path.endsWith("/commands")), false);
assert.equal(calls.some((call) => call.path.endsWith("/runner-jobs")), false);
assert.ok(traceStore.snapshot(traceId).events.some((event) => event.label === "agentrun:dispatch-intent:admitted"));
});
@@ -194,22 +194,24 @@ function createAgentRunFetch(calls, { dispatchIntent }) {
if (method === "POST" && parsed.pathname === "/api/v1/sessions") {
return send({ session: { id: "ses_agentrun_gpt_pika_durable", metadata: {} } });
}
if (method === "POST" && parsed.pathname === "/api/v1/runs") {
if (method === "POST" && /^\/api\/v1\/sessions\/[^/]+\/send$/u.test(parsed.pathname)) {
return send({
id: "run_gpt_pika_durable",
status: "pending",
sessionRef: body.sessionRef,
backendProfile: body.backendProfile
});
}
if (method === "POST" && parsed.pathname === "/api/v1/runs/run_gpt_pika_durable/commands") {
return send({
id: "cmd_gpt_pika_durable",
runId: "run_gpt_pika_durable",
state: "pending",
type: "turn",
seq: 1,
dispatchIntent
decision: "turn",
internalCommandType: "turn",
run: {
id: "run_gpt_pika_durable",
status: "pending",
sessionRef: body.run.sessionRef,
backendProfile: body.run.backendProfile
},
command: {
id: "cmd_gpt_pika_durable",
runId: "run_gpt_pika_durable",
state: "pending",
type: "turn",
seq: 1,
dispatchIntent
}
});
}
throw new Error(`unexpected AgentRun call ${method} ${parsed.pathname}`);
internal/workbench/native-agentrun-application.test.ts
+28 -25
View File
@@ -71,12 +71,11 @@ test("L1 AgentRun application dispatches gpt.pika without polling or projecting
expect(steered).toMatchObject({ accepted: true, status: "running", traceId: "trc_steer_l1_gpt_pika", targetTraceId: input.traceId, steerTraceId: "trc_steer_l1_gpt_pika", requestedDelivery: "steer", delivery: "steer", agentRun: { steerCommandId: "cmd_l1_gpt_pika_steer", targetCommandId: "cmd_l1_gpt_pika" } });
expect(`${calls[0]?.method} ${calls[0]?.path}`).toBe("GET /api/v1/provider-profiles/gpt-pika/models");
expect(`${calls[1]?.method} ${calls[1]?.path}`).toMatch(/^GET \/api\/v1\/sessions\/ses_agentrun_/);
expect(calls.slice(2, 5).map((call) => `${call.method} ${call.path}`)).toEqual([
expect(calls.slice(2, 4).map((call) => `${call.method} ${call.path}`)).toEqual([
"POST /api/v1/sessions",
"POST /api/v1/runs",
"POST /api/v1/runs/run_l1_gpt_pika/commands",
expect.stringMatching(/^POST \/api\/v1\/sessions\/ses_agentrun_[A-Za-z0-9_]+\/send$/),
]);
expect(`${calls[5]?.method} ${calls[5]?.path}`).toMatch(/^POST \/api\/v1\/sessions\/ses_agentrun_[A-Za-z0-9_]+\/send$/);
expect(`${calls[4]?.method} ${calls[4]?.path}`).toMatch(/^POST \/api\/v1\/sessions\/ses_agentrun_[A-Za-z0-9_]+\/send$/);
expect(calls.every((call) => call.authorization === "Bearer test-agentrun-key")).toBe(true);
});
@@ -153,7 +152,7 @@ test("L0 publishes admission, bounded ConnectionRefused retry, and semantic term
if (url.pathname === "/api/v1/sessions") {
return Response.json({ ok: true, data: { session: { id: "ses_agentrun_l0_visible_failure", metadata: {} } } });
}
if (url.pathname === "/api/v1/runs") runCreateAttempts += 1;
if (/^\/api\/v1\/sessions\/[^/]+\/send$/u.test(url.pathname)) runCreateAttempts += 1;
throw Object.assign(new Error("Unable to connect. Is the computer able to access the url?"), { code: "ConnectionRefused" });
}) as typeof fetch,
now: () => "2026-07-20T11:45:01.000Z",
@@ -230,27 +229,31 @@ function fakeAgentRunManager(calls: Array<{ method: string; path: string; author
if (method === "POST" && url.pathname === "/api/v1/sessions") {
return send({ session: { id: "ses_agentrun_l1_gpt_pika", metadata: {} } });
}
if (method === "POST" && url.pathname === "/api/v1/runs") {
expect(body.backendProfile).toBe("gpt-pika");
return send({ id: "run_l1_gpt_pika", status: "pending", sessionRef: body.sessionRef, backendProfile: body.backendProfile });
}
if (method === "POST" && url.pathname === "/api/v1/runs/run_l1_gpt_pika/commands") {
expect(["hi", "first"]).toContain(body.payload.prompt);
expect(body.payload.submittedAt).toBe("2026-07-20T01:13:07.699Z");
if (body.payload.prompt === "hi") {
expect(body.payload.model).toBe("gpt-5.6");
expect(body.payload.reasoningEffort).toBe("high");
}
return send({
id: "cmd_l1_gpt_pika",
runId: "run_l1_gpt_pika",
state: "pending",
type: "turn",
seq: 1,
dispatchIntent: { id: "dispatch_l1_gpt_pika", state: "pending", runnerJobId: "rjob_l1_gpt_pika", attemptCount: 0, durable: true },
});
}
if (method === "POST" && /^\/api\/v1\/sessions\/ses_agentrun_[A-Za-z0-9_]+\/send$/u.test(url.pathname)) {
if (body.payload.requestedDelivery !== "steer") {
expect(body.run.backendProfile).toBe("gpt-pika");
expect(["hi", "first"]).toContain(body.payload.prompt);
expect(body.payload.submittedAt).toBe("2026-07-20T01:13:07.699Z");
if (body.payload.prompt === "hi") {
expect(body.payload.model).toBe("gpt-5.6");
expect(body.payload.reasoningEffort).toBe("high");
}
expect(body.createRunnerJob).toBe(true);
expect(body).not.toHaveProperty("dispatch");
return send({
decision: "turn",
internalCommandType: "turn",
run: { id: "run_l1_gpt_pika", status: "pending", sessionRef: body.run.sessionRef, backendProfile: body.run.backendProfile },
command: {
id: "cmd_l1_gpt_pika",
runId: "run_l1_gpt_pika",
state: "pending",
type: "turn",
seq: 1,
dispatchIntent: { id: "dispatch_l1_gpt_pika", state: "pending", runnerJobId: "rjob_l1_gpt_pika", attemptCount: 0, durable: true },
},
});
}
expect(body.payload.requestedDelivery).toBe("steer");
expect(body.payload.targetTraceId).toBe("trc_l1_gpt_pika_smoke");
if (steerDelivery === "steer") {