fix: harden pi management verification
This commit is contained in:
@@ -0,0 +1,121 @@
|
||||
import { spawn as nodeSpawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import type { AppConfig } from "../config.js";
|
||||
import { secretValue } from "../config/secret-bundle.js";
|
||||
import { clearPrincipalEnvironment } from "../auth/principal.js";
|
||||
import { RpcClient } from "../rpc/rpc-client.js";
|
||||
import { loadPiAuthProviders } from "./auth-providers.js";
|
||||
import { buildPiChildEnv, canonicalPiProvider } from "./provider-credentials.js";
|
||||
import type { PiReasoning } from "./management.js";
|
||||
|
||||
const SMOKE_PROMPT = "Provider health check only. Reply with exactly OK without using tools.";
|
||||
|
||||
export interface PiProviderSmokeRequest {
|
||||
provider: string;
|
||||
model: string;
|
||||
reasoning: PiReasoning;
|
||||
timeoutMs: number;
|
||||
}
|
||||
|
||||
export type PiProviderSmoke = (request: PiProviderSmokeRequest) => Promise<void>;
|
||||
|
||||
interface ProviderSmokeOptions {
|
||||
spawnFn?: (
|
||||
command: string,
|
||||
args: string[],
|
||||
options: { cwd: string; env: NodeJS.ProcessEnv },
|
||||
) => ChildProcessWithoutNullStreams;
|
||||
authProviders?: () => ReadonlySet<string>;
|
||||
}
|
||||
|
||||
export function createPiProviderSmoke(
|
||||
config: AppConfig,
|
||||
options: ProviderSmokeOptions = {},
|
||||
): PiProviderSmoke {
|
||||
const spawnFn = options.spawnFn ?? nodeSpawn;
|
||||
const authProviders = options.authProviders ?? (() => loadPiAuthProviders());
|
||||
|
||||
return async ({ provider, model, reasoning, timeoutMs }): Promise<void> => {
|
||||
let child: ChildProcessWithoutNullStreams | undefined;
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
try {
|
||||
const canonicalProvider = canonicalPiProvider(provider);
|
||||
if (!canonicalProvider || timeoutMs <= 0) throw providerFailure();
|
||||
const env = buildPiChildEnv({
|
||||
provider: canonicalProvider,
|
||||
authProviders: authProviders(),
|
||||
credentialValue: secretValue(config, "THT_MODEL_API_KEY"),
|
||||
credentialFile: config.modelApiKeyFile,
|
||||
});
|
||||
clearPrincipalEnvironment(env);
|
||||
delete env.THT_DATA_ROOT;
|
||||
if (config.dataRoot !== undefined) env.THT_DATA_ROOT = config.dataRoot;
|
||||
|
||||
child = spawnFn(config.piBin, ["--mode", "rpc"], { cwd: config.harnessDir, env });
|
||||
child.stderr.resume();
|
||||
const rpc = new RpcClient(child);
|
||||
const turn = async (): Promise<void> => {
|
||||
requireSuccessfulResponse(await rpc.request({
|
||||
type: "set_model", provider: canonicalProvider, modelId: model,
|
||||
} as object & { type: string }));
|
||||
requireSuccessfulResponse(await rpc.request({
|
||||
type: "set_thinking_level", level: reasoning,
|
||||
} as object & { type: string }));
|
||||
await waitForProviderTurn(rpc, child!);
|
||||
};
|
||||
await Promise.race([
|
||||
turn(),
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => reject(providerTimeout()), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} catch (error) {
|
||||
if (isProviderTimeout(error)) throw providerTimeout();
|
||||
throw providerFailure();
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
if (child) {
|
||||
try { child.kill(); } catch { /* preserve the sanitized smoke outcome */ }
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function waitForProviderTurn(rpc: RpcClient, child: ChildProcessWithoutNullStreams): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let failed = false;
|
||||
rpc.on("event", (event) => {
|
||||
if (event?.type === "message_end" && event.message?.role === "assistant"
|
||||
&& event.message.stopReason === "error") {
|
||||
failed = true;
|
||||
reject(providerFailure());
|
||||
return;
|
||||
}
|
||||
if (event?.type === "agent_end") {
|
||||
const messages = Array.isArray(event.messages) ? event.messages : [];
|
||||
const eventFailed = messages.some((message: any) => (
|
||||
message?.role === "assistant" && message?.stopReason === "error"
|
||||
));
|
||||
if (failed || eventFailed) reject(providerFailure());
|
||||
else resolve();
|
||||
}
|
||||
});
|
||||
child.once("exit", () => reject(providerFailure()));
|
||||
rpc.send({ type: "prompt", message: SMOKE_PROMPT });
|
||||
});
|
||||
}
|
||||
|
||||
function requireSuccessfulResponse(response: any): void {
|
||||
if (!response || response.success !== true) throw providerFailure();
|
||||
}
|
||||
|
||||
function providerFailure(): Error {
|
||||
return new Error("Pi provider smoke check failed");
|
||||
}
|
||||
|
||||
function providerTimeout(): Error {
|
||||
return Object.assign(new Error("Pi smoke check timed out"), { code: "ETIMEDOUT" });
|
||||
}
|
||||
|
||||
function isProviderTimeout(error: unknown): boolean {
|
||||
return Boolean(error && typeof error === "object" && (error as { code?: unknown }).code === "ETIMEDOUT");
|
||||
}
|
||||
Reference in New Issue
Block a user