From 0629ee06e8a8717c7e4cc3b1033d30d24a548e47 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 24 Jul 2026 14:47:45 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=A1=A5=E6=8E=A5Workbench=E4=B8=8ECas?= =?UTF-8?q?eRun?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/hwlab-workbench-api/main.ts | 2 +- internal/harnessrl/contracts.ts | 9 ++ internal/harnessrl/harnessrl.test.ts | 30 ++++- internal/harnessrl/http.ts | 3 +- .../migrations/0001_harnessrl_registry.sql | 3 + internal/harnessrl/registry.ts | 11 +- internal/harnessrl/service.ts | 44 ++++++- internal/harnessrl/transport.ts | 11 +- internal/workbench/caserun-bridge.test.ts | 113 ++++++++++++++++++ internal/workbench/caserun-bridge.ts | 98 +++++++++++++++ internal/workbench/contracts.ts | 11 ++ internal/workbench/dispatcher.ts | 18 ++- .../workbench/native-agentrun-application.ts | 10 ++ internal/workbench/native-application.ts | 11 +- internal/workbench/runtime.ts | 21 +++- tools/src/workbench-caserun-cli.test.ts | 55 +++++++++ tools/src/workbench-cli.ts | 10 +- 17 files changed, 433 insertions(+), 27 deletions(-) create mode 100644 internal/workbench/caserun-bridge.test.ts create mode 100644 internal/workbench/caserun-bridge.ts create mode 100644 tools/src/workbench-caserun-cli.test.ts diff --git a/cmd/hwlab-workbench-api/main.ts b/cmd/hwlab-workbench-api/main.ts index 715ff04b..198d080b 100644 --- a/cmd/hwlab-workbench-api/main.ts +++ b/cmd/hwlab-workbench-api/main.ts @@ -3,7 +3,7 @@ import { createWorkbenchHttpApp } from "../../internal/workbench/http.ts"; import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; import { startHwlabKafkaEventBridge } from "../../internal/cloud/kafka-event-bridge.ts"; -const runtime = workbenchRuntime(process.env, { eventPublisher: null }); +const runtime = workbenchRuntime(process.env, { eventPublisher: null, caseRunTransportMode: "api" }); const kafkaEventBridge = runtime.mode === "agentrun-native" ? startHwlabKafkaEventBridge({ env: process.env }) : null; const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), providerModelCatalog: runtime.application.providerModelCatalog?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, mode: runtime.mode, kafkaEventBridge, close: async () => { await kafkaEventBridge?.stop?.(); await runtime.close(); } }); const host = process.env.WORKBENCH_API_HOST || "0.0.0.0"; diff --git a/internal/harnessrl/contracts.ts b/internal/harnessrl/contracts.ts index a50e5d4d..b0e2a37c 100644 --- a/internal/harnessrl/contracts.ts +++ b/internal/harnessrl/contracts.ts @@ -11,6 +11,12 @@ export type CaseAuthority = { definitionSha256: string; }; +export type CaseRunLaunchContext = { + workbenchSessionRef: { sessionId: string; actorId: string }; + hwpodId: string; + subjectRef?: string; +}; + export type CaseRunEvent = { id: number; runId: string; @@ -29,6 +35,7 @@ export type CaseRunRecord = { stage: string; terminal: boolean; caseAuthority: CaseAuthority; + launchContext: CaseRunLaunchContext | null; caseSnapshotDir: string; caseRepo: string; runDir: string; @@ -45,6 +52,7 @@ export type StartCaseRunInput = { caseId: string; workflowId: string; caseAuthority: CaseAuthority; + launchContext: CaseRunLaunchContext | null; caseSnapshotDir: string; caseRepo: string; runDir: string; @@ -71,6 +79,7 @@ export type CaseRunListItem = { evidenceStatus: string | null; artifactManifestSha256: string | null; caseAuthority: Pick; + launchContext: CaseRunLaunchContext | null; }; export type ActivityResult = { diff --git a/internal/harnessrl/harnessrl.test.ts b/internal/harnessrl/harnessrl.test.ts index dc41e48f..d5da8f2b 100644 --- a/internal/harnessrl/harnessrl.test.ts +++ b/internal/harnessrl/harnessrl.test.ts @@ -46,6 +46,30 @@ test("API restart preserves PostgreSQL-shaped registry read model", async () => assert.equal(record.workflowRunId, "temporal-run-1"); }); +test("Workbench launch context is frozen and conflicting run replay is rejected", async () => { + const fixture = await softwareSmokeFixture(); + const registry = new MemoryRegistry(); + const service = createHarnessRLService({ registry, temporal: new FakeTemporal(), cwd: fixture.root, caseRepo: fixture.root, stateRoot: fixture.stateRoot, caseAuthority: fixture.caseAuthority }); + const launchContext = { + workbenchSessionRef: { sessionId: "ses_caserun_arm2d", actorId: "usr_arm2d" }, + hwpodId: "d601-vm-stm32f405-qemu", + subjectRef: "f909cdaf6f0cd579b1f1b66d2d1eee5235a09fb9", + }; + + const accepted = await service.submit({ caseId: fixture.caseId, runId: "workbench-context", launchContext, runtimeApiUrl: "local://harnessrl" }); + expect(accepted?.launchContext).toEqual(launchContext); + expect((await service.listRuns()).runs[0].launchContext).toEqual(launchContext); + expect((await service.aggregate("workbench-context")).launchContext).toEqual(launchContext); + await expect(service.submit({ + caseId: fixture.caseId, + runId: "workbench-context", + launchContext: { ...launchContext, hwpodId: "another-hwpod" }, + runtimeApiUrl: "local://harnessrl", + })).rejects.toMatchObject({ code: "caserun_run_identity_conflict" }); + await expect(service.submit({ caseId: fixture.caseId, runId: "workbench-context", runtimeApiUrl: "local://harnessrl" })) + .rejects.toMatchObject({ code: "caserun_run_identity_conflict" }); +}); + test("run history lists shared registry results newest first with an opaque cursor", async () => { const fixture = await softwareSmokeFixture(); const registry = new MemoryRegistry(); @@ -92,7 +116,7 @@ test("run history lists shared registry results newest first with an opaque curs test("worker restart reuses idempotent prepare/build/collect activity results", async () => { const fixture = await softwareSmokeFixture(); const registry = new MemoryRegistry(); - await registry.createRun({ runId: "worker-restart", caseId: fixture.caseId, workflowId: "workflow-worker-restart", caseAuthority: fixture.authority, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", "worker-restart"), runtimeApiUrl: "http://127.0.0.1:6667" }); + await registry.createRun({ runId: "worker-restart", caseId: fixture.caseId, workflowId: "workflow-worker-restart", caseAuthority: fixture.authority, launchContext: null, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", "worker-restart"), runtimeApiUrl: "http://127.0.0.1:6667" }); const firstWorker = createHarnessRLActivities({ registry, env: {} }); await firstWorker.markRunning({ runId: "worker-restart" }); const prepared = await firstWorker.prepare({ runId: "worker-restart", identity: "worker-restart:prepare:v1" }); @@ -124,7 +148,7 @@ test("worker uses configured cloud-api authority and resubmits the same identity const registry = new MemoryRegistry(); const runId = "authority-retry"; const identity = `${runId}:agent:v1`; - await registry.createRun({ runId, caseId: fixture.caseId, workflowId: `workflow-${runId}`, caseAuthority: fixture.authority, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", runId), runtimeApiUrl: "https://public.example.test" }); + await registry.createRun({ runId, caseId: fixture.caseId, workflowId: `workflow-${runId}`, caseAuthority: fixture.authority, launchContext: null, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", runId), runtimeApiUrl: "https://public.example.test" }); await registry.saveActivityResult({ runId, activityName: "prepare", identity: `${runId}:prepare:v1`, output: { state: "completed", result: { mode: "hardware", prepared: { run: { runId } } } } }); await registry.saveActivityResult({ runId, activityName: "agent", identity, output: { state: "started", authorityIdentity: identity } }); let submitCount = 0; @@ -148,7 +172,7 @@ test("worker uses configured cloud-api authority and resubmits the same identity test("AgentRun and HWPOD crash gaps reuse stable authoritative operation identities", async () => { const fixture = await hardwareFixture(); const registry = new MemoryRegistry(); - await registry.createRun({ runId: "authority-gap", caseId: fixture.caseId, workflowId: "workflow-authority-gap", caseAuthority: fixture.authority, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", "authority-gap"), runtimeApiUrl: "http://127.0.0.1:6667" }); + await registry.createRun({ runId: "authority-gap", caseId: fixture.caseId, workflowId: "workflow-authority-gap", caseAuthority: fixture.authority, launchContext: null, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", "authority-gap"), runtimeApiUrl: "http://127.0.0.1:6667" }); let agentDispatchCount = 0; let hwpodDispatchCount = 0; let agentLookupCount = 0; diff --git a/internal/harnessrl/http.ts b/internal/harnessrl/http.ts index 67c7fc58..69ecb009 100644 --- a/internal/harnessrl/http.ts +++ b/internal/harnessrl/http.ts @@ -1,4 +1,5 @@ import { caseRunAccepted } from "./transport.ts"; +import type { CaseRunLaunchContext } from "./contracts.ts"; export function createHarnessRLHttpApp(options: { service: ReturnType }) { return { @@ -14,7 +15,7 @@ export function createHarnessRLHttpApp(options: { service: ReturnType 0 ? value : null; +} + function eventRow(row: any): CaseRunEvent { return { id: Number(row.id), runId: row.run_id, status: row.status, stage: row.stage, payload: row.payload ?? {}, createdAt: new Date(row.created_at).toISOString() }; } diff --git a/internal/harnessrl/service.ts b/internal/harnessrl/service.ts index 6717397a..1f7e0a47 100644 --- a/internal/harnessrl/service.ts +++ b/internal/harnessrl/service.ts @@ -6,7 +6,7 @@ import { Buffer } from "node:buffer"; import { mkdir, readdir, stat } from "node:fs/promises"; import path from "node:path"; -import type { CaseRunListItem, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, HarnessRLTemporalGateway } from "./contracts.ts"; +import type { CaseRunLaunchContext, CaseRunListItem, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, HarnessRLTemporalGateway } from "./contracts.ts"; import type { CaseAuthorityResolver } from "./case-authority.ts"; const RUN_ID_PATTERN = /^[A-Za-z0-9_.:-]{3,180}$/u; @@ -41,9 +41,10 @@ export function createHarnessRLService(options: { registry: HarnessRLRegistry; t nextCursor: hasMore && last ? encodeCursor(last) : null }; }, - async submit(input: { caseId: string; runId?: string; runtimeApiUrl: string }) { + async submit(input: { caseId: string; runId?: string; launchContext?: CaseRunLaunchContext; runtimeApiUrl: string }) { const caseId = opaqueId(input.caseId, "invalid_case_id"); const runId = input.runId ? opaqueId(input.runId, "invalid_run_id") : `${caseId}-${Date.now()}-${randomUUID().slice(0, 8)}`; + const launchContext = normalizeLaunchContext(input.launchContext); const workflowId = `harnessrl-caserun-${runId}`; const runDir = path.join(options.stateRoot, "runs", runId); const caseSnapshotDir = path.join(runDir, "case-authority"); @@ -51,14 +52,20 @@ export function createHarnessRLService(options: { registry: HarnessRLRegistry; t if (record && record.caseId !== caseId) { throw codedError("caserun_run_identity_conflict", "runId already belongs to another case", { runId, requestedCaseId: caseId, existingCaseId: record.caseId }); } + if (record && !sameLaunchContext(record.launchContext, launchContext)) { + throw codedError("caserun_run_identity_conflict", "runId already belongs to another Workbench launch context", { runId, requested: launchContext, existing: record.launchContext }); + } if (!record) { const resolved = await options.caseAuthority.resolve(caseId); await mkdir(runDir, { recursive: true }); await resolved.materialize(caseSnapshotDir); - record = await registry.createRun({ runId, caseId, workflowId, caseAuthority: resolved.authority, caseSnapshotDir, caseRepo: options.caseRepo, runDir, runtimeApiUrl: input.runtimeApiUrl }); + record = await registry.createRun({ runId, caseId, workflowId, caseAuthority: resolved.authority, launchContext, caseSnapshotDir, caseRepo: options.caseRepo, runDir, runtimeApiUrl: input.runtimeApiUrl }); + if (record.caseId !== caseId || !sameLaunchContext(record.launchContext, launchContext)) { + throw codedError("caserun_run_identity_conflict", "runId was concurrently admitted with another launch identity", { runId, requestedCaseId: caseId, existingCaseId: record.caseId }); + } } if (!record.workflowRunId) { - await registry.appendEvent(runId, "queued", "queued", { caseId, caseAuthority: record.caseAuthority }); + await registry.appendEvent(runId, "queued", "queued", { caseId, caseAuthority: record.caseAuthority, launchContext: record.launchContext }); const started = await temporal.start({ workflowId, runId }); await registry.updateRun(runId, { workflowRunId: started.workflowRunId }); if (started.reused) await registry.appendEvent(runId, "queued", "workflow-reused", { workflowId, workflowRunId: started.workflowRunId }); @@ -77,7 +84,7 @@ export function createHarnessRLService(options: { registry: HarnessRLRegistry; t async aggregate(runId: string) { const record = await this.getRun(runId); return { ok: true, contractVersion: "hwlab-caserun-aggregate-v1", runId: record.runId, caseId: record.caseId, - status: record.status, stage: record.stage, terminal: record.terminal, caseAuthority: record.caseAuthority, result: record.result, + status: record.status, stage: record.stage, terminal: record.terminal, caseAuthority: record.caseAuthority, launchContext: record.launchContext, result: record.result, sha256: record.result?.artifactManifestSha256 ?? null }; }, async cancel(runId: string) { @@ -114,7 +121,8 @@ function runListItem(record: CaseRunRecord): CaseRunListItem { ref: record.caseAuthority.ref, commit: record.caseAuthority.commit, definitionSha256: record.caseAuthority.definitionSha256 - } + }, + launchContext: record.launchContext }; } @@ -180,6 +188,30 @@ function stringValue(value: unknown): string | null { return typeof value === "string" && value ? value : null; } +function normalizeLaunchContext(value: CaseRunLaunchContext | undefined): CaseRunLaunchContext | null { + if (value === undefined) return null; + if (!value || typeof value !== "object" || Array.isArray(value)) throw codedError("invalid_launch_context", "launchContext must be an object"); + const workbenchSessionRef = value.workbenchSessionRef; + if (!workbenchSessionRef || typeof workbenchSessionRef !== "object" || Array.isArray(workbenchSessionRef)) { + throw codedError("invalid_launch_context", "launchContext.workbenchSessionRef is required"); + } + const sessionId = requiredLaunchText(workbenchSessionRef.sessionId, "launchContext.workbenchSessionRef.sessionId"); + const actorId = requiredLaunchText(workbenchSessionRef.actorId, "launchContext.workbenchSessionRef.actorId"); + const hwpodId = requiredLaunchText(value.hwpodId, "launchContext.hwpodId"); + const subjectRef = String(value.subjectRef ?? "").trim(); + return { workbenchSessionRef: { sessionId, actorId }, hwpodId, ...(subjectRef ? { subjectRef } : {}) }; +} + +function requiredLaunchText(value: unknown, field: string) { + const text = String(value ?? "").trim(); + if (!text || text.length > 500) throw codedError("invalid_launch_context", `${field} is required and must not exceed 500 characters`, { field }); + return text; +} + +function sameLaunchContext(left: CaseRunLaunchContext | null, right: CaseRunLaunchContext | null) { + return JSON.stringify(left) === JSON.stringify(right); +} + function finiteNumber(value: unknown): number | null { const number = Number(value); return Number.isFinite(number) && number >= 0 ? number : null; diff --git a/internal/harnessrl/transport.ts b/internal/harnessrl/transport.ts index a0fa54b0..fba53749 100644 --- a/internal/harnessrl/transport.ts +++ b/internal/harnessrl/transport.ts @@ -1,17 +1,17 @@ // SPEC: PJ2026-010306 嵌入式AgentBench. // SPEC_VERSION: draft-2026-07-24-p0-l0-l1-arm2d. -import type { CaseRunRecord } from "./contracts.ts"; +import type { CaseRunLaunchContext, CaseRunRecord } from "./contracts.ts"; export type CaseRunCommand = - | { kind: "start"; caseId: string; runId?: string } + | { kind: "start"; caseId: string; runId?: string; launchContext?: CaseRunLaunchContext } | { kind: "status"; runId: string } | { kind: "events"; runId: string } | { kind: "aggregate"; runId: string } | { kind: "cancel"; runId: string }; export type HarnessRLService = { - submit(input: { caseId: string; runId?: string; runtimeApiUrl: string }): Promise; + submit(input: { caseId: string; runId?: string; launchContext?: CaseRunLaunchContext; runtimeApiUrl: string }): Promise; getRun(runId: string): Promise; events(runId: string): Promise>; aggregate(runId: string): Promise>; @@ -47,7 +47,7 @@ export function createLocalCaseRunTransport(options: { service: HarnessRLService }, async execute(command) { try { - if (command.kind === "start") return caseRunAccepted(await options.service.submit({ caseId: command.caseId, runId: command.runId, runtimeApiUrl: "local://harnessrl" })); + if (command.kind === "start") return caseRunAccepted(await options.service.submit({ caseId: command.caseId, runId: command.runId, launchContext: command.launchContext, runtimeApiUrl: "local://harnessrl" })); if (command.kind === "status") return options.service.getRun(command.runId); if (command.kind === "events") return options.service.events(command.runId); if (command.kind === "aggregate") return options.service.aggregate(command.runId); @@ -81,7 +81,7 @@ export function createApiCaseRunTransport(options: { baseUrl: string; fetchImpl? ...(command.kind === "start" ? { "content-type": "application/json" } : {}), ...(options.apiKey ? { authorization: `Bearer ${options.apiKey}` } : {}) }, - body: command.kind === "start" ? JSON.stringify({ caseId: command.caseId, ...(command.runId ? { runId: command.runId } : {}) }) : undefined + body: command.kind === "start" ? JSON.stringify({ caseId: command.caseId, ...(command.runId ? { runId: command.runId } : {}), ...(command.launchContext ? { launchContext: command.launchContext } : {}) }) : undefined }); const body = await response.json().catch((error) => { if (controller.signal.aborted) throw error; @@ -114,6 +114,7 @@ export function caseRunAccepted(record: CaseRunRecord | null) { workflowId: record.workflowId, workflowRunId: record.workflowRunId, caseAuthority: record.caseAuthority, + launchContext: record.launchContext, status: record.status, stage: record.stage, statusUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}`, diff --git a/internal/workbench/caserun-bridge.test.ts b/internal/workbench/caserun-bridge.test.ts new file mode 100644 index 00000000..b53ead30 --- /dev/null +++ b/internal/workbench/caserun-bridge.test.ts @@ -0,0 +1,113 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, expect, test } from "bun:test"; + +import type { CaseRunCommand, CaseRunTransport } from "../harnessrl/transport.ts"; +import { createWorkbenchCaseRunLauncher, sessionIdForRun } from "./caserun-bridge.ts"; +import { createWorkbenchDispatcher } from "./dispatcher.ts"; +import { createNativeTestWorkbenchApplication } from "./native-application.ts"; +import { createNativeTestTemporalGateway } from "./native-temporal.ts"; + +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all(temporaryDirectories.splice(0).map((directory) => rm(directory, { recursive: true, force: true }))); +}); + +test("Workbench CaseRun bridge creates one deterministic session and freezes launch identity", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "workbench-caserun-bridge-")); + temporaryDirectories.push(directory); + const application = createNativeTestWorkbenchApplication({ stateFile: path.join(directory, "state.json") }); + const commands: CaseRunCommand[] = []; + const launcher = createWorkbenchCaseRunLauncher({ application, transport: () => fakeTransport(commands) }); + const input = { + actor: { id: "usr_arm2d", role: "user" }, + caseId: "d601-vm-stm32f405-qemu-arm2d-integration", + hwpodId: "d601-vm-stm32f405-qemu", + subjectRef: "f909cdaf6f0cd579b1f1b66d2d1eee5235a09fb9", + runId: "run_arm2d_bridge", + }; + + const first = await launcher.launch(input); + const replay = await launcher.launch(input); + const expectedSessionId = sessionIdForRun(input.runId); + + expect(first).toMatchObject({ + accepted: true, + sessionId: expectedSessionId, + runId: input.runId, + workflowId: `harnessrl-caserun-${input.runId}`, + hwpodId: input.hwpodId, + statusRef: `/v1/caserun/runs/${input.runId}`, + eventsRef: `/v1/caserun/runs/${input.runId}/events`, + aggregateRef: `/v1/caserun/runs/${input.runId}/aggregate`, + harnessTransport: "local", + }); + expect(replay.sessionId).toBe(expectedSessionId); + expect(commands).toHaveLength(2); + expect(commands[0]).toEqual({ + kind: "start", + caseId: input.caseId, + runId: input.runId, + launchContext: { + workbenchSessionRef: { sessionId: expectedSessionId, actorId: input.actor.id }, + hwpodId: input.hwpodId, + subjectRef: input.subjectRef, + }, + }); + const state = await application.snapshot!(); + expect(Object.keys(state.sessions)).toEqual([expectedSessionId]); + expect(state.sessions[expectedSessionId].launchContext).toEqual({ kind: "caserun", caseId: input.caseId, runId: input.runId, hwpodId: input.hwpodId, subjectRef: input.subjectRef }); +}); + +test("Workbench dispatcher exposes CaseRun as an explicit control operation", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "workbench-caserun-dispatcher-")); + temporaryDirectories.push(directory); + const application = createNativeTestWorkbenchApplication({ stateFile: path.join(directory, "state.json") }); + const caseRun = createWorkbenchCaseRunLauncher({ application, transport: () => fakeTransport([]) }); + const dispatch = createWorkbenchDispatcher({ application, temporal: createNativeTestTemporalGateway(application), caseRun, mode: "native-test" }); + + const result = await dispatch({ + operation: "caserun.start", + actor: { id: "usr_arm2d" }, + params: { caseId: "arm2d-integration", hwpodId: "d601-vm-stm32f405-qemu", runId: "run_dispatcher_arm2d" }, + }); + + expect(result).toMatchObject({ + ok: true, + operation: "caserun.start", + mode: "native-test", + data: { + sessionId: sessionIdForRun("run_dispatcher_arm2d"), + runId: "run_dispatcher_arm2d", + workflowId: "harnessrl-caserun-run_dispatcher_arm2d", + }, + }); +}); + +function fakeTransport(commands: CaseRunCommand[]): CaseRunTransport { + return { + mode: "local", + visibility() { return {}; }, + async execute(command) { + commands.push(command); + if (command.kind !== "start") throw new Error("start required"); + return { + ok: true, + runId: command.runId, + workflowId: `harnessrl-caserun-${command.runId}`, + workflowRunId: "temporal-run-1", + status: "queued", + stage: "queued", + caseAuthority: { commit: "a".repeat(40) }, + statusUrl: `/v1/caserun/runs/${command.runId}`, + eventsUrl: `/v1/caserun/runs/${command.runId}/events`, + aggregateUrl: `/v1/caserun/runs/${command.runId}/aggregate`, + cancelUrl: `/v1/caserun/runs/${command.runId}/cancel`, + }; + }, + async close() {}, + }; +} diff --git a/internal/workbench/caserun-bridge.ts b/internal/workbench/caserun-bridge.ts new file mode 100644 index 00000000..85aa8df9 --- /dev/null +++ b/internal/workbench/caserun-bridge.ts @@ -0,0 +1,98 @@ +// SPEC: PJ2026-010306 嵌入式AgentBench. +// SPEC_VERSION: draft-2026-07-24-p0-l0-l1-arm2d. + +import { createHash, randomUUID } from "node:crypto"; + +import type { CaseRunCommand, CaseRunTransport } from "../harnessrl/transport.ts"; +import type { WorkbenchApplication, WorkbenchCaseRunLauncher } from "./contracts.ts"; + +export function createWorkbenchCaseRunLauncher(options: { + application: WorkbenchApplication; + transport: () => CaseRunTransport | Promise; +}): WorkbenchCaseRunLauncher { + return { + async launch(input) { + const caseId = requiredText(input.caseId, "caseId"); + const hwpodId = requiredText(input.hwpodId, "hwpodId"); + const actorId = requiredText(input.actor.id, "actorId"); + const runId = optionalId(input.runId, "runId") ?? `run_${randomUUID()}`; + const sessionId = sessionIdForRun(runId); + const subjectRef = optionalText(input.subjectRef); + const launchContext = { + workbenchSessionRef: { sessionId, actorId }, + hwpodId, + ...(subjectRef ? { subjectRef } : {}), + }; + const created = await options.application.createSession({ + actor: input.actor, + params: { + sessionId, + projectId: "pikasTech/HWLAB", + launchContext: { kind: "caserun", caseId, runId, hwpodId, ...(subjectRef ? { subjectRef } : {}) }, + }, + }); + const actualSessionId = sessionIdFrom(created); + if (actualSessionId !== sessionId) { + throw codedError("workbench_caserun_session_identity_mismatch", "Workbench session creation returned another session identity", { expected: sessionId, actual: actualSessionId }); + } + const command: CaseRunCommand = { kind: "start", caseId, runId, launchContext }; + const transport = await options.transport(); + try { + const accepted = await transport.execute(command); + return { + accepted: true, + sessionId, + runId: requiredText(accepted.runId, "accepted.runId"), + workflowId: requiredText(accepted.workflowId, "accepted.workflowId"), + workflowRunId: optionalText(accepted.workflowRunId), + caseId, + hwpodId, + actorId, + subjectRef: subjectRef || null, + status: accepted.status, + stage: accepted.stage, + caseAuthority: accepted.caseAuthority, + statusRef: requiredText(accepted.statusUrl, "accepted.statusUrl"), + eventsRef: requiredText(accepted.eventsUrl, "accepted.eventsUrl"), + aggregateRef: requiredText(accepted.aggregateUrl, "accepted.aggregateUrl"), + cancelRef: requiredText(accepted.cancelUrl, "accepted.cancelUrl"), + harnessTransport: transport.mode, + }; + } finally { + await transport.close(); + } + }, + }; +} + +export function sessionIdForRun(runId: string) { + return `ses_caserun_${createHash("sha256").update(runId).digest("hex").slice(0, 32)}`; +} + +function sessionIdFrom(value: Record) { + const session = value.session && typeof value.session === "object" && !Array.isArray(value.session) + ? value.session as Record + : null; + return requiredText(session?.sessionId, "session.sessionId"); +} + +function optionalId(value: unknown, field: string) { + const text = optionalText(value); + if (!text) return null; + if (!/^[A-Za-z0-9_.:-]{3,180}$/u.test(text)) throw codedError("invalid_input", `${field} is invalid`, { field }); + return text; +} + +function requiredText(value: unknown, field: string) { + const text = optionalText(value); + if (!text) throw codedError("invalid_input", `${field} is required`, { field }); + return text; +} + +function optionalText(value: unknown) { + return String(value ?? "").trim(); +} + +function codedError(code: string, message: string, details?: unknown) { + return Object.assign(new Error(message), { code, details }); +} diff --git a/internal/workbench/contracts.ts b/internal/workbench/contracts.ts index a33c523d..3aca03ca 100644 --- a/internal/workbench/contracts.ts +++ b/internal/workbench/contracts.ts @@ -8,6 +8,7 @@ export type WorkbenchMode = "native-test" | "agentrun-native" | "temporal"; export type WorkbenchCommand = | { operation: "health" } | { operation: "session.create"; actor: WorkbenchActor; params: Record } + | { operation: "caserun.start"; actor: WorkbenchActor; params: Record } | { operation: "turn.submit"; actor: WorkbenchActor; params: Record; traceId?: string } | { operation: "turn.steer"; actor: WorkbenchActor; traceId: string; params: Record } | { operation: "turn.cancel"; actor: WorkbenchActor; traceId: string; params?: Record }; @@ -20,6 +21,16 @@ export type WorkbenchCommandResult = { error?: { code: string; message: string; details?: unknown }; }; +export interface WorkbenchCaseRunLauncher { + launch(input: { + actor: WorkbenchActor; + caseId: string; + hwpodId: string; + subjectRef?: string; + runId?: string; + }): Promise>; +} + export type WorkbenchTurnInput = { actor: WorkbenchActor; traceId: string; diff --git a/internal/workbench/dispatcher.ts b/internal/workbench/dispatcher.ts index 39313823..b01cb5c1 100644 --- a/internal/workbench/dispatcher.ts +++ b/internal/workbench/dispatcher.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; -import type { WorkbenchApplication, WorkbenchCommand, WorkbenchCommandResult, WorkbenchMode, WorkbenchTemporalGateway } from "./contracts.ts"; +import type { WorkbenchApplication, WorkbenchCaseRunLauncher, WorkbenchCommand, WorkbenchCommandResult, WorkbenchMode, WorkbenchTemporalGateway } from "./contracts.ts"; const SESSION_ID = /^ses_[A-Za-z0-9._:-]{3,180}$/u; const TRACE_ID = /^trc_[A-Za-z0-9._:-]{3,180}$/u; @@ -9,6 +9,7 @@ export function createWorkbenchDispatcher(options: { application: WorkbenchApplication; temporal: WorkbenchTemporalGateway; mode: WorkbenchMode; + caseRun?: WorkbenchCaseRunLauncher; }) { return async function dispatch(command: WorkbenchCommand): Promise { try { @@ -19,6 +20,16 @@ export function createWorkbenchDispatcher(options: { if (command.operation === "session.create") { return success(command.operation, options.mode, await options.application.createSession(command)); } + if (command.operation === "caserun.start") { + if (!options.caseRun) throw codedError("workbench_caserun_unavailable", "Workbench CaseRun bridge is unavailable"); + return success(command.operation, options.mode, await options.caseRun.launch({ + actor: command.actor, + caseId: requiredText(command.params.caseId, "caseId"), + hwpodId: requiredText(command.params.hwpodId, "hwpodId"), + subjectRef: optionalText(command.params.subjectRef), + runId: optionalText(command.params.runId), + })); + } if (command.operation === "turn.submit") { const traceId = validId(command.traceId ?? `trc_${randomUUID()}`, TRACE_ID, "traceId"); const sessionId = validId(command.params.sessionId, SESSION_ID, "sessionId"); @@ -100,6 +111,11 @@ function requiredText(value: unknown, field: string) { return text; } +function optionalText(value: unknown) { + const text = String(value ?? "").trim(); + return text || undefined; +} + function codedError(code: string, message: string, details?: unknown) { return Object.assign(new Error(message), { code, details }); } diff --git a/internal/workbench/native-agentrun-application.ts b/internal/workbench/native-agentrun-application.ts index 664e6d33..66758964 100644 --- a/internal/workbench/native-agentrun-application.ts +++ b/internal/workbench/native-agentrun-application.ts @@ -58,6 +58,13 @@ export function createNativeAgentRunWorkbenchApplication(options: { return mutate((state) => { const sessionId = validOptionalId(params.sessionId, "ses_") ?? `ses_${randomUUID()}`; const conversationId = validOptionalId(params.conversationId, "cnv_") ?? `cnv_${randomUUID()}`; + const existing = state.sessions[sessionId]; + if (existing) { + if (existing.ownerUserId !== actor.id || !sameLaunchContext(existing.launchContext, params.launchContext)) { + throw codedError("session_identity_conflict", "sessionId already belongs to another launch identity"); + } + return { status: "reused", session: existing, valuesRedacted: true, secretMaterialStored: false }; + } const timestamp = now(); const session = { sessionId, @@ -67,6 +74,7 @@ export function createNativeAgentRunWorkbenchApplication(options: { ownerRole: actor.role ?? "user", status: "idle", providerProfile: text(params.providerProfile) || text(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE) || "gpt.pika", + launchContext: objectValue(params.launchContext), createdAt: timestamp, updatedAt: timestamp, mode: "agentrun-native", @@ -370,6 +378,8 @@ async function load(stateFile: string): Promise { return { sessions: object(parsed.sessions), turns: object(parsed.turns) }; } function object(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record> : {}; } +function objectValue(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } +function sameLaunchContext(left: unknown, right: unknown) { return JSON.stringify(objectValue(left)) === JSON.stringify(objectValue(right)); } function text(value: unknown) { return String(value ?? "").trim(); } function validTimestamp(value: unknown) { const timestamp = text(value); return timestamp && Number.isFinite(Date.parse(timestamp)) ? new Date(Date.parse(timestamp)).toISOString() : null; } function requiredText(value: unknown, field: string) { const result = text(value); if (!result) throw codedError("invalid_input", `${field} is required`); return result; } diff --git a/internal/workbench/native-application.ts b/internal/workbench/native-application.ts index 1bf0c5a6..d986b18e 100644 --- a/internal/workbench/native-application.ts +++ b/internal/workbench/native-application.ts @@ -28,7 +28,14 @@ export function createNativeTestWorkbenchApplication(options: { stateFile: strin return update((state) => { const sessionId = validOptionalId(params.sessionId, "ses_") ?? `ses_${randomUUID()}`; const conversationId = validOptionalId(params.conversationId, "cnv_") ?? `cnv_${randomUUID()}`; - const session = { sessionId, conversationId, projectId: text(params.projectId) || "prj_hwpod_workbench", ownerUserId: actor.id, status: "idle", providerProfile: text(params.providerProfile) || "native-test", createdAt: now(), updatedAt: now(), mode: "native-test", messages: [] as Record[] }; + const existing = state.sessions[sessionId]; + if (existing) { + if (existing.ownerUserId !== actor.id || !sameLaunchContext(existing.launchContext, params.launchContext)) { + throw codedError("session_identity_conflict", "sessionId already belongs to another launch identity"); + } + return { status: "reused", session: existing, valuesRedacted: true, secretMaterialStored: false }; + } + const session = { sessionId, conversationId, projectId: text(params.projectId) || "prj_hwpod_workbench", ownerUserId: actor.id, status: "idle", providerProfile: text(params.providerProfile) || "native-test", launchContext: objectValue(params.launchContext), createdAt: now(), updatedAt: now(), mode: "native-test", messages: [] as Record[] }; state.sessions[sessionId] = session; return { status: "created", session, valuesRedacted: true, secretMaterialStored: false }; }); @@ -94,6 +101,8 @@ async function load(stateFile: string): Promise { return { sessions: object(parsed.sessions), turns: object(parsed.turns) }; } function object(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record> : {}; } +function objectValue(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } +function sameLaunchContext(left: unknown, right: unknown) { return JSON.stringify(objectValue(left)) === JSON.stringify(objectValue(right)); } function text(value: unknown) { return String(value ?? "").trim(); } function validOptionalId(value: unknown, prefix: string) { const id = text(value); if (!id) return null; if (!id.startsWith(prefix)) throw codedError("invalid_input", `${prefix.slice(0, -1)} id is invalid`); return id; } function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); } diff --git a/internal/workbench/runtime.ts b/internal/workbench/runtime.ts index 1e5ffa13..5e1114f6 100644 --- a/internal/workbench/runtime.ts +++ b/internal/workbench/runtime.ts @@ -1,5 +1,8 @@ import path from "node:path"; +import { harnessRLRuntime } from "../harnessrl/runtime.ts"; +import { createApiCaseRunTransport, createLocalCaseRunTransport } from "../harnessrl/transport.ts"; +import { createWorkbenchCaseRunLauncher } from "./caserun-bridge.ts"; import { createCloudWorkbenchApplication } from "./cloud-application.ts"; import { createWorkbenchDispatcher } from "./dispatcher.ts"; import { createNativeTestWorkbenchApplication } from "./native-application.ts"; @@ -9,7 +12,7 @@ import { createWorkbenchTemporalGateway } from "./temporal.ts"; import { createWorkbenchKafkaEventPublisher } from "./kafka-event-publisher.ts"; import type { WorkbenchEventPublisher } from "./contracts.ts"; -export function workbenchRuntime(env: Record = process.env, options: { eventPublisher?: WorkbenchEventPublisher | null } = {}) { +export function workbenchRuntime(env: Record = process.env, options: { eventPublisher?: WorkbenchEventPublisher | null; caseRunTransportMode?: "local" | "api" } = {}) { const mode = String(env.WORKBENCH_MODE ?? "").trim(); if (mode !== "native-test" && mode !== "agentrun-native" && mode !== "temporal") throw codedError("workbench_mode_required", "WORKBENCH_MODE must be native-test, agentrun-native, or temporal"); const stateFile = path.resolve(String(env.WORKBENCH_NATIVE_STATE_FILE ?? ".state/workbench-native/state.json")); @@ -33,6 +36,20 @@ export function workbenchRuntime(env: Record = proce retryMaximumAttempts: requiredPositiveInteger(env.WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS, "WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS") }; const temporal = mode === "native-test" ? createNativeTestTemporalGateway(application) : createWorkbenchTemporalGateway({ address: temporalAddress, namespace: temporalNamespace, taskQueue, activity: activity! }); + const caseRunTransportMode = options.caseRunTransportMode ?? "local"; + const caseRun = createWorkbenchCaseRunLauncher({ + application, + async transport() { + if (caseRunTransportMode === "api") { + return createApiCaseRunTransport({ + baseUrl: String(env.WORKBENCH_CASERUN_API_URL ?? ""), + apiKey: String(env.WORKBENCH_CASERUN_API_KEY ?? "") || undefined, + }); + } + const runtime = harnessRLRuntime(env); + return createLocalCaseRunTransport({ service: runtime.service, close: () => runtime.service.close() }); + }, + }); return { mode, stateFile, @@ -42,7 +59,7 @@ export function workbenchRuntime(env: Record = proce eventPublisher, application, temporal, - dispatch: createWorkbenchDispatcher({ application, temporal, mode }), + dispatch: createWorkbenchDispatcher({ application, temporal, mode, caseRun }), async close() { await temporal.close(); await application.close?.(); } }; } diff --git a/tools/src/workbench-caserun-cli.test.ts b/tools/src/workbench-caserun-cli.test.ts new file mode 100644 index 00000000..e0c4002c --- /dev/null +++ b/tools/src/workbench-caserun-cli.test.ts @@ -0,0 +1,55 @@ +import { expect, test } from "bun:test"; + +import { runWorkbenchCli } from "./workbench-cli.ts"; + +test("Workbench CaseRun CLI over-api preserves the shared command DTO", async () => { + let received: Record | null = null; + const server = Bun.serve({ + port: 0, + async fetch(request) { + received = await request.json() as Record; + return Response.json({ ok: true, operation: "caserun.start", mode: "agentrun-native", data: { + accepted: true, + sessionId: "ses_caserun_cli", + runId: "run_caserun_cli", + workflowId: "harnessrl-caserun-run_caserun_cli", + workflowRunId: "temporal-cli", + statusRef: "/v1/caserun/runs/run_caserun_cli", + eventsRef: "/v1/caserun/runs/run_caserun_cli/events", + aggregateRef: "/v1/caserun/runs/run_caserun_cli/aggregate", + } }, { status: 202 }); + }, + }); + try { + const result = await runWorkbenchCli([ + "caserun", "start", + "--case-id", "arm2d-integration", + "--hwpod-id", "d601-vm-stm32f405-qemu", + "--actor-id", "usr_arm2d", + "--subject-ref", "f909cdaf6f0cd579b1f1b66d2d1eee5235a09fb9", + "--run-id", "run_caserun_cli", + "--over-api", + ], { WORKBENCH_API_URL: `http://127.0.0.1:${server.port}` }); + + expect(received).toEqual({ + operation: "caserun.start", + actor: { id: "usr_arm2d", role: "user" }, + params: { + caseId: "arm2d-integration", + hwpodId: "d601-vm-stm32f405-qemu", + subjectRef: "f909cdaf6f0cd579b1f1b66d2d1eee5235a09fb9", + runId: "run_caserun_cli", + }, + }); + expect(result).toMatchObject({ + ok: true, + operation: "caserun.start", + transport: "api", + route: "POST /v1/workbench/commands", + httpStatus: 202, + identity: { sessionId: "ses_caserun_cli", runId: "run_caserun_cli", workflowId: "harnessrl-caserun-run_caserun_cli" }, + }); + } finally { + server.stop(true); + } +}); diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index 10a9d6d3..c0d02b84 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -13,7 +13,7 @@ export async function runWorkbenchCli(argv: string[], env: Record): W if (group === "health") return { operation: "health" }; const actor = { id: required(parsed, "actor-id"), role: parsed.values["actor-role"] ?? "user" }; if (group === "session" && action === "create") return { operation: "session.create", actor, params: compact({ sessionId: parsed.values["session-id"], conversationId: parsed.values["conversation-id"], projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"] }) }; + if (group === "caserun" && action === "start") return { operation: "caserun.start", actor, params: compact({ caseId: required(parsed, "case-id"), hwpodId: required(parsed, "hwpod-id"), subjectRef: parsed.values["subject-ref"], runId: parsed.values["run-id"] }) }; if (group === "turn" && action === "submit") return { operation: "turn.submit", actor, traceId: parsed.values["trace-id"], params: compact({ sessionId: required(parsed, "session-id"), message: required(parsed, "message"), projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"], model: parsed.values.model, reasoningEffort: parsed.values["reasoning-effort"], shortConnection: true }) }; if (group === "turn" && action === "steer") { const targetTraceId = required(parsed, "trace-id"); @@ -305,7 +307,7 @@ function parse(argv: string[]): Parsed { } return parsed; } -function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, traceId: data?.traceId ?? null, targetTraceId: data?.targetTraceId ?? null, steerTraceId: data?.steerTraceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; } +function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, runId: data?.runId ?? null, traceId: data?.traceId ?? null, targetTraceId: data?.targetTraceId ?? null, steerTraceId: data?.steerTraceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; } function record(value: unknown): Record | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } function positiveInteger(value: unknown, fallback: number): number { const parsed = Number.parseInt(String(value ?? ""), 10); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } const eventSemantics = ["user", "backend", "assistant", "terminal", "final"] as const;