mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: complete workspace diagnostic adapters

Browse Source
This commit is contained in:
marcopan
2026-08-04 00:05:33 +02:00
parent ca97bbb9c2
commit 67bb4f6ef9
2 changed files with 111 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/workspaces/diagnostics.ts
+49 -14
View File
@@ -1,7 +1,9 @@
import { randomUUID } from "node:crypto";
import { readFile, realpath } from "node:fs/promises";
import { createConnection } from "node:net";
import { createServer } from "node:net";
import { once } from "node:events";
import { spawn } from "node:child_process";
import { Client } from "pg";
import { MAX_WORKSPACE_DIAGNOSTIC_TIMEOUT_MS } from "../config.js";
import { buildInstallationContract } from "./contracts.js";
@@ -127,6 +129,7 @@ export interface WriteDiagnosticRecordRequest {
timeoutMs: number;
signal: AbortSignal;
credentialFile?: string;
tlsCaFile?: string;
baseUrl?: string;
diagnostic?: RestDiagnosticRequest;
}
@@ -157,6 +160,8 @@ export interface ConcreteDiagnosticAdapterDependencies {
directProtocol?: DirectProtocolFactory;
sshProcess?: SshProcessFactory;
databaseClient?: DatabaseDiagnosticClientFactory;
sshSpawn?: (args: readonly string[]) => { kill(signal?: NodeJS.Signals): boolean };
reserveLoopbackPort?: () => Promise<number>;
}
/**
@@ -211,6 +216,21 @@ async function restHeaders(
return diagnostic.auth === "bearer" ? { authorization: `Bearer ${secret}` } : { "x-api-key": secret };
}
async function reserveLoopbackPort(): Promise<number> {
const server = createServer();
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
try {
const address = server.address();
if (!address || typeof address === "string") throw new Error("SSH tunnel port unavailable");
return address.port;
} finally {
await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
}
/**
* Concrete production adapters deliberately retain only probe metadata. Protocol failures and
* response bodies are discarded at this boundary; callers receive fixed diagnostics instead.
@@ -218,6 +238,22 @@ async function restHeaders(
export function createConcreteDiagnosticAdapters(
dependencies: ConcreteDiagnosticAdapterDependencies = {},
): DiagnosticAdapters {
const spawnSsh = dependencies.sshSpawn ?? ((args: readonly string[]) => spawn("ssh", [...args], { stdio: "ignore" }));
const reserveSshPort = dependencies.reserveLoopbackPort ?? reserveLoopbackPort;
const sshProcess = dependencies.sshProcess ?? {
async start(request: SshTunnelRequest, args: readonly string[]) {
const port = await reserveSshPort();
const resolvedArgs = args.map((argument) => argument === `127.0.0.1:0:${request.targetHost}:${request.targetPort}`
? `127.0.0.1:${port}:${request.targetHost}:${request.targetPort}` : argument);
const child = spawnSsh(resolvedArgs);
const abort = () => { child.kill("SIGTERM"); };
request.signal.addEventListener("abort", abort, { once: true });
return {
tunnel: { host: "127.0.0.1" as const, port },
async close() { request.signal.removeEventListener("abort", abort); child.kill("SIGTERM"); },
};
},
};
const databaseClient = dependencies.databaseClient ?? {
async connect(request: { host: string; port: number; database: string; user: string; credentialFile: string; tlsCaFile: string; signal: AbortSignal }) {
const client = new Client({
@@ -295,14 +331,13 @@ export function createConcreteDiagnosticAdapters(
if (!request.knownHostsFile || !(await secretPresent(request.privateKeyFile))) {
throw new Error("SSH probe failed");
}
if (!dependencies.sshProcess) throw new Error("SSH tunnel process is unavailable");
const args = [
"-N", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=yes",
"-o", `UserKnownHostsFile=${request.knownHostsFile}`, "-i", request.privateKeyFile,
"-p", String(request.sshPort), "-L", `127.0.0.1:0:${request.targetHost}:${request.targetPort}`,
`${request.sshUser}@${request.sshHost}`,
];
const tunnel = await dependencies.sshProcess.start(request, args);
const tunnel = await sshProcess.start(request, args);
try {
return await probe(tunnel.tunnel);
} finally {
@@ -322,7 +357,7 @@ export function createConcreteDiagnosticAdapters(
});
try {
const metadata = await client.query(
"SELECT a.atttypmod - 4 AS dimensions, CASE WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_cosine_ops%' THEN 'cosine' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_l2_ops%' THEN 'l2' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_ip_ops%' THEN 'inner_product' END AS distance FROM pg_attribute a JOIN pg_class c ON c.oid = a.attrelid JOIN pg_namespace n ON n.oid = c.relnamespace LEFT JOIN pg_index i ON i.indrelid = c.oid WHERE n.nspname = $1 AND c.relname = $2 AND a.attnum > 0 AND NOT a.attisdropped AND a.atttypid = (SELECT oid FROM pg_type WHERE typname = 'vector') LIMIT 1",
"SELECT a.atttypmod - 4 AS dimensions, CASE WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_cosine_ops%' THEN 'cosine' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_l2_ops%' THEN 'l2' WHEN pg_get_indexdef(i.indexrelid) LIKE '%vector_ip_ops%' THEN 'inner_product' END AS distance FROM pg_attribute a JOIN pg_class c ON c.oid = a.attrelid JOIN pg_namespace n ON n.oid = c.relnamespace JOIN pg_index i ON i.indrelid = c.oid AND a.attnum = ANY(i.indkey) WHERE n.nspname = $1 AND c.relname = $2 AND a.attnum > 0 AND NOT a.attisdropped AND a.atttypid = (SELECT oid FROM pg_type WHERE typname = 'vector') ORDER BY i.indexrelid LIMIT 1",
[resource.schema, request.collection],
);
const row = metadata.rows[0];
@@ -332,11 +367,12 @@ export function createConcreteDiagnosticAdapters(
await client.end().catch(() => undefined);
}
}
if (request.transport !== "rest_api" || !request.baseUrl || !request.credentialFile || !request.diagnostic
|| !(await secretPresent(request.credentialFile))) throw new Error("vector metadata adapter is unavailable");
if (request.transport !== "rest_api" || !request.baseUrl || !request.diagnostic || request.tlsCaFile) {
throw new Error("vector metadata adapter is unavailable");
}
const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), {
method: request.diagnostic.method,
headers: { authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}` },
headers: await restHeaders(request.diagnostic, request.credentialFile),
signal: request.signal,
redirect: "error",
});
@@ -353,10 +389,10 @@ export function createConcreteDiagnosticAdapters(
};
},
async probeEmbedding(request) {
if (!request.diagnostic || !(await secretPresent(request.credentialFile ?? ""))) throw new Error("embedding probe failed");
if (!request.diagnostic || request.tlsCaFile) throw new Error("embedding probe failed");
const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), {
method: request.diagnostic.method,
headers: { authorization: `Bearer ${await readFile(request.credentialFile!, "utf8")}` },
headers: await restHeaders(request.diagnostic, request.credentialFile),
signal: request.signal,
redirect: "error",
});
@@ -366,12 +402,11 @@ export function createConcreteDiagnosticAdapters(
return { available: true, dimensions: payload[request.diagnostic.response.dimensions] as number };
},
async writeDiagnosticRecord(request) {
if (!request.baseUrl || !request.credentialFile || !request.diagnostic
|| !(await secretPresent(request.credentialFile))) throw new Error("vector write adapter is unavailable");
if (!request.baseUrl || !request.diagnostic || request.tlsCaFile) throw new Error("vector write adapter is unavailable");
const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), {
method: request.diagnostic.method,
headers: {
authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}`,
...await restHeaders(request.diagnostic, request.credentialFile),
"content-type": "application/json",
},
body: JSON.stringify({ operation: "create", id: request.id, collection: request.collection, dimensions: request.dimensions }),
@@ -381,12 +416,11 @@ export function createConcreteDiagnosticAdapters(
if (!response.ok) throw new Error("vector write adapter is unavailable");
},
async removeDiagnosticRecord(request) {
if (!request.baseUrl || !request.credentialFile || !request.diagnostic
|| !(await secretPresent(request.credentialFile))) throw new Error("vector write adapter is unavailable");
if (!request.baseUrl || !request.diagnostic || request.tlsCaFile) throw new Error("vector write adapter is unavailable");
const response = await fetch(resolveDiagnosticUrl(request.baseUrl, request.diagnostic.path).toString(), {
method: request.diagnostic.method,
headers: {
authorization: `Bearer ${(await readFile(request.credentialFile, "utf8")).trim()}`,
...await restHeaders(request.diagnostic, request.credentialFile),
"content-type": "application/json",
},
body: JSON.stringify({ operation: "remove", id: request.id, collection: request.collection }),
@@ -728,6 +762,7 @@ export function createWorkspaceDiagnoser(
timeoutMs: vectorTimeout,
signal: new AbortController().signal,
credentialFile,
tlsCaFile: bindings.vector.values[bindingName(canonical, "VECTOR", "TLS_CA_FILE")],
baseUrl: bindings.vector.values[bindingName(canonical, "VECTOR", "BASE_URL")],
diagnostic: canonical.diagnostics.vector_rest.reversible_probe,
};
backend/test/workspaces-diagnostics.test.ts
+62
View File
@@ -326,6 +326,24 @@ test("uses strict known-host SSH arguments and always closes the temporary tunne
}
});
test("provides a default bounded SSH factory through injected spawn and loopback allocation", async () => {
const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-"));
const privateKeyFile = join(directory, "ssh-key");
await writeFile(privateKeyFile, "test-key\n", { mode: 0o600 });
const kill = vi.fn(() => true);
const sshSpawn = vi.fn(() => ({ kill }));
try {
const adapter = createConcreteDiagnosticAdapters({ sshSpawn, reserveLoopbackPort: async () => 45432 } as any);
await adapter.withSshTunnel({ sshHost: "bastion.example.test", sshPort: 22, sshUser: "tunnel", privateKeyFile, knownHostsFile: "/run/secrets/known-hosts", targetHost: "dwh.internal", targetPort: 5432, localHost: "127.0.0.1", localPort: 0, timeoutMs: 5000, signal: new AbortController().signal }, async () => undefined);
expect(sshSpawn).toHaveBeenCalledWith(expect.arrayContaining([
"StrictHostKeyChecking=yes", "UserKnownHostsFile=/run/secrets/known-hosts", "-L", "127.0.0.1:45432:dwh.internal:5432",
]));
expect(kill).toHaveBeenCalledWith("SIGTERM");
} finally {
await rm(directory, { recursive: true, force: true });
}
});
test("requires a matching embedding model vector and removes its unique write probe", async () => {
const adapters = successfulAdapters();
@@ -454,6 +472,25 @@ test("requires an authenticated TLS database query before direct diagnostics suc
}
});
test("selects the vector index containing the declared vector column for direct metadata", async () => {
const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-"));
const passwordFile = join(directory, "password");
const caFile = join(directory, "ca.pem");
await Promise.all([writeFile(passwordFile, "password\n"), writeFile(caFile, "test-ca\n")]);
const query = vi.fn(async () => ({ rows: [{ dimensions: 768, distance: "cosine" }] }));
const connect = vi.fn(async () => ({ query, end: vi.fn(async () => undefined) }));
try {
const result = await createConcreteDiagnosticAdapters({ databaseClient: { connect } } as any).inspectVector({
transport: "pgvector_direct", host: "127.0.0.1", port: 5432, user: "reader", credentialFile: passwordFile, tlsCaFile: caFile,
resource: { database: "postgres", schema: "vectors" }, collection: "clinical_documents", timeoutMs: 5000, signal: new AbortController().signal,
});
expect(query).toHaveBeenCalledWith(expect.stringContaining("a.attnum = ANY(i.indkey)"), ["vectors", "clinical_documents"]);
expect(result).toMatchObject({ dimensions: 768, distance: "cosine" });
} finally {
await rm(directory, { recursive: true, force: true });
}
});
test("honors a declared unauthenticated REST diagnostic without reading a credential", async () => {
const fetchSpy = vi.fn(async () => new Response(JSON.stringify({ database: "warehouse", schema: "datawarehouse" }), {
status: 200, headers: { "content-type": "application/json" },
@@ -472,6 +509,31 @@ test("honors a declared unauthenticated REST diagnostic without reading a creden
}
});
test("applies declared auth modes and rejects private CA files across vector REST paths", async () => {
const directory = await mkdtemp(join(tmpdir(), "thothii-diagnostic-"));
const keyFile = join(directory, "api-key");
const caFile = join(directory, "ca.pem");
await Promise.all([writeFile(keyFile, "writer-key\n", { mode: 0o600 }), writeFile(caFile, "private-ca\n")]);
const fetchSpy = vi.fn(async () => new Response(JSON.stringify({ collection: "clinical_documents", dimensions: 768, distance: "cosine", model: "embed" }), { status: 200, headers: { "content-type": "application/json" } }));
vi.stubGlobal("fetch", fetchSpy);
const adapter = createConcreteDiagnosticAdapters();
const signal = new AbortController().signal;
try {
await adapter.inspectVector({ transport: "rest_api", baseUrl: "https://vector.example.test", collection: "clinical_documents", timeoutMs: 1, signal, diagnostic: { method: "GET", path: "/metadata", auth: "none", response: { collection: "collection", dimensions: "dimensions", distance: "distance" } } });
await adapter.probeEmbedding({ baseUrl: "https://embed.example.test", model: "embed", timeoutMs: 1, signal, credentialFile: keyFile, diagnostic: { method: "POST", path: "/embed", auth: "x-api-key", response: { model: "model", dimensions: "dimensions" } } });
await adapter.writeDiagnosticRecord({ baseUrl: "https://vector.example.test", credentialFile: keyFile, collection: "clinical_documents", dimensions: 768, id: "diagnostic:test", timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/probe", auth: "none" } });
expect(fetchSpy.mock.calls[0]?.[1]).toMatchObject({ headers: {} });
expect(fetchSpy.mock.calls[1]?.[1]).toMatchObject({ headers: { "x-api-key": "writer-key" } });
expect(fetchSpy.mock.calls[2]?.[1]).toMatchObject({ headers: expect.not.objectContaining({ authorization: expect.anything() }) });
await expect(adapter.inspectVector({ transport: "rest_api", baseUrl: "https://vector.example.test", credentialFile: keyFile, tlsCaFile: caFile, collection: "clinical_documents", timeoutMs: 1, signal, diagnostic: { method: "GET", path: "/metadata", auth: "bearer", response: { collection: "collection", dimensions: "dimensions", distance: "distance" } } })).rejects.toThrow("vector metadata adapter is unavailable");
await expect(adapter.probeEmbedding({ baseUrl: "https://embed.example.test", credentialFile: keyFile, tlsCaFile: caFile, model: "embed", timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/embed", auth: "bearer", response: { model: "model", dimensions: "dimensions" } } })).rejects.toThrow("embedding probe failed");
await expect(adapter.removeDiagnosticRecord({ baseUrl: "https://vector.example.test", credentialFile: keyFile, tlsCaFile: caFile, collection: "clinical_documents", id: "diagnostic:test", dimensions: 768, timeoutMs: 1, signal, diagnostic: { method: "POST", path: "/probe", auth: "bearer" } })).rejects.toThrow("vector write adapter is unavailable");
} finally {
vi.unstubAllGlobals();
await rm(directory, { recursive: true, force: true });
}
});
test("constructs the production diagnoser with the configured timeout and injected adapters", async () => {
const adapters = successfulAdapters();