import { spawn } from "node:child_process"; import { closeSync, constants as fsConstants, openSync } from "node:fs"; import { readdir, readFile } from "node:fs/promises"; import { join } from "node:path"; import { parse } from "yaml"; import { loadConfig } from "./config.js"; import { ThtRunner } from "./tht/tht-runner.js"; import { WorkspaceRegistry } from "./workspaces/registry.js"; import { publishDeterministicRuntimeConfigLease, renderActiveWorkspaceRuntime } from "./workspaces/runtime-config-lease.js"; import { WorkspaceSecretStore } from "./workspaces/secret-store.js"; import { WorkspacePreprocessingService, type WorkspaceOperationResult } from "./workspaces/preprocessing-service.js"; import type { SessionInventoryRow } from "./workspaces/preprocessing-state.js"; export interface WorkspaceMaintenanceIo { stdin: string; stdout: string[]; stderr: string[]; writeStdout(value: string): void; writeStderr(value: string): void; } type Command = "inspect" | "preprocess-dwh" | "schema-suggest-fks" | "schema-check" | "schema-accept" | "index-schema" | "preprocess-evidence" | "preprocess-run" | "vector-inspect" | "vector-rebuild"; function failureResult( operation: string, workspaceId = "", code: WorkspaceOperationResult["code"] = "workspace_not_activatable", ): WorkspaceOperationResult { return { schemaVersion: 1, status: "failed", code, workspaceId, workspaceRevision: "", descriptorBlob: "", operation, completedStages: [], }; } const STATE_ERROR_CODES: Record = { preprocessing_resume_mismatch: "preprocessing_resume_mismatch", preprocessing_conflict: "preprocessing_conflict", effective_config_mismatch: "effective_config_mismatch", }; function boundedJson(result: WorkspaceOperationResult): string { const encoded = JSON.stringify(result); if (Buffer.byteLength(encoded, "utf8") > 1024 * 1024) { return JSON.stringify(failureResult(result.operation || "unknown", result.workspaceId)); } return encoded; } function sanitizeStderr(_error: unknown): string { // Never return raw exception text: it may embed endpoints, tokens, or SQL. return "workspace maintenance failed\n"; } function parseRequest(command: string, stdin: string): Record { const parsed = JSON.parse(stdin) as Record; if (!parsed || typeof parsed !== "object" || Array.isArray(parsed) || parsed.schemaVersion !== 1) { throw new Error("invalid request"); } const allowedByCommand: Record = { inspect: ["schemaVersion", "workspaceId"], "preprocess-dwh": ["schemaVersion", "workspaceId", "resumeRunId"], "schema-suggest-fks": ["schemaVersion", "workspaceId", "fromSql", "assume", "resumeRunId"], "schema-check": ["schemaVersion", "workspaceId", "annotationsYaml", "reviewedCandidatesDigest"], "schema-accept": ["schemaVersion", "workspaceId", "runId", "yes"], "index-schema": ["schemaVersion", "workspaceId", "resumeRunId"], "preprocess-evidence": ["schemaVersion", "workspaceId", "dryRun", "resumeRunId"], "preprocess-run": ["schemaVersion", "workspaceId", "resumeRunId"], "vector-inspect": ["schemaVersion", "workspaceId"], "vector-rebuild": ["schemaVersion", "workspaceId", "collection", "confirm", "destroy"], }; const allowed = allowedByCommand[command]; if (!allowed) throw new Error("unknown command"); if (typeof parsed.workspaceId !== "string") throw new Error("invalid workspace id"); for (const key of Object.keys(parsed)) if (!allowed.includes(key)) throw new Error("unexpected request field"); return parsed; } function exitCodeFor(result: WorkspaceOperationResult): number { if (["succeeded", "unchanged", "dry_run"].includes(result.status)) return 0; if (result.status === "blocked") return 3; return 1; } async function dispatch(command: Command, service: WorkspacePreprocessingService, request: Record): Promise { switch (command) { case "inspect": return await service.inspect({ workspaceId: request.workspaceId as string }); case "preprocess-dwh": return await service.preprocessDwh({ workspaceId: request.workspaceId as string, resumeRunId: request.resumeRunId as string | undefined, }); case "schema-suggest-fks": return await service.suggestFks({ workspaceId: request.workspaceId as string, fromSql: request.fromSql as any, assume: request.assume as any, resumeRunId: request.resumeRunId as string | undefined, }); case "schema-check": return await service.checkSchema({ workspaceId: request.workspaceId as string, annotationsYaml: request.annotationsYaml as string | undefined, reviewedCandidatesDigest: request.reviewedCandidatesDigest as string | undefined, }); case "schema-accept": return await service.acceptSchema({ workspaceId: request.workspaceId as string, runId: request.runId as string, yes: request.yes === true, }); case "index-schema": return await service.indexSchema({ workspaceId: request.workspaceId as string, resumeRunId: request.resumeRunId as string | undefined, }); case "preprocess-evidence": return await service.preprocessEvidence({ workspaceId: request.workspaceId as string, dryRun: request.dryRun as boolean | undefined, resumeRunId: request.resumeRunId as string | undefined, }); case "preprocess-run": return await service.run({ workspaceId: request.workspaceId as string, resumeRunId: request.resumeRunId as string | undefined, }); case "vector-inspect": return await service.vectorInspect({ workspaceId: request.workspaceId as string }); case "vector-rebuild": return await service.vectorRebuild({ workspaceId: request.workspaceId as string, collection: request.collection as string | undefined, confirm: request.confirm as string | undefined, destroy: request.destroy === true, }); } } export async function runWorkspaceMaintenanceCli( argv: readonly string[], service: WorkspacePreprocessingService, io: WorkspaceMaintenanceIo, ): Promise { const command = argv[2]; if (!command) { const result = failureResult("unknown"); io.writeStdout(boundedJson(result)); return 2; } try { const request = parseRequest(command, io.stdin); const result = await dispatch(command as Command, service, request); io.writeStdout(boundedJson(result)); return exitCodeFor(result); } catch (error) { const failureCode = error instanceof Error && "code" in error && typeof (error as { code?: unknown }).code === "string" && (error as { code: string }).code in STATE_ERROR_CODES ? STATE_ERROR_CODES[(error as { code: string }).code] : "workspace_not_activatable"; const result = failureResult(command, (() => { try { return JSON.parse(io.stdin).workspaceId ?? ""; } catch { return ""; } })(), failureCode); io.writeStdout(boundedJson(result)); io.writeStderr(sanitizeStderr(error)); const message = String((error as Error).message ?? ""); const requestError = error instanceof SyntaxError || message === "invalid request" || message === "unknown command" || message === "unexpected request field" || message === "invalid workspace id"; return command in { inspect: true, "preprocess-dwh": true, "schema-suggest-fks": true, "schema-check": true, "schema-accept": true, "index-schema": true, "preprocess-evidence": true, "preprocess-run": true, } ? (requestError ? 2 : 1) : 2; } } async function readSessionInventory(dataRoot: string, workspaceId: string): Promise { const directory = join(dataRoot, "sessions", workspaceId, "sessions"); try { const entries = await readdir(directory, { withFileTypes: true }); const rows: SessionInventoryRow[] = []; for (const entry of entries) { if (!entry.isDirectory() || entry.isSymbolicLink()) continue; try { const source = await readFile(join(directory, entry.name, "session_manifest.yaml"), "utf8"); const manifest = parse(source) as Record; rows.push({ id: entry.name, status: typeof manifest.status === "string" ? manifest.status : "open", archived: manifest.archived === true, workspaceRevision: typeof manifest.workspace_revision === "string" ? manifest.workspace_revision : null, }); } catch { // fail closed at mutation time by ignoring unreadable manifests from the resumable scan } } return rows; } catch { return []; } } function createProductionService(): WorkspacePreprocessingService { const config = loadConfig(process.env, { surface: "workspace-maintenance" }); const registry = new WorkspaceRegistry(config.workspaceRegistry); const workspaceSecretStore = new WorkspaceSecretStore({ root: config.workspaceSecretStoreRoot, runtimeRoot: config.workspaceSecretRuntimeRoot, installationId: config.workspaceRegistry.installationId, }); const runner = new ThtRunner({ thtBin: config.thtBin, harnessDir: config.harnessDir, configPath: process.env.THT_CONFIG ?? "config/tht.yaml", dataRoot: config.dataRoot, runtimeSnapshotRoot: join(config.workspaceRegistry.root, "snapshots", "runtime"), secretRoots: config.workspaceRegistry.secretRoots, secretsFile: config.secretsFile, secretFiles: config.secretFiles, workspaceSecretStore, semanticRuntime: { internalQdrantUrl: config.internalQdrantUrl, internalEmbeddingUrl: config.internalEmbeddingUrl, internalEmbeddingModel: config.internalEmbeddingModel, internalEmbeddingDimensions: config.internalEmbeddingDimensions, }, }); return new WorkspacePreprocessingService({ dataRoot: config.dataRoot ?? "/data", httpPrivateHostAllowlist: (process.env.THT_EVIDENCE_PRIVATE_HOST_ALLOWLIST ?? "") .split(",").map((value) => value.trim()).filter((value) => value.length > 0), acquireActiveRuntime: async (workspaceId) => { const active = await renderActiveWorkspaceRuntime({ workspaceId, registry, registryConfig: config.workspaceRegistry, harnessDir: config.harnessDir, configPath: process.env.THT_CONFIG ?? "config/tht.yaml", dataRoot: config.dataRoot ?? "/data", secretRoots: config.workspaceRegistry.secretRoots, workspaceSecretStore, semanticRuntime: { internalQdrantUrl: config.internalQdrantUrl, internalEmbeddingUrl: config.internalEmbeddingUrl, internalEmbeddingModel: config.internalEmbeddingModel, internalEmbeddingDimensions: config.internalEmbeddingDimensions, }, }); const configLease = await publishDeterministicRuntimeConfigLease({ workspaceId, registry, registryConfig: config.workspaceRegistry, harnessDir: config.harnessDir, configPath: process.env.THT_CONFIG ?? "config/tht.yaml", dataRoot: config.dataRoot ?? "/data", secretRoots: config.workspaceRegistry.secretRoots, semanticRuntime: { internalQdrantUrl: config.internalQdrantUrl, internalEmbeddingUrl: config.internalEmbeddingUrl, internalEmbeddingModel: config.internalEmbeddingModel, internalEmbeddingDimensions: config.internalEmbeddingDimensions, }, workspaceSecretStore, }); return { workspace: active.workspace, workspaceId: active.workspaceId, workspaceRevision: active.workspaceRevision, descriptorBlob: active.descriptorBlob, catalogBlob: active.catalogBlob, configLease, }; }, runChild: async ({ argv, configPath }) => { const configFd = openSync(configPath, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW); try { return await new Promise((resolve) => { const child = spawn(config.thtBin, argv, { cwd: config.harnessDir, env: { ...process.env, ...(config.dataRoot ? { THT_DATA_ROOT: config.dataRoot } : {}) }, stdio: ["ignore", "pipe", "pipe", configFd], }); let stdout = ""; let stderr = ""; child.stdout?.on("data", (chunk: Buffer) => { stdout += chunk.toString("utf8"); }); child.stderr?.on("data", (chunk: Buffer) => { stderr += chunk.toString("utf8"); }); child.on("close", (code) => resolve({ exitCode: code ?? 0, stdout, stderr: stderr.slice(0, 64 * 1024) })); child.on("error", (error) => resolve({ exitCode: 1, stdout, stderr: String(error.message).slice(0, 4096) })); }); } finally { closeSync(configFd); } }, listSessions: async (workspaceId) => await readSessionInventory(config.dataRoot ?? "/data", workspaceId), semanticPreflight: async (workspace) => { const result = await runner.qdrantEnsure(workspace, 30); return result.ok ? { ok: true as const } : { ok: false as const, code: result.code ?? "workspace_not_activatable" }; }, evidencePreflight: async (workspace) => { const result = await runner.qdrantEnsure(workspace, 30, "evidence_maintenance"); return result.ok ? { ok: true as const } : { ok: false as const, code: result.code ?? "workspace_not_activatable" }; }, }); } if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) { const stdout: string[] = []; const stderr: string[] = []; const io: WorkspaceMaintenanceIo = { stdin: await new Promise((resolve) => { let input = ""; process.stdin.setEncoding("utf8"); process.stdin.on("data", (chunk) => { input += chunk; }); process.stdin.on("end", () => resolve(input)); }), stdout, stderr, writeStdout: (value) => { stdout.push(value); }, writeStderr: (value) => { stderr.push(value); }, }; const exitCode = await runWorkspaceMaintenanceCli(process.argv, createProductionService(), io); process.stdout.write(stdout.join("")); if (stderr.length > 0) process.stderr.write(stderr.join("").slice(0, 64 * 1024)); process.exit(exitCode); }