From cf9ac0b08e6163a318705bc3e530495c9e49830c Mon Sep 17 00:00:00 2001 From: lyon Date: Sun, 21 Jun 2026 19:42:26 +0800 Subject: [PATCH] feat: split workbench runtime read service --- cmd/hwlab-workbench-runtime/main.go | 16 + deploy/deploy.schema.json | 1 + deploy/deploy.yaml | 28 + deploy/k8s/base/services.yaml | 25 + deploy/k8s/base/workloads.yaml | 60 ++ .../runtime/boot/hwlab-workbench-runtime.sh | 18 + internal/cloud/workbench-facts-store.ts | 13 +- internal/cloud/workbench-runtime-client.ts | 85 ++ internal/workbenchruntime/service.go | 725 ++++++++++++++++++ scripts/artifact-publish.mjs | 1 + 10 files changed, 966 insertions(+), 6 deletions(-) create mode 100644 cmd/hwlab-workbench-runtime/main.go create mode 100644 deploy/runtime/boot/hwlab-workbench-runtime.sh create mode 100644 internal/cloud/workbench-runtime-client.ts create mode 100644 internal/workbenchruntime/service.go diff --git a/cmd/hwlab-workbench-runtime/main.go b/cmd/hwlab-workbench-runtime/main.go new file mode 100644 index 00000000..4d4caded --- /dev/null +++ b/cmd/hwlab-workbench-runtime/main.go @@ -0,0 +1,16 @@ +package main + +import ( + "context" + "fmt" + "os" + + "github.com/pikasTech/HWLAB/internal/workbenchruntime" +) + +func main() { + if err := workbenchruntime.Run(context.Background()); err != nil { + fmt.Fprintf(os.Stderr, "hwlab-workbench-runtime: %v\n", err) + os.Exit(1) + } +} diff --git a/deploy/deploy.schema.json b/deploy/deploy.schema.json index 385ba631..b7b272ba 100644 --- a/deploy/deploy.schema.json +++ b/deploy/deploy.schema.json @@ -69,6 +69,7 @@ "type": "string", "enum": [ "hwlab-cloud-api", + "hwlab-workbench-runtime", "hwlab-user-billing", "hwlab-cloud-web", "hwlab-gateway", diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index b0bb9275..77e4424f 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -308,6 +308,7 @@ lanes: queryRetryMaxDelayMs: 5000 envReuseServices: - hwlab-cloud-api + - hwlab-workbench-runtime - hwlab-user-billing - hwlab-cloud-web - hwlab-gateway @@ -315,6 +316,7 @@ lanes: - hwlab-agent-skills bootScripts: hwlab-cloud-api: deploy/runtime/boot/hwlab-cloud-api.sh + hwlab-workbench-runtime: deploy/runtime/boot/hwlab-workbench-runtime.sh hwlab-user-billing: deploy/runtime/boot/hwlab-user-billing.sh hwlab-cloud-web: deploy/runtime/boot/hwlab-cloud-web.sh hwlab-gateway: deploy/runtime/boot/hwlab-gateway.sh @@ -367,6 +369,21 @@ lanes: HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_JITTER_MS: "15000" HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_FAILURE_BACKOFF_MS: "120000" observable: true + hwlab-workbench-runtime: + runtimeKind: go-service + entrypoint: cmd/hwlab-workbench-runtime/main.go + artifactKind: go-service + healthPath: /health/ready + healthPort: 6671 + componentPaths: + - go.mod + - go.sum + - cmd/hwlab-workbench-runtime/ + - internal/workbenchruntime/ + - deploy/runtime/boot/hwlab-workbench-runtime.sh + env: + TZ: Asia/Shanghai + observable: true hwlab-user-billing: runtimeKind: go-service entrypoint: cmd/hwlab-user-billing/main.go @@ -499,6 +516,8 @@ lanes: HWLAB_BOOTSTRAP_ADMIN_API_KEY_ID: key_master_server_admin 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_RUNTIME_TIMEOUT_MS: "10000" HWLAB_USER_BILLING_LOGIN_RETRY_ATTEMPTS: "2" HWLAB_USER_BILLING_LOGIN_RETRY_DELAY_MS: "250" HWLAB_USER_BILLING_CODE_AGENT_ENABLED: "1" @@ -522,6 +541,15 @@ lanes: HWLAB_WORKBENCH_EMPTY_SESSION_GC_BATCH_SIZE: "100" HWLAB_WORKBENCH_EMPTY_SESSION_GC_JITTER_MS: "15000" HWLAB_WORKBENCH_EMPTY_SESSION_GC_FAILURE_BACKOFF_MS: "120000" + - serviceId: hwlab-workbench-runtime + env: + HWLAB_WORKBENCH_RUNTIME_DB_URL: secretRef:hwlab-cloud-api-v03-db/database-url + HWLAB_WORKBENCH_RUNTIME_PORT: "6671" + HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX: "8" + HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS: "10000" + HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS: "2000" + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: http://otel-collector.platform-infra.svc.cluster.local:4318/v1/traces + OTEL_SERVICE_NAME: hwlab-workbench-runtime - serviceId: hwlab-user-billing env: HWLAB_USER_BILLING_DB_URL: secretRef:hwlab-cloud-api-v03-db/database-url diff --git a/deploy/k8s/base/services.yaml b/deploy/k8s/base/services.yaml index 2afdc458..19d55301 100644 --- a/deploy/k8s/base/services.yaml +++ b/deploy/k8s/base/services.yaml @@ -27,6 +27,31 @@ ] } }, + { + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": "hwlab-workbench-runtime", + "namespace": "hwlab-dev", + "labels": { + "app.kubernetes.io/name": "hwlab-workbench-runtime", + "hwlab.pikastech.local/service-id": "hwlab-workbench-runtime" + } + }, + "spec": { + "type": "ClusterIP", + "selector": { + "app.kubernetes.io/name": "hwlab-workbench-runtime" + }, + "ports": [ + { + "name": "http", + "port": 6671, + "targetPort": "http" + } + ] + } + }, { "apiVersion": "v1", "kind": "Service", diff --git a/deploy/k8s/base/workloads.yaml b/deploy/k8s/base/workloads.yaml index 746b1fed..8b71960e 100644 --- a/deploy/k8s/base/workloads.yaml +++ b/deploy/k8s/base/workloads.yaml @@ -46,6 +46,66 @@ } } }, + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "hwlab-workbench-runtime", + "namespace": "hwlab-dev", + "labels": { + "app.kubernetes.io/name": "hwlab-workbench-runtime", + "hwlab.pikastech.local/service-id": "hwlab-workbench-runtime" + } + }, + "spec": { + "replicas": 1, + "selector": { + "matchLabels": { + "app.kubernetes.io/name": "hwlab-workbench-runtime" + } + }, + "template": { + "metadata": { + "labels": { + "app.kubernetes.io/name": "hwlab-workbench-runtime", + "hwlab.pikastech.local/service-id": "hwlab-workbench-runtime" + } + }, + "spec": { + "containers": [ + { + "name": "hwlab-workbench-runtime", + "image": "127.0.0.1:5000/hwlab/hwlab-workbench-runtime:af46386", + "ports": [ + { + "name": "http", + "containerPort": 6671 + } + ], + "env": [ + { + "name": "HWLAB_WORKBENCH_RUNTIME_PORT", + "value": "6671" + } + ], + "readinessProbe": { + "httpGet": { + "path": "/health/ready", + "port": "http" + } + }, + "livenessProbe": { + "httpGet": { + "path": "/health/live", + "port": "http" + } + } + } + ] + } + } + } + }, { "apiVersion": "apps/v1", "kind": "Deployment", diff --git a/deploy/runtime/boot/hwlab-workbench-runtime.sh b/deploy/runtime/boot/hwlab-workbench-runtime.sh new file mode 100644 index 00000000..7fe2896e --- /dev/null +++ b/deploy/runtime/boot/hwlab-workbench-runtime.sh @@ -0,0 +1,18 @@ +#!/usr/bin/env sh +set -eu + +cd /workspace/hwlab-boot/repo + +export HWLAB_SERVICE_ID="${HWLAB_SERVICE_ID:-hwlab-workbench-runtime}" +export HWLAB_ARTIFACT_KIND="${HWLAB_ARTIFACT_KIND:-go-service}" +export HWLAB_WORKBENCH_RUNTIME_PORT="${HWLAB_WORKBENCH_RUNTIME_PORT:-6671}" +export PORT="${PORT:-$HWLAB_WORKBENCH_RUNTIME_PORT}" +export GOMODCACHE="${GOMODCACHE:-/opt/hwlab-env/go/pkg/mod}" +export GOCACHE="${GOCACHE:-/opt/hwlab-env/go/cache}" +export GOPROXY="${GOPROXY:-https://goproxy.cn,direct}" +export GOTOOLCHAIN="${GOTOOLCHAIN:-local}" +export CGO_ENABLED="${CGO_ENABLED:-0}" + +mkdir -p "$GOMODCACHE" "$GOCACHE" /tmp/hwlab-workbench-runtime +go build -trimpath -o /tmp/hwlab-workbench-runtime/hwlab-workbench-runtime ./cmd/hwlab-workbench-runtime +exec /tmp/hwlab-workbench-runtime/hwlab-workbench-runtime diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts index 8465c57f..c218a417 100644 --- a/internal/cloud/workbench-facts-store.ts +++ b/internal/cloud/workbench-facts-store.ts @@ -5,6 +5,7 @@ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { safeConversationId, safeSessionId, safeTraceId } from "./server-http-utils.ts"; +import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts"; import { durableTraceStatus, normalizeWorkbenchStatus, TERMINAL_STATUSES, traceTerminalEvidence } from "./workbench-turn-projection.ts"; const TERMINAL_DURABLE_TRACE_CACHE_MAX = 256; @@ -18,15 +19,16 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { const traceSessionCache = options.codeAgentTraceSessionCache instanceof Map ? options.codeAgentTraceSessionCache : null; const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; const runtimeStore = options.runtimeStore ?? null; + const runtimeReader = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger }) ?? runtimeStore; const results = options.codeAgentChatResults ?? null; async function queryFacts(params = {}) { - if (typeof runtimeStore?.queryWorkbenchFacts === "function") { + if (typeof runtimeReader?.queryWorkbenchFacts === "function") { const retryPolicy = workbenchFactsQueryRetryPolicy(options.env ?? process.env); const retryDelaysMs = []; for (let attempt = 1; attempt <= retryPolicy.maxAttempts; attempt += 1) { try { - const result = await runtimeStore.queryWorkbenchFacts(params); + const result = await runtimeReader.queryWorkbenchFacts(params); return { facts: normalizeWorkbenchFactsResult(result?.facts), count: Number.isFinite(Number(result?.count)) ? Number(result.count) : 0, @@ -139,7 +141,7 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { async function traceSnapshot(traceId) { const memory = traceSnapshotSync(traceId); if (hasTraceProjection(memory) && TERMINAL_STATUSES.has(normalizeStatus(memory.status))) return memory; - const durable = await durableTraceSnapshot(runtimeStore, traceId); + const durable = await durableTraceSnapshot(runtimeReader, traceId); if (isProjectionDiagnosticTrace(durable) && hasTraceProjection(memory)) return mergeTraceProjectionDiagnostic(memory, durable); if (durable && shouldPreferDurableTrace(memory, durable)) return durable; if (hasTraceProjection(memory)) return memory; @@ -151,15 +153,14 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { } function subscribeDurableProjection(traceId, listener, pollOptions = {}) { - const runtimeStore = options.runtimeStore ?? null; - if (typeof runtimeStore?.readWorkbenchProjectionOutbox !== "function") return () => {}; + if (typeof runtimeReader?.readWorkbenchProjectionOutbox !== "function") return () => {}; const pollMs = Math.max(500, Number(pollOptions.pollMs ?? 2000)); let cursor = 0; let stopped = false; const poll = async () => { if (stopped) return; try { - const rows = await runtimeStore.readWorkbenchProjectionOutbox({ afterSeq: cursor, limit: 50, traceId }); + const rows = await runtimeReader.readWorkbenchProjectionOutbox({ afterSeq: cursor, limit: 50, traceId }); for (const row of rows) { if (stopped) break; cursor = row.outboxSeq; diff --git a/internal/cloud/workbench-runtime-client.ts b/internal/cloud/workbench-runtime-client.ts new file mode 100644 index 00000000..1831e9f4 --- /dev/null +++ b/internal/cloud/workbench-runtime-client.ts @@ -0,0 +1,85 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影; HWLAB#1832 Workbench runtime Go 微服务拆分. + * 职责: Cloud API 到 hwlab-workbench-runtime 的单一路径只读客户端；配置后不回退本地 PG。 + */ + +const DEFAULT_WORKBENCH_RUNTIME_TIMEOUT_MS = 10000; + +export function createWorkbenchRuntimeClient(options = {}) { + const env = options.env ?? process.env; + const baseUrl = normalizeBaseUrl(env.HWLAB_WORKBENCH_RUNTIME_URL); + if (!baseUrl) return null; + const timeoutMs = boundedInteger(env.HWLAB_WORKBENCH_RUNTIME_TIMEOUT_MS, DEFAULT_WORKBENCH_RUNTIME_TIMEOUT_MS, 1000, 60000); + const fetchFn = options.fetch ?? globalThis.fetch; + return { + serviceId: "hwlab-workbench-runtime", + queryWorkbenchFacts: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/facts", body: params }), + queryAgentTraceEvents: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/trace-events", body: params }), + readWorkbenchProjectionOutbox: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/projection-outbox", body: params }).then((result) => Array.isArray(result?.rows) ? result.rows : []) + }; +} + +async function postRuntime({ fetchFn, baseUrl, timeoutMs, path, body }) { + if (typeof fetchFn !== "function") { + throw runtimeDependencyError({ code: "workbench_runtime_fetch_unavailable", message: "fetch is unavailable for workbench runtime dependency", retryable: false }); + } + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + let response; + try { + response = await fetchFn(new URL(path, baseUrl), { + method: "POST", + headers: { "content-type": "application/json", accept: "application/json" }, + body: JSON.stringify(body ?? {}), + signal: controller.signal + }); + } catch (error) { + throw runtimeDependencyError({ code: error?.name === "AbortError" ? "workbench_runtime_timeout" : "workbench_runtime_unreachable", message: error?.message ?? "workbench runtime request failed", retryable: true, cause: error }); + } finally { + clearTimeout(timer); + } + const payload = await responseJson(response); + if (!response.ok) { + const error = payload?.error ?? {}; + throw runtimeDependencyError({ + code: error.code ?? `workbench_runtime_http_${response.status}`, + message: error.message ?? `workbench runtime request failed with HTTP ${response.status}`, + retryable: response.status >= 500 || error.retryable === true, + status: response.status, + payload + }); + } + return payload; +} + +async function responseJson(response) { + try { return await response.json(); } catch { return null; } +} + +function runtimeDependencyError({ code, message, retryable, status = null, payload = null, cause = null }) { + const error = new Error(message || code || "workbench runtime dependency failed"); + error.name = "WorkbenchRuntimeDependencyError"; + error.code = code ?? "workbench_runtime_dependency_failed"; + if (cause) error.cause = cause; + error.data = { + serviceId: "hwlab-workbench-runtime", + status, + retryable: retryable === true, + transient: retryable === true, + payload, + valuesRedacted: true + }; + return error; +} + +function normalizeBaseUrl(value) { + const text = String(value ?? "").trim(); + if (!text) return null; + return text.endsWith("/") ? text : `${text}/`; +} + +function boundedInteger(value, fallback, min, max) { + const parsed = Number(value); + if (!Number.isFinite(parsed)) return fallback; + return Math.max(min, Math.min(max, Math.trunc(parsed))); +} diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go new file mode 100644 index 00000000..b40e4096 --- /dev/null +++ b/internal/workbenchruntime/service.go @@ -0,0 +1,725 @@ +package workbenchruntime + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + _ "github.com/jackc/pgx/v4/stdlib" +) + +const serviceID = "hwlab-workbench-runtime" + +type Config struct { + Addr string + DatabaseURL string + QueryTimeout time.Duration + ReadinessDelay time.Duration + PoolMax int +} + +type Server struct { + config Config + db *sql.DB + mux *http.ServeMux +} + +type factQuery struct { + Families []string `json:"families"` + FactFamilies []string `json:"factFamilies"` + Limit int `json:"limit"` + AfterProjectedSeq int64 `json:"afterProjectedSeq"` + Order string `json:"order"` + SessionsOrder string `json:"sessionsOrder"` + MessagesOrder string `json:"messagesOrder"` + PartsOrder string `json:"partsOrder"` + TurnsOrder string `json:"turnsOrder"` + CheckpointsOrder string `json:"checkpointsOrder"` + SessionProjection string `json:"sessionProjection"` + SessionID string `json:"sessionId"` + SessionIDs []string `json:"sessionIds"` + TraceID string `json:"traceId"` + TraceIDs []string `json:"traceIds"` + MessageID string `json:"messageId"` + MessageIDs []string `json:"messageIds"` + TurnID string `json:"turnId"` + TurnIDs []string `json:"turnIds"` + Status string `json:"status"` + Statuses []string `json:"statuses"` + OwnerUserID string `json:"ownerUserId"` + OwnerUserIDs []string `json:"ownerUserIds"` + ConversationID string `json:"conversationId"` + ConversationIDs []string `json:"conversationIds"` + EventType string `json:"eventType"` + EventTypes []string `json:"eventTypes"` + ProjectionStatus string `json:"projectionStatus"` + ProjectionStatuses []string `json:"projectionStatuses"` + ProjectionHealth string `json:"projectionHealth"` + ProjectionHealths []string `json:"projectionHealths"` + RunID string `json:"runId"` + RunIDs []string `json:"runIds"` + CommandID string `json:"commandId"` + CommandIDs []string `json:"commandIds"` +} + +type traceEventsQuery struct { + TraceID string `json:"traceId"` + SessionID string `json:"sessionId"` + WorkerSessionID string `json:"workerSessionId"` + Level string `json:"level"` + Limit int `json:"limit"` +} + +type outboxQuery struct { + AfterSeq int64 `json:"afterSeq"` + Limit int `json:"limit"` + TraceID string `json:"traceId"` + SessionID string `json:"sessionId"` +} + +type fieldColumn struct { + field string + values []string + column string +} + +func Run(ctx context.Context) error { + config := configFromEnv() + if strings.TrimSpace(config.DatabaseURL) == "" { + return errors.New("HWLAB_WORKBENCH_RUNTIME_DB_URL or HWLAB_CLOUD_DB_URL is required") + } + db, err := sql.Open("pgx", config.DatabaseURL) + if err != nil { + return err + } + defer db.Close() + if config.PoolMax > 0 { + db.SetMaxOpenConns(config.PoolMax) + db.SetMaxIdleConns(config.PoolMax) + } + server := NewServer(config, db) + httpServer := &http.Server{Addr: config.Addr, Handler: server.mux, ReadHeaderTimeout: 5 * time.Second} + errCh := make(chan error, 1) + go func() { + log.Printf("%s listening addr=%s", serviceID, config.Addr) + errCh <- httpServer.ListenAndServe() + }() + stopCh := make(chan os.Signal, 1) + signal.Notify(stopCh, syscall.SIGINT, syscall.SIGTERM) + select { + case <-ctx.Done(): + case <-stopCh: + case err := <-errCh: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return httpServer.Shutdown(shutdownCtx) +} + +func NewServer(config Config, db *sql.DB) *Server { + server := &Server{config: config, db: db, mux: http.NewServeMux()} + server.routes() + return server +} + +func (s *Server) routes() { + s.mux.HandleFunc("/health/live", s.handleLive) + s.mux.HandleFunc("/health/ready", s.handleReady) + s.mux.HandleFunc("/v1/workbench-runtime/facts", s.handleFacts) + s.mux.HandleFunc("/v1/workbench-runtime/trace-events", s.handleTraceEvents) + s.mux.HandleFunc("/v1/workbench-runtime/projection-outbox", s.handleProjectionOutbox) +} + +func (s *Server) handleLive(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "live", "valuesRedacted": true}) +} + +func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), s.config.ReadinessDelay) + defer cancel() + if err := s.db.PingContext(ctx); err != nil { + writeAPIError(w, http.StatusServiceUnavailable, "workbench_runtime_db_unavailable", "workbench runtime database is unavailable") + return + } + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "ready", "valuesRedacted": true}) +} + +func (s *Server) handleFacts(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w, "POST") + return + } + var query factQuery + if !decodeJSON(w, r, &query) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout) + defer cancel() + facts, err := s.queryFacts(ctx, query) + if err != nil { + writeQueryError(w, err) + return + } + count := 0 + for _, rows := range facts { + count += len(rows) + } + writeJSON(w, http.StatusOK, map[string]any{ + "contractVersion": "workbench-runtime-facts-v1", + "facts": facts, + "count": count, + "persistence": s.persistenceSummary(), + "valuesRedacted": true, + }) +} + +func (s *Server) handleTraceEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w, "POST") + return + } + var query traceEventsQuery + if !decodeJSON(w, r, &query) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout) + defer cancel() + events, err := s.queryAgentTraceEvents(ctx, query) + if err != nil { + writeQueryError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "contractVersion": "workbench-runtime-trace-events-v1", + "events": events, + "count": len(events), + "persistence": s.persistenceSummary(), + "valuesRedacted": true, + }) +} + +func (s *Server) handleProjectionOutbox(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w, "POST") + return + } + var query outboxQuery + if !decodeJSON(w, r, &query) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout) + defer cancel() + rows, err := s.readProjectionOutbox(ctx, query) + if err != nil { + writeQueryError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "contractVersion": "workbench-runtime-projection-outbox-v1", + "rows": rows, + "count": len(rows), + "persistence": s.persistenceSummary(), + "valuesRedacted": true, + }) +} + +func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][]any, error) { + families := factFamilySet(query) + facts := emptyFacts() + var err error + if families["sessions"] { + facts["sessions"], err = s.queryFactRows(ctx, "workbench_sessions", "session_json", query, []fieldColumn{ + {"sessionId", query.SessionIDs, "session_id"}, + {"traceId", query.TraceIDs, "last_trace_id"}, + {"ownerUserId", query.OwnerUserIDs, "owner_user_id"}, + {"conversationId", query.ConversationIDs, "conversation_id"}, + {"status", query.Statuses, "status"}, + }) + if err != nil { return nil, err } + } + if families["messages"] { + facts["messages"], err = s.queryFactRows(ctx, "workbench_messages", "message_json", query, []fieldColumn{ + {"messageId", query.MessageIDs, "message_id"}, + {"sessionId", query.SessionIDs, "session_id"}, + {"traceId", query.TraceIDs, "trace_id"}, + {"turnId", query.TurnIDs, "turn_id"}, + {"status", query.Statuses, "status"}, + }) + if err != nil { return nil, err } + } + if families["parts"] { + facts["parts"], err = s.queryFactRows(ctx, "workbench_parts", "part_json", query, []fieldColumn{ + {"messageId", query.MessageIDs, "message_id"}, + {"sessionId", query.SessionIDs, "session_id"}, + {"traceId", query.TraceIDs, "trace_id"}, + {"turnId", query.TurnIDs, "turn_id"}, + {"status", query.Statuses, "status"}, + }) + if err != nil { return nil, err } + } + if families["turns"] { + facts["turns"], err = s.queryFactRows(ctx, "workbench_turns", "turn_json", query, []fieldColumn{ + {"turnId", query.TurnIDs, "turn_id"}, + {"sessionId", query.SessionIDs, "session_id"}, + {"traceId", query.TraceIDs, "trace_id"}, + {"messageId", query.MessageIDs, "message_id"}, + {"status", query.Statuses, "status"}, + }) + if err != nil { return nil, err } + } + if families["traceEvents"] { + facts["traceEvents"], err = s.queryFactRows(ctx, "workbench_trace_events", "event_json", query, []fieldColumn{ + {"traceId", query.TraceIDs, "trace_id"}, + {"sessionId", query.SessionIDs, "session_id"}, + {"turnId", query.TurnIDs, "turn_id"}, + {"messageId", query.MessageIDs, "message_id"}, + {"eventType", query.EventTypes, "event_type"}, + }) + if err != nil { return nil, err } + } + if families["checkpoints"] { + facts["checkpoints"], err = s.queryFactRows(ctx, "workbench_projection_checkpoints", "checkpoint_json", query, []fieldColumn{ + {"traceId", query.TraceIDs, "trace_id"}, + {"sessionId", query.SessionIDs, "session_id"}, + {"turnId", query.TurnIDs, "turn_id"}, + {"runId", query.RunIDs, "run_id"}, + {"commandId", query.CommandIDs, "command_id"}, + {"projectionStatus", query.ProjectionStatuses, "projection_status"}, + {"projectionHealth", query.ProjectionHealths, "projection_health"}, + }) + if err != nil { return nil, err } + } + return facts, nil +} + +func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn string, query factQuery, fields []fieldColumn) ([]any, error) { + clauses, args := factClauses(query, fields) + limit := boundedLimit(query.Limit, 0, 1000) + if table == "workbench_trace_events" { + if query.AfterProjectedSeq > 0 { + args = append(args, query.AfterProjectedSeq) + clauses = append(clauses, fmt.Sprintf("projected_seq > $%d", len(args))) + } + sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY projected_seq ASC", jsonColumn, table, whereSQL(clauses)) + if limit > 0 { + args = append(args, limit) + sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) + } + return s.queryJSONColumn(ctx, sqlText, args) + } + orderDirection := "ASC" + if factOrder(query, factFamilyForTable(table)) == "updated_desc" { + orderDirection = "DESC" + } + if table == "workbench_sessions" && query.SessionProjection == "summary" { + sqlText := fmt.Sprintf("SELECT %s FROM workbench_sessions%s ORDER BY updated_at %s", sessionSummarySelectClause(), whereSQL(clauses), orderDirection) + if limit > 0 { + args = append(args, limit) + sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) + } + return s.querySessionSummaries(ctx, sqlText, args) + } + sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY updated_at %s", jsonColumn, table, whereSQL(clauses), orderDirection) + if limit > 0 { + args = append(args, limit) + sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) + } + return s.queryJSONColumn(ctx, sqlText, args) +} + +func (s *Server) queryJSONColumn(ctx context.Context, sqlText string, args []any) ([]any, error) { + rows, err := s.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + result := []any{} + for rows.Next() { + var raw []byte + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var item any + if len(raw) > 0 && json.Unmarshal(raw, &item) == nil && item != nil { + result = append(result, item) + } + } + return result, rows.Err() +} + +func (s *Server) querySessionSummaries(ctx context.Context, sqlText string, args []any) ([]any, error) { + rows, err := s.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + result := []any{} + for rows.Next() { + var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString + var projectedSeq, sourceSeq sql.NullInt64 + var terminal, sealed sql.NullBool + var createdAt, updatedAt sql.NullTime + if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil { + return nil, err + } + if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" { + continue + } + item := map[string]any{ + "id": sessionID.String, + "sessionId": sessionID.String, + "ownerUserId": nullableString(ownerUserID), + "projectId": nullableString(projectID), + "conversationId": nullableString(conversationID), + "threadId": nullableString(threadID), + "status": first(nullableStringValue(status), "unknown"), + "lastTraceId": nullableString(lastTraceID), + "projectedSeq": nullableInt(projectedSeq), + "sourceSeq": nullableInt(sourceSeq), + "sourceEventId": nullableString(sourceEventID), + "terminal": nullableBool(terminal), + "sealed": nullableBool(sealed), + "createdAt": nullableTime(createdAt), + "updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)), + "valuesPrinted": false, + "valuesRedacted": true, + } + if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" { + item["providerProfile"] = providerProfile.String + item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false} + } + result = append(result, item) + } + return result, rows.Err() +} + +func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) { + clauses := []string{} + args := []any{} + addTextClause(&clauses, &args, "trace_id", query.TraceID) + addTextClause(&clauses, &args, "agent_session_id", query.SessionID) + addTextClause(&clauses, &args, "worker_session_id", query.WorkerSessionID) + addTextClause(&clauses, &args, "level", query.Level) + sqlText := fmt.Sprintf("SELECT event_json FROM agent_trace_events%s ORDER BY occurred_at ASC, id ASC", whereSQL(clauses)) + if limit := boundedLimit(query.Limit, 0, 1000); limit > 0 { + args = append(args, limit) + sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) + } + return s.queryJSONColumn(ctx, sqlText, args) +} + +func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([]any, error) { + limit := boundedLimit(query.Limit, 100, 500) + args := []any{query.AfterSeq} + clauses := []string{"outbox_seq > $1"} + addTextClause(&clauses, &args, "trace_id", query.TraceID) + addTextClause(&clauses, &args, "session_id", query.SessionID) + args = append(args, limit) + sqlText := fmt.Sprintf("SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE %s ORDER BY outbox_seq ASC LIMIT $%d", strings.Join(clauses, " AND "), len(args)) + rows, err := s.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + result := []any{} + for rows.Next() { + var outboxSeq, projectedSeq, sourceSeq sql.NullInt64 + var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString + var terminal, sealed sql.NullBool + var raw []byte + var createdAt sql.NullTime + if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil { + return nil, err + } + var payload any + if len(raw) > 0 { + _ = json.Unmarshal(raw, &payload) + } + result = append(result, map[string]any{ + "outboxSeq": nullableInt(outboxSeq), + "traceId": nullableString(traceID), + "sessionId": nullableString(sessionID), + "turnId": nullableString(turnID), + "messageId": nullableString(messageID), + "projectedSeq": nullableInt(projectedSeq), + "sourceSeq": nullableInt(sourceSeq), + "sourceEventId": nullableString(sourceEventID), + "commitType": nullableString(commitType), + "terminal": nullableBool(terminal), + "sealed": nullableBool(sealed), + "payload": payload, + "createdAt": nullableTime(createdAt), + "valuesPrinted": false, + }) + } + return result, rows.Err() +} + +func factClauses(query factQuery, fields []fieldColumn) ([]string, []any) { + clauses := []string{} + args := []any{} + for _, field := range fields { + value := singularField(query, field.field) + addTextClause(&clauses, &args, field.column, value) + addTextListClause(&clauses, &args, field.column, field.values) + } + return clauses, args +} + +func addTextClause(clauses *[]string, args *[]any, column string, value string) { + text := strings.TrimSpace(value) + if text == "" { + return + } + *args = append(*args, text) + *clauses = append(*clauses, fmt.Sprintf("%s = $%d", column, len(*args))) +} + +func addTextListClause(clauses *[]string, args *[]any, column string, values []string) { + cleaned := []string{} + for _, value := range values { + if text := strings.TrimSpace(value); text != "" { + cleaned = append(cleaned, text) + } + } + if len(cleaned) == 0 { + return + } + placeholders := []string{} + for _, value := range cleaned { + *args = append(*args, value) + placeholders = append(placeholders, fmt.Sprintf("$%d", len(*args))) + } + *clauses = append(*clauses, fmt.Sprintf("%s IN (%s)", column, strings.Join(placeholders, ", "))) +} + +func singularField(query factQuery, field string) string { + switch field { + case "sessionId": return query.SessionID + case "traceId": return query.TraceID + case "messageId": return query.MessageID + case "turnId": return query.TurnID + case "status": return query.Status + case "ownerUserId": return query.OwnerUserID + case "conversationId": return query.ConversationID + case "eventType": return query.EventType + case "projectionStatus": return query.ProjectionStatus + case "projectionHealth": return query.ProjectionHealth + case "runId": return query.RunID + case "commandId": return query.CommandID + default: return "" + } +} + +func whereSQL(clauses []string) string { + if len(clauses) == 0 { + return "" + } + return " WHERE " + strings.Join(clauses, " AND ") +} + +func emptyFacts() map[string][]any { + return map[string][]any{ + "sessions": []any{}, + "messages": []any{}, + "parts": []any{}, + "turns": []any{}, + "traceEvents": []any{}, + "checkpoints": []any{}, + } +} + +func factFamilySet(query factQuery) map[string]bool { + all := []string{"sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"} + selected := query.Families + if len(selected) == 0 { + selected = query.FactFamilies + } + result := map[string]bool{} + if len(selected) == 0 { + for _, family := range all { result[family] = true } + return result + } + valid := map[string]bool{} + for _, family := range all { valid[family] = true } + for _, family := range selected { + family = strings.TrimSpace(family) + if valid[family] { result[family] = true } + } + if len(result) == 0 { + for _, family := range all { result[family] = true } + } + return result +} + +func factFamilyForTable(table string) string { + switch table { + case "workbench_sessions": return "sessions" + case "workbench_messages": return "messages" + case "workbench_parts": return "parts" + case "workbench_turns": return "turns" + case "workbench_trace_events": return "traceEvents" + case "workbench_projection_checkpoints": return "checkpoints" + default: return "unknown" + } +} + +func factOrder(query factQuery, family string) string { + value := query.Order + switch family { + case "sessions": if query.SessionsOrder != "" { value = query.SessionsOrder } + case "messages": if query.MessagesOrder != "" { value = query.MessagesOrder } + case "parts": if query.PartsOrder != "" { value = query.PartsOrder } + case "turns": if query.TurnsOrder != "" { value = query.TurnsOrder } + case "checkpoints": if query.CheckpointsOrder != "" { value = query.CheckpointsOrder } + } + if value == "updated_desc" { return "updated_desc" } + return "updated_asc" +} + +func sessionSummarySelectClause() string { + return strings.Join([]string{ + "session_id", + "owner_user_id", + "project_id", + "conversation_id", + "thread_id", + "status", + "last_trace_id", + "projected_seq", + "source_seq", + "source_event_id", + "terminal", + "sealed", + "created_at", + "updated_at", + "COALESCE(NULLIF(session_json::jsonb ->> 'providerProfile', ''), NULLIF(session_json::jsonb #>> '{sessionJson,providerProfile}', '')) AS provider_profile", + }, ", ") +} + +func boundedLimit(value int, fallback int, max int) int { + if value <= 0 { + return fallback + } + if max > 0 && value > max { + return max + } + return value +} + +func configFromEnv() Config { + port := first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_PORT"), os.Getenv("PORT"), "6671") + return Config{ + Addr: ":" + port, + DatabaseURL: first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_DB_URL"), os.Getenv("HWLAB_CLOUD_DB_URL")), + QueryTimeout: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS", 10000)) * time.Millisecond, + ReadinessDelay: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS", 2000)) * time.Millisecond, + PoolMax: envInt("HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX", 8), + } +} + +func (s *Server) persistenceSummary() map[string]any { + return map[string]any{"ready": true, "durable": true, "adapter": "postgres", "serviceId": serviceID, "valuesRedacted": true} +} + +func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool { + defer r.Body.Close() + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + if err := decoder.Decode(target); err != nil { + writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must be valid JSON") + return false + } + return true +} + +func writeQueryError(w http.ResponseWriter, err error) { + log.Printf("%s query failed: %v", serviceID, err) + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "ok": false, + "error": map[string]any{ + "code": "workbench_runtime_query_failed", + "message": "workbench runtime query failed", + "retryable": true, + "transient": true, + "causeCode": sqlErrorCode(err), + "valuesRedacted": true, + }, + "valuesRedacted": true, + }) +} + +func writeAPIError(w http.ResponseWriter, status int, code string, message string) { + writeJSON(w, status, map[string]any{"ok": false, "error": map[string]any{"code": code, "message": message, "valuesRedacted": true}, "valuesRedacted": true}) +} + +func methodNotAllowed(w http.ResponseWriter, allow string) { + w.Header().Set("Allow", allow) + writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed") +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} + +func envInt(name string, fallback int) int { + value := strings.TrimSpace(os.Getenv(name)) + if value == "" { return fallback } + parsed, err := strconv.Atoi(value) + if err != nil || parsed <= 0 { return fallback } + return parsed +} + +func first(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { return value } + } + return "" +} + +func nullableString(value sql.NullString) any { + if value.Valid { return value.String } + return nil +} + +func nullableStringValue(value sql.NullString) string { + if value.Valid { return value.String } + return "" +} + +func nullableInt(value sql.NullInt64) any { + if value.Valid { return value.Int64 } + return int64(0) +} + +func nullableBool(value sql.NullBool) bool { + return value.Valid && value.Bool +} + +func nullableTime(value sql.NullTime) string { + if value.Valid { return value.Time.UTC().Format(time.RFC3339Nano) } + return "" +} + +type sqlStateError interface { SQLState() string } + +func sqlErrorCode(err error) string { + if err == nil { return "" } + var stateErr sqlStateError + if errors.As(err, &stateErr) { return stateErr.SQLState() } + return "" +} diff --git a/scripts/artifact-publish.mjs b/scripts/artifact-publish.mjs index 3c4cfdf3..02d67754 100644 --- a/scripts/artifact-publish.mjs +++ b/scripts/artifact-publish.mjs @@ -62,6 +62,7 @@ const ciArtifactIdentityEnvNames = [ const servicePorts = new Map([ ["hwlab-cloud-api", 6667], + ["hwlab-workbench-runtime", 6671], ["hwlab-user-billing", 6670], ["hwlab-cloud-web", 8080], ["hwlab-gateway", 7001],