diff --git a/.gitignore b/.gitignore index e4d6de38..2e8a60db 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,7 @@ dist/ build/ coverage/ tmp/ +core.* # Go *.test diff --git a/.tekton/hwlab-nc01-v03-pac.yaml b/.tekton/hwlab-nc01-v03-pac.yaml index 905a279d..21b1bd12 100644 --- a/.tekton/hwlab-nc01-v03-pac.yaml +++ b/.tekton/hwlab-nc01-v03-pac.yaml @@ -77,7 +77,7 @@ spec: - name: registry-prefix value: 127.0.0.1:5000/hwlab - name: services - value: "hwlab-cloud-api,hwlab-workbench-runtime,hwlab-user-billing,hwlab-project-management,hwlab-tasktree-api,hwlab-tasktree-worker,hwlab-harnessrl-api,hwlab-harnessrl-worker,hwlab-cloud-web,hwlab-gateway,hwlab-edge-proxy,hwlab-agent-skills" + value: "hwlab-cloud-api,hwlab-workbench-runtime,hwlab-user-billing,hwlab-project-management,hwlab-tasktree-api,hwlab-tasktree-worker,hwlab-harnessrl-api,hwlab-harnessrl-worker,hwlab-workbench-api,hwlab-workbench-worker,hwlab-cloud-web,hwlab-gateway,hwlab-edge-proxy,hwlab-agent-skills" - name: base-image value: 127.0.0.1:5000/hwlab/hwlab-node20-base:20-bookworm-slim - name: build-cache-mode diff --git a/cmd/hwlab-workbench-api/main.ts b/cmd/hwlab-workbench-api/main.ts new file mode 100644 index 00000000..039d398d --- /dev/null +++ b/cmd/hwlab-workbench-api/main.ts @@ -0,0 +1,13 @@ +#!/usr/bin/env bun +import { createWorkbenchHttpApp } from "../../internal/workbench/http.ts"; +import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; + +const runtime = workbenchRuntime(); +const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, close: runtime.close }); +const host = process.env.WORKBENCH_API_HOST || "0.0.0.0"; +const port = positivePort(process.env.WORKBENCH_API_PORT || process.env.PORT, 6677); +const server = Bun.serve({ hostname: host, port, fetch: (request) => app.fetch(request) }); +process.stdout.write(`${JSON.stringify({ serviceId: "hwlab-workbench-api", status: "listening", mode: runtime.mode, host, port: server.port, temporalNamespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue })}\n`); +for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, async () => { server.stop(true); await app.close(); process.exit(0); }); + +function positivePort(value: string | undefined, fallback: number) { const parsed = Number.parseInt(String(value ?? fallback), 10); if (!Number.isInteger(parsed) || parsed < 1 || parsed > 65535) throw new Error("WORKBENCH_API_PORT must be a valid port"); return parsed; } diff --git a/cmd/hwlab-workbench-worker/main.ts b/cmd/hwlab-workbench-worker/main.ts new file mode 100644 index 00000000..95a40b77 --- /dev/null +++ b/cmd/hwlab-workbench-worker/main.ts @@ -0,0 +1,33 @@ +#!/usr/bin/env bun +import path from "node:path"; +import { NativeConnection, Worker } from "@temporalio/worker"; + +import { createWorkbenchActivities } from "../../internal/workbench/activities.ts"; +import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; + +const runtime = workbenchRuntime(); +if (runtime.mode !== "temporal") throw Object.assign(new Error("hwlab-workbench-worker requires WORKBENCH_MODE=temporal"), { code: "workbench_worker_temporal_mode_required" }); +const connection = await NativeConnection.connect({ address: runtime.temporalAddress }); +const worker = await Worker.create({ + connection, + namespace: runtime.temporalNamespace, + taskQueue: runtime.taskQueue, + workflowsPath: path.resolve(import.meta.dir, "../../internal/workbench/workflows.ts"), + activities: createWorkbenchActivities(runtime.application) +}); +const healthPort = positivePort(process.env.WORKBENCH_WORKER_HEALTH_PORT, 6678); +const health = Bun.serve({ + hostname: process.env.WORKBENCH_WORKER_HEALTH_HOST || "0.0.0.0", + port: healthPort, + fetch(request) { + const pathname = new URL(request.url).pathname; + return ["/healthz", "/readyz", "/health/live", "/health/ready"].includes(pathname) + ? Response.json({ ok: true, serviceId: "hwlab-workbench-worker", mode: runtime.mode, namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue }) + : new Response("not found", { status: 404 }); + } +}); +process.stdout.write(`${JSON.stringify({ serviceId: "hwlab-workbench-worker", status: "started", mode: runtime.mode, namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue, healthPort: health.port })}\n`); +for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, () => worker.shutdown()); +try { await worker.run(); } finally { health.stop(true); await connection.close(); await runtime.close(); } + +function positivePort(value: string | undefined, fallback: number) { const parsed = Number.parseInt(String(value ?? fallback), 10); if (!Number.isInteger(parsed) || parsed < 1 || parsed > 65535) throw new Error("WORKBENCH_WORKER_HEALTH_PORT must be a valid port"); return parsed; } diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index eabe225c..16be7c94 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -510,6 +510,8 @@ lanes: - hwlab-tasktree-worker - hwlab-harnessrl-api - hwlab-harnessrl-worker + - hwlab-workbench-api + - hwlab-workbench-worker - hwlab-cloud-web - hwlab-gateway - hwlab-edge-proxy @@ -523,6 +525,8 @@ lanes: hwlab-tasktree-worker: deploy/runtime/boot/hwlab-tasktree-worker.sh hwlab-harnessrl-api: deploy/runtime/boot/hwlab-harnessrl-api.sh hwlab-harnessrl-worker: deploy/runtime/boot/hwlab-harnessrl-worker.sh + hwlab-workbench-api: deploy/runtime/boot/hwlab-workbench-api.sh + hwlab-workbench-worker: deploy/runtime/boot/hwlab-workbench-worker.sh hwlab-cloud-web: deploy/runtime/boot/hwlab-cloud-web.sh hwlab-gateway: deploy/runtime/boot/hwlab-gateway.sh hwlab-edge-proxy: deploy/runtime/boot/hwlab-edge-proxy.sh @@ -702,6 +706,40 @@ lanes: TZ: Asia/Shanghai HWLAB_CASERUN_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-dev.svc.cluster.local:6667 observable: true + hwlab-workbench-api: + runtimeKind: bun-command + entrypoint: cmd/hwlab-workbench-api/main.ts + artifactKind: bun-command + healthPath: /health/ready + healthPort: 6677 + componentPaths: + - cmd/hwlab-workbench-api/ + - internal/workbench/ + - internal/cloud/workbench-command-proxy.ts + - tools/src/workbench-cli.ts + - tools/hwlab-cli/bin/hwlab-cli.ts + - package.json + - package-lock.json + - deploy/runtime/boot/hwlab-workbench-api.sh + env: + TZ: Asia/Shanghai + observable: true + hwlab-workbench-worker: + runtimeKind: bun-command + entrypoint: cmd/hwlab-workbench-worker/main.ts + artifactKind: bun-command + healthPath: /health/ready + healthPort: 6678 + componentPaths: + - cmd/hwlab-workbench-worker/ + - internal/workbench/ + - internal/cloud/code-agent-agentrun-adapter.ts + - package.json + - package-lock.json + - deploy/runtime/boot/hwlab-workbench-worker.sh + env: + TZ: Asia/Shanghai + observable: true hwlab-cloud-web: runtimeKind: cloud-web entrypoint: web/hwlab-cloud-web/index.html @@ -929,6 +967,8 @@ lanes: HWLAB_BOOTSTRAP_ADMIN_API_KEY: secretRef:hwlab-v03-master-server-admin-api-key/api-key HWLAB_USER_BILLING_URL: http://hwlab-user-billing.hwlab-v03.svc.cluster.local:6670 HWLAB_WORKBENCH_RUNTIME_URL: http://hwlab-workbench-runtime.hwlab-v03.svc.cluster.local:6671 + HWLAB_WORKBENCH_API_URL: http://hwlab-workbench-api.hwlab-v03.svc.cluster.local:6677 + WORKBENCH_CLOUD_API_AUTHORIZATION: secretRef:hwlab-v03-master-server-admin-api-key/api-key HWLAB_PROJECT_MANAGEMENT_URL: http://hwlab-project-management.hwlab-v03.svc.cluster.local:6672 HWLAB_TASKTREE_URL: http://hwlab-tasktree-api.hwlab-v03.svc.cluster.local:6673 HARNESSRL_API_URL: http://hwlab-harnessrl-api.hwlab-v03.svc.cluster.local:6675 @@ -1087,6 +1127,37 @@ lanes: HWLAB_CASERUN_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667 HWLAB_RUNTIME_INTERNAL_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667 OTEL_SERVICE_NAME: hwlab-harnessrl-worker + - serviceId: hwlab-workbench-api + replicas: 1 + env: + WORKBENCH_MODE: temporal + WORKBENCH_API_PORT: "6677" + WORKBENCH_CLOUD_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667 + WORKBENCH_CLOUD_API_AUTHORIZATION: secretRef:hwlab-v03-master-server-admin-api-key/api-key + WORKBENCH_API_AUTHORIZATION: secretRef:hwlab-v03-master-server-admin-api-key/api-key + WORKBENCH_TEMPORAL_ADDRESS: temporal-frontend.temporal.svc.cluster.local:7233 + WORKBENCH_TEMPORAL_NAMESPACE: unidesk + WORKBENCH_TEMPORAL_TASK_QUEUE: hwlab-v03-workbench + WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS: "30000" + WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000" + WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000" + WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5" + OTEL_SERVICE_NAME: hwlab-workbench-api + - serviceId: hwlab-workbench-worker + replicas: 1 + env: + WORKBENCH_MODE: temporal + WORKBENCH_WORKER_HEALTH_PORT: "6678" + WORKBENCH_CLOUD_API_URL: http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667 + WORKBENCH_CLOUD_API_AUTHORIZATION: secretRef:hwlab-v03-master-server-admin-api-key/api-key + WORKBENCH_TEMPORAL_ADDRESS: temporal-frontend.temporal.svc.cluster.local:7233 + WORKBENCH_TEMPORAL_NAMESPACE: unidesk + WORKBENCH_TEMPORAL_TASK_QUEUE: hwlab-v03-workbench + WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS: "30000" + WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000" + WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000" + WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5" + OTEL_SERVICE_NAME: hwlab-workbench-worker services: - serviceId: hwlab-harnessrl-api namespace: hwlab-dev diff --git a/deploy/k8s/base/services.yaml b/deploy/k8s/base/services.yaml index 4055ed78..16ae87b4 100644 --- a/deploy/k8s/base/services.yaml +++ b/deploy/k8s/base/services.yaml @@ -30,6 +30,20 @@ "ports": [{ "name": "http", "port": 6674, "targetPort": "http" }] } }, + { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": "hwlab-workbench-api", + "namespace": "hwlab-dev", + "labels": { "app.kubernetes.io/name": "hwlab-workbench-api", "hwlab.pikastech.local/service-id": "hwlab-workbench-api" } + }, + "spec": { + "type": "ClusterIP", + "selector": { "app.kubernetes.io/name": "hwlab-workbench-api" }, + "ports": [{ "name": "http", "port": 6677, "targetPort": "http" }] + } + }, { "apiVersion": "v1", "kind": "Service", diff --git a/deploy/k8s/base/workloads.yaml b/deploy/k8s/base/workloads.yaml index 7cb5ad7a..1f22bdc0 100644 --- a/deploy/k8s/base/workloads.yaml +++ b/deploy/k8s/base/workloads.yaml @@ -50,6 +50,54 @@ } } }, + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "hwlab-workbench-api", + "namespace": "hwlab-dev", + "labels": { "app.kubernetes.io/name": "hwlab-workbench-api", "hwlab.pikastech.local/service-id": "hwlab-workbench-api" } + }, + "spec": { + "replicas": 0, + "selector": { "matchLabels": { "app.kubernetes.io/name": "hwlab-workbench-api" } }, + "template": { + "metadata": { "labels": { "app.kubernetes.io/name": "hwlab-workbench-api", "hwlab.pikastech.local/service-id": "hwlab-workbench-api" } }, + "spec": { "containers": [{ + "name": "hwlab-workbench-api", + "image": "127.0.0.1:5000/hwlab/hwlab-workbench-api:dev", + "ports": [{ "name": "http", "containerPort": 6677 }], + "env": [{ "name": "WORKBENCH_API_PORT", "value": "6677" }], + "readinessProbe": { "httpGet": { "path": "/health/ready", "port": "http" } }, + "livenessProbe": { "httpGet": { "path": "/health/live", "port": "http" } } + }] } + } + } + }, + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "hwlab-workbench-worker", + "namespace": "hwlab-dev", + "labels": { "app.kubernetes.io/name": "hwlab-workbench-worker", "hwlab.pikastech.local/service-id": "hwlab-workbench-worker" } + }, + "spec": { + "replicas": 0, + "selector": { "matchLabels": { "app.kubernetes.io/name": "hwlab-workbench-worker" } }, + "template": { + "metadata": { "labels": { "app.kubernetes.io/name": "hwlab-workbench-worker", "hwlab.pikastech.local/service-id": "hwlab-workbench-worker" } }, + "spec": { "containers": [{ + "name": "hwlab-workbench-worker", + "image": "127.0.0.1:5000/hwlab/hwlab-workbench-worker:dev", + "ports": [{ "name": "health", "containerPort": 6678 }], + "env": [{ "name": "WORKBENCH_WORKER_HEALTH_PORT", "value": "6678" }], + "readinessProbe": { "httpGet": { "path": "/health/ready", "port": "health" } }, + "livenessProbe": { "httpGet": { "path": "/health/live", "port": "health" } } + }] } + } + } + }, { "apiVersion": "v1", "kind": "PersistentVolumeClaim", diff --git a/deploy/runtime/boot/hwlab-workbench-api.sh b/deploy/runtime/boot/hwlab-workbench-api.sh new file mode 100755 index 00000000..f24c3400 --- /dev/null +++ b/deploy/runtime/boot/hwlab-workbench-api.sh @@ -0,0 +1,9 @@ +#!/bin/sh +set -eu + +export HWLAB_SERVICE_ID=hwlab-workbench-api +export HWLAB_SERVICE_ENTRYPOINT=cmd/hwlab-workbench-api/main.ts +export WORKBENCH_API_PORT="${WORKBENCH_API_PORT:-6677}" +export PORT="$WORKBENCH_API_PORT" + +exec "${HWLAB_BUN_COMMAND:-bun}" run cmd/hwlab-workbench-api/main.ts diff --git a/deploy/runtime/boot/hwlab-workbench-worker.sh b/deploy/runtime/boot/hwlab-workbench-worker.sh new file mode 100755 index 00000000..6eda2f33 --- /dev/null +++ b/deploy/runtime/boot/hwlab-workbench-worker.sh @@ -0,0 +1,9 @@ +#!/bin/sh +set -eu + +export HWLAB_SERVICE_ID=hwlab-workbench-worker +export HWLAB_SERVICE_ENTRYPOINT=cmd/hwlab-workbench-worker/main.ts +export WORKBENCH_WORKER_HEALTH_PORT="${WORKBENCH_WORKER_HEALTH_PORT:-6678}" +export PORT="$WORKBENCH_WORKER_HEALTH_PORT" + +exec "${HWLAB_BUN_COMMAND:-bun}" run cmd/hwlab-workbench-worker/main.ts diff --git a/internal/cloud/server-code-agent-admission-http.ts b/internal/cloud/server-code-agent-admission-http.ts index 07481b89..b2727161 100644 --- a/internal/cloud/server-code-agent-admission-http.ts +++ b/internal/cloud/server-code-agent-admission-http.ts @@ -294,6 +294,51 @@ export async function handleCodeAgentChatHttp(request, response, options) { sendJson(response, responsePayload.status === "failed" && responsePayload.error?.code === "invalid_params" ? 400 : 200, responsePayload); } +export async function handleWorkbenchTurnAdmissionHttp(request, response, options) { + const body = await readJsonObjectBody(request, options.bodyLimitBytes); + if (!body.ok) { + sendJson(response, 400, manualSessionErrorPayload({ code: body.code, message: body.message, reason: body.reason })); + return; + } + const rawParams = withMdtodoExecutionPrompt(body.value); + const traceId = safeTraceId(rawParams.traceId) || `trc_${randomUUID()}`; + const lifecycle = codeAgentTurnLifecycleFields(traceId, rawParams); + const params = { + ...rawParams, + traceId, + ...lifecycle, + ownerUserId: options.actor?.id, + ownerRole: options.actor?.role + }; + const manualSession = await prepareManualCodeAgentSession({ params, options, traceId, response }); + if (manualSession?.blocked) return; + const admittedParams = stripSyntheticConversationContext(params, traceId, options); + const payload = codeAgentAdmittedFailureBasePayload({ params: admittedParams, options, traceId }); + try { + await recordCodeAgentTurnAdmission({ payload, params: admittedParams, options, traceId }); + } catch (error) { + if (isCodeAgentAdmissionUnavailableError(error)) { + sendJson(response, error.statusCode ?? 503, error.payload); + return; + } + throw error; + } + sendJson(response, 202, { + ok: true, + accepted: true, + status: "admitted", + admissionState: "durable", + traceId, + sessionId: safeSessionId(admittedParams.sessionId), + conversationId: safeConversationId(admittedParams.conversationId), + ...lifecycle, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false, + valuesRedacted: true, + secretMaterialStored: false + }); +} + export async function handleCodeAgentSessionsHttp(request, response, url, options) { const match = url.pathname.match(/^\/v1\/agent\/sessions(?:\/([^/]+)(?:\/(select))?)?$/u); const sessionId = safeSessionId(match?.[1] ? decodeURIComponent(match[1]) : ""); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index a41b4444..84b4b2fb 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1,7 +1,7 @@ // SPEC: PJ2026-0104010803 Workbench唯一投影; PJ2026-010403 API契约; PJ2026-010205 HWLAB接入; PJ2026-0102 Agent编排. // Responsibility: Stable Code Agent HTTP export facade. -export { handleCodeAgentChatHttp, handleCodeAgentSessionsHttp } from "./server-code-agent-admission-http.ts"; +export { handleCodeAgentChatHttp, handleCodeAgentSessionsHttp, handleWorkbenchTurnAdmissionHttp } from "./server-code-agent-admission-http.ts"; export { codeAgentTurnStatusPayload, handleCodeAgentCancelHttp, handleCodeAgentChatResultHttp, handleCodeAgentInspectHttp, handleCodeAgentTurnHttp } from "./server-code-agent-turn-http.ts"; export { handleCodeAgentSteerHttp, handleCodeAgentTraceHttp } from "./server-code-agent-trace-http.ts"; export { createCodeAgentChatResultStore } from "./server-code-agent-http-support.ts"; diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 2395f88f..c1acc0ef 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -63,6 +63,7 @@ import { handleCodeAgentChatResultHttp, handleCodeAgentInspectHttp, handleCodeAgentSessionsHttp, + handleWorkbenchTurnAdmissionHttp, handleCodeAgentSteerHttp, handleCodeAgentTurnHttp, handleCodeAgentTraceHttp @@ -99,6 +100,7 @@ import { handleProjectManagementProxyHttp } from "./project-management-proxy.ts" import { handleTaskTreeProxyHttp } from "./tasktree-proxy.ts"; import { handleDashboardSummaryHttp } from "./server-dashboard-http.ts"; import { HWPOD_NODE_OPS, HWPOD_NODE_OPS_CONTRACT_VERSION } from "../../tools/src/hwpod-node-ops-contract.ts"; +import { proxyWorkbenchCommandHttp, workbenchActivityDispatchRequested, workbenchTemporalProxyEnabled } from "./workbench-command-proxy.ts"; const DEFAULT_BODY_LIMIT_BYTES = 1024 * 1024; const HWLAB_WEB_SESSION_COOKIE = "hwlab_session"; @@ -896,17 +898,43 @@ async function handleRestAdapter(request, response, url, options) { if (request.method === "POST" && url.pathname === "/v1/agent/chat") { const nextOptions = await codeAgentOptions(request, response, options); if (!nextOptions) return; + if (workbenchTemporalProxyEnabled(nextOptions.env) && !workbenchActivityDispatchRequested(request)) { + await proxyWorkbenchCommandHttp(request, response, nextOptions, url.pathname); + return; + } await handleCodeAgentChatHttp(request, response, nextOptions); return; } + if (request.method === "POST" && url.pathname === "/v1/internal/workbench/admit") { + if (!workbenchActivityDispatchRequested(request)) { + sendJson(response, 404, { ok: false, error: { code: "not_found", message: "REST route is not implemented" } }); + return; + } + const nextOptions = await codeAgentOptions(request, response, options, { required: true }); + if (!nextOptions) return; + await handleWorkbenchTurnAdmissionHttp(request, response, nextOptions); + return; + } + if (url.pathname === "/v1/agent/sessions" || url.pathname.startsWith("/v1/agent/sessions/")) { const nextOptions = await codeAgentOptions(request, response, options, { required: true }); if (!nextOptions) return; + if (request.method === "POST" && url.pathname === "/v1/agent/sessions" && workbenchTemporalProxyEnabled(nextOptions.env) && !workbenchActivityDispatchRequested(request)) { + await proxyWorkbenchCommandHttp(request, response, nextOptions, url.pathname); + return; + } await handleCodeAgentSessionsHttp(request, response, url, nextOptions); return; } + if (request.method === "POST" && url.pathname === "/v1/workbench/commands") { + const nextOptions = await codeAgentOptions(request, response, options, { required: true }); + if (!nextOptions) return; + await proxyWorkbenchCommandHttp(request, response, nextOptions, url.pathname); + return; + } + if (request.method === "GET" && url.pathname === "/v1/agent/chat/inspect") { const nextOptions = await codeAgentOptions(request, response, options); if (!nextOptions) return; @@ -938,6 +966,10 @@ async function handleRestAdapter(request, response, url, options) { if (request.method === "POST" && url.pathname === "/v1/agent/chat/cancel") { const nextOptions = await codeAgentOptions(request, response, options); if (!nextOptions) return; + if (workbenchTemporalProxyEnabled(nextOptions.env) && !workbenchActivityDispatchRequested(request)) { + await proxyWorkbenchCommandHttp(request, response, nextOptions, url.pathname); + return; + } await handleCodeAgentCancelHttp(request, response, nextOptions); return; } @@ -1029,7 +1061,7 @@ function navIdForRestPath(pathname, method = "GET") { if (pathname === "/v1/api-keys" || pathname === "/v1/api-keys/default" || pathname.startsWith("/v1/api-keys/")) return "user.apiKeys"; if (pathname === "/v1/users/me/profile" || pathname === "/v1/users/me/password") return "system.settings"; if (pathname === "/v1/workbench/debug/fake-sse" || pathname.startsWith("/v1/workbench/debug/fake-sse/") || pathname === "/v1/workbench/debug/kafka-sse" || pathname.startsWith("/v1/workbench/debug/kafka-sse/")) return "workbench.debug"; - if (pathname === "/v1/workbench/events" || pathname === "/v1/workbench/projection-events" || pathname === "/v1/workbench/launches" || pathname === "/v1/workbench/sessions" || pathname.startsWith("/v1/workbench/sessions/") || pathname.startsWith("/v1/workbench/turns/") || pathname.startsWith("/v1/workbench/traces/")) return "workbench.code"; + if (pathname === "/v1/workbench/commands" || pathname === "/v1/workbench/events" || pathname === "/v1/workbench/projection-events" || pathname === "/v1/workbench/launches" || pathname === "/v1/workbench/sessions" || pathname.startsWith("/v1/workbench/sessions/") || pathname.startsWith("/v1/workbench/turns/") || pathname.startsWith("/v1/workbench/traces/")) return "workbench.code"; if (pathname === "/v1/agent-observer/events" || pathname === "/v1/agent-observer/runs") return "workbench.code"; if (pathname === "/v1/agent/chat" || pathname === "/v1/agent/sessions" || pathname.startsWith("/v1/agent/sessions/") || pathname === "/v1/agent/chat/inspect" || pathname.startsWith("/v1/agent/chat/result/") || pathname.startsWith("/v1/agent/turns/") || pathname.startsWith("/v1/agent/traces/") || pathname === "/v1/agent/chat/cancel" || pathname === "/v1/agent/chat/steer") return "workbench.code"; if (pathname === "/v1/admin/provider-profiles" || pathname.startsWith("/v1/admin/provider-profiles/")) return "admin.providerProfiles"; @@ -1045,6 +1077,25 @@ function navIdForRestPath(pathname, method = "GET") { } async function codeAgentOptions(request, response, options, authOptions = {}) { + if (workbenchActivityDispatchRequested(request)) { + const expected = String(options.env?.WORKBENCH_CLOUD_API_AUTHORIZATION ?? "").trim(); + const actual = String(request.headers?.authorization ?? "").trim(); + const actorId = String(request.headers?.["x-hwlab-actor-id"] ?? "").trim(); + if (!expected || actual !== expected || !actorId) { + sendJson(response, 401, { ok: false, error: { code: "workbench_internal_auth_invalid", message: "Workbench internal activity authentication failed" } }); + return null; + } + return { + ...options, + actor: { id: actorId, role: String(request.headers?.["x-hwlab-actor-role"] ?? "user").trim() || "user" }, + authSession: null, + authApiKey: null, + actorApiKeySecret: null, + userBillingAuth: { active: true, source: "workbench-internal", valuesRedacted: true }, + userBillingToken: null, + userBillingClient: options.userBillingClient ?? options.accessController?.userBilling ?? null + }; + } const auth = await options.accessController.authenticate(request, { required: authOptions.required ?? options.accessController.required }); if (!auth.ok) { sendJson(response, auth.status, auth); diff --git a/internal/cloud/workbench-command-proxy.ts b/internal/cloud/workbench-command-proxy.ts new file mode 100644 index 00000000..89931233 --- /dev/null +++ b/internal/cloud/workbench-command-proxy.ts @@ -0,0 +1,38 @@ +import { readBody, sendJson } from "./server-http-utils.ts"; + +export function workbenchTemporalProxyEnabled(env: Record = process.env) { + return Boolean(String(env.HWLAB_WORKBENCH_API_URL ?? "").trim()); +} + +export function workbenchActivityDispatchRequested(request: any) { + return String(request.headers?.["x-workbench-activity-dispatch"] ?? "").trim() === "1"; +} + +export async function proxyWorkbenchCommandHttp(request: any, response: any, options: any, route: string) { + const baseUrl = String(options.env?.HWLAB_WORKBENCH_API_URL ?? "").trim(); + if (!baseUrl) throw Object.assign(new Error("HWLAB_WORKBENCH_API_URL is required"), { code: "workbench_api_url_required" }); + let bodyText = request.method === "GET" || request.method === "HEAD" ? "" : await readBody(request, options.bodyLimitBytes); + if (route === "/v1/workbench/commands") { + const command = JSON.parse(bodyText || "{}"); + if (!command || typeof command !== "object" || Array.isArray(command)) throw Object.assign(new Error("Workbench command must be a JSON object"), { code: "invalid_json" }); + if (command.operation !== "health") command.actor = { id: String(options.actor?.id ?? ""), role: String(options.actor?.role ?? "user") }; + bodyText = JSON.stringify(command); + } + const headers: Record = { + authorization: String(options.env?.WORKBENCH_CLOUD_API_AUTHORIZATION ?? ""), + "content-type": "application/json", + "x-hwlab-actor-id": String(options.actor?.id ?? ""), + "x-hwlab-actor-role": String(options.actor?.role ?? "user") + }; + const upstream = await (options.fetchImpl ?? fetch)(`${baseUrl.replace(/\/$/u, "")}${route}`, { + method: request.method, + headers, + body: bodyText || undefined + }); + const payload = await upstream.json().catch(() => null); + if (!payload || typeof payload !== "object") { + sendJson(response, 502, { ok: false, error: { code: "workbench_api_invalid_json", message: `Workbench API returned invalid JSON (${upstream.status})` } }); + return; + } + sendJson(response, upstream.status, payload); +} diff --git a/internal/workbench/activities.ts b/internal/workbench/activities.ts new file mode 100644 index 00000000..0c676e45 --- /dev/null +++ b/internal/workbench/activities.ts @@ -0,0 +1,12 @@ +import type { WorkbenchApplication, WorkbenchTurnInput } from "./contracts.ts"; + +export function createWorkbenchActivities(application: WorkbenchApplication) { + return { + async dispatchWorkbenchTurn(input: WorkbenchTurnInput) { + return application.dispatchTurn(input); + }, + async cancelWorkbenchTurn(input: { actor: WorkbenchTurnInput["actor"]; traceId: string; params: Record }) { + return application.cancelTurn(input); + } + }; +} diff --git a/internal/workbench/cloud-application.ts b/internal/workbench/cloud-application.ts new file mode 100644 index 00000000..fb725e64 --- /dev/null +++ b/internal/workbench/cloud-application.ts @@ -0,0 +1,43 @@ +import type { WorkbenchApplication } from "./contracts.ts"; + +export function createCloudWorkbenchApplication(options: { + baseUrl: string; + authorization: string; + fetchImpl?: typeof fetch; +}): WorkbenchApplication { + if (!options.baseUrl) throw codedError("workbench_cloud_api_url_required", "WORKBENCH_CLOUD_API_URL is required"); + if (!options.authorization) throw codedError("workbench_cloud_api_auth_required", "WORKBENCH_CLOUD_API_AUTHORIZATION is required"); + const fetchImpl = options.fetchImpl ?? fetch; + const request = async (method: string, route: string, body: Record | undefined, actor: { id: string; role?: string | null }) => { + const response = await fetchImpl(`${options.baseUrl.replace(/\/$/u, "")}${route}`, { + method, + headers: { + authorization: options.authorization, + "content-type": "application/json", + "x-workbench-activity-dispatch": "1", + "x-hwlab-actor-id": actor.id, + "x-hwlab-actor-role": actor.role ?? "user" + }, + body: body ? JSON.stringify(body) : undefined + }); + const payload = await response.json().catch(() => null) as Record | null; + if (!response.ok || !payload) throw codedError(String((payload?.error as any)?.code ?? "workbench_cloud_api_failed"), String((payload?.error as any)?.message ?? `Cloud API returned ${response.status}`), { status: response.status, route }); + return payload; + }; + return { + async health() { return { serviceId: "hwlab-workbench", status: "ready", cloudApi: "configured" }; }, + async createSession(input) { return request("POST", "/v1/agent/sessions", input.params, input.actor); }, + async admitTurn(input) { + await request("POST", "/v1/internal/workbench/admit", { ...input.params, traceId: input.traceId, sessionId: input.sessionId }, input.actor); + return input; + }, + async dispatchTurn(input) { + return request("POST", "/v1/agent/chat", { ...input.params, traceId: input.traceId, sessionId: input.sessionId, shortConnection: true }, input.actor); + }, + async cancelTurn(input) { + return request("POST", "/v1/agent/chat/cancel", { ...input.params, traceId: input.traceId }, input.actor); + } + }; +} + +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 new file mode 100644 index 00000000..a355cd2f --- /dev/null +++ b/internal/workbench/contracts.ts @@ -0,0 +1,48 @@ +export type WorkbenchActor = { + id: string; + role?: string | null; +}; + +export type WorkbenchCommand = + | { operation: "health" } + | { operation: "session.create"; actor: WorkbenchActor; params: Record } + | { operation: "turn.submit"; actor: WorkbenchActor; params: Record; traceId?: string } + | { operation: "turn.cancel"; actor: WorkbenchActor; traceId: string; params?: Record }; + +export type WorkbenchCommandResult = { + ok: boolean; + operation: WorkbenchCommand["operation"]; + mode: "native-test" | "temporal"; + data?: Record; + error?: { code: string; message: string; details?: unknown }; +}; + +export type WorkbenchTurnInput = { + actor: WorkbenchActor; + traceId: string; + sessionId: string; + params: Record; +}; + +export type WorkbenchActivityConfig = { + startToCloseTimeoutMs: number; + retryInitialIntervalMs: number; + retryMaximumIntervalMs: number; + retryMaximumAttempts: number; +}; + +export interface WorkbenchApplication { + health(): Promise>; + createSession(input: { actor: WorkbenchActor; params: Record }): Promise>; + admitTurn(input: WorkbenchTurnInput): Promise; + dispatchTurn(input: WorkbenchTurnInput): Promise>; + cancelTurn(input: { actor: WorkbenchActor; traceId: string; params: Record }): Promise>; + snapshot?(): Promise<{ sessions: Record>; turns: Record> }>; + close?(): Promise; +} + +export interface WorkbenchTemporalGateway { + submit(input: WorkbenchTurnInput): Promise<{ workflowId: string; workflowRunId: string; reused: boolean }>; + cancel(input: { actor: WorkbenchActor; traceId: string; params: Record }): Promise<{ workflowId: string; workflowRunId: string; reused: boolean }>; + close(): Promise; +} diff --git a/internal/workbench/dispatcher.ts b/internal/workbench/dispatcher.ts new file mode 100644 index 00000000..dbdc36ff --- /dev/null +++ b/internal/workbench/dispatcher.ts @@ -0,0 +1,74 @@ +import { randomUUID } from "node:crypto"; + +import type { WorkbenchApplication, WorkbenchCommand, WorkbenchCommandResult, 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; + +export function createWorkbenchDispatcher(options: { + application: WorkbenchApplication; + temporal: WorkbenchTemporalGateway; + mode: "native-test" | "temporal"; +}) { + return async function dispatch(command: WorkbenchCommand): Promise { + try { + if (command.operation === "health") { + return success(command.operation, options.mode, await options.application.health()); + } + requireActor(command.actor); + if (command.operation === "session.create") { + return success(command.operation, options.mode, await options.application.createSession(command)); + } + if (command.operation === "turn.submit") { + const traceId = validId(command.traceId ?? `trc_${randomUUID()}`, TRACE_ID, "traceId"); + const sessionId = validId(command.params.sessionId, SESSION_ID, "sessionId"); + const admitted = await options.application.admitTurn({ actor: command.actor, traceId, sessionId, params: { ...command.params, traceId, sessionId } }); + const workflow = await options.temporal.submit(admitted); + return success(command.operation, options.mode, { + accepted: true, + status: "running", + traceId, + sessionId, + ...workflow, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false + }); + } + const traceId = validId(command.traceId, TRACE_ID, "traceId"); + const workflow = await options.temporal.cancel({ actor: command.actor, traceId, params: command.params ?? {} }); + return success(command.operation, options.mode, { + accepted: true, + status: "cancel_requested", + traceId, + ...workflow, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false + }); + } catch (error: any) { + return { + ok: false, + operation: command.operation, + mode: options.mode, + error: { code: error?.code ?? "workbench_command_failed", message: error?.message ?? String(error), details: error?.details } + }; + } + }; +} + +function success(operation: WorkbenchCommand["operation"], mode: "native-test" | "temporal", data: Record): WorkbenchCommandResult { + return { ok: true, operation, mode, data }; +} + +function requireActor(actor: { id?: string } | undefined) { + if (!String(actor?.id ?? "").trim()) throw codedError("auth_required", "actor.id is required"); +} + +function validId(value: unknown, pattern: RegExp, field: string) { + const text = String(value ?? "").trim(); + if (!pattern.test(text)) throw codedError("invalid_input", `${field} is invalid`, { field }); + return text; +} + +function codedError(code: string, message: string, details?: unknown) { + return Object.assign(new Error(message), { code, details }); +} diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts new file mode 100644 index 00000000..52cdf3fd --- /dev/null +++ b/internal/workbench/http.ts @@ -0,0 +1,94 @@ +import type { WorkbenchCommand } from "./contracts.ts"; + +export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise; snapshot?: () => Promise<{ sessions: Record>; turns: Record> }>; authorization?: string; close?: () => Promise }) { + return { + async fetch(request: Request): Promise { + try { + const url = new URL(request.url); + if (url.pathname === "/health/live") return json(200, { ok: true, serviceId: "hwlab-workbench-api", status: "live" }); + if (url.pathname === "/health/ready") return resultResponse(await options.dispatch({ operation: "health" })); + if (options.snapshot) { + if (url.pathname === "/auth/session" || url.pathname === "/auth/bootstrap") return json(200, { ok: true, authenticated: true, actor: { id: "usr_native", role: "user" }, mode: "native-test" }); + if (url.pathname === "/auth/login" && request.method === "POST") return Response.json({ ok: true, authenticated: true, actor: { id: "usr_native", role: "user" }, mode: "native-test" }, { status: 200, headers: { "set-cookie": "hwlab_session=native-test; Path=/; HttpOnly; SameSite=Lax", "cache-control": "no-store" } }); + if (url.pathname === "/v1/users/me") return json(200, { ok: true, actor: { id: "usr_native", role: "user" }, authMethod: "native-test" }); + if (url.pathname === "/v1/access/status") return json(200, { ok: true, status: "active", actor: { id: "usr_native", role: "user" }, nav: ["workbench.code"] }); + if (url.pathname === "/v1/provider-profiles") return json(200, { ok: true, items: [{ profile: "native-test", backendProfile: "native-test", configured: true }], count: 1, mode: "native-test" }); + if (url.pathname === "/v1/dashboard/summary") return json(200, { ok: true, status: "ready", mode: "native-test" }); + if (url.pathname === "/v1/workbench/events" && request.method === "GET") return nativeEventStream(); + if (url.pathname === "/v1/workbench/sessions" && request.method === "GET") return nativeSessionList(options, url); + const sessionMatch = /^\/v1\/workbench\/sessions\/([^/]+)(?:\/(messages))?$/u.exec(url.pathname); + if (sessionMatch && request.method === "GET") return nativeSessionDetail(options, decodeURIComponent(sessionMatch[1]), sessionMatch[2] === "messages"); + const turnMatch = /^\/v1\/workbench\/turns\/([^/]+)$/u.exec(url.pathname); + if (turnMatch && request.method === "GET") return nativeTurn(options, decodeURIComponent(turnMatch[1])); + } + if (url.pathname === "/v1/workbench/commands" && request.method === "POST") { requireAuthorization(request, options); return resultResponse(await options.dispatch(await bodyObject(request) as WorkbenchCommand), 202); } + const actor = actorFrom(request, Boolean(options.snapshot)); + if (url.pathname === "/v1/agent/sessions" && request.method === "POST") { requireAuthorization(request, options); return legacyResponse(await options.dispatch({ operation: "session.create", actor, params: await bodyObject(request) }), 201); } + if (url.pathname === "/v1/agent/chat" && request.method === "POST") { + requireAuthorization(request, options); + const params = await bodyObject(request); + return legacyResponse(await options.dispatch({ operation: "turn.submit", actor, params, traceId: text(params.traceId) || undefined }), 202); + } + if (url.pathname === "/v1/agent/chat/cancel" && request.method === "POST") { + requireAuthorization(request, options); + const params = await bodyObject(request); + return legacyResponse(await options.dispatch({ operation: "turn.cancel", actor, traceId: text(params.traceId), params })); + } + return json(404, { ok: false, error: { code: "not_found", message: "Workbench API route was not found" } }); + } catch (error: any) { + return json(400, { ok: false, error: { code: error?.code ?? "workbench_http_error", message: error?.message ?? String(error) } }); + } + }, + close: options.close ?? (async () => {}) + }; +} + +function requireAuthorization(request: Request, options: { snapshot?: unknown; authorization?: string }) { + if (options.snapshot) return; + const expected = text(options.authorization); + const actual = text(request.headers.get("authorization")); + if (!expected || actual !== expected) throw Object.assign(new Error("Workbench API internal authentication failed"), { code: "auth_required" }); +} + +async function nativeSessionList(options: { snapshot?: () => Promise }, url: URL) { + const state = await requiredSnapshot(options); + const owner = text(url.searchParams.get("ownerUserId")); + const sessions = Object.values(state.sessions).filter((session: any) => !owner || session.ownerUserId === owner).sort((left: any, right: any) => String(right.updatedAt).localeCompare(String(left.updatedAt))); + return json(200, { ok: true, status: "ready", sessions, count: sessions.length, mode: "native-test" }); +} +async function nativeSessionDetail(options: { snapshot?: () => Promise }, sessionId: string, messagesOnly: boolean) { + const state = await requiredSnapshot(options); + const session = state.sessions[sessionId]; + if (!session) return json(404, { ok: false, error: { code: "session_not_found", message: "native session was not found" } }); + return json(200, messagesOnly ? { ok: true, sessionId, messages: session.messages ?? [], count: Array.isArray(session.messages) ? session.messages.length : 0, mode: "native-test" } : { ok: true, session, mode: "native-test" }); +} +async function nativeTurn(options: { snapshot?: () => Promise }, traceId: string) { + const state = await requiredSnapshot(options); + const turn = state.turns[traceId]; + return turn ? json(200, { ok: true, ...turn, mode: "native-test" }) : json(404, { ok: false, error: { code: "turn_not_found", message: "native turn was not found" } }); +} +async function requiredSnapshot(options: { snapshot?: () => Promise }) { if (!options.snapshot) throw Object.assign(new Error("native projection is unavailable"), { code: "native_projection_unavailable" }); return options.snapshot(); } +function nativeEventStream() { + const encoder = new TextEncoder(); + let heartbeat: ReturnType | undefined; + return new Response(new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(`event: capability\ndata: ${JSON.stringify({ mode: "native-test", terminalAuthority: "fixture" })}\n\n`)); + heartbeat = setInterval(() => controller.enqueue(encoder.encode(`: native-test-heartbeat ${Date.now()}\n\n`)), 10_000); + }, + cancel() { if (heartbeat) clearInterval(heartbeat); } + }), { status: 200, headers: { "content-type": "text/event-stream", "cache-control": "no-store", connection: "keep-alive" } }); +} + +function actorFrom(request: Request, nativeTest: boolean) { + const id = text(request.headers.get("x-hwlab-actor-id")); + if (!id && nativeTest) return { id: "usr_native", role: "user" }; + if (!id) throw Object.assign(new Error("x-hwlab-actor-id is required"), { code: "auth_required" }); + return { id, role: text(request.headers.get("x-hwlab-actor-role")) || "user" }; +} +function resultResponse(result: any, successStatus = 200) { return json(result?.ok === true ? successStatus : statusFor(result?.error?.code), result); } +function legacyResponse(result: any, successStatus = 200) { return result?.ok === true ? json(successStatus, { ok: true, ...result.data, orchestrationMode: result.mode }) : resultResponse(result, successStatus); } +function statusFor(code: unknown) { const value = String(code ?? ""); if (value === "auth_required") return 401; if (value.includes("not_found")) return 404; if (value.includes("required") || value === "invalid_input") return 400; return 503; } +async function bodyObject(request: Request) { const body = await request.json().catch(() => null); if (!body || typeof body !== "object" || Array.isArray(body)) throw Object.assign(new Error("JSON object body is required"), { code: "invalid_json" }); return body as Record; } +function text(value: unknown) { return String(value ?? "").trim(); } +function json(status: number, body: unknown) { return Response.json(body, { status, headers: { "cache-control": "no-store" } }); } diff --git a/internal/workbench/native-application.ts b/internal/workbench/native-application.ts new file mode 100644 index 00000000..306b064f --- /dev/null +++ b/internal/workbench/native-application.ts @@ -0,0 +1,78 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; + +import type { WorkbenchApplication, WorkbenchTurnInput } from "./contracts.ts"; + +type NativeState = { + sessions: Record>; + turns: Record>; +}; + +export function createNativeTestWorkbenchApplication(options: { stateFile: string; now?: () => string }): WorkbenchApplication { + const now = options.now ?? (() => new Date().toISOString()); + const update = async (fn: (state: NativeState) => T | Promise) => { + const state = await load(options.stateFile); + const result = await fn(state); + await mkdir(path.dirname(options.stateFile), { recursive: true }); + await writeFile(options.stateFile, `${JSON.stringify(state, null, 2)}\n`, "utf8"); + return result; + }; + return { + async snapshot() { return load(options.stateFile); }, + async health() { + const state = await load(options.stateFile); + return { serviceId: "hwlab-workbench-native-test", status: "ready", sessionCount: Object.keys(state.sessions).length, turnCount: Object.keys(state.turns).length }; + }, + async createSession({ actor, params }) { + 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[] }; + state.sessions[sessionId] = session; + return { status: "created", session, valuesRedacted: true, secretMaterialStored: false }; + }); + }, + async admitTurn(input) { + return update((state) => { + const session = state.sessions[input.sessionId]; + if (!session || session.ownerUserId !== input.actor.id) throw codedError("session_not_found", "session is not visible to the actor"); + const createdAt = now(); + const message = text(input.params.message); + state.turns[input.traceId] = { traceId: input.traceId, sessionId: input.sessionId, ownerUserId: input.actor.id, status: "admitted", message, createdAt, updatedAt: createdAt, terminal: false }; + const messages = Array.isArray(session.messages) ? session.messages as Record[] : []; + messages.push({ messageId: `msg_user_${input.traceId.slice(4)}`, role: "user", content: message, traceId: input.traceId, status: "completed", createdAt }); + session.messages = messages; + session.status = "running"; + session.lastTraceId = input.traceId; + session.updatedAt = createdAt; + return input; + }); + }, + async dispatchTurn(input: WorkbenchTurnInput) { + return update((state) => { + const turn = state.turns[input.traceId]; + if (!turn) throw codedError("turn_not_admitted", "turn must be admitted before dispatch"); + Object.assign(turn, { status: "running", updatedAt: now(), runId: `run_native_${input.traceId.slice(4)}`, commandId: `cmd_native_${input.traceId.slice(4)}` }); + return { accepted: true, status: "running", traceId: input.traceId, sessionId: input.sessionId, runId: turn.runId, commandId: turn.commandId, mode: "native-test", terminal: false }; + }); + }, + async cancelTurn(input) { + return update((state) => { + const turn = state.turns[input.traceId]; + if (!turn || turn.ownerUserId !== input.actor.id) throw codedError("turn_not_found", "turn is not visible to the actor"); + Object.assign(turn, { status: "cancel_requested", updatedAt: now() }); + return { accepted: true, status: "cancel_requested", traceId: input.traceId, terminal: false, mode: "native-test" }; + }); + } + }; +} + +async function load(stateFile: string): Promise { + const parsed = JSON.parse(await readFile(stateFile, "utf8").catch(() => '{"sessions":{},"turns":{}}')); + 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 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/native-temporal.ts b/internal/workbench/native-temporal.ts new file mode 100644 index 00000000..0b462371 --- /dev/null +++ b/internal/workbench/native-temporal.ts @@ -0,0 +1,15 @@ +import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts.ts"; + +export function createNativeTestTemporalGateway(application: WorkbenchApplication): WorkbenchTemporalGateway { + return { + async submit(input) { + await application.dispatchTurn(input); + return { workflowId: `native-workbench-turn-${input.traceId}`, workflowRunId: `native-${input.traceId}`, reused: false }; + }, + async cancel(input) { + await application.cancelTurn(input); + return { workflowId: `native-workbench-cancel-${input.traceId}`, workflowRunId: `native-cancel-${input.traceId}`, reused: false }; + }, + async close() {} + }; +} diff --git a/internal/workbench/runtime.ts b/internal/workbench/runtime.ts new file mode 100644 index 00000000..866c2d0d --- /dev/null +++ b/internal/workbench/runtime.ts @@ -0,0 +1,40 @@ +import path from "node:path"; + +import { createCloudWorkbenchApplication } from "./cloud-application.ts"; +import { createWorkbenchDispatcher } from "./dispatcher.ts"; +import { createNativeTestWorkbenchApplication } from "./native-application.ts"; +import { createNativeTestTemporalGateway } from "./native-temporal.ts"; +import { createWorkbenchTemporalGateway } from "./temporal.ts"; + +export function workbenchRuntime(env: Record = process.env) { + const mode = String(env.WORKBENCH_MODE ?? "").trim(); + if (mode !== "native-test" && mode !== "temporal") throw codedError("workbench_mode_required", "WORKBENCH_MODE must be native-test or temporal"); + const stateFile = path.resolve(String(env.WORKBENCH_NATIVE_STATE_FILE ?? ".state/workbench-native/state.json")); + const application = mode === "native-test" + ? createNativeTestWorkbenchApplication({ stateFile }) + : createCloudWorkbenchApplication({ baseUrl: String(env.WORKBENCH_CLOUD_API_URL ?? ""), authorization: String(env.WORKBENCH_CLOUD_API_AUTHORIZATION ?? "") }); + const temporalAddress = String(env.WORKBENCH_TEMPORAL_ADDRESS ?? env.TEMPORAL_ADDRESS ?? ""); + const temporalNamespace = String(env.WORKBENCH_TEMPORAL_NAMESPACE ?? "unidesk"); + const taskQueue = String(env.WORKBENCH_TEMPORAL_TASK_QUEUE ?? "hwlab-v03-workbench"); + const activity = mode === "native-test" ? null : { + startToCloseTimeoutMs: requiredPositiveInteger(env.WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS, "WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS"), + retryInitialIntervalMs: requiredPositiveInteger(env.WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS, "WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS"), + retryMaximumIntervalMs: requiredPositiveInteger(env.WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS, "WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS"), + 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! }); + return { + mode, + stateFile, + temporalAddress, + temporalNamespace, + taskQueue, activity, + application, + temporal, + dispatch: createWorkbenchDispatcher({ application, temporal, mode }), + async close() { await temporal.close(); await application.close?.(); } + }; +} + +function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); } +function requiredPositiveInteger(value: string | undefined, name: string) { const parsed = Number.parseInt(String(value ?? ""), 10); if (!Number.isSafeInteger(parsed) || parsed <= 0) throw codedError("workbench_temporal_config_invalid", `${name} must be a positive integer`); return parsed; } diff --git a/internal/workbench/temporal.ts b/internal/workbench/temporal.ts new file mode 100644 index 00000000..e0d997ca --- /dev/null +++ b/internal/workbench/temporal.ts @@ -0,0 +1,42 @@ +import { Client, Connection } from "@temporalio/client"; + +import type { WorkbenchActivityConfig, WorkbenchTemporalGateway, WorkbenchTurnInput } from "./contracts.ts"; + +export function createWorkbenchTemporalGateway(options: { address: string; namespace: string; taskQueue: string; activity: WorkbenchActivityConfig }): WorkbenchTemporalGateway { + if (!options.address) throw codedError("workbench_temporal_address_required", "WORKBENCH_TEMPORAL_ADDRESS is required"); + let connection: Connection | undefined; + let client: Client | undefined; + const getClient = async () => { + connection ??= await Connection.connect({ address: options.address }); + client ??= new Client({ connection, namespace: options.namespace }); + return client; + }; + const start = async (workflowType: string, workflowId: string, input: unknown) => { + const temporal = await getClient(); + try { + const handle = await temporal.workflow.start(workflowType, { taskQueue: options.taskQueue, workflowId, args: [input] }); + return { workflowId, workflowRunId: handle.firstExecutionRunId, reused: false }; + } catch (error: any) { + if (error?.name !== "WorkflowExecutionAlreadyStartedError") throw error; + const description = await temporal.workflow.getHandle(workflowId).describe(); + return { workflowId, workflowRunId: description.runId, reused: true }; + } + }; + return { + submit(input: WorkbenchTurnInput) { + return start("workbenchTurnWorkflow", `workbench-turn-${input.traceId}`, { ...input, activity: options.activity }); + }, + cancel(input) { + return start("workbenchCancelWorkflow", `workbench-cancel-${input.traceId}`, { ...input, activity: options.activity }); + }, + async close() { + await connection?.close(); + connection = undefined; + client = undefined; + } + }; +} + +function codedError(code: string, message: string) { + return Object.assign(new Error(message), { code }); +} diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts new file mode 100644 index 00000000..f46ed86e --- /dev/null +++ b/internal/workbench/workbench.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, test } from "bun:test"; + +import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts.ts"; +import { createWorkbenchDispatcher } from "./dispatcher.ts"; +import { createCloudWorkbenchApplication } from "./cloud-application.ts"; + +describe("Workbench dispatcher", () => { + test("submits through Temporal without synthesizing terminal state", async () => { + const calls: string[] = []; + const application: WorkbenchApplication = { + async health() { return { status: "ready" }; }, + async createSession() { return { session: { sessionId: "ses_test" } }; }, + async admitTurn(input) { calls.push("admit"); return input; }, + async dispatchTurn() { throw new Error("API dispatcher must not dispatch directly"); }, + async cancelTurn() { throw new Error("API dispatcher must not cancel directly"); } + }; + const temporal: WorkbenchTemporalGateway = { + async submit(input) { calls.push("temporal"); return { workflowId: `workbench-turn-${input.traceId}`, workflowRunId: "run-1", reused: false }; }, + async cancel() { return { workflowId: "cancel", workflowRunId: "run-2", reused: false }; }, + async close() {} + }; + const dispatch = createWorkbenchDispatcher({ application, temporal, mode: "temporal" }); + const result = await dispatch({ operation: "turn.submit", actor: { id: "usr_test" }, traceId: "trc_test", params: { sessionId: "ses_test", message: "hi" } }); + expect(calls).toEqual(["admit", "temporal"]); + expect(result).toMatchObject({ ok: true, mode: "temporal", data: { status: "running", traceId: "trc_test", terminalAuthority: "hwlab.event.v1", resultSynthesized: false } }); + }); + + test("cancel is a durable Temporal command", async () => { + const application = stubApplication(); + const temporal: WorkbenchTemporalGateway = { + async submit() { throw new Error("not used"); }, + async cancel(input) { return { workflowId: `workbench-cancel-${input.traceId}`, workflowRunId: "run-cancel", reused: false }; }, + async close() {} + }; + const result = await createWorkbenchDispatcher({ application, temporal, mode: "temporal" })({ operation: "turn.cancel", actor: { id: "usr_test" }, traceId: "trc_test" }); + expect(result).toMatchObject({ ok: true, data: { status: "cancel_requested", resultSynthesized: false } }); + }); +}); + +describe("Workbench Cloud application adapter", () => { + test("persists admission before dispatch through distinct internal routes", async () => { + const calls: Array<{ path: string; authorization: string | null }> = []; + const application = createCloudWorkbenchApplication({ + baseUrl: "http://cloud.internal", + authorization: "unified-internal-key", + fetchImpl: (async (url: string | URL | Request, init?: RequestInit) => { + calls.push({ path: new URL(String(url)).pathname, authorization: new Headers(init?.headers).get("authorization") }); + return Response.json({ ok: true, accepted: true, status: "admitted" }); + }) as typeof fetch + }); + const input = { actor: { id: "usr_test" }, traceId: "trc_test", sessionId: "ses_test", params: { sessionId: "ses_test", message: "hi" } }; + await application.admitTurn(input); + await application.dispatchTurn(input); + expect(calls).toEqual([ + { path: "/v1/internal/workbench/admit", authorization: "unified-internal-key" }, + { path: "/v1/agent/chat", authorization: "unified-internal-key" } + ]); + }); +}); + +function stubApplication(): WorkbenchApplication { + return { + async health() { return {}; }, + async createSession() { return {}; }, + async admitTurn(input) { return input; }, + async dispatchTurn() { return {}; }, + async cancelTurn() { return {}; } + }; +} diff --git a/internal/workbench/workflows.ts b/internal/workbench/workflows.ts new file mode 100644 index 00000000..9bc63efe --- /dev/null +++ b/internal/workbench/workflows.ts @@ -0,0 +1,27 @@ +import { proxyActivities } from "@temporalio/workflow"; + +import type { WorkbenchActivityConfig, WorkbenchTurnInput } from "./contracts.ts"; + +type WorkbenchActivities = { + dispatchWorkbenchTurn(input: WorkbenchTurnInput): Promise>; + cancelWorkbenchTurn(input: { actor: WorkbenchTurnInput["actor"]; traceId: string; params: Record }): Promise>; +}; + +export async function workbenchTurnWorkflow(input: WorkbenchTurnInput & { activity: WorkbenchActivityConfig }) { + const activities = workbenchActivities(input.activity); + const dispatch = await activities.dispatchWorkbenchTurn(input); + return { accepted: true, traceId: input.traceId, dispatch, terminalAuthority: "hwlab.event.v1", resultSynthesized: false }; +} + +export async function workbenchCancelWorkflow(input: { actor: WorkbenchTurnInput["actor"]; traceId: string; params: Record; activity: WorkbenchActivityConfig }) { + const activities = workbenchActivities(input.activity); + const cancel = await activities.cancelWorkbenchTurn(input); + return { accepted: true, traceId: input.traceId, cancel, terminalAuthority: "hwlab.event.v1", resultSynthesized: false }; +} + +function workbenchActivities(activity: WorkbenchActivityConfig) { + return proxyActivities({ + startToCloseTimeout: `${activity.startToCloseTimeoutMs} milliseconds`, + retry: { initialInterval: `${activity.retryInitialIntervalMs} milliseconds`, maximumInterval: `${activity.retryMaximumIntervalMs} milliseconds`, maximumAttempts: activity.retryMaximumAttempts } + }); +} diff --git a/package.json b/package.json index ecf3259c..319ff18b 100644 --- a/package.json +++ b/package.json @@ -16,6 +16,10 @@ "harnessrl:worker:dev": "bun --watch cmd/hwlab-harnessrl-worker/main.ts", "harnessrl:native:smoke": "bun test internal/harnessrl/harnessrl.test.ts internal/cloud/server-caserun-http.test.ts", "harnessrl:native:l1-smoke": "node scripts/harnessrl-native-l1-smoke.mjs", + "workbench:api:dev": "bun --watch cmd/hwlab-workbench-api/main.ts", + "workbench:worker:dev": "bun --watch cmd/hwlab-workbench-worker/main.ts", + "workbench:web:dev": "cd web/hwlab-cloud-web && WORKBENCH_NATIVE_API_URL=http://127.0.0.1:6677 WORKBENCH_VITE_USE_POLLING=1 bun run dev", + "workbench:native:smoke": "bun test internal/workbench/workbench.test.ts && bun run scripts/workbench-native-smoke.mjs", "workbench:p1:inventory": "node scripts/workbench-long-reliability-p1-inventory.mjs", "dev-runtime-base:build": "node scripts/dev-runtime-base-image.mjs", "cloud-api:smoke": "node scripts/run-bun.mjs scripts/cloud-api-runtime-smoke.mjs", diff --git a/scripts/workbench-native-smoke.mjs b/scripts/workbench-native-smoke.mjs new file mode 100755 index 00000000..cb57925d --- /dev/null +++ b/scripts/workbench-native-smoke.mjs @@ -0,0 +1,68 @@ +#!/usr/bin/env node +import { spawn } from "node:child_process"; +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +const root = path.resolve(import.meta.dirname, ".."); +const stateDir = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-native-")); +const children = []; +const logs = []; + +try { + const api = start(["bun", "cmd/hwlab-workbench-api/main.ts"], { + ...process.env, + WORKBENCH_MODE: "native-test", + WORKBENCH_NATIVE_STATE_FILE: path.join(stateDir, "state.json"), + WORKBENCH_API_HOST: "127.0.0.1", + WORKBENCH_API_PORT: "6677" + }, "api"); + children.push(api); + await waitFor("http://127.0.0.1:6677/health/ready"); + + const web = start(["bun", "run", "workbench:web:dev"], process.env, "web"); + children.push(web); + await waitFor("http://127.0.0.1:5173/workbench", { acceptHtml: true }); + + const commandHealth = await requiredJson(await fetch("http://127.0.0.1:6677/v1/workbench/commands", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ operation: "health" }) }), "command.health"); + if (commandHealth.mode !== "native-test") throw new Error("native command transport did not expose mode=native-test"); + + const actorHeaders = { "content-type": "application/json", cookie: "hwlab_session=native-test" }; + const sessionResponse = await fetch("http://127.0.0.1:6677/v1/agent/sessions", { method: "POST", headers: actorHeaders, body: JSON.stringify({ providerProfile: "native-test" }) }); + const session = await requiredJson(sessionResponse, "session.create"); + const sessionId = session.session?.sessionId; + if (!sessionId) throw new Error("native session.create returned no sessionId"); + + const traceId = "trc_native_smoke"; + const submit = await requiredJson(await fetch("http://127.0.0.1:6677/v1/agent/chat", { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId, message: "native smoke" }) }), "turn.submit"); + if (submit.orchestrationMode !== "native-test" || submit.resultSynthesized !== false) throw new Error("native submit mode/authority contract failed"); + const cancel = await requiredJson(await fetch("http://127.0.0.1:6677/v1/agent/chat/cancel", { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId }) }), "turn.cancel"); + if (cancel.status !== "cancel_requested") throw new Error("native cancel was not accepted"); + + process.stdout.write(`${JSON.stringify({ ok: true, mode: "native-test", apiUrl: "http://127.0.0.1:6677", webUrl: "http://127.0.0.1:5173/workbench", sessionId, traceId, services: { api: "independent", web: "vite-hmr" }, terminalAuthority: "fixture-only", resultSynthesized: false })}\n`); +} catch (error) { + process.stderr.write(`${JSON.stringify({ ok: false, error: { code: "workbench_native_smoke_failed", message: error instanceof Error ? error.message : String(error) }, logs: logs.slice(-20) })}\n`); + process.exitCode = 1; +} finally { + for (const child of children.reverse()) child.kill("SIGTERM"); + await Promise.all(children.map((child) => child.exitCode !== null || child.signalCode !== null ? Promise.resolve() : new Promise((resolve) => child.once("exit", resolve)))); + await rm(stateDir, { recursive: true, force: true }); +} + +function start(command, env, service) { + const child = spawn(command[0], command.slice(1), { cwd: root, env, stdio: ["ignore", "pipe", "pipe"] }); + child.stdout.on("data", (chunk) => append(service, "stdout", chunk)); + child.stderr.on("data", (chunk) => append(service, "stderr", chunk)); + return child; +} +function append(service, stream, chunk) { for (const line of String(chunk).split("\n").filter(Boolean)) logs.push({ service, stream, line: line.slice(0, 500) }); if (logs.length > 100) logs.splice(0, logs.length - 100); } +async function waitFor(url, options = {}) { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + const response = await fetch(url).catch(() => null); + if (response?.ok && (!options.acceptHtml || String(response.headers.get("content-type")).includes("text/html"))) return; + await new Promise((resolve) => setTimeout(resolve, 250)); + } + throw new Error(`timed out waiting for ${url}`); +} +async function requiredJson(response, operation) { const body = await response.json().catch(() => null); if (!response.ok || !body) throw new Error(`${operation} failed with HTTP ${response.status}`); return body; } diff --git a/tools/hwlab-cli/bin/hwlab-cli.ts b/tools/hwlab-cli/bin/hwlab-cli.ts index 6b4c5493..c6ce93ba 100644 --- a/tools/hwlab-cli/bin/hwlab-cli.ts +++ b/tools/hwlab-cli/bin/hwlab-cli.ts @@ -12,6 +12,16 @@ if (argv[0] === "tasktree") { console.log(JSON.stringify({ ok: false, operation: "tasktree", error: { code: (error as any)?.code ?? "tasktree_cli_error", message: (error as any)?.message ?? String(error) } }, null, 2)); process.exitCode = 1; } +} else if (argv[0] === "workbench") { + const { runWorkbenchCli } = await import("../../src/workbench-cli.ts"); + try { + const result = await runWorkbenchCli(argv.slice(1)); + console.log(JSON.stringify(result, null, 2)); + process.exitCode = result?.ok === false ? 1 : 0; + } catch (error: any) { + console.log(JSON.stringify({ ok: false, operation: "workbench", error: { code: error?.code ?? "workbench_cli_error", message: error?.message ?? String(error) } }, null, 2)); + process.exitCode = 1; + } } else if (argv[0] === "kafka") { const { mainKafkaCli } = await import("../../src/hwlab-cli/kafka-regenerate.ts"); await mainKafkaCli(argv.slice(1)); diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts new file mode 100644 index 00000000..262b6fa1 --- /dev/null +++ b/tools/src/workbench-cli.ts @@ -0,0 +1,81 @@ +import type { WorkbenchCommand } from "../../internal/workbench/contracts.ts"; +import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; +import { workbenchNativeServiceCommand } from "./workbench-native-service.ts"; + +export async function runWorkbenchCli(argv: string[], env: Record = process.env) { + const parsed = parse(argv); + if (parsed.help || parsed.positionals.length === 0) return help(); + if (parsed.positionals[0] === "service") return workbenchNativeServiceCommand({ service: parsed.positionals[1], action: parsed.positionals[2] ?? "status", cwd: process.cwd(), env }); + const command = commandFrom(parsed, env); + if (parsed.overApi) return overApi(command, parsed, env); + const runtime = workbenchRuntime(env); + try { + const result = await runtime.dispatch(command); + return { ...result, transport: "local", route: "application-dispatcher", identity: identity(result.data) }; + } finally { + await runtime.close(); + } +} + +function help() { + return { + ok: true, + operation: "help", + mode: "local-dispatcher-default", + usage: [ + "hwlab-cli workbench health [--over-api]", + "hwlab-cli workbench session create --actor-id ID [--provider-profile PROFILE] [--over-api]", + "hwlab-cli workbench turn submit --actor-id ID --session-id ID --message TEXT [--trace-id ID] [--over-api]", + "hwlab-cli workbench turn cancel --actor-id ID --trace-id ID [--over-api]", + "hwlab-cli workbench service api|worker|web start|stop|restart|status|logs" + ], + transportContract: "--over-api only changes transport; --overapi is unsupported", + localConfig: ["WORKBENCH_MODE", "WORKBENCH_NATIVE_STATE_FILE"], + apiConfig: ["WORKBENCH_API_URL", "HWLAB_API_KEY"] + }; +} + +function commandFrom(parsed: Parsed, env: Record): WorkbenchCommand { + const [group, action] = parsed.positionals; + 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 === "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"], shortConnection: true }) }; + if (group === "turn" && action === "cancel") return { operation: "turn.cancel", actor, traceId: required(parsed, "trace-id"), params: compact({ sessionId: parsed.values["session-id"] }) }; + throw codedError("unsupported_workbench_command", `unsupported Workbench command: ${group} ${action ?? ""}`.trim()); +} + +async function overApi(command: WorkbenchCommand, parsed: Parsed, env: Record) { + const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL; + if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api"); + const headers: Record = { "content-type": "application/json" }; + const apiKey = env.HWLAB_API_KEY; + if (apiKey) headers.authorization = `Bearer ${apiKey}`; + const route = "/v1/workbench/commands"; + const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${route}`, { method: "POST", headers, body: JSON.stringify(command) }); + const body = await response.json().catch(() => null) as Record | null; + if (!body) throw codedError("workbench_api_invalid_json", `Workbench API returned invalid JSON (${response.status})`); + return { ...body, transport: "api", baseUrl, route: `POST ${route}`, httpStatus: response.status, identity: identity(body.data) }; +} + +type Parsed = { positionals: string[]; values: Record; overApi: boolean; help: boolean }; +function parse(argv: string[]): Parsed { + const parsed: Parsed = { positionals: [], values: {}, overApi: false, help: false }; + for (let index = 0; index < argv.length; index += 1) { + const token = argv[index]; + if (token === "--overapi") throw codedError("unsupported_option", "--overapi was retired; use --over-api"); + if (token === "--over-api") parsed.overApi = true; + else if (token === "--help" || token === "-h") parsed.help = true; + else if (token.startsWith("--")) { + const value = argv[index + 1]; + if (!value || value.startsWith("--")) throw codedError("option_value_required", `${token} requires a value`); + parsed.values[token.slice(2)] = value; + index += 1; + } else parsed.positionals.push(token); + } + return parsed; +} +function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, traceId: data?.traceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; } +function compact(value: Record) { return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)); } +function required(parsed: Parsed, key: string) { const value = parsed.values[key]; if (!value) throw codedError("option_required", `--${key} is required`); return value; } +function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); } diff --git a/tools/src/workbench-native-service.ts b/tools/src/workbench-native-service.ts new file mode 100644 index 00000000..0e7c6c4e --- /dev/null +++ b/tools/src/workbench-native-service.ts @@ -0,0 +1,88 @@ +import { spawn } from "node:child_process"; +import { closeSync, openSync } from "node:fs"; +import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import path from "node:path"; + +type ServiceName = "api" | "worker" | "web"; + +export async function workbenchNativeServiceCommand(input: { service: string; action: string; cwd: string; env: Record }) { + const service = validService(input.service); + const stateDir = path.resolve(input.cwd, input.env.WORKBENCH_NATIVE_SERVICE_STATE_DIR ?? ".state/workbench-native/services"); + const stateFile = path.join(stateDir, `${service}.json`); + const logFile = path.join(stateDir, `${service}.log`); + if (input.action === "status") return status(service, stateFile, logFile); + if (input.action === "logs") return logs(service, stateFile, logFile); + if (input.action === "stop") return stop(service, stateFile, logFile); + if (input.action === "restart") { + await stop(service, stateFile, logFile); + return start(service, stateDir, stateFile, logFile, input.cwd, input.env); + } + if (input.action === "start") return start(service, stateDir, stateFile, logFile, input.cwd, input.env); + throw codedError("unsupported_service_action", `unsupported Workbench service action: ${input.action}`); +} + +async function start(service: ServiceName, stateDir: string, stateFile: string, logFile: string, cwd: string, env: Record) { + const current = await readState(stateFile); + if (current?.pid && alive(current.pid)) throw codedError("service_already_running", `${service} is already running`, { pid: current.pid, correctionCommand: `hwlab-cli workbench service ${service} stop` }); + await mkdir(stateDir, { recursive: true }); + const command = serviceCommand(service, env); + const logFd = openSync(logFile, "w"); + const child = spawn(command[0], command.slice(1), { + cwd, + env: serviceEnv(service, env), + detached: true, + stdio: ["ignore", logFd, logFd] + }); + closeSync(logFd); + child.unref(); + const state = { service, pid: child.pid, status: "starting", command, cwd, logFile, healthUrl: healthUrl(service, env), startedAt: new Date().toISOString() }; + await writeFile(stateFile, `${JSON.stringify(state, null, 2)}\n`, "utf8"); + return { ok: true, operation: `workbench.service.${service}.start`, ...state, nextCommand: `hwlab-cli workbench service ${service} status` }; +} + +async function stop(service: ServiceName, stateFile: string, logFile: string) { + const state = await readState(stateFile); + if (!state?.pid || !alive(state.pid)) { await rm(stateFile, { force: true }); return { ok: true, operation: `workbench.service.${service}.stop`, service, status: "stopped", alreadyStopped: true, logFile }; } + try { process.kill(-state.pid, "SIGTERM"); } catch (error: any) { if (error?.code !== "ESRCH") throw error; } + await rm(stateFile, { force: true }); + return { ok: true, operation: `workbench.service.${service}.stop`, service, status: "stopped", pid: state.pid, logFile }; +} + +async function status(service: ServiceName, stateFile: string, logFile: string) { + const state = await readState(stateFile); + const processRunning = Boolean(state?.pid && alive(state.pid)); + const health = processRunning ? await probeHealth(state.healthUrl ?? healthUrl(service, {})) : { ok: false, status: null, error: "process-not-running" }; + const running = processRunning && health.ok; + return { ok: running, operation: `workbench.service.${service}.status`, service, status: running ? "running" : processRunning ? "degraded" : "stopped", pid: processRunning ? state.pid : null, health, stateFile, logFile, nextCommand: running ? `hwlab-cli workbench service ${service} logs` : `hwlab-cli workbench service ${service} logs` }; +} + +async function logs(service: ServiceName, stateFile: string, logFile: string) { + const state = await status(service, stateFile, logFile); + const lines = (await readFile(logFile, "utf8").catch(() => "")).split("\n").filter(Boolean).slice(-80); + return { ...state, operation: `workbench.service.${service}.logs`, lines, truncated: lines.length === 80 }; +} + +function serviceCommand(service: ServiceName, env: Record) { + const watch = env.WORKBENCH_BACKEND_WATCH === "1" ? ["--watch"] : []; + if (service === "api") return ["bun", ...watch, "cmd/hwlab-workbench-api/main.ts"]; + if (service === "worker") return ["bun", ...watch, "cmd/hwlab-workbench-worker/main.ts"]; + return ["bun", "run", "workbench:web:dev"]; +} +function serviceEnv(service: ServiceName, env: Record) { + if (service === "api") return { ...process.env, ...env, WORKBENCH_MODE: env.WORKBENCH_MODE ?? "native-test", WORKBENCH_API_HOST: env.WORKBENCH_API_HOST ?? "127.0.0.1", WORKBENCH_API_PORT: env.WORKBENCH_API_PORT ?? "6677" }; + if (service === "web") return { ...process.env, ...env, WORKBENCH_NATIVE_API_URL: env.WORKBENCH_NATIVE_API_URL ?? "http://127.0.0.1:6677" }; + return { ...process.env, ...env, WORKBENCH_MODE: "temporal" }; +} +function validService(value: string): ServiceName { if (value === "api" || value === "worker" || value === "web") return value; throw codedError("invalid_service", "service must be api, worker, or web"); } +async function readState(file: string) { return JSON.parse(await readFile(file, "utf8").catch(() => "null")); } +function alive(pid: number) { try { process.kill(pid, 0); return true; } catch { return false; } } +function healthUrl(service: ServiceName, env: Record) { + if (service === "api") return `http://${env.WORKBENCH_API_HOST ?? "127.0.0.1"}:${env.WORKBENCH_API_PORT ?? "6677"}/health/ready`; + if (service === "worker") return `http://${env.WORKBENCH_WORKER_HEALTH_HOST ?? "127.0.0.1"}:${env.WORKBENCH_WORKER_HEALTH_PORT ?? "6678"}/health/ready`; + return `http://${env.WORKBENCH_WEB_HOST ?? "127.0.0.1"}:${env.WORKBENCH_WEB_PORT ?? "5173"}/workbench`; +} +async function probeHealth(url: string) { + try { const response = await fetch(url, { signal: AbortSignal.timeout(1000) }); return { ok: response.ok, status: response.status, url }; } + catch (error) { return { ok: false, status: null, url, error: error instanceof Error ? error.message : String(error) }; } +} +function codedError(code: string, message: string, details?: unknown) { return Object.assign(new Error(message), { code, details }); } diff --git a/web/hwlab-cloud-web/vite.config.ts b/web/hwlab-cloud-web/vite.config.ts index 06055ca4..2d7eb6da 100644 --- a/web/hwlab-cloud-web/vite.config.ts +++ b/web/hwlab-cloud-web/vite.config.ts @@ -6,10 +6,15 @@ import { defineConfig, loadEnv } from "vite"; export default defineConfig(({ mode }) => { const env = loadEnv(mode, process.cwd(), ""); + const nativeApiUrl = String(process.env.WORKBENCH_NATIVE_API_URL ?? env.WORKBENCH_NATIVE_API_URL ?? "").trim(); const nativeCaseRunEnabled = (process.env.HWLAB_CASERUN_NATIVE_TEST ?? env.HWLAB_CASERUN_NATIVE_TEST) === "1"; const nativeCaseRunTarget = process.env.HWLAB_CASERUN_NATIVE_URL ?? env.HWLAB_CASERUN_NATIVE_URL ?? "http://127.0.0.1:4316"; const nativeCaseRunAgent = new Agent({ keepAlive: true }); + if (nativeApiUrl && nativeCaseRunEnabled) { + throw new Error("Workbench and CaseRun native modes cannot share one Vite server"); + } + return { plugins: [vue()], resolve: { @@ -17,24 +22,34 @@ export default defineConfig(({ mode }) => { "@": fileURLToPath(new URL("./src", import.meta.url)) } }, - server: nativeCaseRunEnabled ? { - watch: { - usePolling: true, - interval: 300 - }, - proxy: { - "/auth": { - target: nativeCaseRunTarget, - changeOrigin: false, - agent: nativeCaseRunAgent - }, - "/v1/caserun": { - target: nativeCaseRunTarget, - changeOrigin: false, - agent: nativeCaseRunAgent - } - } - } : undefined, + server: { + host: env.WORKBENCH_WEB_HOST || "127.0.0.1", + port: Number(env.WORKBENCH_WEB_PORT || 5173), + strictPort: true, + watch: nativeCaseRunEnabled || env.WORKBENCH_VITE_USE_POLLING === "1" + ? { usePolling: true, interval: nativeCaseRunEnabled ? 300 : 500 } + : undefined, + proxy: nativeCaseRunEnabled + ? { + "/auth": { + target: nativeCaseRunTarget, + changeOrigin: false, + agent: nativeCaseRunAgent + }, + "/v1/caserun": { + target: nativeCaseRunTarget, + changeOrigin: false, + agent: nativeCaseRunAgent + } + } + : nativeApiUrl + ? { + "/v1": { target: nativeApiUrl, changeOrigin: false }, + "/auth": { target: nativeApiUrl, changeOrigin: false }, + "/health": { target: nativeApiUrl, changeOrigin: false } + } + : undefined + }, build: { outDir: "dist", emptyOutDir: true,