334 lines
10 KiB
TypeScript
334 lines
10 KiB
TypeScript
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { afterEach, expect, test, vi } from "vitest";
|
|
import { parseWorkspaceYaml } from "../src/workspaces/schema.js";
|
|
import { PreprocessingStateStore } from "../src/workspaces/preprocessing-state.js";
|
|
import {
|
|
WorkspacePreprocessingService,
|
|
type ChildProcessRequest,
|
|
} from "../src/workspaces/preprocessing-service.js";
|
|
|
|
const roots: string[] = [];
|
|
|
|
afterEach(() => {
|
|
roots.splice(0).forEach((root) => rmSync(root, { recursive: true, force: true }));
|
|
});
|
|
|
|
const semanticRuntime = {
|
|
internalQdrantUrl: "http://qdrant:6333",
|
|
internalEmbeddingUrl: "http://embedding:11434",
|
|
internalEmbeddingModel: "qwen3-embedding:0.6b",
|
|
internalEmbeddingDimensions: 1024,
|
|
};
|
|
|
|
const baseWorkspace = parseWorkspaceYaml(`workspace:
|
|
schema_version: 3
|
|
id: psd-clinical
|
|
name: Runtime Lease
|
|
language: en
|
|
dwh:
|
|
engine: postgres
|
|
database: analytics
|
|
schema: mart
|
|
supported_transports: [postgres_direct]
|
|
semantic_index:
|
|
vector_store:
|
|
engine: qdrant
|
|
collection: psd-clinical
|
|
dimensions: 1024
|
|
distance: cosine
|
|
embedding:
|
|
provider: ollama_internal
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
llm_policy:
|
|
allowed: [zai/glm-5.2]
|
|
`);
|
|
const filesystemWorkspace = parseWorkspaceYaml(`${baseWorkspace ? '' : ''}workspace:
|
|
schema_version: 3
|
|
id: fs-workspace
|
|
name: Filesystem
|
|
language: en
|
|
dwh:
|
|
engine: postgres
|
|
database: analytics
|
|
schema: mart
|
|
supported_transports: [postgres_direct]
|
|
semantic_index:
|
|
vector_store:
|
|
engine: qdrant
|
|
collection: fs-workspace
|
|
dimensions: 1024
|
|
distance: cosine
|
|
embedding:
|
|
provider: ollama_internal
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
llm_policy:
|
|
allowed: [zai/glm-5.2]
|
|
evidence:
|
|
source:
|
|
type: filesystem
|
|
uri: fs-workspace/evidence
|
|
`);
|
|
const privateHttpWorkspace = parseWorkspaceYaml(`workspace:
|
|
schema_version: 3
|
|
id: http-workspace
|
|
name: Http
|
|
language: en
|
|
dwh:
|
|
engine: postgres
|
|
database: analytics
|
|
schema: mart
|
|
supported_transports: [postgres_direct]
|
|
semantic_index:
|
|
vector_store:
|
|
engine: qdrant
|
|
collection: http-workspace
|
|
dimensions: 1024
|
|
distance: cosine
|
|
embedding:
|
|
provider: ollama_internal
|
|
model: qwen3-embedding:0.6b
|
|
dimensions: 1024
|
|
llm_policy:
|
|
allowed: [zai/glm-5.2]
|
|
evidence:
|
|
source:
|
|
type: http
|
|
uris: [http://127.0.0.1/private.md]
|
|
authentication: none
|
|
connect_timeout_ms: 1000
|
|
read_timeout_ms: 2000
|
|
max_bytes: 100
|
|
max_redirects: 0
|
|
allow_private_hosts: true
|
|
max_cache_bytes: 100
|
|
`);
|
|
|
|
function runtime(workspace = baseWorkspace, workspaceId = workspace.workspace.id) {
|
|
return {
|
|
workspace,
|
|
workspaceId,
|
|
workspaceRevision: "a".repeat(40),
|
|
descriptorBlob: "b".repeat(40),
|
|
catalogBlob: "c".repeat(40),
|
|
configLease: {
|
|
path: `/data/sessions/${workspaceId}/preprocessing/runtime-config/${"a".repeat(40)}.yaml`,
|
|
workspaceId,
|
|
workspaceRevision: "a".repeat(40),
|
|
descriptorBlob: "b".repeat(40),
|
|
catalogBlob: "c".repeat(40),
|
|
configDigest: "sha256:config",
|
|
bindingDigest: "sha256:bindings",
|
|
release: () => undefined,
|
|
},
|
|
};
|
|
}
|
|
|
|
function fixture(workspace = baseWorkspace) {
|
|
const dataRoot = mkdtempSync(join(tmpdir(), "tht-preprocessing-service-"));
|
|
roots.push(dataRoot);
|
|
const requests: ChildProcessRequest[] = [];
|
|
const runChild = vi.fn(async (request: ChildProcessRequest) => {
|
|
requests.push(request);
|
|
return { exitCode: 0, stdout: JSON.stringify({ status: "succeeded" }), stderr: "" };
|
|
});
|
|
const service = new WorkspacePreprocessingService({
|
|
dataRoot,
|
|
acquireActiveRuntime: async () => runtime(workspace),
|
|
runChild,
|
|
listSessions: async () => [],
|
|
semanticPreflight: async () => ({ ok: true }),
|
|
});
|
|
return { dataRoot, runChild, requests, service };
|
|
}
|
|
|
|
test("preprocess dwh uses fixed argv and resumes outer state without rerunning a completed stage", async () => {
|
|
const f = fixture();
|
|
f.runChild.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({ status: "succeeded", run_id: "d".repeat(32) }),
|
|
stderr: "",
|
|
});
|
|
|
|
const first = await f.service.preprocessDwh({ workspaceId: "psd-clinical" });
|
|
expect(first).toMatchObject({
|
|
status: "succeeded",
|
|
code: "ok",
|
|
operation: "preprocess dwh",
|
|
completedStages: ["dwh"],
|
|
childRuns: { dwh: "d".repeat(32) },
|
|
});
|
|
expect((f.runChild.mock.calls[0]![0] as ChildProcessRequest).argv).toEqual([
|
|
"preprocess", "dwh", "--steps", "introspect,lsh", "--json", "-c", "/dev/fd/3",
|
|
]);
|
|
|
|
const second = await f.service.preprocessDwh({
|
|
workspaceId: "psd-clinical",
|
|
resumeRunId: first.runId!,
|
|
});
|
|
expect(second.status).toBe("unchanged");
|
|
expect(f.runChild).toHaveBeenCalledTimes(1);
|
|
});
|
|
|
|
test("schema suggest-fks publishes a candidate artifact and blocks full runs for manual review", async () => {
|
|
const f = fixture();
|
|
f.runChild
|
|
.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({ status: "succeeded", run_id: "d".repeat(32) }),
|
|
stderr: "",
|
|
})
|
|
.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({
|
|
status: "succeeded",
|
|
candidate_count: 1,
|
|
candidate_digest: "sha256:" + "e".repeat(64),
|
|
candidate_yaml: "tables: []\n",
|
|
}),
|
|
stderr: "",
|
|
});
|
|
|
|
const result = await f.service.run({ workspaceId: "psd-clinical" });
|
|
expect(result).toMatchObject({
|
|
status: "blocked",
|
|
code: "manual_review_required",
|
|
completedStages: ["dwh", "fk_suggest"],
|
|
});
|
|
expect(f.runChild.mock.calls.map(([request]) => (request as ChildProcessRequest).argv[0])).toEqual(["preprocess", "schema"]);
|
|
});
|
|
|
|
test("schema check requires the exact candidate digest, stages annotations via temp file, and persists the review", async () => {
|
|
const f = fixture();
|
|
f.runChild.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({
|
|
status: "succeeded",
|
|
candidate_count: 1,
|
|
candidate_digest: "sha256:" + "e".repeat(64),
|
|
candidate_yaml: "tables: []\n",
|
|
}),
|
|
stderr: "",
|
|
});
|
|
const suggest = await f.service.suggestFks({ workspaceId: "psd-clinical" });
|
|
await expect(f.service.checkSchema({
|
|
workspaceId: "psd-clinical",
|
|
annotationsYaml: "tables: {}\n",
|
|
reviewedCandidatesDigest: "sha256:" + "f".repeat(64),
|
|
})).resolves.toMatchObject({ status: "failed", code: "annotation_invalid" });
|
|
|
|
const reviewedDigest = suggest.artifactIdentities![0]!.digest;
|
|
let stagedPath = "";
|
|
f.runChild.mockImplementationOnce(async (request: ChildProcessRequest) => {
|
|
stagedPath = request.argv[request.argv.indexOf("--annotations") + 1]!;
|
|
expect(readFileSync(stagedPath, "utf8")).toBe("tables: {}\n");
|
|
expect(request.argv).toEqual([
|
|
"schema", "check", "--annotations", stagedPath,
|
|
"--reviewed-candidates", reviewedDigest,
|
|
"--json", "-c", "/dev/fd/3",
|
|
]);
|
|
return {
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({
|
|
status: "succeeded",
|
|
orphan_count: 0,
|
|
annotations_digest: "sha256:annotations",
|
|
reviewed_candidates_digest: reviewedDigest,
|
|
}),
|
|
stderr: "",
|
|
};
|
|
});
|
|
|
|
const checked = await f.service.checkSchema({
|
|
workspaceId: "psd-clinical",
|
|
annotationsYaml: "tables: {}\n",
|
|
reviewedCandidatesDigest: reviewedDigest,
|
|
});
|
|
expect(checked).toMatchObject({ status: "succeeded", code: "ok" });
|
|
expect(() => readFileSync(stagedPath, "utf8")).toThrow();
|
|
const state = new PreprocessingStateStore({ dataRoot: f.dataRoot, workspaceId: "psd-clinical" });
|
|
expect(state.readFkReview(suggest.runId!)?.reviewedCandidatesDigest).toBe(reviewedDigest);
|
|
});
|
|
|
|
test("index schema fails closed when semantic preflight refuses the collection", async () => {
|
|
const dataRoot = mkdtempSync(join(tmpdir(), "tht-preprocessing-service-"));
|
|
roots.push(dataRoot);
|
|
const runChild = vi.fn();
|
|
const service = new WorkspacePreprocessingService({
|
|
dataRoot,
|
|
acquireActiveRuntime: async () => runtime(baseWorkspace),
|
|
runChild,
|
|
listSessions: async () => [],
|
|
semanticPreflight: async () => ({ ok: false, code: "semantic_index_incompatible" }),
|
|
});
|
|
|
|
const result = await service.indexSchema({ workspaceId: "psd-clinical" });
|
|
expect(result).toMatchObject({ status: "failed", code: "semantic_index_incompatible" });
|
|
expect(runChild).not.toHaveBeenCalled();
|
|
});
|
|
|
|
test("evidence stops before child execution for filesystem sources and refuses private HTTP hosts outside the allowlist", async () => {
|
|
const filesystem = fixture(filesystemWorkspace);
|
|
const blocked = await filesystem.service.preprocessEvidence({ workspaceId: "fs-workspace" });
|
|
expect(blocked).toMatchObject({ status: "blocked", code: "evidence_materialization_required" });
|
|
expect(filesystem.runChild).not.toHaveBeenCalled();
|
|
|
|
const httpDataRoot = mkdtempSync(join(tmpdir(), "tht-preprocessing-service-"));
|
|
roots.push(httpDataRoot);
|
|
const httpService = new WorkspacePreprocessingService({
|
|
dataRoot: httpDataRoot,
|
|
acquireActiveRuntime: async () => runtime(privateHttpWorkspace, "http-workspace"),
|
|
runChild: vi.fn(),
|
|
listSessions: async () => [],
|
|
semanticPreflight: async () => ({ ok: true }),
|
|
httpPrivateHostAllowlist: ["metadata.internal"],
|
|
});
|
|
|
|
const refused = await httpService.preprocessEvidence({ workspaceId: "http-workspace" });
|
|
expect(refused).toMatchObject({ status: "failed", code: "egress_policy_refused" });
|
|
});
|
|
|
|
test("full runs follow the explicit order and finish unchanged when no Evidence source exists", async () => {
|
|
const f = fixture();
|
|
f.runChild
|
|
.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({ status: "succeeded", run_id: "d".repeat(32) }),
|
|
stderr: "",
|
|
})
|
|
.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({
|
|
status: "succeeded",
|
|
candidate_count: 0,
|
|
candidate_digest: "sha256:" + "0".repeat(64),
|
|
candidate_yaml: "tables: []\n",
|
|
}),
|
|
stderr: "",
|
|
})
|
|
.mockResolvedValueOnce({
|
|
exitCode: 0,
|
|
stdout: JSON.stringify({
|
|
status: "succeeded",
|
|
counts: { added: 1, updated: 0, deleted: 0, unchanged: 0 },
|
|
}),
|
|
stderr: "",
|
|
});
|
|
|
|
const result = await f.service.run({ workspaceId: "psd-clinical" });
|
|
expect(result).toMatchObject({
|
|
status: "succeeded",
|
|
code: "ok",
|
|
completedStages: ["dwh", "fk_suggest", "schema_index"],
|
|
warnings: ["workspace has no Evidence source"],
|
|
});
|
|
expect(f.runChild.mock.calls.map(([request]) => (request as ChildProcessRequest).argv.slice(0, 2).join(" "))).toEqual([
|
|
"preprocess dwh",
|
|
"schema suggest-fks",
|
|
"vector index-schema",
|
|
]);
|
|
});
|