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

fix: close internal semantic review gaps

Browse Source
This commit is contained in:
marcopan
2026-08-08 23:27:58 +02:00
parent 43d8063922
commit 2c4534d968
18 changed files with 806 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/routes/sessions.ts
+34 -8
View File
@@ -8,7 +8,7 @@ import type { PrincipalContext } from "../auth/principal.js";
import type { ReadinessManager } from "../runtime/readiness-manager.js";
import type { ListModelsFn } from "./meta.js";
import type { WorkspaceRegistry } from "../workspaces/registry.js";
import type { WorkspaceDescriptor } from "../workspaces/schema.js";
import { validateOperationalWorkspace, type WorkspaceDescriptor } from "../workspaces/schema.js";
import type { MaintenanceBarrier } from "../runtime/maintenance-gate.js";
const BOOTSTRAP_FAILURE_MESSAGE =
@@ -108,7 +108,11 @@ export function sessionRoutes(
const isNotFound = (error: unknown) =>
/not found|non trovata|inesistente|404/i.test(error instanceof Error ? error.message : String(error));
type LocatedSession = { manifest: any; workspaceConfigPath: string };
type LocatedSession = {
manifest: any;
workspaceConfigPath: string;
workspace?: WorkspaceDescriptor;
};
const workspaceRevisionUnavailable = () => Object.assign(
new Error("workspace revision unavailable"), { code: "workspace_revision_unavailable" },
@@ -168,7 +172,12 @@ export function sessionRoutes(
if (!saved.workspace_id || !saved.workspace_revision) return located;
try {
const pinned = await d.workspaceRegistry.readPinned(saved.workspace_id, saved.workspace_revision);
return { ...located, workspaceConfigPath: pinned.workspaceConfigPath ?? (pinned as any).revision?.snapshotPath };
const workspace = validateOperationalWorkspace(pinned.workspace);
return {
...located,
workspace,
workspaceConfigPath: pinned.workspaceConfigPath ?? (pinned as any).revision?.snapshotPath,
};
} catch {
throw workspaceRevisionUnavailable();
}
@@ -319,6 +328,7 @@ export function sessionRoutes(
let workspaceConfigPath: string | undefined;
let workspaceId: string | undefined;
let workspaceRevision: string | undefined;
let workspaceDescriptor: WorkspaceDescriptor | undefined;
let allowedModels: readonly string[] | undefined;
if (requestedWorkspaceId) {
try {
@@ -344,6 +354,7 @@ export function sessionRoutes(
workspaceConfigPath = resolved.revision.snapshotPath;
workspaceId = resolved.revision.id;
workspaceRevision = resolved.revision.commit;
workspaceDescriptor = resolved.workspace;
allowedModels = resolved.workspace.llm_policy.allowed;
} catch {
return reply.code(409).send({
@@ -362,8 +373,13 @@ export function sessionRoutes(
// runtime owned by this principal, while runtimes belonging to other users remain intact.
// Optional chaining preserves the deliberately narrow manager stubs used by route tests.
for (const id of d.mgr.teardownForPrincipal?.(principal) ?? []) boundRuntimes.delete(id);
const ensure = await d.readiness.ensure(workspaceConfigPath ?? "", principal);
if (!ensure.ok) return reply.code(503).send({ error: READINESS_FAILURE_MESSAGE });
const ensure = await d.readiness.ensure(
workspaceConfigPath ?? "", principal, workspaceDescriptor,
);
if (!ensure.ok) return reply.code(503).send({
error: READINESS_FAILURE_MESSAGE,
...(ensure.code ? { code: ensure.code } : {}),
});
// Local-only: verify the DWH is reachable BEFORE creating the session, so a dropped
// VPN surfaces as an up-front alert instead of a session that spawns Pi and then dies
// in bootstrap retrieval. `code` lets the client show a specific message.
@@ -546,7 +562,12 @@ export function sessionRoutes(
workspace_id?: string; workspace_revision?: string;
};
let workspaceConfigPath: string;
try { workspaceConfigPath = (await resolveSessionWorkspace(located)).workspaceConfigPath; }
let workspaceDescriptor: WorkspaceDescriptor | undefined;
try {
const resolved = await resolveSessionWorkspace(located);
workspaceConfigPath = resolved.workspaceConfigPath;
workspaceDescriptor = resolved.workspace;
}
catch { return unavailableWorkspaceReply(reply); }
try { settings = await d.getSettings(principal); } catch { return storageFailure(reply); }
// This check belongs inside the per-session lock: a preceding cold Resume may have
@@ -558,8 +579,13 @@ export function sessionRoutes(
return reply.code(200).send({ id, alreadyActive: true });
}
}
const ensure = await d.readiness.ensure(workspaceConfigPath ?? "", principal);
if (!ensure.ok) return reply.code(503).send({ error: READINESS_FAILURE_MESSAGE });
const ensure = await d.readiness.ensure(
workspaceConfigPath ?? "", principal, workspaceDescriptor,
);
if (!ensure.ok) return reply.code(503).send({
error: READINESS_FAILURE_MESSAGE,
...(ensure.code ? { code: ensure.code } : {}),
});
const options = {
provider: saved?.provider,
model: saved?.model,
backend/src/routes/workspaces.ts
+17 -3
View File
@@ -19,6 +19,7 @@ import {
parseWorkspaceYaml,
serializeWorkspaceYaml,
validateCanonicalWorkspace,
validateOperationalWorkspace,
type CanonicalWorkspace,
type WorkspaceDescriptor,
} from "../workspaces/schema.js";
@@ -327,9 +328,22 @@ export function workspaceRoutes(app: FastifyInstance, deps: WorkspaceRoutesDeps)
app.post("/workspaces/:id/test", async (request, reply) => {
try {
const { id } = z.object({ id: workspaceId }).parse(request.params);
const { workspace } = await deps.registry.read(id);
const bindings = resolveRuntimeBindings(workspace, process.env, deps.config.secretRoots);
return await deps.diagnose(workspace, bindings, { writeProbe: false });
const { workspace, revision } = await deps.registry.read(id);
if (revision.state !== "operational") {
throw new WorkspaceRegistryError(
"workspace_not_activatable", "Workspace requires explicit migration",
);
}
let operational: CanonicalWorkspace;
try {
operational = validateOperationalWorkspace(workspace);
} catch {
throw new WorkspaceRegistryError(
"workspace_not_activatable", "Workspace requires explicit migration",
);
}
const bindings = resolveRuntimeBindings(operational, process.env, deps.config.secretRoots);
return await deps.diagnose(operational, bindings, { writeProbe: false });
} catch (error) {
return errorReply(reply, error);
}
backend/src/runtime/readiness-manager.ts
+29 -5
View File
@@ -1,9 +1,16 @@
import type { OllamaEnsureResult, ThtRunner } from "../tht/tht-runner.js";
import type {
OllamaEnsureResult,
SemanticReadinessCode,
ThtRunner,
} from "../tht/tht-runner.js";
import type { PrincipalContext } from "../auth/principal.js";
import type { WorkspaceDescriptor } from "../workspaces/schema.js";
export type ReadinessResult = OllamaEnsureResult & { code?: SemanticReadinessCode };
interface ReadyEntry {
expiresAt: number;
result: OllamaEnsureResult;
result: ReadinessResult;
}
/**
@@ -11,7 +18,7 @@ interface ReadyEntry {
* Failures are deliberately not cached so a submit can retry after a transient outage.
*/
export class ReadinessManager {
private inFlight = new Map<string, Promise<OllamaEnsureResult>>();
private inFlight = new Map<string, Promise<ReadinessResult>>();
private ready = new Map<string, ReadyEntry>();
constructor(
@@ -21,7 +28,11 @@ export class ReadinessManager {
private now: () => number = Date.now,
) {}
ensure(workspace = "", principal?: PrincipalContext): Promise<OllamaEnsureResult> {
ensure(
workspace = "",
principal?: PrincipalContext,
descriptor?: WorkspaceDescriptor,
): Promise<ReadinessResult> {
const key = `${principal?.issuer ?? ""}\0${principal?.subject ?? ""}\0${workspace}`;
const cached = this.ready.get(key);
if (cached && cached.expiresAt > this.now()) return Promise.resolve(cached.result);
@@ -32,7 +43,20 @@ export class ReadinessManager {
const runner = principal && typeof (this.tht as any).withPrincipal === "function"
? this.tht.withPrincipal(principal) : this.tht;
const pending = runner.ollamaEnsure(workspace, this.timeoutSec)
const pending = (async (): Promise<ReadinessResult> => {
try {
if (descriptor) {
const qdrant = await runner.qdrantEnsure(descriptor, this.timeoutSec);
if (!qdrant.ok) return qdrant;
}
const ollama = await runner.ollamaEnsure(workspace, this.timeoutSec);
return ollama.ok
? ollama
: { ...ollama, code: "workspace_not_activatable" };
} catch {
return { ok: false, code: "workspace_not_activatable" };
}
})()
.then((result) => {
if (result.ok) {
this.ready.set(key, { result, expiresAt: this.now() + this.ttlMs });
backend/src/tht/tht-runner.ts
+73 -2
View File
@@ -15,7 +15,11 @@ import {
type RuntimePaths,
type SemanticRuntimeConfig,
} from "../workspaces/runtime-renderer.js";
import { parseWorkspaceYaml } from "../workspaces/schema.js";
import {
parseWorkspaceYaml,
validateOperationalWorkspace,
type WorkspaceDescriptor,
} from "../workspaces/schema.js";
export interface ThtConfig extends SecretBundleConfig {
thtBin: string;
@@ -25,6 +29,7 @@ export interface ThtConfig extends SecretBundleConfig {
runtimeSnapshotRoot?: string;
secretRoots?: readonly string[];
semanticRuntime: SemanticRuntimeConfig;
qdrantRequest?: typeof fetch;
}
export interface RuntimeConfigLease {
@@ -64,6 +69,24 @@ export interface OllamaEnsureResult {
model_name?: string;
}
export type SemanticReadinessCode = "workspace_not_activatable" | "semantic_index_incompatible";
export interface QdrantEnsureResult {
ok: boolean;
code?: SemanticReadinessCode;
}
const REQUIRED_QDRANT_PAYLOAD_INDEXES = [
"content_hash",
"document_id",
"kind",
"record_key",
"record_kind",
"vector_generation",
"workspace_id",
"workspace_revision",
] as const;
interface RuntimeSnapshot {
path: string;
dev: number;
@@ -136,7 +159,7 @@ export class ThtRunner {
if (before.dev !== after.dev || before.ino !== after.ino || before.size !== after.size) {
throw new Error("workspace snapshot changed while reading");
}
const workspace = parseWorkspaceYaml(source);
const workspace = validateOperationalWorkspace(parseWorkspaceYaml(source));
if (workspace.workspace.id !== identity.workspaceId) {
throw new Error("workspace snapshot identity does not match its path");
}
@@ -537,4 +560,52 @@ export class ThtRunner {
error: parsed?.error ?? (stderr.trim() || `tht ollama ensure exit ${code}`),
};
}
async qdrantEnsure(
workspace: WorkspaceDescriptor,
timeoutSec: number,
): Promise<QdrantEnsureResult> {
let descriptor;
try {
descriptor = validateOperationalWorkspace(workspace);
} catch {
return { ok: false, code: "workspace_not_activatable" };
}
const collection = descriptor.semantic_index.vector_store;
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), Math.max(1, timeoutSec) * 1000);
try {
const url = new URL(
`/collections/${encodeURIComponent(collection.collection)}`,
this.cfg.semanticRuntime.internalQdrantUrl,
);
const request = this.cfg.qdrantRequest ?? fetch;
const response = await request(url.toString(), { method: "GET", signal: controller.signal });
if (response.status === 404) {
return { ok: false, code: "semantic_index_incompatible" };
}
if (!response.ok) return { ok: false, code: "workspace_not_activatable" };
const body = await response.json() as any;
const result = body?.result;
const vectors = result?.config?.params?.vectors;
const payloadSchema = result?.payload_schema;
const configurationMatches = vectors
&& vectors.size === collection.dimensions
&& typeof vectors.distance === "string"
&& vectors.distance.toLowerCase() === collection.distance;
const indexesMatch = payloadSchema
&& typeof payloadSchema === "object"
&& REQUIRED_QDRANT_PAYLOAD_INDEXES.every(
(field) => payloadSchema[field]?.data_type === "keyword",
);
return configurationMatches && indexesMatch
? { ok: true }
: { ok: false, code: "semantic_index_incompatible" };
} catch {
return { ok: false, code: "workspace_not_activatable" };
} finally {
clearTimeout(timer);
controller.abort();
}
}
}
backend/src/workspaces/registry.ts
+8 -2
View File
@@ -13,6 +13,7 @@ import {
isCanonicalWorkspace,
parseWorkspaceYaml,
serializeWorkspaceYaml,
validateOperationalWorkspace,
type CanonicalWorkspace,
type WorkspaceDescriptor,
} from "./schema.js";
@@ -216,7 +217,9 @@ export class WorkspaceRegistry {
}
let workspace: WorkspaceDescriptor;
try {
workspace = parseWorkspaceYaml(await readFile(revision.snapshotPath, "utf8"));
workspace = validateOperationalWorkspace(
parseWorkspaceYaml(await readFile(revision.snapshotPath, "utf8")),
);
} catch (error) {
throw workspaceError(error);
}
@@ -259,7 +262,10 @@ export class WorkspaceRegistry {
const snapshotPath = this.snapshotPath(safeCommit(commit), id);
try {
const source = await readFile(snapshotPath, "utf8");
return { workspace: parseWorkspaceYaml(source), workspaceConfigPath: snapshotPath };
return {
workspace: validateOperationalWorkspace(parseWorkspaceYaml(source)),
workspaceConfigPath: snapshotPath,
};
} catch (error) {
throw workspaceError(error);
}