open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2026-09-20 16:07:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(ql3): add bounded copilot product client

Browse Source
This commit is contained in:
whyour
2026-08-16 02:15:21 +08:00
parent e5a3d83a8d
commit 0e85cbdeb6
12 changed files with 2123 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ql3-cluster-admin/package.json
+6
View File
@@ -374,6 +374,11 @@
"types": "./dist/plugin-package/publisher/pluginPackagePublisherTrustManagement.d.ts",
"require": "./dist/plugin-package/publisher/pluginPackagePublisherTrustManagement.js",
"default": "./dist/plugin-package/publisher/pluginPackagePublisherTrustManagement.js"
},
"./copilot-client": {
"types": "./dist/copilot-client/client.d.ts",
"require": "./dist/copilot-client/client.js",
"default": "./dist/copilot-client/client.js"
}
},
"files": [
@@ -388,6 +393,7 @@
},
"bin": {
"ql3-cluster-admin": "dist/product-cli/cli.js",
"ql3-copilot-client": "dist/copilot-client/cli.js",
"ql3-plugin-package-recover": "dist/plugin-package/recovery/pluginPackageRecoveryCli.js",
"ql3-plugin-package-manage": "dist/plugin-package/management/pluginPackageManagementCli.js",
"ql3-plugin-package-client": "dist/plugin-package/management/pluginPackageManagementClientCli.js",
packages/ql3-cluster-admin/src/copilot-client/cli.ts
+96
View File
@@ -0,0 +1,96 @@
#!/usr/bin/env node
import {
ClusterCopilotClientRemoteError,
executeClusterCopilotClient,
} from './client';
const USAGE =
'Usage: ql3-copilot-client --config=/absolute/client.json --command=/absolute/command.json --credential=/absolute/credential';
function argumentsFrom(argv: readonly string[]): Readonly<{
configFile: string;
commandFile: string;
credentialFile: string;
}> | null {
if (argv.length !== 3) return null;
const values = new Map<string, string>();
for (const argument of argv) {
const match = /^--(config|command|credential)=(\/.+)$/.exec(argument);
if (!match || values.has(match[1]!)) return null;
values.set(match[1]!, match[2]!);
}
if (
!values.has('config') ||
!values.has('command') ||
!values.has('credential')
) {
return null;
}
return Object.freeze({
configFile: values.get('config')!,
commandFile: values.get('command')!,
credentialFile: values.get('credential')!,
});
}
function failureFact(error: unknown): Readonly<Record<string, unknown>> {
const candidate = error as { readonly code?: unknown };
return Object.freeze({
schemaVersion: 1,
component: 'qinglong3-cluster-copilot-client',
event: 'command_failed',
code:
typeof candidate?.code === 'string'
? candidate.code
: 'QL3_CLUSTER_COPILOT_CLIENT_FAILED',
...(error instanceof ClusterCopilotClientRemoteError
? {
statusCode: error.statusCode,
responseCode: error.responseCode,
requestId: error.requestId,
...(error.retryAfterSeconds === null
? {}
: { retryAfterSeconds: error.retryAfterSeconds }),
}
: {}),
});
}
async function run(argv: readonly string[]): Promise<void> {
if (argv.length === 1 && (argv[0] === '--help' || argv[0] === '-h')) {
process.stdout.write(`${USAGE}\n`);
return;
}
const paths = argumentsFrom(argv);
if (!paths) {
process.stderr.write(
`${JSON.stringify({
schemaVersion: 1,
component: 'qinglong3-cluster-copilot-client',
event: 'usage_invalid',
code: 'QL3_CLUSTER_COPILOT_CLIENT_USAGE_INVALID',
})}\n`,
);
process.exitCode = 64;
return;
}
try {
const result = await executeClusterCopilotClient(paths);
process.stdout.write(
`${JSON.stringify({
schemaVersion: 1,
component: 'qinglong3-cluster-copilot-client',
event: 'command_completed',
operation: result.operation,
requestId: result.requestId,
result: result.result,
})}\n`,
);
} catch (error) {
process.stderr.write(`${JSON.stringify(failureFact(error))}\n`);
process.exitCode = 1;
}
}
void run(process.argv.slice(2));
packages/ql3-cluster-admin/src/copilot-client/client.ts
+636
View File
@@ -0,0 +1,636 @@
import { randomUUID, X509Certificate } from 'node:crypto';
import { request as httpsRequest } from 'node:https';
import { isIP, type LookupFunction } from 'node:net';
import { TextDecoder } from 'node:util';
import {
ClusterPluginPackageManagementClientConfigurationError,
readCanonicalFile,
} from '../management-support/managementClientConfiguration';
import {
InvalidClusterCopilotClientCommandError,
InvalidClusterCopilotClientResponseError,
normalizeClusterCopilotClientCommand,
prepareClusterCopilotClientRequest,
validateClusterCopilotClientResponse,
type ClusterCopilotClientCommand,
type ClusterCopilotClientOperation,
} from './contracts';
export {
CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_CANCELLATION_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_INSPECTION_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_OUTPUT_READ_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_RESPONSE_SCHEMA,
InvalidClusterCopilotClientCommandError,
InvalidClusterCopilotClientResponseError,
normalizeClusterCopilotClientCommand,
prepareClusterCopilotClientRequest,
validateClusterCopilotClientResponse,
} from './contracts';
export type {
ClusterCopilotClientCommand,
ClusterCopilotClientOperation,
ClusterCopilotClientPreparedRequest,
} from './contracts';
export const CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA =
'qinglong/cluster-copilot-client-config@v1' as const;
export interface ClusterCopilotClientPaths {
readonly configFile: string;
readonly commandFile: string;
readonly credentialFile: string;
}
export interface ClusterCopilotClientOptions {
readonly createRequestId?: () => string;
readonly lookup?: LookupFunction;
}
export interface ClusterCopilotClientResult {
readonly schemaVersion: 1;
readonly operation: ClusterCopilotClientOperation;
readonly requestId: string;
readonly result: Readonly<Record<string, unknown>>;
}
export interface ClusterCopilotClientConfigurationSummary {
readonly schemaVersion: 1;
readonly transport: 'https';
readonly clientCertificate: 'forbidden';
}
export interface ClusterCopilotClientReadiness {
readonly schemaVersion: 1;
readonly transport: 'https';
readonly ready: boolean;
}
interface PreparedClusterCopilotClientConfiguration {
readonly endpoint: URL;
readonly servername: string;
readonly port: number;
readonly requestTimeoutMs: number;
readonly caBytes: Buffer;
dispose(): void;
}
interface JsonResponse {
readonly statusCode: number;
readonly headers: Readonly<Record<string, string | string[] | undefined>>;
readonly rawHeaders: readonly string[];
readonly body: unknown;
}
export class ClusterCopilotClientConfigurationError extends TypeError {
readonly code = 'QL3_CLUSTER_COPILOT_CLIENT_CONFIG_INVALID';
constructor() {
super('Cluster Copilot client configuration is invalid');
this.name = 'ClusterCopilotClientConfigurationError';
}
}
export class ClusterCopilotClientRequestError extends Error {
readonly code = 'QL3_CLUSTER_COPILOT_CLIENT_REQUEST_FAILED';
constructor(options?: ErrorOptions) {
super('Cluster Copilot client request failed', options);
this.name = 'ClusterCopilotClientRequestError';
}
}
export class ClusterCopilotClientRemoteError extends Error {
readonly code = 'QL3_CLUSTER_COPILOT_CLIENT_REMOTE_REJECTED';
constructor(
readonly statusCode: number,
readonly responseCode: string,
readonly requestId: string,
readonly retryAfterSeconds: number | null,
) {
super('Cluster Copilot server rejected the request');
this.name = 'ClusterCopilotClientRemoteError';
}
}
const MAXIMUM_CONFIG_BYTES = 16 * 1024;
const MAXIMUM_CA_BYTES = 256 * 1024;
const MAXIMUM_COMMAND_BYTES = 16 * 1024;
const MAXIMUM_CREDENTIAL_BYTES = 256;
const MAXIMUM_RESPONSE_BYTES = 2 * 1024 * 1024;
const MAXIMUM_READINESS_RESPONSE_BYTES = 1_024;
const DNS_NAME =
/^(?=.{1,253}$)[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*$/;
const API_CREDENTIAL =
/^ql3c_[A-Za-z0-9][A-Za-z0-9._:-]{0,63}_[A-Za-z0-9_-]{43}$/;
const RESPONSE_CODE = /^[a-z][a-z0-9_]{0,127}$/;
function configurationFailure(): never {
throw new ClusterCopilotClientConfigurationError();
}
function exact(
value: unknown,
keys: readonly string[],
): Record<string, unknown> {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
return configurationFailure();
}
const actual = Object.keys(value).sort();
const expected = [...keys].sort();
if (
actual.length !== expected.length ||
actual.some((key, index) => key !== expected[index])
) {
return configurationFailure();
}
return value as Record<string, unknown>;
}
function decodeJson(bytes: Buffer, kind: 'config' | 'command'): unknown {
try {
return JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(bytes));
} catch {
if (kind === 'command') {
throw new InvalidClusterCopilotClientCommandError();
}
return configurationFailure();
}
}
function prepareConfiguration(
configFile: string,
): PreparedClusterCopilotClientConfiguration {
let configBytes: Buffer | undefined;
let caBytes: Buffer | undefined;
try {
configBytes = readCanonicalFile(
configFile,
MAXIMUM_CONFIG_BYTES,
'private',
);
const config = exact(decodeJson(configBytes, 'config'), [
'caFile',
'endpoint',
'requestTimeoutMs',
'schema',
'servername',
]);
if (
config.schema !== CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA ||
typeof config.endpoint !== 'string' ||
typeof config.servername !== 'string' ||
!DNS_NAME.test(config.servername) ||
isIP(config.servername) !== 0 ||
typeof config.caFile !== 'string' ||
!Number.isSafeInteger(config.requestTimeoutMs) ||
(config.requestTimeoutMs as number) < 1_000 ||
(config.requestTimeoutMs as number) > 120_000
) {
return configurationFailure();
}
let endpoint: URL;
try {
endpoint = new URL(config.endpoint);
} catch {
return configurationFailure();
}
if (
endpoint.protocol !== 'https:' ||
endpoint.username !== '' ||
endpoint.password !== '' ||
endpoint.search !== '' ||
endpoint.hash !== '' ||
endpoint.pathname !== '/' ||
endpoint.hostname !== config.servername ||
isIP(endpoint.hostname) !== 0
) {
return configurationFailure();
}
const port = endpoint.port === '' ? 443 : Number(endpoint.port);
if (!Number.isSafeInteger(port) || port < 1 || port > 65_535) {
return configurationFailure();
}
caBytes = readCanonicalFile(
config.caFile,
MAXIMUM_CA_BYTES,
'public-integrity',
);
try {
new X509Certificate(caBytes);
} catch {
return configurationFailure();
}
let disposed = false;
return Object.freeze({
endpoint,
servername: config.servername,
port,
requestTimeoutMs: config.requestTimeoutMs as number,
caBytes,
dispose() {
if (disposed) return;
disposed = true;
caBytes?.fill(0);
},
});
} catch (error) {
caBytes?.fill(0);
if (error instanceof ClusterCopilotClientConfigurationError) throw error;
throw new ClusterCopilotClientConfigurationError();
} finally {
configBytes?.fill(0);
}
}
function validateOptions(
options: ClusterCopilotClientOptions | undefined,
): Readonly<ClusterCopilotClientOptions> {
if (options === undefined) return Object.freeze({});
if (
!options ||
typeof options !== 'object' ||
Array.isArray(options) ||
Object.keys(options).some(
(key) => key !== 'createRequestId' && key !== 'lookup',
) ||
(options.createRequestId !== undefined &&
typeof options.createRequestId !== 'function') ||
(options.lookup !== undefined && typeof options.lookup !== 'function')
) {
return configurationFailure();
}
return Object.freeze({ ...options });
}
function rawHeaderCount(rawHeaders: readonly string[], name: string): number {
let count = 0;
for (let index = 0; index < rawHeaders.length; index += 2) {
if (rawHeaders[index]?.toLowerCase() === name) count += 1;
}
return count;
}
function responseHeadersValid(
response: JsonResponse,
byteLength: number,
): boolean {
const contentLength = response.headers['content-length'];
return (
rawHeaderCount(response.rawHeaders, 'content-type') === 1 &&
response.headers['content-type'] === 'application/json; charset=utf-8' &&
response.headers['content-encoding'] === undefined &&
rawHeaderCount(response.rawHeaders, 'content-length') <= 1 &&
(contentLength === undefined ||
(typeof contentLength === 'string' &&
/^(?:0|[1-9][0-9]*)$/.test(contentLength) &&
Number(contentLength) === byteLength))
);
}
function readinessStatus(value: unknown): 'ready' | 'not_ready' {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
throw new ClusterCopilotClientRequestError();
}
const record = value as Record<string, unknown>;
if (
Object.keys(record).length !== 1 ||
!Object.hasOwn(record, 'status') ||
(record.status !== 'ready' && record.status !== 'not_ready')
) {
throw new ClusterCopilotClientRequestError();
}
return record.status;
}
function requestJson(
prepared: PreparedClusterCopilotClientConfiguration,
request: Readonly<{
method: 'GET' | 'POST';
path: string;
requestId?: string;
authorization?: string;
body?: Buffer;
}>,
maximumResponseBytes: number,
options: Readonly<ClusterCopilotClientOptions>,
): Promise<Readonly<JsonResponse>> {
return new Promise((resolve, reject) => {
let settled = false;
const chunks: Buffer[] = [];
let length = 0;
const clearChunks = (): void => {
for (const chunk of chunks) chunk.fill(0);
};
const finish = (error: unknown, result?: Readonly<JsonResponse>): void => {
if (settled) return;
settled = true;
if (error) reject(error);
else resolve(result!);
};
const outgoing = httpsRequest(
{
protocol: 'https:',
hostname: prepared.endpoint.hostname,
port: prepared.port,
path: request.path,
method: request.method,
servername: prepared.servername,
ca: prepared.caBytes,
minVersion: 'TLSv1.3',
maxVersion: 'TLSv1.3',
rejectUnauthorized: true,
agent: false,
...(options.lookup === undefined ? {} : { lookup: options.lookup }),
headers: {
accept: 'application/json',
'accept-encoding': 'identity',
connection: 'close',
...(request.authorization === undefined
? {}
: { authorization: request.authorization }),
...(request.requestId === undefined
? {}
: { 'x-request-id': request.requestId }),
...(request.body === undefined
? {}
: {
'content-type': 'application/json; charset=utf-8',
'content-length': String(request.body.length),
}),
},
},
(incoming) => {
incoming.once('aborted', () => {
clearChunks();
finish(new ClusterCopilotClientRequestError());
});
incoming.once('error', (cause) => {
clearChunks();
finish(new ClusterCopilotClientRequestError({ cause }));
});
incoming.on('data', (chunk: Buffer | string) => {
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
length += bytes.length;
if (length > maximumResponseBytes) {
bytes.fill(0);
clearChunks();
const error = new ClusterCopilotClientRequestError();
incoming.destroy();
outgoing.destroy(error);
finish(error);
return;
}
chunks.push(bytes);
});
incoming.once('end', () => {
const bytes = Buffer.concat(chunks, length);
try {
const provisional: JsonResponse = Object.freeze({
statusCode: incoming.statusCode ?? 0,
headers: incoming.headers,
rawHeaders: Object.freeze([...incoming.rawHeaders]),
body: null,
});
if (!responseHeadersValid(provisional, bytes.length)) {
throw new ClusterCopilotClientRequestError();
}
let body: unknown;
try {
body = JSON.parse(
new TextDecoder('utf-8', { fatal: true }).decode(bytes),
);
} catch (cause) {
throw new ClusterCopilotClientRequestError({ cause });
}
finish(
undefined,
Object.freeze({ ...provisional, body }),
);
} catch (error) {
finish(
error instanceof ClusterCopilotClientRequestError
? error
: new ClusterCopilotClientRequestError({
cause: error instanceof Error ? error : undefined,
}),
);
} finally {
bytes.fill(0);
clearChunks();
}
});
},
);
outgoing.setTimeout(prepared.requestTimeoutMs, () => {
outgoing.destroy(new ClusterCopilotClientRequestError());
});
outgoing.once('error', (cause) => {
clearChunks();
finish(
cause instanceof ClusterCopilotClientRequestError
? cause
: new ClusterCopilotClientRequestError({ cause }),
);
});
outgoing.end(request.body);
});
}
function responseRequestId(
response: JsonResponse,
expected: string,
): string {
const value = response.headers['x-request-id'];
if (
rawHeaderCount(response.rawHeaders, 'x-request-id') !== 1 ||
typeof value !== 'string' ||
value !== expected
) {
throw new ClusterCopilotClientRequestError();
}
return value;
}
function retryAfterSeconds(value: string | string[] | undefined): number | null {
if (typeof value !== 'string' || !/^[1-9][0-9]{0,3}$/.test(value)) {
return null;
}
const seconds = Number(value);
return seconds <= 3_600 ? seconds : null;
}
function remoteCode(value: unknown): string {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
throw new ClusterCopilotClientRequestError();
}
const record = value as Record<string, unknown>;
const keys = Object.keys(record).sort();
if (
keys.length < 1 ||
keys.length > 3 ||
keys[0] !== 'code' ||
keys.some((key) => key !== 'code' && key !== 'reason' && key !== 'schema') ||
typeof record.code !== 'string' ||
!RESPONSE_CODE.test(record.code)
) {
throw new ClusterCopilotClientRequestError();
}
return record.code;
}
export function validateClusterCopilotClientConfiguration(
configFile: string,
): Readonly<ClusterCopilotClientConfigurationSummary> {
const prepared = prepareConfiguration(configFile);
try {
return Object.freeze({
schemaVersion: 1,
transport: 'https',
clientCertificate: 'forbidden',
});
} finally {
prepared.dispose();
}
}
export async function probeClusterCopilotClientReadiness(
configFile: string,
options?: ClusterCopilotClientOptions,
): Promise<Readonly<ClusterCopilotClientReadiness>> {
const normalizedOptions = validateOptions(options);
const prepared = prepareConfiguration(configFile);
try {
const response = await requestJson(
prepared,
Object.freeze({ method: 'GET', path: '/readyz' }),
MAXIMUM_READINESS_RESPONSE_BYTES,
normalizedOptions,
);
const status = readinessStatus(response.body);
const ready = response.statusCode === 200 && status === 'ready';
const notReady =
response.statusCode === 503 && status === 'not_ready';
if (!ready && !notReady) throw new ClusterCopilotClientRequestError();
return Object.freeze({ schemaVersion: 1, transport: 'https', ready });
} catch (error) {
if (
error instanceof ClusterCopilotClientConfigurationError ||
error instanceof ClusterCopilotClientRequestError
) {
throw error;
}
throw new ClusterCopilotClientRequestError({
cause: error instanceof Error ? error : undefined,
});
} finally {
prepared.dispose();
}
}
export async function executeClusterCopilotClient(
paths: ClusterCopilotClientPaths,
options?: ClusterCopilotClientOptions,
): Promise<Readonly<ClusterCopilotClientResult>> {
const normalizedOptions = validateOptions(options);
exact(paths, ['commandFile', 'configFile', 'credentialFile']);
let commandBytes: Buffer | undefined;
let credentialBytes: Buffer | undefined;
let bodyBytes: Buffer | undefined;
let prepared: PreparedClusterCopilotClientConfiguration | undefined;
try {
prepared = prepareConfiguration(paths.configFile);
commandBytes = readCanonicalFile(
paths.commandFile,
MAXIMUM_COMMAND_BYTES,
'private',
);
credentialBytes = readCanonicalFile(
paths.credentialFile,
MAXIMUM_CREDENTIAL_BYTES,
'private',
);
const command = normalizeClusterCopilotClientCommand(
decodeJson(commandBytes, 'command'),
);
if (credentialBytes.some((byte) => byte > 0x7f)) {
return configurationFailure();
}
const credential = credentialBytes.toString('ascii');
if (!API_CREDENTIAL.test(credential)) return configurationFailure();
const transportRequestId =
command.operation === 'inspect' || command.operation === 'output'
? (normalizedOptions.createRequestId ?? randomUUID)()
: undefined;
const request = prepareClusterCopilotClientRequest(
command,
transportRequestId,
);
if (request.body !== null) {
bodyBytes = Buffer.from(JSON.stringify(request.body), 'utf8');
if (
bodyBytes.length < 2 ||
bodyBytes.length > MAXIMUM_COMMAND_BYTES
) {
return configurationFailure();
}
}
const response = await requestJson(
prepared,
Object.freeze({
method: request.method,
path: request.path,
requestId: request.requestId,
authorization: `Bearer ${credential}`,
...(bodyBytes === undefined ? {} : { body: bodyBytes }),
}),
MAXIMUM_RESPONSE_BYTES,
normalizedOptions,
);
const requestId = responseRequestId(response, request.requestId);
if (request.acceptedStatusCodes.includes(response.statusCode)) {
return Object.freeze({
schemaVersion: 1,
operation: command.operation,
requestId,
result: validateClusterCopilotClientResponse(response.body, command),
});
}
if (response.statusCode >= 400 && response.statusCode <= 599) {
throw new ClusterCopilotClientRemoteError(
response.statusCode,
remoteCode(response.body),
requestId,
retryAfterSeconds(response.headers['retry-after']),
);
}
throw new ClusterCopilotClientRequestError();
} catch (error) {
if (
error instanceof ClusterPluginPackageManagementClientConfigurationError
) {
throw new ClusterCopilotClientConfigurationError();
}
if (
error instanceof ClusterCopilotClientConfigurationError ||
error instanceof InvalidClusterCopilotClientCommandError ||
error instanceof InvalidClusterCopilotClientResponseError ||
error instanceof ClusterCopilotClientRequestError ||
error instanceof ClusterCopilotClientRemoteError
) {
throw error;
}
throw new ClusterCopilotClientRequestError({
cause: error instanceof Error ? error : undefined,
});
} finally {
bodyBytes?.fill(0);
commandBytes?.fill(0);
credentialBytes?.fill(0);
prepared?.dispose();
}
}
packages/ql3-cluster-admin/src/copilot-client/contracts.ts
+550
View File
@@ -0,0 +1,550 @@
import { CLUSTER_RUN_CANCELLATION_SCHEMA } from '@qinglong/runtime-core/cluster-run-cancellation';
export const CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA =
'qinglong/cluster-copilot-client-command@v1' as const;
export const CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA =
'qinglong/cluster-copilot-failure-diagnosis-request@v1' as const;
export const CLUSTER_COPILOT_FAILURE_DIAGNOSIS_RESPONSE_SCHEMA =
'qinglong/cluster-copilot-failure-diagnosis-response@v1' as const;
export const CLUSTER_COPILOT_FAILURE_DIAGNOSIS_INSPECTION_RESPONSE_SCHEMA =
'qinglong/cluster-copilot-failure-diagnosis-inspection-response@v1' as const;
export const CLUSTER_COPILOT_FAILURE_DIAGNOSIS_OUTPUT_READ_RESPONSE_SCHEMA =
'qinglong/cluster-copilot-failure-diagnosis-output-read-response@v1' as const;
export const CLUSTER_COPILOT_FAILURE_DIAGNOSIS_CANCELLATION_RESPONSE_SCHEMA =
'qinglong/cluster-copilot-failure-diagnosis-cancellation-response@v1' as const;
export type ClusterCopilotClientOperation =
| 'diagnose'
| 'inspect'
| 'output'
| 'cancel';
interface ClusterCopilotClientCommandBase {
readonly schema: typeof CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA;
readonly operation: ClusterCopilotClientOperation;
readonly projectId: string;
readonly sourceRunId: string;
readonly requestId: string;
}
export type ClusterCopilotClientCommand =
| (ClusterCopilotClientCommandBase &
Readonly<{ readonly operation: 'diagnose'; readonly traceId: string }>)
| (ClusterCopilotClientCommandBase &
Readonly<{ readonly operation: 'inspect' | 'output' }>)
| (ClusterCopilotClientCommandBase &
Readonly<{ readonly operation: 'cancel'; readonly mutationId: string }>);
export interface ClusterCopilotClientPreparedRequest {
readonly method: 'GET' | 'POST';
readonly path: string;
readonly requestId: string;
readonly body: Readonly<Record<string, unknown>> | null;
readonly acceptedStatusCodes: readonly number[];
}
export class InvalidClusterCopilotClientCommandError extends TypeError {
readonly code = 'QL3_CLUSTER_COPILOT_CLIENT_COMMAND_INVALID';
constructor() {
super('Cluster Copilot client command is invalid');
this.name = 'InvalidClusterCopilotClientCommandError';
}
}
export class InvalidClusterCopilotClientResponseError extends Error {
readonly code = 'QL3_CLUSTER_COPILOT_CLIENT_RESPONSE_INVALID';
constructor() {
super('Cluster Copilot client response is invalid');
this.name = 'InvalidClusterCopilotClientResponseError';
}
}
const IDENTITY = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
const RUN_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,35}$/;
const DIGEST = /^[0-9a-f]{64}$/;
const OUTCOMES = new Set(['succeeded', 'failed', 'timed_out', 'cancelled']);
const STAGES = new Set(['model', 'tool', 'log', 'deadline', 'cancellation']);
const REASONS = new Set([
'tool_failed',
'tool_timed_out',
'log_not_found',
'log_pending',
'log_missing',
'log_retired',
'tool_budget_exhausted',
'deadline_exceeded',
'cancellation_requested',
]);
const FINISH_REASONS = new Set([
'stop',
'length',
'content_filter',
'tool_call',
'unknown',
]);
const CANCELLATION_STATUSES = new Set([
'accepted',
'already_requested',
'already_terminal',
]);
const CANCELLATION_REASONS = new Set([
'user',
'policy',
'shutdown',
'reconcile',
'timeout',
]);
function invalidCommand(): never {
throw new InvalidClusterCopilotClientCommandError();
}
function invalidResponse(): never {
throw new InvalidClusterCopilotClientResponseError();
}
function exact(
value: unknown,
keys: readonly string[],
): Record<string, unknown> {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
return invalidResponse();
}
const actual = Object.keys(value).sort();
const expected = [...keys].sort();
if (
actual.length !== expected.length ||
actual.some((key, index) => key !== expected[index])
) {
return invalidResponse();
}
return value as Record<string, unknown>;
}
function exactCommand(
value: unknown,
keys: readonly string[],
): Record<string, unknown> {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
return invalidCommand();
}
const actual = Object.keys(value).sort();
const expected = [...keys].sort();
if (
actual.length !== expected.length ||
actual.some((key, index) => key !== expected[index])
) {
return invalidCommand();
}
return value as Record<string, unknown>;
}
function nonNegativeInteger(value: unknown): value is number {
return Number.isSafeInteger(value) && (value as number) >= 0;
}
function identity(value: unknown): value is string {
return typeof value === 'string' && IDENTITY.test(value);
}
function runId(value: unknown): value is string {
return typeof value === 'string' && RUN_ID.test(value);
}
export function normalizeClusterCopilotClientCommand(
value: unknown,
): Readonly<ClusterCopilotClientCommand> {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
return invalidCommand();
}
const candidate = value as Record<string, unknown>;
const operation = candidate.operation;
if (
operation !== 'diagnose' &&
operation !== 'inspect' &&
operation !== 'output' &&
operation !== 'cancel'
) {
return invalidCommand();
}
const record = exactCommand(value, [
'operation',
'projectId',
'requestId',
'schema',
'sourceRunId',
...(operation === 'diagnose' ? ['traceId'] : []),
...(operation === 'cancel' ? ['mutationId'] : []),
]);
if (
record.schema !== CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA ||
!identity(record.projectId) ||
!runId(record.sourceRunId) ||
!identity(record.requestId) ||
(operation === 'diagnose' && !identity(record.traceId)) ||
(operation === 'cancel' && !identity(record.mutationId))
) {
return invalidCommand();
}
return Object.freeze({ ...record }) as Readonly<ClusterCopilotClientCommand>;
}
function targetPath(command: Readonly<ClusterCopilotClientCommand>): string {
const target = `/api/v3/projects/${encodeURIComponent(
command.projectId,
)}/runs/${encodeURIComponent(
command.sourceRunId,
)}/copilot/failure-diagnoses`;
return command.operation === 'diagnose'
? target
: `${target}/${encodeURIComponent(command.requestId)}`;
}
export function prepareClusterCopilotClientRequest(
command: Readonly<ClusterCopilotClientCommand>,
readRequestId?: string,
): Readonly<ClusterCopilotClientPreparedRequest> {
const normalized = normalizeClusterCopilotClientCommand(command);
if (
(normalized.operation === 'inspect' || normalized.operation === 'output') &&
!identity(readRequestId)
) {
return invalidCommand();
}
const target = targetPath(normalized);
if (normalized.operation === 'diagnose') {
return Object.freeze({
method: 'POST',
path: target,
requestId: normalized.requestId,
body: Object.freeze({
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA,
traceId: normalized.traceId,
}),
acceptedStatusCodes: Object.freeze([200, 201]),
});
}
if (normalized.operation === 'cancel') {
return Object.freeze({
method: 'POST',
path: `${target}/cancellation`,
requestId: normalized.mutationId,
body: Object.freeze({
schema: CLUSTER_RUN_CANCELLATION_SCHEMA,
mutationId: normalized.mutationId,
}),
acceptedStatusCodes: Object.freeze([200, 202]),
});
}
return Object.freeze({
method: 'GET',
path: normalized.operation === 'output' ? `${target}/output` : target,
requestId: readRequestId!,
body: null,
acceptedStatusCodes: Object.freeze([200]),
});
}
function exactTarget(
value: Record<string, unknown>,
command: Readonly<ClusterCopilotClientCommand>,
): boolean {
return (
value.projectId === command.projectId &&
value.sourceRunId === command.sourceRunId &&
value.requestId === command.requestId
);
}
function usage(
value: unknown,
settled: boolean,
): Readonly<Record<string, unknown>> {
const hasCostMicros =
value !== null &&
typeof value === 'object' &&
!Array.isArray(value) &&
Object.hasOwn(value, 'costMicros');
const record = exact(
value,
settled
? ['costMicros', 'currency', 'inputTokens', 'outputTokens', 'totalTokens']
: [
'inputTokens',
'outputTokens',
'totalTokens',
...(hasCostMicros ? ['costMicros'] : []),
],
);
if (
!nonNegativeInteger(record.inputTokens) ||
!nonNegativeInteger(record.outputTokens) ||
!nonNegativeInteger(record.totalTokens) ||
record.totalTokens !== record.inputTokens + record.outputTokens ||
(settled &&
!(
(record.currency === null && record.costMicros === null) ||
(record.currency === 'USD' && nonNegativeInteger(record.costMicros))
)) ||
(!settled &&
record.costMicros !== undefined &&
!nonNegativeInteger(record.costMicros))
) {
return invalidResponse();
}
return Object.freeze({ ...record });
}
function diagnosisResponse(
value: unknown,
command: Extract<ClusterCopilotClientCommand, { readonly operation: 'diagnose' }>,
): Readonly<Record<string, unknown>> {
const record = exact(value, [
'diagnosisRunId',
'outcome',
'outputArtifact',
'reason',
'replayed',
'requestId',
'schema',
'sourceRunId',
'stage',
'status',
]);
if (
record.schema !== CLUSTER_COPILOT_FAILURE_DIAGNOSIS_RESPONSE_SCHEMA ||
record.requestId !== command.requestId ||
record.sourceRunId !== command.sourceRunId ||
!runId(record.diagnosisRunId) ||
(record.status !== 'created' && record.status !== 'existing') ||
record.replayed !== (record.status === 'existing') ||
typeof record.outcome !== 'string' ||
!OUTCOMES.has(record.outcome) ||
typeof record.stage !== 'string' ||
!STAGES.has(record.stage) ||
(record.stage === 'model') !== (record.reason === null) ||
(record.reason !== null &&
(typeof record.reason !== 'string' || !REASONS.has(record.reason)))
) {
return invalidResponse();
}
let outputArtifact: Readonly<Record<string, unknown>> | null = null;
if (record.outputArtifact !== null) {
const artifact = exact(record.outputArtifact, [
'artifactDigest',
'artifactId',
]);
if (
!identity(artifact.artifactId) ||
typeof artifact.artifactDigest !== 'string' ||
!DIGEST.test(artifact.artifactDigest) ||
record.stage !== 'model' ||
record.outcome !== 'succeeded'
) {
return invalidResponse();
}
outputArtifact = Object.freeze({ ...artifact });
} else if (record.stage === 'model' && record.outcome === 'succeeded') {
return invalidResponse();
}
return Object.freeze({ ...record, outputArtifact });
}
function inspectionResponse(
value: unknown,
command: Readonly<ClusterCopilotClientCommand>,
): Readonly<Record<string, unknown>> {
const record = exact(value, [
'admittedAtMs',
'diagnosisRunId',
'finalizedAtMs',
'outcome',
'outputAvailable',
'projectId',
'reason',
'requestId',
'schema',
'sourceRunId',
'stage',
'status',
'usage',
]);
if (
record.schema !==
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_INSPECTION_RESPONSE_SCHEMA ||
!exactTarget(record, command) ||
!runId(record.diagnosisRunId) ||
!nonNegativeInteger(record.admittedAtMs) ||
typeof record.outputAvailable !== 'boolean'
) {
return invalidResponse();
}
if (record.status === 'running') {
if (
record.outcome !== null ||
record.stage !== null ||
record.reason !== null ||
record.outputAvailable !== false ||
record.finalizedAtMs !== null ||
record.usage !== null
) {
return invalidResponse();
}
return Object.freeze({ ...record });
}
if (
record.status !== 'terminal' ||
typeof record.outcome !== 'string' ||
!OUTCOMES.has(record.outcome) ||
typeof record.stage !== 'string' ||
!STAGES.has(record.stage) ||
!nonNegativeInteger(record.finalizedAtMs) ||
record.finalizedAtMs < record.admittedAtMs ||
(record.stage === 'model') !== (record.reason === null) ||
(record.reason !== null &&
(typeof record.reason !== 'string' || !REASONS.has(record.reason))) ||
record.outputAvailable !==
(record.stage === 'model' && record.outcome === 'succeeded')
) {
return invalidResponse();
}
return Object.freeze({
...record,
usage: record.usage === null ? null : usage(record.usage, true),
});
}
function outputResponse(
value: unknown,
command: Readonly<ClusterCopilotClientCommand>,
): Readonly<Record<string, unknown>> {
const record = exact(value, [
'diagnosisRunId',
'projectId',
'reference',
'requestId',
'result',
'schema',
'sourceRunId',
'status',
]);
const reference = exact(record.reference, [
'artifactDigest',
'artifactId',
'contentDigest',
'outputBytes',
'sealedAtMs',
]);
const result = exact(record.result, ['finishReason', 'text', 'usage']);
if (
record.schema !==
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_OUTPUT_READ_RESPONSE_SCHEMA ||
record.status !== 'available' ||
!exactTarget(record, command) ||
!runId(record.diagnosisRunId) ||
!identity(reference.artifactId) ||
typeof reference.artifactDigest !== 'string' ||
!DIGEST.test(reference.artifactDigest) ||
typeof reference.contentDigest !== 'string' ||
!DIGEST.test(reference.contentDigest) ||
!nonNegativeInteger(reference.outputBytes) ||
reference.outputBytes > 1024 * 1024 ||
!nonNegativeInteger(reference.sealedAtMs) ||
typeof result.text !== 'string' ||
Buffer.byteLength(result.text, 'utf8') !== reference.outputBytes ||
typeof result.finishReason !== 'string' ||
!FINISH_REASONS.has(result.finishReason)
) {
return invalidResponse();
}
return Object.freeze({
...record,
reference: Object.freeze({ ...reference }),
result: Object.freeze({
...result,
usage: usage(result.usage, false),
}),
});
}
function cancellationResponse(
value: unknown,
command: Readonly<ClusterCopilotClientCommand>,
): Readonly<Record<string, unknown>> {
const record = exact(value, [
'cancelReason',
'cancelRequestedAtMs',
'convergence',
'diagnosisRunId',
'eventSequence',
'outcome',
'projectId',
'requestId',
'runStatus',
'runVersion',
'schema',
'sourceRunId',
'status',
]);
if (
record.schema !==
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_CANCELLATION_RESPONSE_SCHEMA ||
!exactTarget(record, command) ||
!runId(record.diagnosisRunId) ||
typeof record.status !== 'string' ||
!CANCELLATION_STATUSES.has(record.status) ||
!nonNegativeInteger(record.runVersion) ||
!nonNegativeInteger(record.eventSequence) ||
record.runVersion !== record.eventSequence ||
!(
(record.cancelRequestedAtMs === null && record.cancelReason === null) ||
(nonNegativeInteger(record.cancelRequestedAtMs) &&
typeof record.cancelReason === 'string' &&
CANCELLATION_REASONS.has(record.cancelReason))
)
) {
return invalidResponse();
}
if (
record.convergence === 'model_in_flight' &&
(record.runStatus !== 'running' ||
record.outcome !== null ||
record.cancelRequestedAtMs === null)
) {
return invalidResponse();
}
if (
record.convergence === 'terminal' &&
(typeof record.runStatus !== 'string' ||
!OUTCOMES.has(record.runStatus) ||
record.outcome !== record.runStatus)
) {
return invalidResponse();
}
if (
record.convergence !== 'model_in_flight' &&
record.convergence !== 'terminal'
) {
return invalidResponse();
}
return Object.freeze({ ...record });
}
export function validateClusterCopilotClientResponse(
value: unknown,
command: Readonly<ClusterCopilotClientCommand>,
): Readonly<Record<string, unknown>> {
const normalized = normalizeClusterCopilotClientCommand(command);
if (normalized.operation === 'diagnose') {
return diagnosisResponse(value, normalized);
}
if (normalized.operation === 'inspect') {
return inspectionResponse(value, normalized);
}
if (normalized.operation === 'output') {
return outputResponse(value, normalized);
}
return cancellationResponse(value, normalized);
}
packages/ql3-cluster-admin/src/product-cli/productCommand.ts
+6
View File
@@ -34,6 +34,12 @@ const SEMVER_PATTERN =
export const QINGLONG3_CLUSTER_PRODUCT_COMMANDS: readonly QingLong3ClusterProductCommandDefinition[] =
Object.freeze([
Object.freeze({
name: 'copilot',
binary: 'ql3-copilot-client',
target: 'copilot-client/cli.js',
description: 'diagnose, inspect, read or cancel Runs through the API',
}),
Object.freeze({
name: 'package',
binary: 'ql3-plugin-package-client',
packages/ql3-cluster-admin/src/product-cli/productContext.ts
+33 -7
View File
@@ -10,6 +10,10 @@ import {
import { isAbsolute } from 'node:path';
import { TextDecoder } from 'node:util';
import {
probeClusterCopilotClientReadiness,
validateClusterCopilotClientConfiguration,
} from '../copilot-client/client';
import {
validateClusterAuthenticatedManagementClientConfiguration,
} from '../management-support/pluginPackageManagementClient';
@@ -24,6 +28,7 @@ const MAXIMUM_CONTEXT_BYTES = 64 * 1024;
const MAXIMUM_PATH_BYTES = 4_096;
const CONTROL_PATTERN = /[\u0000-\u001f\u007f]/u;
const CONTEXT_COMMANDS = Object.freeze([
'copilot',
'package',
'package-kubernetes',
'worker-credential',
@@ -80,7 +85,10 @@ export interface QingLong3ClusterProductContextProbe {
}
const CONTEXT_COMMAND_CLIENT_KINDS: Readonly<
Record<ContextCommandName, ClusterAuthenticatedManagementClientKind>
Record<
Exclude<ContextCommandName, 'copilot'>,
ClusterAuthenticatedManagementClientKind
>
> = Object.freeze({
package: 'package',
'package-kubernetes': 'package',
@@ -317,12 +325,23 @@ export async function validateQingLong3ClusterProductContext(
for (const name of CONTEXT_COMMANDS) {
const command = context.commands[name];
if (command === undefined) continue;
const clientKind = CONTEXT_COMMAND_CLIENT_KINDS[name];
const https = validateClusterAuthenticatedManagementClientConfiguration(
command.configFile,
clientKind,
);
if (name === 'package-kubernetes') {
if (name === 'copilot') {
const https = validateClusterCopilotClientConfiguration(
command.configFile,
);
commands.push(
Object.freeze({
name,
transport: https.transport,
clientCertificate: https.clientCertificate,
}),
);
} else if (name === 'package-kubernetes') {
const https =
validateClusterAuthenticatedManagementClientConfiguration(
command.configFile,
CONTEXT_COMMAND_CLIENT_KINDS[name],
);
const kubernetes =
await validateClusterPluginPackageManagementKubernetesConfiguration(
command.kubernetesFile!,
@@ -336,6 +355,11 @@ export async function validateQingLong3ClusterProductContext(
}),
);
} else {
const https =
validateClusterAuthenticatedManagementClientConfiguration(
command.configFile,
CONTEXT_COMMAND_CLIENT_KINDS[name],
);
commands.push(
Object.freeze({
name,
@@ -378,6 +402,8 @@ export async function probeQingLong3ClusterProductContext(
command.configFile,
command.kubernetesFile!,
)
: name === 'copilot'
? await probeClusterCopilotClientReadiness(command.configFile)
: await probeClusterAuthenticatedManagementClientReadiness(
command.configFile,
CONTEXT_COMMAND_CLIENT_KINDS[name],
packages/ql3-cluster-admin/test/copilotClient.test.cjs
+639
View File
@@ -0,0 +1,639 @@
const assert = require('node:assert/strict');
const { spawn } = require('node:child_process');
const fs = require('node:fs');
const os = require('node:os');
const path = require('node:path');
const { test } = require('node:test');
const { createServer } = require('node:https');
const packageRoot = path.resolve(__dirname, '..');
const cliPath = path.join(packageRoot, 'dist', 'copilot-client', 'cli.js');
const fixtureRoot = path.resolve(
packageRoot,
'../ql3-cluster-control/test/fixtures/mtls',
);
const caFixture = path.join(fixtureRoot, 'ca-cert.pem');
const certificateFixture = path.join(fixtureRoot, 'server-cert.pem');
const privateKeyFixture = path.join(fixtureRoot, 'server-key.pem');
const {
CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
ClusterCopilotClientRemoteError,
executeClusterCopilotClient,
probeClusterCopilotClientReadiness,
validateClusterCopilotClientConfiguration,
} = require('../dist/copilot-client/client.js');
const {
CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_CANCELLATION_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_INSPECTION_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_OUTPUT_READ_RESPONSE_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA,
CLUSTER_COPILOT_FAILURE_DIAGNOSIS_RESPONSE_SCHEMA,
normalizeClusterCopilotClientCommand,
prepareClusterCopilotClientRequest,
validateClusterCopilotClientResponse,
} = require('../dist/copilot-client/contracts.js');
const credential = `ql3c_credential-1_${Buffer.alloc(32, 7).toString('base64url')}`;
const baseCommand = {
schema: CLUSTER_COPILOT_CLIENT_COMMAND_SCHEMA,
projectId: 'project-1',
sourceRunId: 'source-run-1',
requestId: 'diagnosis-request-1',
};
function privateFile(directory, name, contents) {
const filePath = path.join(directory, name);
fs.writeFileSync(filePath, contents, { mode: 0o600 });
return fs.realpathSync(filePath);
}
function temporaryDirectory(t) {
const directory = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'ql3-copilot-client-')),
);
t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
return directory;
}
function configuration(directory, port) {
const caFile = privateFile(
directory,
'ca.pem',
fs.readFileSync(caFixture),
);
return privateFile(
directory,
'client.json',
JSON.stringify({
schema: CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
endpoint: `https://localhost:${port}/`,
servername: 'localhost',
caFile,
requestTimeoutMs: 2_000,
}),
);
}
function commandFile(directory, name, operation, extra = {}) {
return privateFile(
directory,
`${name}.json`,
JSON.stringify({ ...baseCommand, operation, ...extra }),
);
}
function jsonResponse(response, statusCode, requestId, body, headers = {}) {
const bytes = Buffer.from(JSON.stringify(body));
response.writeHead(statusCode, {
'content-type': 'application/json; charset=utf-8',
'content-length': String(bytes.length),
...(requestId === null ? {} : { 'x-request-id': requestId }),
...headers,
});
response.end(bytes);
}
async function startServer(handler) {
const server = createServer(
{
key: fs.readFileSync(privateKeyFixture),
cert: fs.readFileSync(certificateFixture),
minVersion: 'TLSv1.3',
maxVersion: 'TLSv1.3',
},
handler,
);
await new Promise((resolvePromise, reject) => {
server.once('error', reject);
server.listen(0, '127.0.0.1', resolvePromise);
});
return {
port: server.address().port,
close: () =>
new Promise((resolvePromise, reject) => {
server.close((error) =>
error ? reject(error) : resolvePromise(),
);
}),
};
}
function runCli(args) {
return new Promise((resolvePromise, reject) => {
const child = spawn(process.execPath, [cliPath, ...args], {
cwd: packageRoot,
stdio: ['ignore', 'pipe', 'pipe'],
});
const stdout = [];
const stderr = [];
child.stdout.on('data', (chunk) => stdout.push(chunk));
child.stderr.on('data', (chunk) => stderr.push(chunk));
child.once('error', reject);
child.once('close', (status, signal) => {
resolvePromise({
status,
signal,
stdout: Buffer.concat(stdout).toString('utf8'),
stderr: Buffer.concat(stderr).toString('utf8'),
});
});
});
}
function diagnoseResponse(overrides = {}) {
return {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_RESPONSE_SCHEMA,
requestId: baseCommand.requestId,
status: 'created',
replayed: false,
sourceRunId: baseCommand.sourceRunId,
diagnosisRunId: 'diagnosis-run-1',
outcome: 'succeeded',
stage: 'model',
reason: null,
outputArtifact: {
artifactId: 'cdo:artifact-1',
artifactDigest: 'a'.repeat(64),
},
...overrides,
};
}
function inspectionResponse(overrides = {}) {
return {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_INSPECTION_RESPONSE_SCHEMA,
status: 'running',
projectId: baseCommand.projectId,
sourceRunId: baseCommand.sourceRunId,
requestId: baseCommand.requestId,
diagnosisRunId: 'diagnosis-run-1',
outcome: null,
stage: null,
reason: null,
outputAvailable: false,
admittedAtMs: 100,
finalizedAtMs: null,
usage: null,
...overrides,
};
}
function outputResponse() {
return {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_OUTPUT_READ_RESPONSE_SCHEMA,
status: 'available',
projectId: baseCommand.projectId,
sourceRunId: baseCommand.sourceRunId,
requestId: baseCommand.requestId,
diagnosisRunId: 'diagnosis-run-1',
reference: {
artifactId: 'cdo:artifact-1',
artifactDigest: 'a'.repeat(64),
contentDigest: 'b'.repeat(64),
outputBytes: Buffer.byteLength('diagnosis'),
sealedAtMs: 200,
},
result: {
text: 'diagnosis',
finishReason: 'stop',
usage: { inputTokens: 3, outputTokens: 2, totalTokens: 5 },
},
};
}
function cancellationResponse() {
return {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_CANCELLATION_RESPONSE_SCHEMA,
status: 'accepted',
convergence: 'terminal',
projectId: baseCommand.projectId,
sourceRunId: baseCommand.sourceRunId,
requestId: baseCommand.requestId,
diagnosisRunId: 'diagnosis-run-1',
runStatus: 'cancelled',
outcome: 'cancelled',
runVersion: 7,
eventSequence: 7,
cancelRequestedAtMs: 500,
cancelReason: 'user',
};
}
test('normalizes only the four bounded commands and derives exact requests', () => {
const commands = [
{ ...baseCommand, operation: 'diagnose', traceId: 'trace-1' },
{ ...baseCommand, operation: 'inspect' },
{ ...baseCommand, operation: 'output' },
{
...baseCommand,
operation: 'cancel',
mutationId: '11111111-1111-4111-8111-111111111111',
},
];
for (const command of commands) {
const normalized = normalizeClusterCopilotClientCommand(command);
assert.deepEqual(normalized, command);
assert.equal(Object.isFrozen(normalized), true);
}
assert.deepEqual(prepareClusterCopilotClientRequest(commands[0]), {
method: 'POST',
path: '/api/v3/projects/project-1/runs/source-run-1/copilot/failure-diagnoses',
requestId: baseCommand.requestId,
body: {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA,
traceId: 'trace-1',
},
acceptedStatusCodes: [200, 201],
});
assert.deepEqual(
prepareClusterCopilotClientRequest(commands[2], 'transport-read-1'),
{
method: 'GET',
path: '/api/v3/projects/project-1/runs/source-run-1/copilot/failure-diagnoses/diagnosis-request-1/output',
requestId: 'transport-read-1',
body: null,
acceptedStatusCodes: [200],
},
);
assert.throws(() =>
normalizeClusterCopilotClientCommand({ ...commands[0], model: 'private' }),
);
assert.throws(() =>
prepareClusterCopilotClientRequest(commands[1], '../invalid'),
);
});
test('validates exact target-bound response state for every operation', () => {
const diagnose = { ...baseCommand, operation: 'diagnose', traceId: 'trace-1' };
const inspect = { ...baseCommand, operation: 'inspect' };
const output = { ...baseCommand, operation: 'output' };
const cancel = {
...baseCommand,
operation: 'cancel',
mutationId: '11111111-1111-4111-8111-111111111111',
};
assert.equal(
validateClusterCopilotClientResponse(diagnoseResponse(), diagnose).status,
'created',
);
assert.equal(
validateClusterCopilotClientResponse(inspectionResponse(), inspect).status,
'running',
);
assert.equal(
validateClusterCopilotClientResponse(
inspectionResponse({
status: 'terminal',
outcome: 'succeeded',
stage: 'model',
outputAvailable: true,
finalizedAtMs: 200,
usage: {
inputTokens: 3,
outputTokens: 2,
totalTokens: 5,
currency: 'USD',
costMicros: 7,
},
}),
inspect,
).status,
'terminal',
);
assert.equal(
validateClusterCopilotClientResponse(outputResponse(), output).result.text,
'diagnosis',
);
assert.equal(
validateClusterCopilotClientResponse(cancellationResponse(), cancel)
.status,
'accepted',
);
assert.equal(
validateClusterCopilotClientResponse(
diagnoseResponse({ status: 'existing', replayed: true }),
diagnose,
).replayed,
true,
);
assert.equal(
validateClusterCopilotClientResponse(
{
...cancellationResponse(),
status: 'already_requested',
convergence: 'model_in_flight',
runStatus: 'running',
outcome: null,
runVersion: 6,
eventSequence: 6,
},
cancel,
).convergence,
'model_in_flight',
);
assert.throws(() =>
validateClusterCopilotClientResponse(
diagnoseResponse({ schema: 'qinglong/drift@v1' }),
diagnose,
),
);
assert.throws(() =>
validateClusterCopilotClientResponse(
diagnoseResponse({ projectId: 'widened' }),
diagnose,
),
);
assert.throws(() =>
validateClusterCopilotClientResponse(
inspectionResponse({ projectId: 'other-project' }),
inspect,
),
);
const invalidOutput = outputResponse();
invalidOutput.reference.outputBytes += 1;
assert.throws(() =>
validateClusterCopilotClientResponse(invalidOutput, output),
);
assert.throws(() =>
validateClusterCopilotClientResponse(
cancellationResponse(),
{ ...cancel, requestId: 'other-request' },
),
);
});
test('uses TLS 1.3, Bearer credential and exact request identities end to end', async (t) => {
const seen = [];
const server = await startServer((request, response) => {
const chunks = [];
request.on('data', (chunk) => chunks.push(chunk));
request.on('end', () => {
const body = Buffer.concat(chunks).toString('utf8');
seen.push({
method: request.method,
url: request.url,
authorization: request.headers.authorization,
requestId: request.headers['x-request-id'],
body: body === '' ? null : JSON.parse(body),
peerCertificate: request.socket.getPeerCertificate(),
});
const requestId = request.headers['x-request-id'];
if (request.url === '/readyz') {
assert.equal(request.headers.authorization, undefined);
jsonResponse(response, 200, null, { status: 'ready' });
} else if (request.method === 'POST' && request.url.endsWith('/cancellation')) {
jsonResponse(response, 202, requestId, cancellationResponse());
} else if (request.method === 'POST') {
jsonResponse(response, 201, requestId, diagnoseResponse());
} else if (request.url.endsWith('/output')) {
jsonResponse(response, 200, requestId, outputResponse());
} else {
jsonResponse(response, 200, requestId, inspectionResponse());
}
});
});
t.after(() => server.close());
const directory = temporaryDirectory(t);
const configFile = configuration(directory, server.port);
const credentialFile = privateFile(directory, 'credential', credential);
assert.deepEqual(validateClusterCopilotClientConfiguration(configFile), {
schemaVersion: 1,
transport: 'https',
clientCertificate: 'forbidden',
});
assert.deepEqual(await probeClusterCopilotClientReadiness(configFile), {
schemaVersion: 1,
transport: 'https',
ready: true,
});
const operations = [
['diagnose', { traceId: 'trace-1' }, baseCommand.requestId],
['inspect', {}, 'transport-read-1'],
['output', {}, 'transport-read-2'],
[
'cancel',
{ mutationId: '11111111-1111-4111-8111-111111111111' },
'11111111-1111-4111-8111-111111111111',
],
];
for (const [operation, extra, expectedRequestId] of operations) {
const result = await executeClusterCopilotClient(
{
configFile,
commandFile: commandFile(directory, operation, operation, extra),
credentialFile,
},
{ createRequestId: () => expectedRequestId },
);
assert.equal(result.operation, operation);
assert.equal(result.requestId, expectedRequestId);
}
assert.equal(seen.length, 5);
for (const request of seen.slice(1)) {
assert.equal(request.authorization, `Bearer ${credential}`);
assert.equal(request.peerCertificate.subject, undefined);
}
assert.deepEqual(seen[1].body, {
schema: CLUSTER_COPILOT_FAILURE_DIAGNOSIS_REQUEST_SCHEMA,
traceId: 'trace-1',
});
assert.equal(seen[2].body, null);
assert.equal(seen[3].body, null);
assert.deepEqual(seen[4].body, {
schema: 'qinglong/run-cancellation@v1',
mutationId: '11111111-1111-4111-8111-111111111111',
});
});
test('fails closed on weak files, request-id drift and low-sensitive remote errors', async (t) => {
let mode = 'readiness-drift';
const server = await startServer((request, response) => {
request.resume();
request.on('end', () => {
if (mode === 'readiness-drift') {
jsonResponse(response, 200, null, { status: 'ready', widened: true });
return;
}
if (mode === 'drift') {
jsonResponse(response, 201, 'wrong-request', diagnoseResponse());
return;
}
jsonResponse(
response,
429,
request.headers['x-request-id'],
{ code: 'copilot_rate_limited', reason: 'private detail' },
{ 'retry-after': '30' },
);
});
});
t.after(() => server.close());
const directory = temporaryDirectory(t);
const configFile = configuration(directory, server.port);
const command = commandFile(directory, 'diagnose', 'diagnose', {
traceId: 'trace-1',
});
const credentialFile = privateFile(directory, 'credential', credential);
const paths = { configFile, commandFile: command, credentialFile };
await assert.rejects(probeClusterCopilotClientReadiness(configFile), {
code: 'QL3_CLUSTER_COPILOT_CLIENT_REQUEST_FAILED',
});
mode = 'drift';
await assert.rejects(executeClusterCopilotClient(paths), {
code: 'QL3_CLUSTER_COPILOT_CLIENT_REQUEST_FAILED',
});
mode = 'remote';
await assert.rejects(
executeClusterCopilotClient(paths),
(error) => {
assert.equal(error instanceof ClusterCopilotClientRemoteError, true);
assert.equal(error.statusCode, 429);
assert.equal(error.responseCode, 'copilot_rate_limited');
assert.equal(error.requestId, baseCommand.requestId);
assert.equal(error.retryAfterSeconds, 30);
assert.equal(JSON.stringify(error).includes('private detail'), false);
return true;
},
);
const cli = await runCli([
`--config=${configFile}`,
`--command=${command}`,
`--credential=${credentialFile}`,
]);
assert.equal(cli.status, 1);
assert.equal(cli.stdout, '');
assert.deepEqual(JSON.parse(cli.stderr), {
schemaVersion: 1,
component: 'qinglong3-cluster-copilot-client',
event: 'command_failed',
code: 'QL3_CLUSTER_COPILOT_CLIENT_REMOTE_REJECTED',
statusCode: 429,
responseCode: 'copilot_rate_limited',
requestId: baseCommand.requestId,
retryAfterSeconds: 30,
});
assert.equal(cli.stderr.includes(credential), false);
assert.equal(cli.stderr.includes('private detail'), false);
assert.equal(cli.stderr.includes(directory), false);
fs.chmodSync(credentialFile, 0o644);
await assert.rejects(executeClusterCopilotClient(paths), {
code: 'QL3_CLUSTER_COPILOT_CLIENT_CONFIG_INVALID',
});
});
test('CLI help, usage and explicit output success remain deterministic', async (t) => {
const help = await runCli(['--help']);
assert.equal(help.status, 0);
assert.match(help.stdout, /^Usage: ql3-copilot-client /);
assert.equal(help.stderr, '');
const usage = await runCli([]);
assert.equal(usage.status, 64);
assert.equal(usage.stdout, '');
assert.equal(
JSON.parse(usage.stderr).code,
'QL3_CLUSTER_COPILOT_CLIENT_USAGE_INVALID',
);
const server = await startServer((request, response) => {
request.resume();
request.on('end', () => {
jsonResponse(
response,
200,
request.headers['x-request-id'],
outputResponse(),
);
});
});
t.after(() => server.close());
const directory = temporaryDirectory(t);
const result = await runCli([
`--config=${configuration(directory, server.port)}`,
`--command=${commandFile(directory, 'output', 'output')}`,
`--credential=${privateFile(directory, 'credential', credential)}`,
]);
assert.equal(result.status, 0, result.stderr);
assert.equal(result.stderr, '');
const fact = JSON.parse(result.stdout);
assert.equal(fact.operation, 'output');
assert.equal(fact.result.result.text, 'diagnosis');
assert.equal(fact.component, 'qinglong3-cluster-copilot-client');
});
test('rejects response framing drift, oversized bodies, aborts and timeouts', async (t) => {
let mode = 'content-type';
const server = await startServer((request, response) => {
response.on('error', () => {});
request.resume();
request.on('end', () => {
if (mode === 'timeout') return;
if (mode === 'abort') {
response.writeHead(200, {
'content-type': 'application/json; charset=utf-8',
'x-request-id': request.headers['x-request-id'],
});
response.write('{"schema":');
response.destroy();
return;
}
if (mode === 'oversized') {
response.writeHead(200, {
'content-type': 'application/json; charset=utf-8',
'x-request-id': request.headers['x-request-id'],
});
response.end(Buffer.alloc(2 * 1024 * 1024 + 1, 0x20));
return;
}
const body = Buffer.from(JSON.stringify(diagnoseResponse()));
response.writeHead(201, {
'content-type': 'text/plain',
'content-length': String(body.length),
'x-request-id': request.headers['x-request-id'],
});
response.end(body);
});
});
t.after(() => server.close());
const directory = temporaryDirectory(t);
const caFile = privateFile(
directory,
'ca.pem',
fs.readFileSync(caFixture),
);
const configFile = privateFile(
directory,
'client.json',
JSON.stringify({
schema: CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
endpoint: `https://localhost:${server.port}/`,
servername: 'localhost',
caFile,
requestTimeoutMs: 1_000,
}),
);
const paths = {
configFile,
commandFile: commandFile(directory, 'diagnose', 'diagnose', {
traceId: 'trace-1',
}),
credentialFile: privateFile(directory, 'credential', credential),
};
for (const failureMode of [
'content-type',
'oversized',
'abort',
'timeout',
]) {
mode = failureMode;
await assert.rejects(executeClusterCopilotClient(paths), {
code: 'QL3_CLUSTER_COPILOT_CLIENT_REQUEST_FAILED',
});
}
});
packages/ql3-cluster-admin/test/productCli.test.cjs
+116 -2
View File
@@ -54,6 +54,9 @@ const {
const {
validateClusterAuthenticatedManagementClientConfiguration,
} = require('../dist/management-support/pluginPackageManagementClient.js');
const {
CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
} = require('../dist/copilot-client/client.js');
function runCli(args) {
return spawnSync(process.execPath, [cliPath, ...args], {
@@ -121,6 +124,41 @@ async function startReadinessServer(status) {
};
}
async function startCopilotReadinessServer(status) {
const server = createServer(
{
key: fs.readFileSync(localhostPrivateKeyFixture),
cert: fs.readFileSync(localhostCertificateFixture),
minVersion: 'TLSv1.3',
maxVersion: 'TLSv1.3',
},
(request, response) => {
assert.equal(request.method, 'GET');
assert.equal(request.url, '/readyz');
assert.equal(request.headers.authorization, undefined);
const body = Buffer.from(JSON.stringify({ status: status.value }));
response.writeHead(status.value === 'ready' ? 200 : 503, {
'content-type': 'application/json; charset=utf-8',
'content-length': String(body.length),
});
response.end(body);
},
);
await new Promise((resolvePromise, reject) => {
server.once('error', reject);
server.listen(0, '127.0.0.1', resolvePromise);
});
return {
port: server.address().port,
close: () =>
new Promise((resolvePromise, reject) => {
server.close((error) =>
error ? reject(error) : resolvePromise(),
);
}),
};
}
function privateFile(directory, name, contents) {
const filePath = path.join(directory, name);
fs.writeFileSync(filePath, contents, { mode: 0o600 });
@@ -193,6 +231,17 @@ function validContextFixture(t) {
'/api/v3/plugin-packages/management',
'forbidden',
);
const copilotConfig = privateFile(
directory,
'copilot-client.json',
JSON.stringify({
schema: CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
endpoint: 'https://copilot.example.test:8443/',
servername: 'copilot.example.test',
caFile,
requestTimeoutMs: 1_000,
}),
);
const kubeconfigFile = privateFile(
directory,
'kubeconfig.json',
@@ -236,6 +285,7 @@ function validContextFixture(t) {
}),
);
const commands = {
copilot: { configFile: copilotConfig },
package: { configFile: packageConfig },
'package-kubernetes': { configFile: packageConfig, kubernetesFile },
'worker-credential': {
@@ -283,7 +333,7 @@ function validContextFixture(t) {
test('catalog exposes only reviewed remote clients from the same package', () => {
assert.equal(manifest.bin['ql3-cluster-admin'], 'dist/product-cli/cli.js');
assert.equal(QINGLONG3_CLUSTER_PRODUCT_COMMANDS.length, 7);
assert.equal(QINGLONG3_CLUSTER_PRODUCT_COMMANDS.length, 8);
assert.equal(
new Set(QINGLONG3_CLUSTER_PRODUCT_COMMANDS.map(({ name }) => name)).size,
QINGLONG3_CLUSTER_PRODUCT_COMMANDS.length,
@@ -324,6 +374,7 @@ test('help and version are bounded installation-derived product facts', () => {
const help = qingLong3ClusterProductHelp();
assert.match(help, /^Usage: ql3-cluster-admin <command> \[arguments\]/);
assert.match(help, /\n run\s+retry or stop Runs/);
assert.match(help, /\n copilot\s+diagnose, inspect, read or cancel Runs/);
assert.match(help, /Server, migration, recovery, executor and key-custody/);
assert.equal(help.includes('plugin-package-manage'), false);
assert.equal(
@@ -558,8 +609,9 @@ test('validates the complete operator context offline without operational author
schemaVersion: 1,
component: 'qinglong3-cluster-product-cli',
event: 'context_valid',
commandCount: 7,
commandCount: 8,
commands: [
{ name: 'copilot', transport: 'https', clientCertificate: 'forbidden' },
{ name: 'package', transport: 'https', clientCertificate: 'forbidden' },
{
name: 'package-kubernetes',
@@ -671,6 +723,68 @@ test('probes a context with fixed read-only readiness semantics and exit status'
assert.equal(fact.commands[0].status, 'not_ready');
});
test('probes Copilot context through its separate unauthenticated readiness contract', async (t) => {
const status = { value: 'ready' };
const server = await startCopilotReadinessServer(status);
t.after(() => server.close());
const directory = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'ql3-copilot-probe-context-')),
);
t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
const caFile = privateFile(
directory,
'ca.pem',
fs.readFileSync(localhostCaFixture),
);
const configFile = privateFile(
directory,
'copilot.json',
JSON.stringify({
schema: CLUSTER_COPILOT_CLIENT_CONFIG_SCHEMA,
endpoint: `https://localhost:${server.port}/`,
servername: 'localhost',
caFile,
requestTimeoutMs: 1_000,
}),
);
const contextFile = privateFile(
directory,
'operator-context.json',
JSON.stringify({
schemaVersion: 1,
commands: { copilot: { configFile } },
}),
);
const ready = await runCliAsync([
'context',
'probe',
`--context=${contextFile}`,
]);
assert.equal(ready.status, 0, ready.stderr);
assert.equal(ready.stderr, '');
assert.deepEqual(JSON.parse(ready.stdout), {
schemaVersion: 1,
component: 'qinglong3-cluster-product-cli',
event: 'context_probed',
commandCount: 1,
commands: [{ name: 'copilot', transport: 'https', status: 'ready' }],
allReady: true,
requestMethod: 'GET',
requestPath: '/readyz',
mutation: false,
});
status.value = 'not_ready';
const notReady = await runCliAsync([
'context',
'probe',
`--context=${contextFile}`,
]);
assert.equal(notReady.status, 69);
assert.equal(JSON.parse(notReady.stdout).allReady, false);
});
test('context validation fails closed for invalid client configuration and syntax', (t) => {
const fixture = validContextFixture(t);
fs.writeFileSync(fixture.commands.run.configFile, '{}', { mode: 0o600 });