From 5eeeff5b356bb3bfd7a244082bc54350105148a7 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 24 Jul 2026 17:19:02 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=94=B1AgentRun=E9=80=89=E6=8B=A9?= =?UTF-8?q?=E5=8E=9F=E7=94=9Frunner=20placement?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cloud/code-agent-agentrun-adapter.ts | 36 ++++++++-- ...e-agent-agentrun-durable-admission.test.ts | 70 ++++++++++--------- .../native-agentrun-application.test.ts | 53 +++++++------- 3 files changed, 94 insertions(+), 65 deletions(-) diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 9c29a35a..50019065 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -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 } }); diff --git a/internal/cloud/code-agent-agentrun-durable-admission.test.ts b/internal/cloud/code-agent-agentrun-durable-admission.test.ts index aa0c1a5b..b8fb5045 100644 --- a/internal/cloud/code-agent-agentrun-durable-admission.test.ts +++ b/internal/cloud/code-agent-agentrun-durable-admission.test.ts @@ -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}`); diff --git a/internal/workbench/native-agentrun-application.test.ts b/internal/workbench/native-agentrun-application.test.ts index 9fbd7455..ed356b4d 100644 --- a/internal/workbench/native-agentrun-application.test.ts +++ b/internal/workbench/native-agentrun-application.test.ts @@ -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") {