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

fix: render registry workspaces for harness

Browse Source
This commit is contained in:
marcopan
2026-08-05 16:03:45 +02:00
parent ece9cfda50
commit bd798b1c96
21 changed files with 659 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/app.ts
+2 -1
View File
@@ -55,6 +55,7 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
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,
});
@@ -141,7 +142,7 @@ export function buildApp(config: AppConfig, deps?: BuildAppDeps): FastifyInstanc
app.get("/internal/maintenance/status", async (req, reply) => {
return maintenanceBarrier.status();
});
sqlRoutes(app, { tht: tht as ThtRunner, getSettings });
sqlRoutes(app, { tht: tht as ThtRunner, getSettings, workspaceRegistry });
metaRoutes(app, { harnessDir: config.harnessDir, listModels });
workspaceRoutes(app, { registry: workspaceRegistry, config: config.workspaceRegistry, diagnose: workspaceDiagnoser });
settingsRoutes(app, { cfg: config, listModels, getSettings });
backend/src/pi/pi-process-manager.ts
+31 -7
View File
@@ -2,7 +2,7 @@ import { spawn as nodeSpawn, type ChildProcessWithoutNullStreams } from "node:ch
import type { AppConfig } from "../config.js";
import { RpcClient } from "../rpc/rpc-client.js";
import { SessionBridge } from "../bridge/session-bridge.js";
import type { ThtRunner } from "../tht/tht-runner.js";
import type { RuntimeConfigLease, ThtRunner } from "../tht/tht-runner.js";
import { buildPiChildEnv, canonicalPiProvider } from "./provider-credentials.js";
import { loadPiAuthProviders } from "./auth-providers.js";
import { secretValue } from "../config/secret-bundle.js";
@@ -14,6 +14,7 @@ export interface SessionRuntime {
bridge: SessionBridge;
child: ChildProcessWithoutNullStreams;
ownerKey?: string;
releaseRuntimeConfig?: () => void;
}
export interface RuntimeOptions {
@@ -24,6 +25,7 @@ export interface RuntimeOptions {
question?: string;
mode?: "new" | "resume";
principal?: PrincipalContext;
runtimeConfig?: RuntimeConfigLease;
}
/** Injectable child-process boundary; callbacks may ignore arguments in simpler tests. */
@@ -37,6 +39,7 @@ export class PiProcessManager {
private runtimes = new Map<string, SessionRuntime>();
private spawnFn: (
sessionId: string, author: string, provider: string | undefined, principal?: PrincipalContext,
runtimeConfigPath?: string,
) => ChildProcessWithoutNullStreams;
private loadAuthProviders: (agentDir: string) => ReadonlySet<string>;
@@ -47,16 +50,17 @@ export class PiProcessManager {
this.loadAuthProviders = opts?.authProviders
?? ((agentDir) => loadPiAuthProviders({ agentDir }));
if (opts?.spawnFn) {
this.spawnFn = (sessionId, author, provider, principal) =>
this.spawnPi(opts.spawnFn!, sessionId, author, provider, principal);
this.spawnFn = (sessionId, author, provider, principal, runtimeConfigPath) =>
this.spawnPi(opts.spawnFn!, sessionId, author, provider, principal, runtimeConfigPath);
} else {
this.spawnFn = (sessionId, author, provider, principal) =>
this.spawnPi(nodeSpawn, sessionId, author, provider, principal);
this.spawnFn = (sessionId, author, provider, principal, runtimeConfigPath) =>
this.spawnPi(nodeSpawn, sessionId, author, provider, principal, runtimeConfigPath);
}
}
private spawnPi(
spawnFn: SpawnFn, sessionId: string, author: string, provider: string | undefined, principal?: PrincipalContext,
spawnFn: SpawnFn, sessionId: string, author: string, provider: string | undefined,
principal?: PrincipalContext, runtimeConfigPath?: string,
): ChildProcessWithoutNullStreams {
// This is the final shared boundary for createFor(), spawnFor(), and resume(). Validate
// before auth-provider inspection, then make Pi consume the exact copied bytes rather than
@@ -94,6 +98,7 @@ export class PiProcessManager {
}
delete env.THT_DATA_ROOT;
if (this.cfg.dataRoot !== undefined) env.THT_DATA_ROOT = this.cfg.dataRoot;
if (runtimeConfigPath !== undefined) env.THT_CONFIG = runtimeConfigPath;
// pi 0.73 removed `--approve`: rpc mode is headless and its argv is intentionally minimal.
child = spawnFn(this.cfg.piBin, ["--mode", "rpc"], {
cwd: this.cfg.harnessDir,
@@ -132,15 +137,31 @@ export class PiProcessManager {
// SIGTERM to the in-flight Pi process and lose its pending gate.
const existing = this.runtimes.get(sessionId);
if (existing) {
o.runtimeConfig?.release();
throw new Error(`session runtime already active: ${sessionId}`);
}
if (o.principal) this.teardownForPrincipal(o.principal);
if (this.runtimes.size >= this.cfg.maxPiProcesses) {
o.runtimeConfig?.release();
throw new Error("max Pi processes reached");
}
const author = o.author ?? "dev@local";
const provider = canonicalPiProvider(o.provider ?? this.cfg.defaults.provider);
const child = this.spawnFn(sessionId, author, provider, o.principal);
let child: ChildProcessWithoutNullStreams;
try {
child = this.spawnFn(sessionId, author, provider, o.principal, o.runtimeConfig?.path);
} catch (error) {
o.runtimeConfig?.release();
throw error;
}
let runtimeConfigReleased = false;
const releaseRuntimeConfig = () => {
if (runtimeConfigReleased) return;
runtimeConfigReleased = true;
o.runtimeConfig?.release();
};
child.once("exit", releaseRuntimeConfig);
child.once("close", releaseRuntimeConfig);
let rt: SessionRuntime | undefined;
try {
const rpc = new RpcClient(child);
@@ -150,6 +171,7 @@ export class PiProcessManager {
bridge,
child,
ownerKey: o.principal ? `${o.principal.issuer}\0${o.principal.subject}` : undefined,
...(o.runtimeConfig ? { releaseRuntimeConfig } : {}),
};
rt = runtime;
bridge.beginTurn();
@@ -184,6 +206,7 @@ export class PiProcessManager {
return runtime;
} catch (error) {
if (rt && this.runtimes.get(sessionId) === rt) this.runtimes.delete(sessionId);
releaseRuntimeConfig();
try { child.kill(); } catch { /* preserve the initialization error */ }
throw error;
}
@@ -251,6 +274,7 @@ export class PiProcessManager {
// Delete before signalling the child so its asynchronous exit cannot be mistaken for a
// crash, and so a replacement installed by a later lifecycle operation is never targeted.
this.runtimes.delete(id);
expected.releaseRuntimeConfig?.();
expected.child.kill();
return true;
}
backend/src/routes/sessions.ts
+19 -5
View File
@@ -77,6 +77,12 @@ export function sessionRoutes(
return typeof runner.withPrincipal === "function" ? runner.withPrincipal(principal) : runner;
};
const optionsWithRuntimeConfig = (runner: any, workspaceConfigPath: string | undefined, options: any) => (
workspaceConfigPath && typeof runner.acquireWorkspaceRuntime === "function"
? { ...options, runtimeConfig: runner.acquireWorkspaceRuntime(workspaceConfigPath) }
: options
);
const maintenanceReply = (reply: any) => reply.code(503).send({
code: "maintenance",
error: "Session admission is temporarily paused for maintenance. Try again shortly.",
@@ -413,9 +419,11 @@ export function sessionRoutes(
principal,
question: b.question,
};
let runtimeOptions = options;
let rt: ReturnType<PiProcessManager["createFor"]> | undefined;
try {
rt = d.mgr.createFor(id, options);
runtimeOptions = optionsWithRuntimeConfig(runner, workspaceConfigPath, options);
rt = d.mgr.createFor(id, runtimeOptions);
bindRuntime(id, rt, runner, workspaceConfigPath);
} catch (error) {
if (rt) d.mgr.teardownIfCurrent(id, rt);
@@ -430,9 +438,9 @@ export function sessionRoutes(
}
info(id, "Session created");
bootstrap(
id, rt, runner, workspaceConfigPath, d.mgr.configure(rt, options),
id, rt, runner, workspaceConfigPath, d.mgr.configure(rt, runtimeOptions),
runner.searchPack(b.question, id, workspaceConfigPath),
() => d.mgr.start(id, rt, options),
() => d.mgr.start(id, rt, runtimeOptions),
);
return { id };
} finally {
@@ -560,6 +568,7 @@ export function sessionRoutes(
principal,
mode: "resume" as const,
};
let runtimeOptions = options;
// Reopening is validation, not the transport commit point. Keep the old hub intact if
// persistence cannot be reopened.
@@ -589,7 +598,8 @@ export function sessionRoutes(
if (boundRuntimes.get(id) === current) boundRuntimes.delete(id);
d.mgr.teardownIfCurrent(id, current);
}
rt = d.mgr.createFor(id, options);
runtimeOptions = optionsWithRuntimeConfig(runner, workspaceConfigPath, options);
rt = d.mgr.createFor(id, runtimeOptions);
bindRuntime(id, rt, runner, workspaceConfigPath);
} catch {
// A created-but-unbound runtime is not usable. The old hub remains attached because
@@ -605,7 +615,11 @@ export function sessionRoutes(
// immediately before the first event produced by the new Resume.
d.hub.clear(id);
info(id, "Resuming session");
bootstrap(id, rt, runner, workspaceConfigPath, d.mgr.configure(rt, options), null, () => d.mgr.start(id, rt, options));
bootstrap(
id, rt, runner, workspaceConfigPath,
d.mgr.configure(rt, runtimeOptions), null,
() => d.mgr.start(id, rt, runtimeOptions),
);
return reply.code(200).send({ id, alreadyActive: false });
});
});
backend/src/routes/sql.ts
+40 -11
View File
@@ -3,22 +3,49 @@ import type { ThtRunner } from "../tht/tht-runner.js";
import { getPrincipal } from "../auth/auth.js";
import type { PrincipalContext } from "../auth/principal.js";
import type { Settings } from "../settings/settings-store.js";
import type { WorkspaceRegistry } from "../workspaces/registry.js";
export function sqlRoutes(app: FastifyInstance, deps: {
tht: ThtRunner; getSettings: (principal: PrincipalContext) => Promise<Settings>;
workspaceRegistry: WorkspaceRegistry;
}): void {
const runnerFor = (principal: PrincipalContext): any => {
const runner = deps.tht as any;
return typeof runner.withPrincipal === "function" ? runner.withPrincipal(principal) : runner;
};
const authorize = async (principal: PrincipalContext, id: string, workspace?: string) => {
try {
const runner = runnerFor(principal);
if (typeof runner.sessionShow !== "function") return {};
return await runner.sessionShow(id, workspace);
const isNotFound = (error: unknown) => /not found|non trovata|inesistente|404/i.test(
error instanceof Error ? error.message : String(error),
);
const locate = async (principal: PrincipalContext, id: string, legacyWorkspace?: string) => {
const runner = runnerFor(principal);
if (typeof runner.sessionShow !== "function") return { manifest: {}, workspace: legacyWorkspace };
const registry = deps.workspaceRegistry as Partial<WorkspaceRegistry>;
const revisions = typeof registry.listRetainedSnapshots === "function"
? await registry.listRetainedSnapshots.call(deps.workspaceRegistry)
: await deps.workspaceRegistry.list();
for (const revision of revisions) {
if (revision.state !== "operational") continue;
try {
const manifest = await runner.sessionShow(id, revision.snapshotPath);
if (!manifest) continue;
const saved = manifest as { workspace_id?: string; workspace_revision?: string };
if (saved.workspace_id && saved.workspace_revision) {
const pinned = await deps.workspaceRegistry.readPinned(saved.workspace_id, saved.workspace_revision);
return {
manifest,
workspace: pinned.workspaceConfigPath ?? (pinned as any).revision?.snapshotPath,
};
}
return { manifest, workspace: revision.snapshotPath };
} catch (error) {
if (!isNotFound(error)) throw error;
}
}
catch (error) {
if (/not found|non trovata|inesistente|404/i.test(error instanceof Error ? error.message : String(error))) return undefined;
try {
const manifest = await runner.sessionShow(id, legacyWorkspace);
return manifest ? { manifest, workspace: legacyWorkspace } : undefined;
} catch (error) {
if (isNotFound(error)) return undefined;
throw error;
}
};
@@ -30,8 +57,9 @@ export function sqlRoutes(app: FastifyInstance, deps: {
try {
principal = getPrincipal(req);
const settings = await deps.getSettings(principal);
workspace = settings.workspace;
if (!await authorize(principal, id, settings.workspace)) return reply.code(404).send({ error: "session not found" });
const located = await locate(principal, id, settings.workspace);
if (!located) return reply.code(404).send({ error: "session not found" });
workspace = located.workspace;
} catch {
return reply.code(503).send({ error: "session storage is unavailable" });
}
@@ -49,8 +77,9 @@ export function sqlRoutes(app: FastifyInstance, deps: {
try {
principal = getPrincipal(req);
const settings = await deps.getSettings(principal);
workspace = settings.workspace;
if (!await authorize(principal, id, settings.workspace)) return reply.code(404).send({ error: "session not found" });
const located = await locate(principal, id, settings.workspace);
if (!located) return reply.code(404).send({ error: "session not found" });
workspace = located.workspace;
} catch {
return reply.code(503).send({ error: "session storage is unavailable" });
}
backend/src/tht/tht-runner.ts
+119 -3
View File
@@ -4,9 +4,13 @@ import {
closeSync, constants as fsConstants, existsSync, fchmodSync, fstatSync, fsyncSync, lstatSync, mkdirSync,
openSync, readFileSync, readSync, realpathSync, statSync, unlinkSync, writeFileSync,
} from "node:fs";
import { dirname, isAbsolute, join, relative } from "node:path";
import { dirname, isAbsolute, join, relative, resolve } from "node:path";
import { parseAllDocuments } from "yaml";
import { clearPrincipalEnvironment, principalEnvironment, type PrincipalContext } from "../auth/principal.js";
import { secretValue, type SecretBundleConfig } from "../config/secret-bundle.js";
import { resolveRuntimeBindings } from "../workspaces/bindings.js";
import { renderRuntimeConfig, type RuntimeInstallationOverlay, type RuntimePaths } from "../workspaces/runtime-renderer.js";
import { parseWorkspaceYaml } from "../workspaces/schema.js";
export interface ThtConfig extends SecretBundleConfig {
thtBin: string;
@@ -14,6 +18,14 @@ export interface ThtConfig extends SecretBundleConfig {
configPath: string;
dataRoot?: string;
runtimeSnapshotRoot?: string;
secretRoots?: readonly string[];
}
export interface RuntimeConfigLease {
path: string;
workspaceId: string;
workspaceRevision: string;
release(): void;
}
export interface SessionRow {
@@ -87,18 +99,110 @@ export class ThtRunner {
return ["-c", this.cfg.configPath];
}
private assertWorkspaceSnapshot(path: string): void {
private assertWorkspaceSnapshot(path: string): { workspaceId: string; workspaceRevision: string } {
if (!this.cfg.runtimeSnapshotRoot) throw new Error("workspace snapshot root is not configured");
const snapshotsRoot = dirname(this.cfg.runtimeSnapshotRoot);
const pathRelative = relative(snapshotsRoot, path);
const match = /^([0-9a-f]{40})\/([a-z][a-z0-9-]{2,62})\.yaml$/.exec(pathRelative);
if (
pathRelative.startsWith("..") || isAbsolute(pathRelative)
|| !/^[0-9a-f]{40}\/[a-z][a-z0-9-]{2,62}\.yaml$/.test(pathRelative)
|| !match
) throw new Error("config path is not a trusted runtime snapshot");
const entry = lstatSync(path);
if (!entry.isFile() || entry.isSymbolicLink()) {
throw new Error("config path is not a trusted runtime snapshot");
}
return { workspaceRevision: match[1], workspaceId: match[2] };
}
private readCanonicalWorkspaceSnapshot(path: string): {
workspace: ReturnType<typeof parseWorkspaceYaml>;
workspaceId: string;
workspaceRevision: string;
} {
const identity = this.assertWorkspaceSnapshot(path);
const fd = openSync(path, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW);
try {
const before = fstatSync(fd);
if (!before.isFile()) throw new Error("workspace snapshot is not a file");
const source = readFileSync(fd, "utf8");
const after = fstatSync(fd);
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);
if (workspace.workspace.id !== identity.workspaceId) {
throw new Error("workspace snapshot identity does not match its path");
}
return { workspace, ...identity };
} finally {
closeSync(fd);
}
}
private runtimePaths(workspaceId: string): RuntimePaths {
if (!this.cfg.dataRoot || !isAbsolute(this.cfg.dataRoot)) {
throw new Error("registry workspace runtime requires an absolute data root");
}
// The portable stack persists one `sessions` store at <dataRoot>/sessions. Keep every
// workspace's mutable harness roots below that mounted boundary.
const root = join(this.cfg.dataRoot, "sessions", workspaceId);
return {
sessions: join(root, "sessions"),
artifacts: join(root, "artifacts"),
indexes: join(root, "indexes"),
};
}
private installationOverlay(): RuntimeInstallationOverlay {
const path = isAbsolute(this.cfg.configPath)
? this.cfg.configPath
: resolve(this.cfg.harnessDir, this.cfg.configPath);
if (!existsSync(path)) return {};
const documents = parseAllDocuments(readFileSync(path, "utf8"), { uniqueKeys: true });
if (documents.length !== 1) throw new Error("installation config must contain one YAML document");
const document = documents[0];
if (document.errors.length > 0 || document.warnings.length > 0) {
throw new Error("installation config contains invalid YAML");
}
const parsed = document.toJSON();
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
throw new Error("installation config must be a YAML mapping");
}
const source = parsed as Record<string, unknown>;
return {
...(source.session_storage === undefined ? {} : { session_storage: source.session_storage }),
...(source.profile === undefined ? {} : { profile: source.profile }),
};
}
/** Render one immutable canonical registry revision into a backend-owned harness config. */
acquireWorkspaceRuntime(workspaceConfigPath: string): RuntimeConfigLease {
const canonical = this.readCanonicalWorkspaceSnapshot(workspaceConfigPath);
const bindings = resolveRuntimeBindings(
canonical.workspace,
process.env,
this.cfg.secretRoots ?? [],
);
const config = renderRuntimeConfig(
canonical.workspace,
bindings,
this.runtimePaths(canonical.workspaceId),
canonical,
this.installationOverlay(),
);
const path = this.createRuntimeSnapshot(config);
let released = false;
return {
path,
workspaceId: canonical.workspaceId,
workspaceRevision: canonical.workspaceRevision,
release: () => {
if (released) return;
released = true;
this.cleanupRuntimeSnapshot(path);
},
};
}
private runtimeSnapshotDirectory(): string {
@@ -228,6 +332,18 @@ export class ThtRunner {
run(
args: string[], workspaceConfigPath?: string, timeoutMs: number = ThtRunner.DEFAULT_TIMEOUT_MS,
): Promise<{ code: number; stdout: string; stderr: string }> {
if (
workspaceConfigPath && isAbsolute(workspaceConfigPath)
&& !this.runtimeSnapshots.has(workspaceConfigPath)
) {
let runtime: RuntimeConfigLease;
try {
runtime = this.acquireWorkspaceRuntime(workspaceConfigPath);
} catch (error) {
return Promise.reject(error);
}
return this.run(args, runtime.path, timeoutMs).finally(runtime.release);
}
return new Promise((resolve) => {
const env: NodeJS.ProcessEnv = { ...process.env };
delete env.THT_DATA_ROOT;
backend/src/workspaces/runtime-renderer.ts
+22
View File
@@ -10,6 +10,16 @@ export interface RuntimePaths {
indexes: string;
}
export interface RuntimeIdentity {
workspaceId: string;
workspaceRevision: string;
}
export interface RuntimeInstallationOverlay {
session_storage?: unknown;
profile?: unknown;
}
function seconds(timeoutMs: number | undefined): number | undefined {
return timeoutMs === undefined ? undefined : Math.max(1, Math.ceil(timeoutMs / 1_000));
}
@@ -73,6 +83,8 @@ export function renderRuntimeConfig(
workspace: WorkspaceDescriptor,
bindings: RuntimeBindings,
paths: RuntimePaths,
identity?: RuntimeIdentity,
installation: RuntimeInstallationOverlay = {},
): string {
const canonical = validateCanonicalWorkspace(workspace);
if ([...bindings.dwh.missing, ...bindings.vector.missing, ...bindings.embedding.missing].length > 0) {
@@ -113,10 +125,20 @@ export function renderRuntimeConfig(
if (embeddingTimeout !== undefined) embedding.timeout = embeddingTimeout;
const rendered: Record<string, unknown> = {
...(identity ? {
runtime_identity: {
workspace_id: identity.workspaceId,
workspace_revision: identity.workspaceRevision,
},
} : {}),
...(installation.session_storage === undefined
? {} : { session_storage: installation.session_storage }),
...(installation.profile === undefined ? {} : { profile: installation.profile }),
language: canonical.workspace.language,
database,
vector_db: vectorDb,
embeddings: embedding,
roots: paths,
paths,
};
if (dwhDirect) {
backend/src/workspaces/schema.ts
+3 -2
View File
@@ -266,8 +266,9 @@ export function parseWorkspaceYaml(source: string): WorkspaceDescriptor {
const documents = parseAllDocuments(source, { uniqueKeys: true });
if (documents.length !== 1) throw new Error("Workspace YAML must contain exactly one document");
const document = documents[0];
if (document.errors.length > 0) {
throw new Error(`Invalid workspace YAML: ${document.errors.map((error) => error.message).join("; ")}`);
if (document.errors.length > 0 || document.warnings.length > 0) {
throw new Error(`Invalid workspace YAML: ${[...document.errors, ...document.warnings]
.map((error) => error.message).join("; ")}`);
}
return validateWorkspaceDescriptor(document.toJSON());
}