Files
pikasTech-unidesk/scripts/src/platform-infra-sub2api-codex/runtime.ts
T
2026-07-14 15:29:20 +02:00

3000 lines
157 KiB
TypeScript

import type { UniDeskConfig } from "../config";
import type { RenderedCliResult } from "../output";
import { desiredAccountNames } from "./accounts";
import { readCodexPoolConfig } from "./config";
import { isRecord, stringValue } from "./config-utils";
import { protectedManualAccountsForTarget } from "./public-exposure";
import { renderSub2ApiTempUnschedulableCredentials } from "./redaction";
import { boolField, compactCapture, parseJsonOutput, runRemoteCodexPoolScript } from "./remote";
import { renderedCliResult, renderTable } from "./render";
import { parseRuntimeOptions, type RuntimeOptions } from "./runtime-options";
import { codexPoolRuntimeTarget } from "./runtime-target";
export async function codexPoolRuntime(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown> | RenderedCliResult> {
const options = parseRuntimeOptions(args);
const pool = readCodexPoolConfig();
const target = codexPoolRuntimeTarget(options.targetId);
const selectedTemplate = options.template === null ? null : pool.runtime.templatesById[options.template];
if (options.template !== null && selectedTemplate === undefined) {
throw new Error(`unknown runtime template ${options.template}; available: ${pool.runtime.templates.map((item) => item.id).join(", ")}`);
}
const payload = {
action: options.action,
account: options.account,
accounts: options.accounts,
group: options.group,
allGroups: options.allGroups,
platform: options.platform,
pageToken: options.pageToken,
kind: options.kind,
confirm: options.confirm,
full: options.full,
since: options.since,
tail: options.tail,
groupName: pool.groupName,
adminEmailDefault: pool.adminEmailDefault,
managedAccountNames: desiredAccountNames(pool),
protectedManualAccountNames: protectedManualAccountsForTarget(pool, target).map((item) => item.accountName),
templates: pool.runtime.templates.map((template) => ({
id: template.id,
kind: template.kind,
description: template.description,
credentials: renderSub2ApiTempUnschedulableCredentials(template.tempUnschedulable),
})),
selectedTemplate: selectedTemplate === null || selectedTemplate === undefined ? null : {
id: selectedTemplate.id,
kind: selectedTemplate.kind,
credentials: renderSub2ApiTempUnschedulableCredentials(selectedTemplate.tempUnschedulable),
},
};
const result = await runRemoteCodexPoolScript(config, "runtime", runtimeScript(payload, target), target);
const parsed = parseJsonOutput(result.stdout);
const ok = result.exitCode === 0 && boolField(parsed, "ok", false);
if (!options.full && !options.raw && !options.json && (options.action === "apply" || options.action === "delete")) {
return renderRuntimeMutationResult(ok, options, parsed, result.exitCode);
}
if (options.raw) {
return {
ok,
action: "platform-infra-sub2api-codex-pool-runtime",
remote: compactCapture(result, { full: parsed === null || result.exitCode !== 0 }),
parsed,
valuesPrinted: false,
};
}
const response = {
ok,
action: "platform-infra-sub2api-codex-pool-runtime",
target: {
id: target.id,
route: target.route,
namespace: target.namespace,
runtimeMode: target.runtimeMode,
endpoint: target.serviceDns,
},
request: {
operation: options.action,
account: options.account,
accounts: options.accounts,
group: options.group,
allGroups: options.allGroups,
platform: options.platform,
pageToken: options.pageToken,
template: options.template,
kind: options.kind,
confirm: options.confirm,
since: options.action === "errors" || options.action === "infrastructure" ? options.since : undefined,
tail: options.action === "errors" || options.action === "infrastructure" ? options.tail : undefined,
},
runtime: compactRuntimeErrors(parsed, options),
remote: compactCapture(result, { full: result.exitCode !== 0 || parsed === null }),
disclosure: options.action === "list"
? {
get: "bun scripts/cli.ts platform-infra sub2api codex-pool runtime get --account <name-or-id>",
errors: "bun scripts/cli.ts platform-infra sub2api codex-pool runtime errors [--account <name-or-id>] [--since 24h]",
}
: undefined,
valuesPrinted: false,
};
return options.full || options.json ? response : renderRuntimeResult(response, options);
}
function compactRuntimeErrors(value: unknown, options: RuntimeOptions): unknown {
if (options.action !== "errors" || options.full || value === null || typeof value !== "object" || Array.isArray(value)) return value;
const runtime = value as Record<string, unknown>;
const allGroups = runtimeRecord(runtime.allGroups);
if (allGroups !== null) {
const nextPageToken = typeof allGroups.nextPageToken === "string" ? allGroups.nextPageToken : null;
const platformFlag = options.platform === null ? "" : ` --platform ${options.platform}`;
const baseCommand = `bun scripts/cli.ts platform-infra sub2api codex-pool runtime errors --target ${options.targetId} --since ${options.since}`;
return {
ok: runtime.ok,
operation: runtime.operation,
mutation: runtime.mutation,
endpoint: runtime.endpoint,
allGroups,
disclosure: {
group: `${baseCommand} --group <group-name-or-id>${platformFlag}`,
nextPage: nextPageToken === null ? null : `${baseCommand} --all-groups${platformFlag} --page-token ${nextPageToken}`,
note: "The overview uses stable group-id pagination. Use group drilldown for account root causes and request indexes.",
},
valuesPrinted: false,
};
}
const errors = runtimeRecord(runtime.errors);
if (errors === null) return value;
const effectiveness = runtimeRecord(errors.policyEffectiveness) ?? {};
const nativeOps = runtimeRecord(errors.nativeOps) ?? {};
const availability = runtimeRecord(nativeOps.accountAvailability) ?? {};
const availabilityGroup = runtimeRecord(availability.group) ?? {};
const concurrency = runtimeRecord(nativeOps.concurrency) ?? {};
const concurrencyGroup = runtimeRecord(concurrency.group) ?? {};
const sink = runtimeRecord(nativeOps.systemLogSink) ?? {};
const sourceStatus = runtimeRecord(effectiveness.sourceStatus) ?? {};
const drilldown = runtimeRecord(effectiveness.requestDrilldown) ?? {};
const requestIndexes = Object.fromEntries(Object.entries(drilldown).flatMap(([key, item]) => {
if (!Array.isArray(item)) return [];
const ids = item.filter((entry): entry is string => typeof entry === "string");
if (ids.length === 0) return [];
return [[key, {
count: ids.length,
sampleRequestId: ids[0],
moreAvailable: ids.length > 1,
}]];
}));
const baseCommand = `bun scripts/cli.ts platform-infra sub2api codex-pool runtime errors --target ${options.targetId} --since ${options.since}`;
const groupFlag = options.group === null ? "" : ` --group ${options.group}`;
const platformFlag = options.platform === null ? "" : ` --platform ${options.platform}`;
if (options.json && options.account === null) {
return {
ok: runtime.ok,
operation: runtime.operation,
mutation: runtime.mutation,
group: runtime.group,
endpoint: runtime.endpoint,
errors: {
window: errors.window,
nativeOps: {
source: nativeOps.source,
overview: nativeOps.overview,
accountQuality: compactAccountQuality(nativeOps.accountQuality, false),
accountAvailability: {
status: availability.status,
totalAccounts: availabilityGroup.total_accounts,
availableCount: availabilityGroup.available_count,
rateLimitCount: availabilityGroup.rate_limit_count,
errorCount: availabilityGroup.error_count,
unavailableAccounts: Array.isArray(availability.unavailableAccounts)
? availability.unavailableAccounts.map((item) => {
const account = runtimeRecord(item) ?? {};
return {
accountId: account.accountId,
accountName: account.accountName,
status: account.status,
reason: account.error,
};
})
: [],
},
concurrency: {
status: concurrency.status,
currentInUse: concurrencyGroup.current_in_use,
maxCapacity: concurrencyGroup.max_capacity,
loadPercentage: concurrencyGroup.load_percentage,
waitingInQueue: concurrencyGroup.waiting_in_queue,
},
},
},
disclosure: {
account: `${baseCommand}${groupFlag}${platformFlag} --account <name-or-id>`,
trace: "bun scripts/cli.ts platform-infra sub2api codex-pool trace --target <target-id> --request-id <request-id>",
note: "This group overview keeps every account name and quality summary. Use account drilldown for request indexes and complete failure evidence.",
},
valuesPrinted: false,
};
}
return {
ok: runtime.ok,
operation: runtime.operation,
mutation: runtime.mutation,
group: runtime.group,
endpoint: runtime.endpoint,
errors: {
window: errors.window,
nativeOps: {
source: nativeOps.source,
overview: nativeOps.overview,
responseHeaderTimeout: options.account === null || !options.json ? nativeOps.responseHeaderTimeout : undefined,
requestTiming: options.account === null || !options.json ? compactRequestTiming(nativeOps.requestTiming) : undefined,
customerErrors: compactCustomerErrors(nativeOps.customerErrors, options.account !== null || !options.json),
accountQuality: compactAccountQuality(nativeOps.accountQuality, options.account !== null || !options.json),
accountAvailability: {
status: availability.status,
totalAccounts: availabilityGroup.total_accounts,
availableCount: availabilityGroup.available_count,
rateLimitCount: availabilityGroup.rate_limit_count,
errorCount: availabilityGroup.error_count,
unavailableAccounts: Array.isArray(availability.unavailableAccounts)
? availability.unavailableAccounts.map((item) => {
const account = runtimeRecord(item) ?? {};
return {
accountId: account.accountId,
accountName: account.accountName,
status: account.status,
reason: account.error,
};
})
: [],
},
concurrency: {
status: concurrency.status,
currentInUse: concurrencyGroup.current_in_use,
maxCapacity: concurrencyGroup.max_capacity,
loadPercentage: concurrencyGroup.load_percentage,
waitingInQueue: concurrencyGroup.waiting_in_queue,
},
systemLogSink: {
status: sink.status,
queueDepth: sink.queueDepth,
droppedCount: sink.droppedCount,
writeFailedCount: sink.writeFailedCount,
},
},
rootCauseAnalysis: errors.rootCauseAnalysis,
policyEffectiveness: {
source: effectiveness.source,
scope: effectiveness.scope,
attribution: effectiveness.attribution,
assessment: effectiveness.assessment,
ruleMatches: effectiveness.ruleMatches,
failover: effectiveness.failover,
forwardFailure: effectiveness.forwardFailure,
selectFailedEvents: effectiveness.selectFailedEvents,
requestIndexes,
sourceStatus: {
status: sourceStatus.status,
fallbackUsed: sourceStatus.fallbackUsed,
fallbackReason: sourceStatus.fallbackReason,
},
},
},
disclosure: {
complete: `${baseCommand}${groupFlag}${platformFlag}${options.account === null ? "" : " --account <same-account-selector>"} --full`,
account: `${baseCommand}${groupFlag}${platformFlag} --account <name-or-id>`,
trace: "bun scripts/cli.ts platform-infra sub2api codex-pool trace --target <target-id> --request-id <request-id>",
note: "Use requestIndexes for trace drilldown and --full for the complete finite index.",
},
valuesPrinted: false,
};
}
function compactRequestTiming(value: unknown): unknown {
const timing = runtimeRecord(value);
if (timing === null) return value;
return {
status: timing.status,
reason: timing.reason,
source: timing.source,
ttft: timing.ttft,
duration: timing.duration,
observedFloorSeconds: timing.observedFloorSeconds,
attribution: timing.attribution,
};
}
function compactCustomerErrors(value: unknown, detailed: boolean): unknown {
const profile = runtimeRecord(value);
if (profile === null) return value;
return {
status: profile.status,
reason: profile.reason,
source: profile.source,
total: profile.total,
modelBuckets: runtimeRecords(profile.modelBuckets).slice(0, 5),
accountBuckets: detailed ? runtimeRecords(profile.accountBuckets).slice(0, 5) : undefined,
statusBuckets: runtimeRecords(profile.statusBuckets).slice(0, 5),
streamBuckets: runtimeRecords(profile.streamBuckets).slice(0, 5),
};
}
function compactAccountQuality(value: unknown, detailed: boolean): unknown {
const quality = runtimeRecord(value);
if (quality === null) return value;
return {
source: quality.source,
scope: quality.scope,
window: quality.window,
scoreRule: (() => {
const rule = runtimeRecord(quality.scoreRule) ?? {};
return detailed ? {
reliabilityWeight: rule.reliabilityWeight,
latencyWeight: rule.latencyWeight,
availabilityWeight: rule.availabilityWeight,
failureDefinition: rule.failureDefinition,
missingMetric: rule.missingMetric,
} : {
reliabilityWeight: rule.reliabilityWeight,
latencyWeight: rule.latencyWeight,
availabilityWeight: rule.availabilityWeight,
};
})(),
accounts: runtimeRecords(quality.accounts).map((account) => detailed ? {
accountId: account.accountId,
accountName: account.accountName,
status: account.status,
currentlyAvailable: account.currentlyAvailable,
score: account.score,
grade: account.grade,
assessment: account.assessment,
confidence: account.confidence,
observedAttempts: account.observedAttempts,
successRequests: account.successRequests,
failureRequests: account.failureRequests,
failureRate: account.failureRate,
firstTokenSamples: account.firstTokenSamples,
firstTokenCoverage: account.firstTokenCoverage,
ttftP95Ms: account.ttftP95Ms,
customerErrorRequests: account.customerErrorRequests,
failoverRequests: account.failoverRequests,
failoverRecovered: account.failoverRecovered,
failoverFailed: account.failoverFailed,
sameAccountRetryEvents: account.sameAccountRetryEvents,
tempUnschedulableEvents: account.tempUnschedulableEvents,
forwardFailedRequests: account.forwardFailedRequests,
reasons: account.reasons,
} : {
accountId: account.accountId,
accountName: account.accountName,
currentlyAvailable: account.currentlyAvailable,
score: account.score,
grade: account.grade,
confidence: account.confidence,
observedAttempts: account.observedAttempts,
failureRate: account.failureRate,
ttftP95Ms: account.ttftP95Ms,
customerErrorRequests: account.customerErrorRequests,
failoverRequests: account.failoverRequests,
failoverRecovered: account.failoverRecovered,
failoverFailed: account.failoverFailed,
}),
attribution: detailed ? quality.attribution : undefined,
};
}
function runtimeRecord(value: unknown): Record<string, unknown> | null {
return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
}
function runtimeRecords(value: unknown): Record<string, unknown>[] {
return Array.isArray(value) ? value.flatMap((item) => {
const record = runtimeRecord(item);
return record === null ? [] : [record];
}) : [];
}
function runtimeText(value: unknown): string {
if (value === null || value === undefined || value === "") return "-";
if (typeof value === "boolean") return value ? "true" : "false";
if (Array.isArray(value)) return value.map(runtimeText).join(",");
return String(value).replace(/\s+/gu, " ").trim();
}
function runtimeShort(value: unknown, max = 42): string {
const text = runtimeText(value);
return text.length <= max ? text : `${text.slice(0, max - 1)}…`;
}
function runtimeRate(value: unknown): string {
return typeof value === "number" && Number.isFinite(value) ? `${(value * 100).toFixed(2)}%` : "-";
}
function renderRuntimeResult(response: Record<string, unknown>, options: RuntimeOptions): RenderedCliResult {
const runtime = runtimeRecord(response.runtime) ?? {};
const ok = response.ok === true;
const lines: string[] = [];
if (options.action === "errors") renderRuntimeErrors(lines, runtime, options);
else if (options.action === "infrastructure") renderRuntimeInfrastructure(lines, runtime);
else if (options.action === "list") renderRuntimeAccountList(lines, runtime);
else if (options.action === "get") renderRuntimeAccountDetail(lines, runtime);
else renderRuntimeMutation(lines, runtime, options);
lines.push("", `JSON: bun scripts/cli.ts platform-infra sub2api codex-pool runtime ${options.action} <same-options> --json`);
return renderedCliResult(ok, `platform-infra sub2api codex-pool runtime ${options.action}`, lines.join("\n"));
}
function renderRuntimeErrors(lines: string[], runtime: Record<string, unknown>, options: RuntimeOptions): void {
const allGroups = runtimeRecord(runtime.allGroups);
if (allGroups !== null) {
const totals = runtimeRecord(allGroups.pageTotals) ?? {};
const timeout = runtimeRecord(allGroups.responseHeaderTimeout) ?? {};
const timing = runtimeRecord(allGroups.requestTiming) ?? {};
const groups = runtimeRecords(allGroups.groups);
lines.push(`SUB2API RUNTIME ERRORS ok=${runtime.ok === true ? "true" : "false"} scope=all-groups since=${options.since}`);
lines.push(renderTable([
["GROUPS", "RETURNED", "REQUESTS", "CLIENT_ERR", "ERR_RATE", "UPSTREAM_ERR", "UP_RATE", "ATTENTION"],
[
runtimeText(allGroups.totalGroups), runtimeText(allGroups.returnedGroups), runtimeText(totals.requestCount),
runtimeText(totals.errorCount), runtimeRate(totals.errorRate), runtimeText(totals.upstreamErrorCount),
runtimeRate(totals.upstreamErrorRate), runtimeText(totals.groupsNeedingAttention),
],
]));
lines.push(renderTable([
["OPENAI_HEADER_TIMEOUT", "SOURCE", "GENERIC_TIMEOUT", "GENERIC_APPLIES_TO_OPENAI", "PAGE_MAX_TTFT_P99_MS", "OBSERVED_FLOOR_SECONDS"],
[
timeout.disabled === true ? "disabled" : `${runtimeText(timeout.effectiveSeconds)}s`, runtimeText(timeout.source),
`${runtimeText(timeout.genericResponseHeaderTimeoutSeconds)}s`, runtimeText(timeout.genericAppliesToOpenAI),
runtimeText(timing.maxTtftP99Ms), runtimeText(timing.observedFloorSeconds),
],
]));
if (typeof timing.attribution === "string") lines.push(`NOTE: ${timing.attribution}`);
lines.push("", "GROUPS");
lines.push(renderTable([
["NAME", "ID", "PLATFORM", "STATUS", "REQ", "CLIENT_ERR", "ERR_RATE", "UP_ERR", "UP_RATE", "TTFT_P99_MS", "DUR_P99_MS", "ACCOUNTS", "AVAILABLE", "USE/CAP", "QUEUE", "ATTN"],
...groups.map((group) => [
runtimeShort(group.groupName, 28), runtimeText(group.groupId), runtimeText(group.platform), runtimeText(group.status),
runtimeText(group.requestCount), runtimeText(group.errorCount), runtimeRate(group.errorRate), runtimeText(group.upstreamErrorCount),
runtimeRate(group.upstreamErrorRate), runtimeText(group.ttftP99Ms), runtimeText(group.durationP99Ms), runtimeText(group.totalAccounts), runtimeText(group.availableCount),
`${runtimeText(group.currentInUse)}/${runtimeText(group.maxCapacity)}`, runtimeText(group.waitingInQueue), group.needsAttention === true ? "yes" : "no",
]),
]));
const disclosure = runtimeRecord(runtime.disclosure) ?? {};
lines.push("", `Detail: ${runtimeText(disclosure.group)}`);
if (typeof disclosure.nextPage === "string") lines.push(`Next: ${disclosure.nextPage}`);
return;
}
const group = runtimeRecord(runtime.group) ?? {};
const errors = runtimeRecord(runtime.errors) ?? {};
const nativeOps = runtimeRecord(errors.nativeOps) ?? {};
const overview = runtimeRecord(nativeOps.overview) ?? {};
const availability = runtimeRecord(nativeOps.accountAvailability) ?? {};
const concurrency = runtimeRecord(nativeOps.concurrency) ?? {};
const timeout = runtimeRecord(nativeOps.responseHeaderTimeout) ?? {};
const timing = runtimeRecord(nativeOps.requestTiming) ?? {};
const ttft = runtimeRecord(timing.ttft) ?? {};
const duration = runtimeRecord(timing.duration) ?? {};
const customerErrors = runtimeRecord(nativeOps.customerErrors) ?? {};
const accountQuality = runtimeRecord(nativeOps.accountQuality) ?? {};
lines.push(`SUB2API RUNTIME ERRORS ok=${runtime.ok === true ? "true" : "false"} scope=group since=${options.since}`);
lines.push(renderTable([
["GROUP", "ID", "PLATFORM", "REQUESTS", "CLIENT_ERR", "ERR_RATE", "UP_RATE", "BUSINESS_LIMIT", "HEALTH"],
[
runtimeShort(group.name, 30), runtimeText(group.id), runtimeText(group.platform), runtimeText(overview.requestCount),
runtimeText(overview.errorCount), runtimeRate(overview.errorRate), runtimeRate(overview.upstreamErrorRate),
runtimeText(overview.businessLimitedCount), runtimeText(overview.healthScore),
],
]));
lines.push(renderTable([
["OPENAI_HEADER_TIMEOUT", "SOURCE", "TTFT_P50_MS", "TTFT_P95_MS", "TTFT_P99_MS", "TTFT_MAX_MS", "DURATION_P99_MS", "DURATION_MAX_MS"],
[
timeout.disabled === true ? "disabled" : `${runtimeText(timeout.effectiveSeconds)}s`, runtimeText(timeout.source),
runtimeText(ttft.p50_ms), runtimeText(ttft.p95_ms), runtimeText(ttft.p99_ms), runtimeText(ttft.max_ms),
runtimeText(duration.p99_ms), runtimeText(duration.max_ms),
],
]));
const candidates = runtimeRecords(timing.candidates);
if (candidates.length > 0) lines.push("", "TIMEOUT CANDIDATES", renderTable([
["SECONDS", "OBSERVED_FLOOR", "TTFT_P99_HEADROOM_MS", "TTFT_P99_BELOW", "ALL_DUR>=", "ERROR_DUR>=", "ASSESSMENT"],
...candidates.map((candidate) => [
runtimeText(candidate.seconds), runtimeText(candidate.observedFloor), runtimeText(candidate.ttftP99HeadroomMs), runtimeText(candidate.ttftP99Below),
runtimeText(candidate.totalDurationAtLeastCount), runtimeText(candidate.errorDurationAtLeastCount), runtimeText(candidate.assessment),
]),
]));
if (typeof timing.attribution === "string") lines.push(`NOTE: ${timing.attribution}`);
const modelTiming = runtimeRecords(timing.modelTiming);
if (modelTiming.length > 0) lines.push("", "MODEL TIMING", renderTable([
["MODEL", "REQUESTS", "FIRST_TOKEN_ROWS", "COVERAGE", "AVG_TTFT_MS", "AVG_DURATION_MS", "AVG_TOKENS_S"],
...modelTiming.map((item) => [runtimeShort(item.model, 32), runtimeText(item.requestCount), runtimeText(item.requestsWithFirstToken), runtimeRate(item.firstTokenCoverage), runtimeText(item.avgFirstTokenMs), runtimeText(item.avgDurationMs), runtimeText(item.avgTokensPerSec)]),
]));
const qualityAccounts = runtimeRecords(accountQuality.accounts);
if (qualityAccounts.length > 0) {
lines.push("", "ACCOUNT QUALITY", renderTable([
["ACCOUNT", "ID", "SCORE", "GRADE", "QUALITY", "CONF", "ATTEMPTS", "SUCCESS", "FAIL", "FAIL_RATE", "TTFT_P95_MS", "TTFT_N", "AVAILABLE", "REASONS"],
...qualityAccounts.map((item) => [
runtimeText(item.accountName), runtimeText(item.accountId), runtimeText(item.score), runtimeText(item.grade), runtimeText(item.assessment), runtimeText(item.confidence),
runtimeText(item.observedAttempts), runtimeText(item.successRequests), runtimeText(item.failureRequests), runtimeRate(item.failureRate),
runtimeText(item.ttftP95Ms), runtimeText(item.firstTokenSamples), runtimeText(item.currentlyAvailable),
runtimeShort(Array.isArray(item.reasons) ? item.reasons.join(",") : "", 58),
]),
]));
const controlRows = qualityAccounts.filter((item) => [
item.failoverRequests, item.sameAccountRetryEvents, item.tempUnschedulableEvents, item.forwardFailedRequests, item.customerErrorRequests,
].some((value) => typeof value === "number" && value > 0));
if (controlRows.length > 0) lines.push("ACCOUNT FAILURE CONTROL", renderTable([
["ACCOUNT", "ID", "FAILOVER", "RECOVERED", "FAILED", "MISSING", "RETRY", "TEMP_UNSCHED", "FORWARD_FAIL", "CUSTOMER_ERR"],
...controlRows.map((item) => [
runtimeText(item.accountName), runtimeText(item.accountId), runtimeText(item.failoverRequests), runtimeText(item.failoverRecovered),
runtimeText(item.failoverFailed), runtimeText(item.failoverOutcomeMissing), runtimeText(item.sameAccountRetryEvents),
runtimeText(item.tempUnschedulableEvents), runtimeText(item.forwardFailedRequests), runtimeText(item.customerErrorRequests),
]),
]));
const scoreRule = runtimeRecord(accountQuality.scoreRule) ?? {};
lines.push(`SCORE RULE reliability=${runtimeText(scoreRule.reliabilityWeight)} latency=${runtimeText(scoreRule.latencyWeight)} availability=${runtimeText(scoreRule.availabilityWeight)} confidence=separate`);
if (typeof accountQuality.attribution === "string") lines.push(`NOTE: ${accountQuality.attribution}`);
}
const errorModels = runtimeRecords(customerErrors.modelBuckets);
const errorAccounts = runtimeRecords(customerErrors.accountBuckets);
const errorDimensions = [
...runtimeRecords(customerErrors.statusBuckets).map((item) => ["status", runtimeText(item.value), runtimeText(item.count)]),
...runtimeRecords(customerErrors.streamBuckets).map((item) => ["mode", runtimeText(item.value), runtimeText(item.count)]),
...runtimeRecords(customerErrors.phaseBuckets).map((item) => ["phase", runtimeText(item.value), runtimeText(item.count)]),
...runtimeRecords(customerErrors.endpointBuckets).map((item) => ["endpoint", runtimeText(item.value), runtimeText(item.count)]),
...runtimeRecords(customerErrors.pathBuckets).map((item) => ["path", runtimeText(item.value), runtimeText(item.count)]),
];
if (errorModels.length > 0) lines.push("", `CUSTOMER ERRORS source=${runtimeText(customerErrors.source)} total=${runtimeText(customerErrors.total)}`, renderTable([
["MODEL", "COUNT"], ...errorModels.map((item) => [runtimeShort(item.value, 40), runtimeText(item.count)]),
]));
if (errorAccounts.length > 0) lines.push(renderTable([
["ACCOUNT_ID", "COUNT"], ...errorAccounts.map((item) => [runtimeText(item.value), runtimeText(item.count)]),
]));
if (errorDimensions.length > 0) lines.push(renderTable([["DIMENSION", "VALUE", "COUNT"], ...errorDimensions]));
const errorSamples = runtimeRecords(customerErrors.samples);
if (errorSamples.length > 0) lines.push("CUSTOMER ERROR INDEX", renderTable([
["REQUEST_ID", "MODEL", "STATUS", "ACCOUNT", "MODE", "PATH", "MESSAGE"],
...errorSamples.map((item) => [runtimeText(item.requestId), runtimeShort(item.model, 26), runtimeText(item.statusCode), runtimeText(item.accountId), runtimeText(item.mode), runtimeShort(item.path, 34), runtimeShort(item.message, 48)]),
]));
lines.push(renderTable([
["ACCOUNTS", "AVAILABLE", "RATE_LIMIT", "ERROR", "IN_USE", "CAPACITY", "QUEUE"],
[
runtimeText(availability.totalAccounts), runtimeText(availability.availableCount), runtimeText(availability.rateLimitCount),
runtimeText(availability.errorCount), runtimeText(concurrency.currentInUse), runtimeText(concurrency.maxCapacity), runtimeText(concurrency.waitingInQueue),
],
]));
const unavailable = runtimeRecords(availability.unavailableAccounts);
if (unavailable.length > 0) {
lines.push("", "UNAVAILABLE ACCOUNTS", renderTable([
["ACCOUNT", "ID", "STATUS", "REASON"],
...unavailable.map((account) => [runtimeShort(account.accountName, 36), runtimeText(account.accountId), runtimeText(account.status), runtimeShort(account.reason, 70)]),
]));
}
const root = runtimeRecord(errors.rootCauseAnalysis) ?? {};
const causes = runtimeRecords(root.causeBuckets);
if (causes.length > 0) {
lines.push("", `ROOT CAUSES scope=${runtimeText(root.scope)}`, renderTable([
["CAUSE", "COUNT"],
...causes.map((cause) => [runtimeText(cause.cause), runtimeText(cause.count)]),
]));
}
const policy = runtimeRecord(errors.policyEffectiveness) ?? {};
const rules = runtimeRecord(policy.ruleMatches) ?? {};
const failover = runtimeRecord(policy.failover) ?? {};
const forward = runtimeRecord(policy.forwardFailure) ?? {};
lines.push("", `POLICY EFFECTIVENESS scope=${runtimeText(policy.scope)}`);
lines.push(renderTable([
["TEMP_UNSCHED", "FAILOVER", "SUCCESS", "FAILED", "MISSING", "FORWARD_200", "FORWARD_ERR", "SELECT_FAIL"],
[
runtimeText(rules.tempUnschedulableEvents), runtimeText(failover.requests), runtimeText(failover.succeeded),
runtimeText(failover.failed), runtimeText(failover.outcomeMissing), runtimeText(forward.withHttpSuccess),
runtimeText(forward.withHttpError), runtimeText(policy.selectFailedEvents),
],
]));
const indexes = runtimeRecord(policy.requestIndexes) ?? {};
const indexRows = Object.entries(indexes).flatMap(([kind, value]) => {
const item = runtimeRecord(value);
return item === null ? [] : [[kind, runtimeText(item.count), runtimeText(item.sampleRequestId), item.moreAvailable === true ? "yes" : "no"]];
});
if (indexRows.length > 0) lines.push("", "REQUEST INDEX", renderTable([["TYPE", "COUNT", "SAMPLE_REQUEST_ID", "MORE"], ...indexRows]));
if (typeof root.attribution === "string") lines.push("", `NOTE: ${root.attribution}`);
if (typeof policy.attribution === "string") lines.push(`NOTE: ${policy.attribution}`);
const disclosure = runtimeRecord(runtime.disclosure) ?? {};
lines.push("", `Account: ${runtimeText(disclosure.account)}`, `Trace: ${runtimeText(disclosure.trace)}`, `Full: ${runtimeText(disclosure.complete)}`);
}
function renderRuntimeAccountList(lines: string[], runtime: Record<string, unknown>): void {
const group = runtimeRecord(runtime.group) ?? {};
const accounts = runtimeRecords(runtime.accounts);
lines.push(`SUB2API RUNTIME ACCOUNTS ok=${runtime.ok === true ? "true" : "false"} group=${runtimeText(group.name)} count=${accounts.length}`);
lines.push(renderTable([
["NAME", "ID", "MANAGEMENT", "STATUS", "SCHED", "CONCURRENCY", "LOAD", "PRIORITY", "TEMP_RULES", "TEMPLATE"],
...accounts.map((account) => {
const temp = runtimeRecord(account.tempUnschedulable) ?? {};
return [
runtimeShort(account.accountName, 36), runtimeText(account.accountId), runtimeText(account.management), runtimeText(account.status),
runtimeText(account.schedulable), `${runtimeText(account.currentConcurrency)}/${runtimeText(account.concurrencyLimit)}`,
runtimeText(account.loadFactor), runtimeText(account.priority), runtimeText(temp.ruleCount), runtimeText(account.matchingTemplate),
];
}),
]));
lines.push("", "Detail: bun scripts/cli.ts platform-infra sub2api codex-pool runtime get --account <name-or-id>");
}
function renderRuntimeAccountDetail(lines: string[], runtime: Record<string, unknown>): void {
const account = runtimeRecord(runtime.account) ?? {};
const temp = runtimeRecord(account.tempUnschedulable) ?? {};
lines.push(`SUB2API RUNTIME ACCOUNT ok=${runtime.ok === true ? "true" : "false"}`);
lines.push(renderTable([
["NAME", "ID", "MANAGEMENT", "STATUS", "SCHED", "PROXY", "CONCURRENCY", "LOAD", "PRIORITY", "TEMPLATE"],
[
runtimeShort(account.accountName, 42), runtimeText(account.accountId), runtimeText(account.management), runtimeText(account.status),
runtimeText(account.schedulable), runtimeText(account.proxyId), `${runtimeText(account.currentConcurrency)}/${runtimeText(account.concurrencyLimit)}`,
runtimeText(account.loadFactor), runtimeText(account.priority), runtimeText(account.matchingTemplate),
],
]));
lines.push(renderTable([
["TEMP_ENABLED", "RULES", "STATUS_CODES", "POOL_MODE"],
[runtimeText(temp.enabled), runtimeText(temp.ruleCount), runtimeText(temp.statusCodes), runtimeText(account.poolModeEnabled)],
]));
}
function renderRuntimeInfrastructure(lines: string[], runtime: Record<string, unknown>): void {
const account = runtimeRecord(runtime.account) ?? {};
const infra = runtimeRecord(runtime.infrastructure) ?? {};
const proxy = runtimeRecord(infra.accountProxy) ?? {};
const app = runtimeRecord(infra.appRuntime) ?? {};
const route = runtimeRecord(infra.defaultRoute) ?? {};
const dns = runtimeRecord(infra.dns) ?? {};
const sockets = runtimeRecord(infra.proxySockets) ?? {};
const listener = runtimeRecord(infra.proxyEndpointOwner) ?? {};
const serviceLogs = runtimeRecord(infra.proxyServiceEvidence) ?? {};
const logs = runtimeRecord(infra.recentNetworkEvidence) ?? {};
lines.push(`SUB2API RUNTIME INFRASTRUCTURE ok=${runtime.ok === true ? "true" : "false"} since=${runtimeText(logs.window)}`);
if (runtime.ok !== true && typeof runtime.error === "string") lines.push(`ERROR: ${runtime.error}`);
lines.push(renderTable([
["ACCOUNT", "ID", "TYPE", "STATUS", "SCHED", "PROXY_ID", "PROXY_NAME", "PROTOCOL", "PROXY_ENDPOINT", "PROXY_STATUS"],
[
runtimeText(account.accountName), runtimeText(account.accountId), runtimeText(account.type), runtimeText(account.status),
runtimeText(account.schedulable), runtimeText(proxy.id), runtimeText(proxy.name), runtimeText(proxy.protocol),
`${runtimeText(proxy.host)}:${runtimeText(proxy.port)}`, runtimeText(proxy.status),
],
]));
lines.push("", "APP RUNTIME", renderTable([
["MODE", "CONTAINER", "STATE", "RESTARTS", "OOM", "NETWORK_MODE", "STARTED_AT"],
[
runtimeText(app.runtimeMode), runtimeText(app.name), runtimeText(app.state), runtimeText(app.restartCount),
runtimeText(app.oomKilled), runtimeText(app.networkMode), runtimeText(app.startedAt),
],
]));
const envRows = runtimeRecords(infra.proxyEnvironment);
if (envRows.length > 0) lines.push("PROXY ENVIRONMENT", renderTable([
["VARIABLE", "PRESENT", "SCHEME", "HOST", "PORT", "CREDENTIALS_REDACTED"],
...envRows.map((item) => [runtimeText(item.name), runtimeText(item.present), runtimeText(item.scheme), runtimeText(item.host), runtimeText(item.port), runtimeText(item.credentialsRedacted)]),
]));
lines.push("NETWORK", renderTable([
["DEFAULT_IFACE", "DEFAULT_GATEWAY", "DNS_NAMESERVERS", "DNS_SEARCH", "PROXY_RESOLVED_IPS", "SOCKETS", "ESTABLISHED"],
[
runtimeText(route.interface), runtimeText(route.gateway), runtimeText(dns.nameservers), runtimeText(dns.search),
runtimeText(sockets.resolvedIps), runtimeText(sockets.total), runtimeText(sockets.established),
],
]));
const listeners = runtimeRecords(listener.listeners);
if (listeners.length > 0) lines.push("PROXY ENDPOINT OWNER", renderTable([
["LISTEN", "PROCESS", "PID", "UNIT", "ACTIVE", "RESTARTS", "TCP_PEERS", "UDP_PEERS"],
...listeners.map((item) => [runtimeText(item.localEndpoint), runtimeText(item.processName), runtimeText(item.pid), runtimeText(item.systemdUnit), runtimeText(item.activeState), runtimeText(item.serviceRestarts), runtimeText(item.establishedPeers), runtimeText(item.udpPeers)]),
]));
const serviceBuckets = runtimeRecords(serviceLogs.buckets);
if (serviceBuckets.length > 0) lines.push("PROXY SERVICE EVIDENCE", renderTable([
["CAUSE", "COUNT"],
...serviceBuckets.map((item) => [runtimeText(item.cause), runtimeText(item.count)]),
]));
const serviceSamples = runtimeRecords(serviceLogs.samples);
if (serviceSamples.length > 0) lines.push("PROXY SERVICE INDEX", renderTable([
["AT", "CAUSE", "MESSAGE"],
...serviceSamples.map((item) => [runtimeText(item.at), runtimeText(item.cause), runtimeShort(item.message, 100)]),
]));
const interfaces = runtimeRecords(infra.interfaces);
if (interfaces.length > 0) lines.push("INTERFACES", renderTable([
["NAME", "RX_BYTES", "TX_BYTES", "RX_ERRORS", "TX_ERRORS", "RX_DROPPED", "TX_DROPPED"],
...interfaces.map((item) => [runtimeText(item.name), runtimeText(item.rxBytes), runtimeText(item.txBytes), runtimeText(item.rxErrors), runtimeText(item.txErrors), runtimeText(item.rxDropped), runtimeText(item.txDropped)]),
]));
const buckets = runtimeRecords(logs.buckets);
lines.push("", `RECENT NETWORK EVIDENCE matched=${runtimeText(logs.matchedEvents)} source=${runtimeText(logs.source)}`);
if (buckets.length > 0) lines.push(renderTable([
["CAUSE", "COUNT"],
...buckets.map((item) => [runtimeText(item.cause), runtimeText(item.count)]),
]));
const samples = runtimeRecords(logs.samples);
if (samples.length > 0) lines.push("EVIDENCE INDEX", renderTable([
["AT", "REQUEST_ID", "CAUSE", "STATUS", "MESSAGE"],
...samples.map((item) => [runtimeText(item.at), runtimeText(item.requestId), runtimeText(item.cause), runtimeText(item.statusCode), runtimeShort(item.message, 90)]),
]));
if (typeof infra.attribution === "string") lines.push(`NOTE: ${infra.attribution}`);
}
function renderRuntimeMutation(lines: string[], runtime: Record<string, unknown>, options: RuntimeOptions): void {
const plans = runtimeRecords(runtime.plans);
if (plans.length > 0) {
const summary = runtimeRecord(runtime.summary) ?? {};
lines.push(`SUB2API RUNTIME ${options.action.toUpperCase()} ok=${runtime.ok === true ? "true" : "false"} mode=${runtimeText(runtime.mode)} mutation=${runtimeText(runtime.mutation)}`);
lines.push(renderTable([
["ACCOUNT", "ID", "CHANGE", "TEMPLATE", "BEFORE_RULES", "AFTER_RULES", "BEFORE_CODES", "AFTER_CODES", "RESULT"],
...plans.map((plan) => {
const account = runtimeRecord(plan.account) ?? {};
const before = runtimeRecord(plan.before) ?? {};
const after = runtimeRecord(plan.after) ?? {};
return [
runtimeShort(account.accountName, 36), runtimeText(account.accountId), runtimeText(plan.change), runtimeText(plan.template),
runtimeText(before.ruleCount), runtimeText(after.ruleCount), runtimeText(before.statusCodes), runtimeText(after.statusCodes), runtimeText(plan.result),
];
}),
]));
lines.push(`SUMMARY selected=${runtimeText(summary.selected)} changed=${runtimeText(summary.changed)} succeeded=${runtimeText(summary.succeeded)} failed=${runtimeText(summary.failed)}`);
return;
}
const plan = runtimeRecord(runtime.plan) ?? {};
const account = runtimeRecord(plan.account) ?? runtimeRecord(runtime.account) ?? {};
const before = runtimeRecord(plan.before) ?? {};
const after = runtimeRecord(plan.after) ?? {};
lines.push(`SUB2API RUNTIME ${options.action.toUpperCase()} ok=${runtime.ok === true ? "true" : "false"} mode=${runtimeText(runtime.mode)} mutation=${runtimeText(runtime.mutation)}`);
lines.push(renderTable([
["ACCOUNT", "ID", "CHANGE", "TEMPLATE", "BEFORE_RULES", "AFTER_RULES", "BEFORE_CODES", "AFTER_CODES"],
[
runtimeShort(account.accountName, 36), runtimeText(account.accountId), runtimeText(plan.change), runtimeText(plan.template),
runtimeText(before.ruleCount), runtimeText(after.ruleCount), runtimeText(before.statusCodes), runtimeText(after.statusCodes),
],
]));
}
function renderRuntimeMutationResult(ok: boolean, options: RuntimeOptions, parsed: Record<string, unknown> | null, exitCode: number): RenderedCliResult {
const summary = parsed !== null && isRecord(parsed.summary) ? parsed.summary : {};
const items = parsed !== null && Array.isArray(parsed.items) ? parsed.items.filter(isRecord) : [];
const lines = [[
"SUB2API RUNTIME",
`operation=${options.action}`,
`mode=${stringValue(parsed?.mode) ?? (options.confirm ? "confirmed" : "dry-run")}`,
`ok=${ok ? "true" : "false"}`,
`selected=${summary.selected ?? items.length}`,
`changed=${summary.changed ?? "?"}`,
`writeSucceeded=${summary.writeSucceeded ?? "?"}`,
`writeFailed=${summary.writeFailed ?? "?"}`,
`reconciled=${summary.reconciled ?? "?"}`,
`mismatched=${summary.mismatched ?? "?"}`,
].join(" ")];
if (items.length > 0) {
lines.push(renderTable([
["ACCOUNT", "ID", "CHANGE", "WRITE", "RECONCILED", "BEFORE", "DESIRED", "ACTUAL", "BEFORE_CODES", "DESIRED_CODES", "ACTUAL_CODES", "FAILURE"],
...items.map((item) => {
const write = isRecord(item.write) ? item.write : {};
const before = isRecord(item.before) ? item.before : {};
const desired = isRecord(item.desired) ? item.desired : {};
const actual = isRecord(item.actual) ? item.actual : {};
const codes = (value: unknown): string => Array.isArray(value) ? value.join(",") || "-" : "-";
return [
stringValue(item.accountName) ?? "-",
String(item.accountId ?? "-"),
stringValue(item.change) ?? "-",
stringValue(write.status) ?? "-",
item.reconciled === true ? "true" : item.reconciled === false ? "false" : "-",
String(before.ruleCount ?? "-"),
String(desired.ruleCount ?? "-"),
String(actual.ruleCount ?? "-"),
codes(before.statusCodes),
codes(desired.statusCodes),
codes(actual.statusCodes),
stringValue(write.error) ?? stringValue(item.reconcileError) ?? "-",
];
}),
]));
} else if (parsed !== null && typeof parsed.error === "string") {
lines.push(`ERROR ${parsed.error}`);
} else if (parsed === null) {
lines.push(`ERROR runtime output unavailable remote_exit=${exitCode}`);
}
return {
ok,
command: "platform-infra sub2api codex-pool runtime",
renderedText: lines.join("\n"),
contentType: "text/plain",
projection: parsed ?? undefined,
};
}
function pyJson(value: unknown): string {
return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
}
function runtimeScript(payload: unknown, target: ReturnType<typeof codexPoolRuntimeTarget>): string {
return `
set -u
python3 - <<'PY'
import json
import re
import socket
import struct
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from urllib.parse import quote, urlsplit
RUNTIME_MODE = ${pyJson(target.runtimeMode)}
NAMESPACE = ${pyJson(target.namespace)}
HOST_DOCKER_APP_PORT = ${pyJson(target.hostDockerAppPort)}
HOST_DOCKER_ENV_PATH = ${pyJson(target.hostDockerEnvPath)}
HOST_DOCKER_APP_CONTAINER = "sub2api-app"
APP_SECRET_NAME = ${pyJson(target.appSecretName)}
PAYLOAD = ${pyJson(payload)}
def run(cmd, input_bytes=None):
return subprocess.run(cmd, input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
def text(value, limit=1000):
if isinstance(value, bytes):
value = value.decode("utf-8", errors="replace")
return str(value)[-limit:]
def docker(args):
proc = run(["docker", *args])
if proc.returncode == 0:
return proc
sudo_proc = run(["sudo", "-n", "docker", *args])
return sudo_proc if sudo_proc.returncode == 0 else proc
def kubectl(args, input_bytes=None):
return run(["kubectl", *args], input_bytes)
def kube_json(args, label):
proc = kubectl([*args, "-o", "json"])
if proc.returncode != 0:
raise RuntimeError(f"{label} failed: {text(proc.stderr)}")
return json.loads(proc.stdout.decode("utf-8"))
def read_host_env():
try:
with open(HOST_DOCKER_ENV_PATH, "r", encoding="utf-8") as handle:
lines = handle.read().splitlines()
except PermissionError:
proc = run(["sudo", "-n", "cat", HOST_DOCKER_ENV_PATH])
if proc.returncode != 0:
raise RuntimeError("read host env failed: " + text(proc.stderr))
lines = proc.stdout.decode("utf-8", errors="replace").splitlines()
values = {}
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
key, value = stripped.split("=", 1)
value = value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
value = value[1:-1]
values[key.strip()] = value
return values
def select_app_pod():
if RUNTIME_MODE == "host-docker":
return HOST_DOCKER_APP_CONTAINER
pods = kube_json(["-n", NAMESPACE, "get", "pods", "-l", "app.kubernetes.io/name=sub2api"], "sub2api pods").get("items") or []
running = [item for item in pods if (item.get("status") or {}).get("phase") == "Running"]
if not running:
raise RuntimeError("sub2api app pod not found")
return running[0]["metadata"]["name"]
APP_POD = select_app_pod()
def config_value(name, key, fallback=None):
if RUNTIME_MODE == "host-docker":
return read_host_env().get(key) or fallback
data = kube_json(["-n", NAMESPACE, "get", "configmap", name], f"configmap/{name}").get("data") or {}
return data.get(key) or fallback
def secret_value(name, key):
if RUNTIME_MODE == "host-docker":
return read_host_env().get(key)
import base64
data = kube_json(["-n", NAMESPACE, "get", "secret", name], f"secret/{name}").get("data") or {}
return base64.b64decode(data[key]).decode("utf-8") if key in data else None
def parse_curl(proc):
output = proc.stdout.decode("utf-8", errors="replace")
marker = "\\n__HTTP_CODE__:"
pos = output.rfind(marker)
if pos < 0:
return {"ok": False, "httpStatus": 0, "json": None, "body": output, "stderr": text(proc.stderr)}
body = output[:pos]
try:
status = int(output[pos + len(marker):].strip()[-3:])
except ValueError:
status = 0
try:
parsed = json.loads(body) if body.strip() else None
except json.JSONDecodeError:
parsed = None
return {"ok": proc.returncode == 0 and 200 <= status < 300, "httpStatus": status, "json": parsed, "body": body, "stderr": text(proc.stderr)}
def curl_api(method, path, bearer=None, payload=None):
body = b"" if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
script = r'''
set -eu
method="$1"
url="$2"
token="\${3:-}"
tmp="$(mktemp)"
trap 'rm -f "$tmp"' EXIT
cat > "$tmp"
if [ -n "$token" ] && [ "$method" = GET ] && [ ! -s "$tmp" ]; then
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H "Authorization: Bearer $token" "$url"
elif [ -n "$token" ]; then
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' -H "Authorization: Bearer $token" --data-binary @"$tmp" "$url"
elif [ "$method" = GET ] && [ ! -s "$tmp" ]; then
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" "$url"
else
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' --data-binary @"$tmp" "$url"
fi
'''
if RUNTIME_MODE == "host-docker":
proc = run(["sh", "-c", script, "sh", method, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}{path}", bearer or ""], body)
else:
proc = run(["kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD, "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or ""], body)
return parse_curl(proc)
def data_of(response, label):
parsed = response.get("json")
code = parsed.get("code") if isinstance(parsed, dict) else None
if response.get("ok") is not True or (code is not None and code != 0):
message = parsed.get("message") if isinstance(parsed, dict) else text(response.get("body"), 500)
raise RuntimeError(f"{label} failed: http={response.get('httpStatus')} message={message}")
return parsed.get("data") if isinstance(parsed, dict) and "data" in parsed else parsed
def items_of(data):
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ("items", "groups", "accounts"):
if isinstance(data.get(key), list):
return data[key]
return []
def login():
email = config_value("sub2api-config", "ADMIN_EMAIL", PAYLOAD["adminEmailDefault"])
password = secret_value(APP_SECRET_NAME, "ADMIN_PASSWORD")
if not password:
raise RuntimeError("ADMIN_PASSWORD missing")
data = data_of(curl_api("POST", "/api/v1/auth/login", payload={"email": email, "password": password}), "admin login")
token = None
if isinstance(data, dict):
token = data.get("access_token") or data.get("token")
if not token:
raise RuntimeError("admin login response has no token")
return token
def list_groups(token, platform=None):
suffix = "?platform=" + quote(str(platform), safe="") if platform else ""
return items_of(data_of(curl_api("GET", "/api/v1/admin/groups/all" + suffix, bearer=token), "list groups"))
def list_group_accounts(token, group_id, platform=None):
path = f"/api/v1/admin/accounts?group_id={group_id}&page=1&page_size=500"
if platform:
path += "&platform=" + quote(str(platform), safe="")
return items_of(data_of(curl_api("GET", path, bearer=token), "list group accounts"))
def account_detail(token, account_id):
return data_of(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), "get account")
def normalize_policy(credentials):
if not isinstance(credentials, dict):
credentials = {}
enabled = credentials.get("temp_unschedulable_enabled") is True
raw = credentials.get("temp_unschedulable_rules")
if isinstance(raw, str):
try:
raw = json.loads(raw)
except json.JSONDecodeError:
raw = []
rules = raw if isinstance(raw, list) else []
codes = sorted(set(item.get("error_code") for item in rules if isinstance(item, dict) and isinstance(item.get("error_code"), int)))
return {"enabled": enabled, "ruleCount": len(rules), "statusCodes": codes, "rules": rules}
def template_summary(template):
policy = normalize_policy(template.get("credentials"))
return {"id": template.get("id"), "kind": template.get("kind"), "description": template.get("description"), **{key: policy[key] for key in ("enabled", "ruleCount", "statusCodes")}}
def matching_template(credentials):
current = normalize_policy(credentials)
for template in PAYLOAD["templates"]:
if normalize_policy(template.get("credentials")) == current:
return template.get("id")
return None
def account_summary(detail, include_rules=False):
credentials = detail.get("credentials") if isinstance(detail.get("credentials"), dict) else {}
policy = normalize_policy(credentials)
name = detail.get("name")
management = "yaml-managed" if name in PAYLOAD["managedAccountNames"] else ("yaml-protected-manual" if name in PAYLOAD["protectedManualAccountNames"] else "runtime-manual")
result = {
"accountName": name,
"accountId": detail.get("id"),
"management": management,
"type": detail.get("type"),
"status": detail.get("status"),
"schedulable": detail.get("schedulable"),
"proxyId": detail.get("proxy_id"),
"concurrencyLimit": detail.get("concurrency"),
"currentConcurrency": detail.get("current_concurrency"),
"loadFactor": detail.get("load_factor"),
"priority": detail.get("priority"),
"tempUnschedulable": {key: policy[key] for key in ("enabled", "ruleCount", "statusCodes")},
"matchingTemplate": matching_template(credentials),
"poolModeEnabled": credentials.get("pool_mode") is True,
"credentialsPrinted": False,
}
if include_rules:
result["tempUnschedulable"]["rules"] = policy["rules"]
return result
def list_proxies(token):
return items_of(data_of(curl_api("GET", "/api/v1/admin/proxies?page=1&page_size=500", bearer=token), "list proxies"))
def proxy_summary(token, proxy_id):
if proxy_id is None:
return {"configured": False, "route": "direct", "credentialsPrinted": False}
matches = [item for item in list_proxies(token) if isinstance(item, dict) and item.get("id") == proxy_id]
if len(matches) != 1:
return {"configured": True, "id": proxy_id, "status": "record-not-found", "credentialsPrinted": False}
item = matches[0]
return {
"configured": True,
"route": "account-proxy",
"id": item.get("id"),
"name": item.get("name"),
"protocol": item.get("protocol"),
"host": item.get("host"),
"port": item.get("port"),
"status": item.get("status"),
"fallbackMode": item.get("fallback_mode"),
"lastCheckAt": item.get("last_check_at"),
"lastError": text(item.get("last_error"), 240) if item.get("last_error") else None,
"credentialsPrinted": False,
}
def proxy_environment_summary():
rows = []
for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"):
value = app_text(["sh", "-c", f"printenv {name} 2>/dev/null || true"])
parsed = urlsplit(value) if value else None
rows.append({
"name": name,
"present": bool(value),
"scheme": parsed.scheme if parsed else None,
"host": parsed.hostname if parsed else None,
"port": parsed.port if parsed else None,
"credentialsRedacted": bool(parsed and (parsed.username or parsed.password)),
})
no_proxy = app_text(["sh", "-c", "printenv NO_PROXY 2>/dev/null || printenv no_proxy 2>/dev/null || true"])
return rows, {
"present": bool(no_proxy),
"entryCount": len([item for item in no_proxy.split(",") if item.strip()]),
"hyueapiPresent": any(item.strip() in ("hyueapi.com", ".hyueapi.com") for item in no_proxy.split(",")),
"valuesPrinted": False,
}
def app_runtime_summary():
if RUNTIME_MODE == "host-docker":
proc = docker(["inspect", APP_POD])
if proc.returncode != 0:
return {"runtimeMode": RUNTIME_MODE, "name": APP_POD, "status": "unavailable", "reason": text(proc.stderr, 240)}
values = json.loads(proc.stdout.decode("utf-8"))
item = values[0] if values else {}
state = item.get("State") or {}
host_config = item.get("HostConfig") or {}
networks = ((item.get("NetworkSettings") or {}).get("Networks") or {})
return {
"runtimeMode": RUNTIME_MODE,
"name": APP_POD,
"state": state.get("Status"),
"health": ((state.get("Health") or {}).get("Status")),
"restartCount": item.get("RestartCount"),
"oomKilled": state.get("OOMKilled"),
"startedAt": state.get("StartedAt"),
"networkMode": host_config.get("NetworkMode"),
"networks": [
{"name": name, "ipAddress": value.get("IPAddress"), "gateway": value.get("Gateway")}
for name, value in sorted(networks.items()) if isinstance(value, dict)
],
}
pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], "sub2api pod")
spec = pod.get("spec") or {}
status = pod.get("status") or {}
restarts = sum(item.get("restartCount") or 0 for item in status.get("containerStatuses") or [] if isinstance(item, dict))
return {
"runtimeMode": RUNTIME_MODE,
"name": APP_POD,
"state": status.get("phase"),
"restartCount": restarts,
"oomKilled": None,
"startedAt": status.get("startTime"),
"networkMode": "host" if spec.get("hostNetwork") is True else "pod",
"nodeName": spec.get("nodeName"),
"podIp": status.get("podIP"),
"dnsPolicy": spec.get("dnsPolicy"),
}
def dns_summary():
content = app_text(["cat", "/etc/resolv.conf"])
nameservers = []
search = []
options = []
for line in content.splitlines():
fields = line.strip().split()
if not fields or fields[0].startswith("#"):
continue
if fields[0] == "nameserver": nameservers.extend(fields[1:])
elif fields[0] == "search": search.extend(fields[1:])
elif fields[0] == "options": options.extend(fields[1:])
return {"nameservers": nameservers, "search": search, "options": options}
def default_route_summary():
content = app_text(["cat", "/proc/net/route"])
for line in content.splitlines()[1:]:
fields = line.split()
if len(fields) < 4 or fields[1] != "00000000":
continue
try:
gateway = socket.inet_ntoa(struct.pack("<I", int(fields[2], 16)))
except (ValueError, OSError):
gateway = None
return {"interface": fields[0], "gateway": gateway, "flags": fields[3]}
return {"interface": None, "gateway": None, "status": "default-route-not-found"}
def interface_summaries():
script = 'for path in /sys/class/net/*; do [ -d "$path/statistics" ] || continue; name="$(basename "$path")"; printf "%s" "$name"; for key in rx_bytes tx_bytes rx_errors tx_errors rx_dropped tx_dropped; do value="$(cat "$path/statistics/$key" 2>/dev/null || printf 0)"; printf "\\t%s" "$value"; done; printf "\\n"; done'
rows = []
for line in app_text(["sh", "-c", script]).splitlines():
fields = line.split("\t")
if len(fields) != 7:
continue
values = []
for value in fields[1:]:
try:
values.append(int(value))
except ValueError:
values.append(None)
rows.append({
"name": fields[0], "rxBytes": values[0], "txBytes": values[1],
"rxErrors": values[2], "txErrors": values[3], "rxDropped": values[4], "txDropped": values[5],
})
return rows
def proxy_socket_summary(proxy):
host = proxy.get("host") if isinstance(proxy, dict) else None
port = proxy.get("port") if isinstance(proxy, dict) else None
if not isinstance(host, str) or not isinstance(port, int):
return {"status": "not-applicable", "resolvedIps": [], "total": 0, "established": 0}
try:
resolved = sorted(set(item[4][0] for item in socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)))
resolution_status = "resolved"
except OSError as exc:
resolved = []
resolution_status = "failed: " + text(exc, 160)
state_names = {"01": "established", "02": "syn-sent", "03": "syn-recv", "04": "fin-wait-1", "05": "fin-wait-2", "06": "time-wait", "08": "close-wait", "09": "last-ack", "0A": "listen", "0B": "closing"}
states = {}
content = app_text(["cat", "/proc/net/tcp"])
for line in content.splitlines()[1:]:
fields = line.split()
if len(fields) < 4 or ":" not in fields[2]:
continue
address_hex, port_hex = fields[2].split(":", 1)
try:
address = socket.inet_ntoa(struct.pack("<I", int(address_hex, 16)))
remote_port = int(port_hex, 16)
except (ValueError, OSError):
continue
if address not in resolved or remote_port != port:
continue
state = state_names.get(fields[3], fields[3])
states[state] = states.get(state, 0) + 1
return {
"status": resolution_status,
"resolvedIps": resolved,
"port": port,
"total": sum(states.values()),
"established": states.get("established", 0),
"states": [{"state": key, "count": value} for key, value in sorted(states.items())],
"attribution": "Socket counts are a query-time snapshot. DNS resolution is performed without opening an application connection.",
}
def proxy_endpoint_owner(proxy):
host = proxy.get("host") if isinstance(proxy, dict) else None
port = proxy.get("port") if isinstance(proxy, dict) else None
if RUNTIME_MODE != "host-docker" or host not in ("127.0.0.1", "localhost", "::1") or not isinstance(port, int):
return {"status": "not-applicable", "listeners": []}
proc = run(["ss", "-ltnpH"])
if proc.returncode != 0:
return {"status": "unavailable", "reason": text(proc.stderr, 240), "listeners": []}
rows = []
for line in proc.stdout.decode("utf-8", errors="replace").splitlines():
fields = line.split()
if len(fields) < 4:
continue
local_endpoint = fields[3]
try:
local_port = int(local_endpoint.rsplit(":", 1)[1])
except (ValueError, IndexError):
continue
if local_port != port:
continue
process_match = re.search(r'\\("([^\"]+)",pid=(\\d+)', line)
process_name = process_match.group(1) if process_match else None
pid = int(process_match.group(2)) if process_match else None
executable = None
peers = []
udp_peers = []
systemd_unit = None
active_state = None
service_restarts = None
if isinstance(pid, int):
exe_proc = run(["readlink", f"/proc/{pid}/exe"])
if exe_proc.returncode == 0:
executable = exe_proc.stdout.decode("utf-8", errors="replace").strip()
peer_proc = run(["ss", "-tnpH"])
if peer_proc.returncode == 0:
for peer_line in peer_proc.stdout.decode("utf-8", errors="replace").splitlines():
if f"pid={pid}," not in peer_line:
continue
peer_fields = peer_line.split()
if len(peer_fields) >= 5 and peer_fields[0] == "ESTAB":
peers.append(peer_fields[4])
udp_proc = run(["ss", "-unpH"])
if udp_proc.returncode == 0:
for udp_line in udp_proc.stdout.decode("utf-8", errors="replace").splitlines():
if f"pid={pid}," not in udp_line:
continue
udp_fields = udp_line.split()
if len(udp_fields) >= 5 and udp_fields[4] not in ("*:*", "0.0.0.0:*"):
udp_peers.append(udp_fields[4])
cgroup_proc = run(["cat", f"/proc/{pid}/cgroup"])
cgroup_text = cgroup_proc.stdout.decode("utf-8", errors="replace") if cgroup_proc.returncode == 0 else ""
unit_match = re.search(r'/([^/]+\\.service)(?:/|$)', cgroup_text)
systemd_unit = unit_match.group(1) if unit_match else None
if systemd_unit:
show_proc = run(["systemctl", "show", systemd_unit, "--property=ActiveState", "--property=SubState", "--property=NRestarts"])
if show_proc.returncode == 0:
service_values = dict(line.split("=", 1) for line in show_proc.stdout.decode("utf-8", errors="replace").splitlines() if "=" in line)
active_state = "/".join(value for value in (service_values.get("ActiveState"), service_values.get("SubState")) if value)
service_restarts = service_values.get("NRestarts")
rows.append({
"localEndpoint": local_endpoint,
"processName": process_name,
"pid": pid,
"executable": executable,
"establishedPeers": sorted(set(peers))[:12],
"udpPeers": sorted(set(udp_peers))[:12],
"systemdUnit": systemd_unit,
"activeState": active_state,
"serviceRestarts": service_restarts,
})
return {"status": "available", "listeners": rows, "credentialsPrinted": False}
def proxy_service_evidence(owner):
listeners = owner.get("listeners") if isinstance(owner, dict) else []
units = sorted(set(item.get("systemdUnit") for item in listeners if isinstance(item, dict) and item.get("systemdUnit")))
if not units:
return {"status": "not-applicable", "buckets": [], "samples": []}
buckets = {}
samples = []
for unit in units:
proc = run(["journalctl", "--unit", unit, "--since", since_start_time(PAYLOAD.get("since")), "--no-pager", "--output", "json", "--lines", "2000"])
if proc.returncode != 0:
continue
for line in proc.stdout.decode("utf-8", errors="replace").splitlines():
try:
item = json.loads(line)
except json.JSONDecodeError:
continue
message = str(item.get("MESSAGE") or "")
lowered = message.lower()
priority = int(item.get("PRIORITY")) if str(item.get("PRIORITY") or "").isdigit() else 7
cause = None
if "reconnect" in lowered or "retry" in lowered: cause = "reconnect"
elif "disconnect" in lowered or "closed" in lowered: cause = "disconnected"
elif "timeout" in lowered or "deadline" in lowered: cause = "timeout"
elif "error" in lowered or "failed" in lowered or priority <= 3: cause = "service-error"
if cause is None:
continue
buckets[cause] = buckets.get(cause, 0) + 1
if len(samples) < (30 if PAYLOAD.get("full") is True else 12):
samples.append({"at": item.get("__REALTIME_TIMESTAMP"), "cause": cause, "message": text(message, 400)})
return {
"status": "available",
"units": units,
"buckets": [{"cause": key, "count": value} for key, value in sorted(buckets.items(), key=lambda pair: (-pair[1], pair[0]))],
"samples": samples,
}
def classify_network_message(message, status_code):
lowered = message.lower()
checks = (
("dns-failure", ("no such host", "name resolution", "temporary failure in name resolution")),
("proxy-connect-failure", ("proxyconnect", "connect tunnel", "http connect")),
("connection-reset", ("connection reset", "reset by peer")),
("unexpected-eof", ("unexpected eof", " eof")),
("broken-pipe", ("broken pipe",)),
("connection-refused", ("connection refused",)),
("network-unreachable", ("network is unreachable", "no route to host")),
("transport-timeout", ("i/o timeout", "tls handshake timeout", "context deadline exceeded")),
("context-canceled", ("context canceled", "context cancelled")),
("stream-incomplete", ("missing terminal event", "stream usage incomplete", "response failed")),
)
for cause, markers in checks:
if any(marker in lowered for marker in markers):
return cause
return "upstream-503" if status_code == 503 or " 503" in lowered else None
def recent_network_evidence(account_id):
lines = runtime_log_lines()
buckets = {}
samples = []
matched = 0
sample_limit = 30 if PAYLOAD.get("full") is True else 12
for raw in lines:
candidate = raw[raw.find("{"):] if "{" in raw else raw
try:
item = json.loads(candidate)
except json.JSONDecodeError:
item = {}
extra = item.get("extra") if isinstance(item.get("extra"), dict) else {}
observed_account = item.get("account_id") if item.get("account_id") is not None else extra.get("account_id")
if str(observed_account) != str(account_id):
continue
status_code = item.get("status_code") if isinstance(item.get("status_code"), int) else extra.get("status_code")
if not isinstance(status_code, int):
upstream_status = extra.get("upstream_status")
status_code = upstream_status if isinstance(upstream_status, int) else None
message_parts = [item.get("message"), item.get("error"), extra.get("error"), extra.get("message")]
message = " | ".join(str(value) for value in message_parts if value not in (None, ""))
cause = classify_network_message(message, status_code)
if cause is None:
continue
matched += 1
buckets[cause] = buckets.get(cause, 0) + 1
if len(samples) < sample_limit:
samples.append({
"at": item.get("time") or item.get("created_at") or extra.get("_at"),
"requestId": item.get("request_id") or extra.get("request_id"),
"cause": cause,
"statusCode": status_code,
"message": text(message, 360),
})
return {
"source": "sub2api-target-runtime-logs",
"window": PAYLOAD.get("since"),
"scannedLines": len(lines),
"matchedEvents": matched,
"buckets": [{"cause": key, "count": value} for key, value in sorted(buckets.items(), key=lambda pair: (-pair[1], pair[0]))],
"samples": samples,
}
def infrastructure_snapshot(token, detail):
proxy = proxy_summary(token, detail.get("proxy_id"))
proxy_env, no_proxy = proxy_environment_summary()
owner = proxy_endpoint_owner(proxy)
return {
"source": "sub2api-admin-api-and-target-runtime",
"accountProxy": proxy,
"appRuntime": app_runtime_summary(),
"proxyEnvironment": proxy_env,
"noProxy": no_proxy,
"dns": dns_summary(),
"defaultRoute": default_route_summary(),
"interfaces": interface_summaries(),
"proxySockets": proxy_socket_summary(proxy),
"proxyEndpointOwner": owner,
"proxyServiceEvidence": proxy_service_evidence(owner),
"recentNetworkEvidence": recent_network_evidence(detail.get("id")),
"attribution": "The account proxy record controls this account's transport. Container proxy environment is reported separately. Interface counters and sockets are query-time snapshots; no upstream account test or application request is sent.",
"valuesPrinted": False,
}
def policy_summary(policy, include_rules=False):
keys = ("enabled", "ruleCount", "statusCodes", "rules") if include_rules else ("enabled", "ruleCount", "statusCodes")
return {key: policy[key] for key in keys}
def find_account(token, accounts, selector):
exact = [item for item in accounts if str(item.get("id")) == selector or item.get("name") == selector]
if len(exact) != 1:
raise RuntimeError("account-not-found-or-ambiguous: " + selector)
return account_detail(token, exact[0]["id"])
def runtime_log_lines():
since = PAYLOAD.get("since", "24h")
tail = str(PAYLOAD.get("tail", 50000))
if RUNTIME_MODE == "host-docker":
proc = docker(["logs", "--since", since, "--tail", tail, APP_POD])
else:
proc = kubectl(["-n", NAMESPACE, "logs", APP_POD, "--since=" + since, "--tail=" + tail])
if proc.returncode != 0:
raise RuntimeError("read Sub2API runtime logs failed: " + text(proc.stderr))
output = proc.stdout + b"\\n" + proc.stderr
return output.decode("utf-8", errors="replace").splitlines()
def app_exec(args):
if RUNTIME_MODE == "host-docker":
return docker(["exec", APP_POD, *args])
return kubectl(["-n", NAMESPACE, "exec", APP_POD, "--", *args])
def app_text(args):
proc = app_exec(args)
return proc.stdout.decode("utf-8", errors="replace").strip() if proc.returncode == 0 else ""
def config_scalar(path, key):
script = r'''awk -v wanted="$2" '
BEGIN { in_gateway=0 }
/^gateway:[[:space:]]*($|#)/ { in_gateway=1; next }
in_gateway && /^[^[:space:]#]/ { exit }
in_gateway && $0 ~ "^[[:space:]]+" wanted ":[[:space:]]*" {
value=$0
sub("^[[:space:]]+" wanted ":[[:space:]]*", "", value)
sub("[[:space:]]*#.*$", "", value)
gsub("^[[:space:]\\\"'\''']+|[[:space:]\\\"'\''']+$", "", value)
print value
exit
}' "$1"'''
return app_text(["sh", "-c", script, "sh", path, key])
def runtime_response_header_timeout():
env_openai = app_text(["sh", "-c", "printenv GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT 2>/dev/null || true"])
env_generic = app_text(["sh", "-c", "printenv GATEWAY_RESPONSE_HEADER_TIMEOUT 2>/dev/null || true"])
config_file = app_text(["sh", "-c", "printenv CONFIG_FILE 2>/dev/null || true"])
paths = [value for value in (config_file, "/app/data/config.yaml", "/etc/sub2api/config.yaml", "/app/config.yaml") if value]
config_openai = ""
config_generic = ""
config_source = None
for path in dict.fromkeys(paths):
openai_value = config_scalar(path, "openai_response_header_timeout")
generic_value = config_scalar(path, "response_header_timeout")
if openai_value or generic_value:
config_openai = openai_value
config_generic = generic_value
config_source = path
break
def non_negative_int(value):
try:
parsed = int(str(value).strip())
return parsed if parsed >= 0 else None
except (TypeError, ValueError):
return None
openai_configured = non_negative_int(config_openai)
openai_env = non_negative_int(env_openai)
generic_configured = non_negative_int(config_generic)
generic_env = non_negative_int(env_generic)
effective = openai_configured if openai_configured is not None else openai_env if openai_env is not None else 0
source = "config-file:" + config_source if openai_configured is not None else "process-env" if openai_env is not None else "sub2api-default"
generic = generic_configured if generic_configured is not None else generic_env if generic_env is not None else 600
return {
"effectiveSeconds": effective,
"disabled": effective == 0,
"source": source,
"genericResponseHeaderTimeoutSeconds": generic,
"genericAppliesToOpenAI": False,
"configFileChecked": config_source,
"valuesPrinted": False,
}
def compact_error(value):
if not isinstance(value, str):
return ""
compact = " ".join(value.replace("\x00", "").split())
compact = re.sub(r"(?i)\\bBearer\\s+[A-Za-z0-9._~+/=-]+", "Bearer [REDACTED]", compact)
compact = re.sub(r"\\bsk-[A-Za-z0-9_-]{8,}\\b", "sk-[REDACTED]", compact)
compact = re.sub(r"(?i)\\b(api[_ -]?key|token)[=:]\\s*[^\\s,;]+", r"\\1=[REDACTED]", compact)
return compact[:240]
def since_start_time(value):
match = re.fullmatch(r"(\\d+)([mhd])", str(value or ""))
if not match:
raise RuntimeError("unsupported runtime error window: " + str(value))
amount = int(match.group(1))
unit = match.group(2)
duration = timedelta(minutes=amount) if unit == "m" else timedelta(hours=amount) if unit == "h" else timedelta(days=amount)
return (datetime.now(timezone.utc) - duration).isoformat().replace("+00:00", "Z")
def ops_error_cause(value):
lower = value.lower()
markers = {
"concurrency": ("concurrency limit", "concurrent request", "too many concurrent", "maximum concurrency", "max concurrency"),
"overload": ("overloaded", "server busy", "temporarily unavailable", "capacity exhausted"),
"model-route-unavailable": ("no available channel for model", "unsupported model", "model not found", "model_not_found"),
"context-window": ("context window", "context length", "maximum context"),
"continuation-payload": ("encrypted_content", "missing_required_parameter", "missing required parameter"),
"invalid-request": ("invalid_request", "invalid request"),
"rate-limit": ("rate limit", "too many requests"),
"access-denied": ("forbidden", "access denied", "permission denied"),
"not-found": ("not found", "does not exist"),
"incomplete-stream": ("ended before a terminal event", "missing terminal event", "before response.completed"),
"transport": ("do request failed", "unexpected eof", "connection reset", "stream read error", "stream closed", "internal_error; received from peer"),
"upstream-api-key-validation": ("failed to validate api key", "api key validation failed"),
"authentication": ("unauthorized", "invalid token", "token revoked", "invalid_grant"),
"generic-upstream-failure": ("upstream request failed", "upstream gateway error"),
}
for cause, values in markers.items():
if any(marker in lower for marker in values):
return cause
if re.match(r"recovered upstream error \\d+\\b", lower):
return "status-only-upstream-failure"
return "unknown"
def ops_error_analysis(token, account, log_statuses):
account_id = account.get("id")
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
path = f"/api/v1/admin/ops/upstream-errors?account_id={account_id}&view=all&page=1&page_size=500&start_time={start_time}"
response = curl_api("GET", path, bearer=token)
try:
data = data_of(response, "list account ops errors")
except Exception as exc:
return {
"source": "ops-upstream-errors",
"status": "unavailable",
"logObservedStatusCodes": log_statuses,
"reason": compact_error(str(exc)),
"assessment": "insufficient-evidence",
}
rows = items_of(data)
total = data.get("total", len(rows)) if isinstance(data, dict) else len(rows)
detail_limit = 24 if PAYLOAD.get("full") is True else 8
analyzed = []
for row in rows:
message = compact_error(row.get("message") if isinstance(row, dict) else "")
analyzed.append({"row": row, "evidence": message, "cause": ops_error_cause(message)})
for entry in analyzed[:detail_limit]:
row = entry["row"]
error_id = row.get("id") if isinstance(row, dict) else None
if not isinstance(error_id, int):
continue
detail_response = curl_api("GET", f"/api/v1/admin/ops/upstream-errors/{error_id}", bearer=token)
try:
detail = data_of(detail_response, "get account 500 ops error")
except Exception:
continue
fields = []
for key in ("message", "upstream_error_message", "upstream_error_detail", "error_body", "upstream_errors"):
value = detail.get(key) if isinstance(detail, dict) else None
if isinstance(value, str) and value.strip():
fields.append(value)
evidence = compact_error(" ".join(fields))
entry["evidence"] = evidence
entry["cause"] = ops_error_cause(evidence)
causes = {}
statuses = {}
messages = {}
models = {}
hours = {}
stream_modes = {"stream": 0, "synchronous": 0, "unknown": 0}
concurrency_count = 0
for entry in analyzed:
row = entry["row"]
error_id = row.get("id") if isinstance(row, dict) else None
evidence = entry["evidence"]
cause = entry["cause"]
causes[cause] = causes.get(cause, 0) + 1
list_message = compact_error(row.get("message") if isinstance(row, dict) else "")
if list_message:
messages[list_message] = messages.get(list_message, 0) + 1
status_code = row.get("status_code") if isinstance(row, dict) else None
if isinstance(status_code, int):
statuses[str(status_code)] = statuses.get(str(status_code), 0) + 1
if cause == "concurrency":
concurrency_count += 1
model = str(row.get("model") or "unknown") if isinstance(row, dict) else "unknown"
models[model] = models.get(model, 0) + 1
created_at = str(row.get("created_at") or "") if isinstance(row, dict) else ""
hour = created_at[:13] + ":00" if len(created_at) >= 13 else "unknown"
hours[hour] = hours.get(hour, 0) + 1
stream = row.get("stream") if isinstance(row, dict) else None
stream_key = "stream" if stream is True else "synchronous" if stream is False else "unknown"
stream_modes[stream_key] += 1
sample = {
"errorId": error_id,
"createdAt": row.get("created_at") if isinstance(row, dict) else None,
"requestId": row.get("request_id") if isinstance(row, dict) else None,
"model": row.get("model") if isinstance(row, dict) else None,
"stream": row.get("stream") if isinstance(row, dict) else None,
"cause": cause,
"evidence": evidence,
}
entry["sample"] = sample
analyzed_count = len(analyzed)
if analyzed_count == 0:
assessment = "insufficient-evidence"
elif concurrency_count == analyzed_count:
assessment = "concurrency-related"
elif concurrency_count > 0:
assessment = "mixed"
else:
assessment = "no-direct-concurrency-evidence"
dominant_cause = None
dominant_count = 0
if causes:
dominant_cause, dominant_count = max(causes.items(), key=lambda pair: pair[1])
dominant_share = round(dominant_count / analyzed_count, 4) if analyzed_count > 0 else None
if dominant_cause == "upstream-api-key-validation":
finding = "The upstream gateway returned HTTP 500 while validating its inbound API key; investigate its key lookup and storage dependencies first."
elif dominant_cause == "concurrency":
finding = "Direct concurrency-limit messages dominate the observed Ops records."
elif dominant_cause is not None:
finding = "The dominant observed Ops cause is " + dominant_cause + "."
else:
finding = "No structured HTTP 500 Ops records were available for cause analysis."
if concurrency_count > 0:
concurrency_action = "consider-lower-limit-after-controlled-comparison"
else:
concurrency_action = "do-not-lower-limit-as-primary-fix"
ranked_messages = sorted(messages.items(), key=lambda pair: (-pair[1], pair[0]))
message_limit = 20 if PAYLOAD.get("full") is True else 5
return {
"source": "ops-upstream-errors",
"status": "available",
"logObservedStatusCodes": log_statuses,
"opsRecordCount": total,
"analyzedRecordCount": analyzed_count,
"detailRecordCount": min(detail_limit, analyzed_count),
"recordsTruncated": total > analyzed_count,
"statusBuckets": [{"statusCode": int(code), "count": count} for code, count in sorted(statuses.items(), key=lambda pair: int(pair[0]))],
"causeBuckets": [{"cause": cause, "count": count} for cause, count in sorted(causes.items())],
"messageBuckets": [{"message": message, "count": count} for message, count in ranked_messages[:message_limit]],
"concurrencyRelatedRecordCount": concurrency_count,
"assessment": assessment,
"rootCause": {
"dominantCause": dominant_cause,
"dominantCount": dominant_count,
"dominantShare": dominant_share,
"finding": finding,
},
"errorRateEvidence": {
"errorRecordCount": total,
"requestAttemptDenominator": None,
"errorRate": None,
"reason": "The available Ops endpoint lists errors but does not provide all routing attempts for this account.",
},
"concurrencyRecommendation": {
"action": concurrency_action,
"reason": "No error-time concurrency history is available; only explicit concurrency error text is treated as direct evidence.",
},
"runtimeConcurrency": {
"limit": account.get("concurrency"),
"current": account.get("current_concurrency"),
"loadFactor": account.get("load_factor"),
"historicalAtErrorAvailable": False,
},
**({
"modelBuckets": [{"model": model, "count": count} for model, count in sorted(models.items(), key=lambda pair: (-pair[1], pair[0]))],
"hourBuckets": [{"hour": hour, "count": count} for hour, count in sorted(hours.items())],
"streamBuckets": [{"mode": mode, "count": count} for mode, count in stream_modes.items() if count > 0],
"samples": [entry["sample"] for entry in analyzed[:detail_limit]],
} if PAYLOAD.get("full") is True else {
"samples": [entry["sample"] for entry in analyzed[:1]],
}),
"disclosure": "Ops records are independently sampled evidence and are not positionally joined to account_upstream_error log lines.",
}
def gateway_request_path(value):
path = str(value or "").split("?", 1)[0].rstrip("/")
return path in (
"/responses", "/v1/responses", "/responses/compact", "/v1/responses/compact",
"/messages", "/v1/messages", "/chat/completions", "/v1/chat/completions",
)
def native_ops_snapshot(token, group_id, platform=None):
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
base_query = f"group_id={group_id}&start_time={start_time}{platform_query}"
def optional(path, label):
response = curl_api("GET", path, bearer=token)
try:
return {"status": "available", "data": data_of(response, label)}
except Exception as exc:
return {"status": "unavailable", "reason": compact_error(str(exc))}
overview_result = optional(f"/api/v1/admin/ops/dashboard/overview?{base_query}&query_mode=raw", "get ops dashboard overview")
availability_result = optional(f"/api/v1/admin/ops/account-availability?group_id={group_id}{platform_query}", "get ops account availability")
concurrency_result = optional(f"/api/v1/admin/ops/concurrency?group_id={group_id}{platform_query}", "get ops concurrency")
sink_result = optional("/api/v1/admin/ops/system-logs/health", "get ops system log health")
overview = overview_result.get("data") if overview_result.get("status") == "available" else None
overview_summary = None
if isinstance(overview, dict):
overview_summary = {
key: overview.get(key)
for key in (
"start_time", "end_time", "health_score", "success_count", "error_count_total",
"business_limited_count", "error_count_sla", "request_count_total", "request_count_sla",
"error_rate", "upstream_error_rate", "upstream_error_count_excl_429_529",
"upstream_429_count", "upstream_529_count",
"duration", "ttft",
)
}
availability = availability_result.get("data") if availability_result.get("status") == "available" else None
availability_summary = None
if isinstance(availability, dict):
group_values = availability.get("group") if isinstance(availability.get("group"), dict) else {}
account_values = availability.get("account") if isinstance(availability.get("account"), dict) else {}
selected_group = group_values.get(str(group_id)) or group_values.get(group_id)
unavailable = []
for value in account_values.values():
if not isinstance(value, dict) or value.get("group_id") != group_id or value.get("is_available") is True:
continue
unavailable.append({
"accountId": value.get("account_id"),
"accountName": value.get("account_name"),
"status": value.get("status"),
"rateLimited": value.get("is_rate_limited"),
"rateLimitRemainingSeconds": value.get("rate_limit_remaining_sec"),
"overloaded": value.get("is_overloaded"),
"overloadRemainingSeconds": value.get("overload_remaining_sec"),
"hasError": value.get("has_error"),
"error": compact_error(value.get("error_message")),
})
availability_summary = {
"enabled": availability.get("enabled"),
"timestamp": availability.get("timestamp"),
"group": selected_group,
"unavailableAccounts": unavailable[:20 if PAYLOAD.get("full") is True else 8],
"unavailableAccountCount": max(
len(unavailable),
max(0, selected_group.get("total_accounts") - selected_group.get("available_count"))
if isinstance(selected_group, dict)
and isinstance(selected_group.get("total_accounts"), int)
and isinstance(selected_group.get("available_count"), int)
else 0,
),
"unavailableAccountDetailsComplete": (
not isinstance(selected_group, dict)
or not isinstance(selected_group.get("total_accounts"), int)
or not isinstance(selected_group.get("available_count"), int)
or len(unavailable) >= max(0, selected_group.get("total_accounts") - selected_group.get("available_count"))
),
}
concurrency = concurrency_result.get("data") if concurrency_result.get("status") == "available" else None
concurrency_summary = None
if isinstance(concurrency, dict):
group_values = concurrency.get("group") if isinstance(concurrency.get("group"), dict) else {}
account_values = concurrency.get("account") if isinstance(concurrency.get("account"), dict) else {}
selected_group = group_values.get(str(group_id)) or group_values.get(group_id)
loaded = []
for value in account_values.values():
if not isinstance(value, dict) or value.get("group_id") != group_id:
continue
if (value.get("current_in_use") or 0) <= 0 and (value.get("waiting_in_queue") or 0) <= 0:
continue
loaded.append({
"accountId": value.get("account_id"),
"accountName": value.get("account_name"),
"currentInUse": value.get("current_in_use"),
"maxCapacity": value.get("max_capacity"),
"loadPercentage": value.get("load_percentage"),
"waitingInQueue": value.get("waiting_in_queue"),
})
concurrency_summary = {
"enabled": concurrency.get("enabled"),
"timestamp": concurrency.get("timestamp"),
"group": selected_group,
"activeAccounts": sorted(loaded, key=lambda item: (-(item.get("loadPercentage") or 0), str(item.get("accountName") or "")))[:20 if PAYLOAD.get("full") is True else 8],
}
sink = sink_result.get("data") if sink_result.get("status") == "available" else None
return {
"source": "sub2api-native-admin-ops",
"overview": {"status": overview_result.get("status"), "summary": overview_summary, "reason": overview_result.get("reason")},
"accountAvailability": {"status": availability_result.get("status"), "summary": availability_summary, "reason": availability_result.get("reason")},
"concurrency": {"status": concurrency_result.get("status"), "summary": concurrency_summary, "reason": concurrency_result.get("reason")},
"systemLogSink": {"status": sink_result.get("status"), "health": sink if isinstance(sink, dict) else None, "reason": sink_result.get("reason")},
}
def native_request_timing(token, group_id, platform, overview):
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
base = f"group_id={group_id}&start_time={start_time}{platform_query}"
all_counts = {}
error_counts = {}
status = "available"
reason = None
duration = overview.get("duration") if isinstance(overview.get("duration"), dict) else {}
ttft = overview.get("ttft") if isinstance(overview.get("ttft"), dict) else {}
ttft_p99 = ttft.get("p99_ms")
observed_floor = max(60, ((ttft_p99 + 29999) // 15000) * 15) if isinstance(ttft_p99, int) else None
candidate_seconds = sorted(set([30, 45, 60] + ([observed_floor] if isinstance(observed_floor, int) else [])))
error_duration_response = curl_api("GET", f"/api/v1/admin/ops/requests?{base}&page=1&page_size=1&kind=error&min_duration_ms=0", bearer=token)
try:
error_duration_data = data_of(error_duration_response, "get error duration coverage")
error_duration_record_count = error_duration_data.get("total") if isinstance(error_duration_data, dict) else None
except Exception as exc:
error_duration_record_count = None
status = "unavailable"
reason = compact_error(str(exc))
for seconds in candidate_seconds:
threshold = seconds * 1000
for kind, target in ((None, all_counts), ("error", error_counts)):
kind_query = "&kind=error" if kind else ""
response = curl_api("GET", f"/api/v1/admin/ops/requests?{base}&page=1&page_size=1&min_duration_ms={threshold}{kind_query}", bearer=token)
try:
data = data_of(response, "get long request count")
value = data.get("total") if isinstance(data, dict) else None
target[threshold] = None if kind == "error" and error_duration_record_count == 0 else value
except Exception as exc:
status = "unavailable"
reason = compact_error(str(exc))
target[threshold] = None
candidates = []
for seconds in candidate_seconds:
headroom = seconds * 1000 - ttft_p99 if isinstance(ttft_p99, int) else None
candidates.append({
"seconds": seconds,
"observedFloor": seconds == observed_floor,
"ttftP99HeadroomMs": headroom,
"ttftP99Below": headroom >= 0 if isinstance(headroom, int) else None,
"totalDurationAtLeastCount": all_counts.get(seconds * 1000),
"errorDurationAtLeastCount": error_counts.get(seconds * 1000),
"assessment": "observed-ttft-p99-plus-15s-rounded" if seconds == observed_floor else "ttft-p99-within-candidate" if isinstance(headroom, int) and headroom >= 0 else "ttft-p99-exceeds-candidate" if isinstance(headroom, int) else "insufficient-ttft-evidence",
})
since = str(PAYLOAD.get("since") or "")
token_range = {"30m": "30m", "60m": "1h", "1h": "1h", "24h": "1d", "1d": "1d"}.get(since)
model_timing = []
model_timing_status = "unsupported-window"
if token_range:
response = curl_api("GET", f"/api/v1/admin/ops/dashboard/openai-token-stats?group_id={group_id}&time_range={token_range}&top_n=100{platform_query}", bearer=token)
try:
data = data_of(response, "get OpenAI model token stats")
rows = items_of(data)
for row in rows:
if not isinstance(row, dict):
continue
request_count = row.get("request_count")
first_count = row.get("requests_with_first_token")
model_timing.append({
"model": row.get("model"), "requestCount": request_count,
"requestsWithFirstToken": first_count,
"firstTokenCoverage": round(first_count / request_count, 6) if isinstance(first_count, int) and isinstance(request_count, int) and request_count > 0 else None,
"avgFirstTokenMs": row.get("avg_first_token_ms"), "avgDurationMs": row.get("avg_duration_ms"),
"avgTokensPerSec": row.get("avg_tokens_per_sec"),
})
model_timing.sort(key=lambda item: (-(item.get("avgFirstTokenMs") or -1), str(item.get("model") or "")))
model_timing_status = "available"
except Exception as exc:
model_timing_status = "unavailable"
reason = compact_error(str(exc))
return {
"status": status,
"reason": reason,
"source": "sub2api-native-admin-ops-dashboard-and-requests",
"ttft": ttft,
"duration": duration,
"observedFloorSeconds": observed_floor,
"errorDurationRecordCount": error_duration_record_count,
"modelTimingStatus": model_timing_status,
"modelTimingWindow": token_range,
"modelTiming": model_timing[:20 if PAYLOAD.get("full") is True else 8],
"candidates": candidates,
"attribution": "TTFT covers successful streaming requests with first_token_ms. Total and error duration are end-to-end request durations, not response-header wait or guaranteed failover lead time. The observed floor is TTFT P99 plus 15 seconds rounded up to 15 seconds; it is not a deployment recommendation. A dash in ERROR_DUR>= means error duration coverage is unavailable.",
}
def native_customer_error_profile(token, group_id, platform):
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
page = 1
page_size = 500
rows = []
total = None
try:
while True:
response = curl_api("GET", f"/api/v1/admin/ops/request-errors?group_id={group_id}&start_time={start_time}&view=all&page={page}&page_size={page_size}{platform_query}", bearer=token)
data = data_of(response, "get customer-visible request errors")
page_rows = items_of(data)
rows.extend(page_rows)
if total is None and isinstance(data, dict) and isinstance(data.get("total"), int):
total = data.get("total")
if not page_rows or len(page_rows) < page_size or (isinstance(total, int) and len(rows) >= total):
break
page += 1
except Exception as exc:
return {"status": "unavailable", "source": "sub2api-native-admin-ops-request-errors", "reason": compact_error(str(exc))}
if total is None:
total = len(rows)
def buckets(key_fn):
counts = {}
for row in rows:
value = key_fn(row) if isinstance(row, dict) else None
label = str(value) if value not in (None, "") else "unknown"
counts[label] = counts.get(label, 0) + 1
return [{"value": key, "count": count} for key, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))[:20 if PAYLOAD.get("full") is True else 8]]
samples = []
account_metrics = {}
seen = set()
for row in rows:
if not isinstance(row, dict):
continue
account_id = row.get("account_id")
account_key = str(account_id) if account_id is not None else "unknown"
if account_key not in account_metrics:
account_metrics[account_key] = {
"accountId": account_id, "customerErrorEvents": 0, "customerErrorRequestIds": set(), "statusCodes": {},
}
metric = account_metrics[account_key]
metric["customerErrorEvents"] += 1
request_id = row.get("request_id")
if isinstance(request_id, str) and request_id:
metric["customerErrorRequestIds"].add(request_id)
status_code = row.get("status_code")
if isinstance(status_code, int):
status_key = str(status_code)
metric["statusCodes"][status_key] = metric["statusCodes"].get(status_key, 0) + 1
if not request_id:
continue
key = (row.get("requested_model") or row.get("model"), row.get("status_code"), row.get("account_id"), row.get("stream"), row.get("request_path") or row.get("inbound_endpoint"))
if key in seen:
continue
seen.add(key)
samples.append({
"requestId": row.get("request_id"), "model": key[0], "statusCode": key[1], "accountId": key[2],
"mode": "stream" if key[3] is True else "sync" if key[3] is False else "unknown", "path": key[4],
"message": compact_error(row.get("message") or row.get("error_message")),
})
if len(samples) >= (30 if PAYLOAD.get("full") is True else 15):
break
return {
"status": "available", "source": "sub2api-native-admin-ops-request-errors", "total": total,
"returned": len(rows), "pagesRead": page, "recordsTruncated": False,
"modelBuckets": buckets(lambda row: row.get("requested_model") or row.get("model")),
"accountBuckets": buckets(lambda row: row.get("account_id")),
"statusBuckets": buckets(lambda row: row.get("status_code")),
"streamBuckets": buckets(lambda row: "stream" if row.get("stream") is True else "sync" if row.get("stream") is False else "unknown"),
"phaseBuckets": buckets(lambda row: row.get("phase")),
"endpointBuckets": buckets(lambda row: row.get("inbound_endpoint") or row.get("request_path")),
"pathBuckets": buckets(lambda row: row.get("request_path") or row.get("inbound_endpoint")),
"accountMetrics": [
{
"accountId": item["accountId"], "customerErrorEvents": item["customerErrorEvents"],
"customerErrorRequests": len(item["customerErrorRequestIds"]),
"_requestIds": sorted(item["customerErrorRequestIds"]),
"statusBuckets": [
{"statusCode": int(code), "count": count}
for code, count in sorted(item["statusCodes"].items(), key=lambda pair: int(pair[0]))
],
}
for item in sorted(account_metrics.values(), key=lambda item: int(item.get("accountId") or 0))
],
"samples": samples,
}
def parse_native_time(value):
if not isinstance(value, str) or not value:
return None
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
except ValueError:
return None
def metric_percentile(values, fraction):
ordered = sorted(value for value in values if isinstance(value, (int, float)) and value >= 0)
if not ordered:
return None
position = (len(ordered) - 1) * fraction
lower = int(position)
upper = min(lower + 1, len(ordered) - 1)
return round(ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower))
def account_usage_metrics(token, accounts, group_id):
start_at = parse_native_time(since_start_time(PAYLOAD.get("since")))
end_at = datetime.now(timezone.utc)
result = {}
for account in accounts:
account_id = account.get("id")
key = str(account_id)
rows = []
page = 1
status = "available"
reason = None
try:
while True:
path = (
f"/api/v1/admin/usage?page={page}&page_size=100&group_id={group_id}&account_id={account_id}"
f"&start_date={start_at.date().isoformat()}&end_date={end_at.date().isoformat()}&timezone=UTC"
"&sort_by=created_at&sort_order=desc"
)
data = data_of(curl_api("GET", path, bearer=token), "list account usage")
page_rows = items_of(data)
reached_start = False
for row in page_rows:
created_at = parse_native_time(row.get("created_at") if isinstance(row, dict) else None)
if created_at is None or created_at < start_at or created_at > end_at:
if created_at is not None and created_at < start_at:
reached_start = True
continue
rows.append(row)
if not page_rows or len(page_rows) < 100 or reached_start:
break
page += 1
except Exception as exc:
status = "unavailable"
reason = compact_error(str(exc))
stream_rows = [row for row in rows if isinstance(row, dict) and row.get("stream") is True]
ttft_values = [row.get("first_token_ms") for row in stream_rows if isinstance(row.get("first_token_ms"), int)]
duration_values = [row.get("duration_ms") for row in rows if isinstance(row.get("duration_ms"), int)]
models = {}
for row in rows:
model = str(row.get("model") or "unknown")
models[model] = models.get(model, 0) + 1
result[key] = {
"status": status, "reason": reason, "source": "sub2api-native-admin-usage",
"pagesRead": page, "successRequests": len(rows), "streamSuccessRequests": len(stream_rows),
"firstTokenSamples": len(ttft_values),
"firstTokenCoverage": round(len(ttft_values) / len(stream_rows), 6) if stream_rows else None,
"ttftP50Ms": metric_percentile(ttft_values, 0.50), "ttftP95Ms": metric_percentile(ttft_values, 0.95),
"ttftP99Ms": metric_percentile(ttft_values, 0.99), "ttftMaxMs": max(ttft_values) if ttft_values else None,
"durationP95Ms": metric_percentile(duration_values, 0.95),
"modelBuckets": [{"model": model, "count": count} for model, count in sorted(models.items(), key=lambda item: (-item[1], item[0]))[:8]],
}
return result
def reliability_score_points(failure_rate):
if not isinstance(failure_rate, (int, float)):
return None
return round(60 * (1 - min(max(failure_rate, 0), 0.20) / 0.20), 2)
def latency_score_points(ttft_p95_ms):
if not isinstance(ttft_p95_ms, (int, float)):
return None
return round(25 * (1 - min(max(ttft_p95_ms - 10000, 0), 170000) / 170000), 2)
def account_quality_scores(token, accounts, group_id, native_ops, effectiveness):
usage_by_account = account_usage_metrics(token, accounts, group_id)
customer_profile = native_ops.get("customerErrors") if isinstance(native_ops.get("customerErrors"), dict) else {}
customer_by_account = {str(item.get("accountId")): item for item in customer_profile.get("accountMetrics") or [] if isinstance(item, dict)}
policy_by_account = {str(item.get("accountId")): item for item in effectiveness.get("accountPolicyMetrics") or [] if isinstance(item, dict)}
availability = ((native_ops.get("accountAvailability") or {}).get("summary") or {})
unavailable_ids = set(str(item.get("accountId")) for item in availability.get("unavailableAccounts") or [] if isinstance(item, dict))
rows = []
for account in accounts:
account_id = account.get("id")
key = str(account_id)
usage = usage_by_account.get(key) or {}
customer = customer_by_account.get(key) or {}
policy = policy_by_account.get(key) or {}
policy_failure_ids = set(policy.pop("_failureRequestIds", []) or [])
customer_failure_ids = set(customer.pop("_requestIds", []) or [])
failure_requests = len(policy_failure_ids | customer_failure_ids)
success_requests = usage.get("successRequests") if isinstance(usage.get("successRequests"), int) else 0
observed_attempts = success_requests + failure_requests
failure_rate = round(failure_requests / observed_attempts, 6) if observed_attempts > 0 else None
reliability_points = reliability_score_points(failure_rate)
ttft_samples = usage.get("firstTokenSamples") if isinstance(usage.get("firstTokenSamples"), int) else 0
latency_points = latency_score_points(usage.get("ttftP95Ms")) if ttft_samples >= 5 else None
current_available = key not in unavailable_ids and account.get("status") == "active" and account.get("schedulable") is not False
availability_points = 15 if current_available else 8 if account.get("status") == "active" else 0
earned = (reliability_points or 0) + (latency_points or 0) + availability_points
available_weight = (60 if reliability_points is not None else 0) + (25 if latency_points is not None else 0) + 15
score = round(earned / available_weight * 100, 1) if observed_attempts > 0 and available_weight > 0 else None
score_comparable = observed_attempts >= 10 and ttft_samples >= 5
grade = "insufficient"
if isinstance(score, (int, float)) and (score_comparable or (score < 60 and observed_attempts >= 10)):
grade = "A" if score >= 90 else "B" if score >= 80 else "C" if score >= 70 else "D" if score >= 60 else "E"
assessment = "preferred" if grade == "A" else "healthy" if grade == "B" else "watch" if grade == "C" else "degraded" if grade == "D" else "poor" if grade == "E" else "insufficient-evidence"
confidence = "high" if observed_attempts >= 50 and ttft_samples >= 20 else "medium" if observed_attempts >= 10 and ttft_samples >= 5 else "low"
reasons = []
if isinstance(failure_rate, (int, float)) and failure_rate >= 0.10:
reasons.append("failure-rate>=10%")
elif isinstance(failure_rate, (int, float)) and failure_rate >= 0.03:
reasons.append("failure-rate>=3%")
if isinstance(usage.get("ttftP95Ms"), int) and usage.get("ttftP95Ms") >= 120000:
reasons.append("ttft-p95>=120s")
if policy.get("failoverRequests", 0) > 0:
reasons.append("upstream-failover-triggered")
if policy.get("sameAccountRetryEvents", 0) > 0:
reasons.append("same-account-retry-observed")
if not current_available:
reasons.append("currently-unavailable")
if ttft_samples < 5:
reasons.append("ttft-evidence-insufficient")
if observed_attempts < 10:
reasons.append("request-evidence-insufficient")
rows.append({
"accountId": account_id, "accountName": account.get("name"), "status": account.get("status"),
"schedulable": account.get("schedulable"), "currentlyAvailable": current_available,
"score": score, "grade": grade, "assessment": assessment, "confidence": confidence,
"scoreComparable": score_comparable,
"observedAttempts": observed_attempts, "successRequests": success_requests,
"failureRequests": failure_requests, "failureRate": failure_rate,
"streamSuccessRequests": usage.get("streamSuccessRequests"), "firstTokenSamples": ttft_samples,
"firstTokenCoverage": usage.get("firstTokenCoverage"), "ttftP50Ms": usage.get("ttftP50Ms"),
"ttftP95Ms": usage.get("ttftP95Ms"), "ttftP99Ms": usage.get("ttftP99Ms"), "ttftMaxMs": usage.get("ttftMaxMs"),
"durationP95Ms": usage.get("durationP95Ms"), "modelBuckets": usage.get("modelBuckets"),
"customerErrorRequests": customer.get("customerErrorRequests", 0),
"failoverRequests": policy.get("failoverRequests", 0), "failoverRecovered": policy.get("failoverRecovered", 0),
"failoverFailed": policy.get("failoverFailed", 0), "failoverOutcomeMissing": policy.get("failoverOutcomeMissing", 0),
"sameAccountRetryEvents": policy.get("sameAccountRetryEvents", 0),
"tempUnschedulableEvents": policy.get("tempUnschedulableEvents", 0),
"forwardFailedRequests": policy.get("forwardFailedRequests", 0),
"upstreamStatusBuckets": policy.get("upstreamStatusBuckets", []),
"scoreComponents": {"reliability": reliability_points, "latency": latency_points, "availability": availability_points, "availableWeight": available_weight},
"reasons": reasons, "usageStatus": usage.get("status"), "usageReason": usage.get("reason"),
})
rows.sort(key=lambda item: (item.get("score") is None, -(item.get("score") or -1), int(item.get("accountId") or 0)))
return {
"source": "sub2api-native-admin-usage-request-errors-and-system-logs",
"scope": "group-account-window", "window": PAYLOAD.get("since"),
"scoreRule": {
"reliabilityWeight": 60, "latencyWeight": 25, "availabilityWeight": 15,
"failureDefinition": "unique request ids with account failover, forward_failed, or customer-visible error",
"reliability": "linear from 60 points at 0% observed failures to 0 points at 20%",
"latency": "requires at least 5 first-token samples; linear from 25 points at <=10s TTFT P95 to 0 points at >=180s",
"availability": "15 points when active, schedulable, and currently available; 8 when active but unavailable; otherwise 0",
"missingMetric": "missing latency is excluded from the denominator and lowers confidence; it is not treated as zero latency",
},
"accounts": rows,
"attribution": "Success denominators and first-token latency come from native admin usage records. Failures are deduplicated by request id across native request-errors and indexed failover/forward events. Scores are analytical and never mutate scheduling.",
}
def all_group_error_overview(token, groups):
page_size = 20
cursor = int(PAYLOAD.get("pageToken") or 0)
ordered = sorted(
[item for item in groups if isinstance(item, dict) and isinstance(item.get("id"), int)],
key=lambda item: item["id"],
)
remaining = [item for item in ordered if item["id"] > cursor]
page = remaining[:page_size]
summaries = []
totals = {
"requestCount": 0,
"errorCount": 0,
"businessLimitedCount": 0,
"upstreamErrorCount": 0,
"groupsNeedingAttention": 0,
}
for group in page:
group_id = group["id"]
platform = PAYLOAD.get("platform") or group.get("platform")
snapshot = native_ops_snapshot(token, group_id, platform)
overview = ((snapshot.get("overview") or {}).get("summary") or {})
availability = ((snapshot.get("accountAvailability") or {}).get("summary") or {})
availability_group = availability.get("group") if isinstance(availability.get("group"), dict) else {}
concurrency = ((snapshot.get("concurrency") or {}).get("summary") or {})
concurrency_group = concurrency.get("group") if isinstance(concurrency.get("group"), dict) else {}
request_count = overview.get("request_count_total")
error_count = overview.get("error_count_total")
business_limited = overview.get("business_limited_count")
upstream_error_count = sum(
value for value in (
overview.get("upstream_error_count_excl_429_529"),
overview.get("upstream_429_count"),
overview.get("upstream_529_count"),
) if isinstance(value, int)
)
unavailable_count = availability.get("unavailableAccountCount")
needs_attention = (
(isinstance(error_count, int) and error_count > 0)
or upstream_error_count > 0
or (isinstance(unavailable_count, int) and unavailable_count > 0)
)
totals["requestCount"] += request_count if isinstance(request_count, int) else 0
totals["errorCount"] += error_count if isinstance(error_count, int) else 0
totals["businessLimitedCount"] += business_limited if isinstance(business_limited, int) else 0
totals["upstreamErrorCount"] += upstream_error_count
totals["groupsNeedingAttention"] += 1 if needs_attention else 0
summaries.append({
"groupId": group_id,
"groupName": group.get("name"),
"platform": platform,
"status": group.get("status"),
"requestCount": request_count,
"errorCount": error_count,
"errorRate": overview.get("error_rate"),
"upstreamErrorCount": upstream_error_count,
"upstreamErrorRate": overview.get("upstream_error_rate"),
"businessLimitedCount": business_limited,
"ttftP99Ms": ((overview.get("ttft") or {}).get("p99_ms") if isinstance(overview.get("ttft"), dict) else None),
"durationP99Ms": ((overview.get("duration") or {}).get("p99_ms") if isinstance(overview.get("duration"), dict) else None),
"totalAccounts": availability_group.get("total_accounts"),
"availableCount": availability_group.get("available_count"),
"rateLimitCount": availability_group.get("rate_limit_count"),
"errorAccountCount": availability_group.get("error_count"),
"unavailableAccountCount": unavailable_count,
"currentInUse": concurrency_group.get("current_in_use"),
"maxCapacity": concurrency_group.get("max_capacity"),
"waitingInQueue": concurrency_group.get("waiting_in_queue"),
"needsAttention": needs_attention,
"opsStatus": {
"overview": (snapshot.get("overview") or {}).get("status"),
"accountAvailability": (snapshot.get("accountAvailability") or {}).get("status"),
"concurrency": (snapshot.get("concurrency") or {}).get("status"),
},
})
next_token = str(page[-1]["id"]) if len(remaining) > page_size and page else None
totals["errorRate"] = round(totals["errorCount"] / totals["requestCount"], 6) if totals["requestCount"] > 0 else None
totals["upstreamErrorRate"] = round(totals["upstreamErrorCount"] / totals["requestCount"], 6) if totals["requestCount"] > 0 else None
ttft_p99_values = [item.get("ttftP99Ms") for item in summaries if isinstance(item.get("ttftP99Ms"), int)]
max_ttft_p99 = max(ttft_p99_values) if ttft_p99_values else None
observed_floor = max(60, ((max_ttft_p99 + 29999) // 15000) * 15) if isinstance(max_ttft_p99, int) else None
return {
"scope": "all-groups-overview",
"window": {"since": PAYLOAD.get("since")},
"platform": PAYLOAD.get("platform"),
"totalGroups": len(ordered),
"returnedGroups": len(summaries),
"pageSize": page_size,
"pageToken": str(cursor) if cursor > 0 else None,
"nextPageToken": next_token,
"pageTotals": totals,
"responseHeaderTimeout": runtime_response_header_timeout(),
"requestTiming": {
"maxTtftP99Ms": max_ttft_p99,
"observedFloorSeconds": observed_floor,
"scope": "returned-groups-page",
"attribution": "The observed floor is the maximum returned-group TTFT P99 plus 15 seconds rounded up to 15 seconds. It is analytical only and must not be treated as a safe timeout.",
},
"groups": summaries,
"disclosure": "Page totals cover only returned groups. Follow nextPageToken for remaining groups and use --group for account root causes and request indexes.",
}
def native_system_log_lines(token, platform=None):
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
markers = (
"account_temp_unschedulable", "openai.upstream_failover_switching",
"openai.account_select_failed", "openai.forward_failed",
"openai.pool_mode_same_account_retry", "gateway.failover_same_account_retry",
"http request completed",
)
rows_by_id = {}
queries = []
for marker in markers:
page = 1
returned = 0
total = None
try:
while True:
path = f"/api/v1/admin/ops/system-logs?page={page}&page_size=200&start_time={start_time}&q={quote(marker, safe='')}{platform_query}"
data = data_of(curl_api("GET", path, bearer=token), "list ops system logs")
rows = items_of(data)
returned += len(rows)
if total is None and isinstance(data, dict) and isinstance(data.get("total"), int):
total = data.get("total")
for row in rows:
if not isinstance(row, dict):
continue
row_id = row.get("id")
key = str(row_id) if row_id is not None else json.dumps(row, sort_keys=True, default=str)
rows_by_id[key] = row
if not rows or len(rows) < 200 or (isinstance(total, int) and returned >= total):
break
page += 1
except Exception as exc:
return None, {
"status": "unavailable",
"source": "sub2api-native-admin-ops-system-logs",
"reason": compact_error(str(exc)),
"fallback": "runtime-container-logs",
}
queries.append({"marker": marker, "returned": returned, "total": total if total is not None else returned, "pagesRead": page, "truncated": False})
lines = []
for row in sorted(rows_by_id.values(), key=lambda item: str(item.get("created_at") or "")):
extra = dict(row.get("extra") or {}) if isinstance(row.get("extra"), dict) else {}
for source_key, target_key in (
("request_id", "request_id"), ("client_request_id", "client_request_id"),
("account_id", "account_id"), ("platform", "platform"), ("model", "model"),
):
if extra.get(target_key) is None and row.get(source_key) is not None:
extra[target_key] = row.get(source_key)
message = str(row.get("message") or "")
lines.append(str(row.get("created_at") or "") + "\\t" + message + "\\t" + json.dumps(extra, separators=(",", ":"), default=str))
return lines, {
"status": "available",
"source": "sub2api-native-admin-ops-system-logs",
"uniqueEventCount": len(lines),
"queries": queries,
"fallbackUsed": False,
}
def policy_effectiveness(lines, accounts, source, group_id=None):
account_names = {str(item.get("id")): item.get("name") for item in accounts}
account_ids = set(account_names.keys())
account_policy = {
key: {
"accountId": item.get("id"), "accountName": item.get("name"),
"tempUnschedulableEvents": 0, "failoverEvents": 0, "failoverRequestIds": set(),
"failoverRecoveredRequestIds": set(), "failoverFailedRequestIds": set(), "failoverOutcomeMissingRequestIds": set(),
"forwardFailedEvents": 0, "forwardFailedRequestIds": set(), "sameAccountRetryEvents": 0,
"upstreamStatusCodes": {},
}
for key, item in ((str(item.get("id")), item) for item in accounts)
}
requests = {}
counts = {
"tempUnschedulableEvents": 0,
"failoverEvents": 0,
"selectFailedEvents": 0,
"forwardFailedEvents": 0,
"sameAccountRetryEvents": 0,
"gatewayCompletedEvents": 0,
"gatewayFinalErrorEvents": 0,
}
missing_request_ids = {
"tempUnschedulable": 0,
"failover": 0,
"selectFailed": 0,
"forwardFailed": 0,
"sameAccountRetry": 0,
"gatewayCompleted": 0,
}
temp_accounts = {}
temp_samples = []
failover_samples = []
select_failed_samples = []
final_statuses = {}
event_line_count = 0
def request_entry(request_id):
if not isinstance(request_id, str) or not request_id:
return None
if request_id not in requests:
requests[request_id] = {
"requestId": request_id,
"tempUnschedulableCount": 0,
"failoverCount": 0,
"selectFailedCount": 0,
"forwardFailedCount": 0,
"sameAccountRetryCount": 0,
"finalStatus": None,
}
return requests[request_id]
for line in lines:
if not any(marker in line for marker in (
"account_temp_unschedulable", "openai.upstream_failover_switching",
"openai.account_select_failed", "openai.forward_failed",
"openai.pool_mode_same_account_retry", "gateway.failover_same_account_retry",
"http request completed",
)):
continue
json_start = line.find("{")
if json_start < 0:
continue
try:
item = json.loads(line[json_start:])
except json.JSONDecodeError:
continue
if not isinstance(item, dict):
continue
at = line.split("\\t", 1)[0].strip() or None
request_id = item.get("request_id")
account_id = item.get("account_id")
event_group_id = item.get("group_id")
if event_group_id is not None and group_id is not None and str(event_group_id) != str(group_id):
continue
if account_id is not None and str(account_id) not in account_ids:
continue
explicitly_scoped = account_id is not None or event_group_id is not None
if not explicitly_scoped and (not isinstance(request_id, str) or request_id not in requests):
continue
request = request_entry(request_id)
base = {
"at": at,
"requestId": request_id,
"accountId": account_id,
"accountName": account_names.get(str(account_id)),
}
raw_account_stat = account_policy.get(str(account_id))
account_stat = raw_account_stat if group_id is None or event_group_id is not None else None
if "account_temp_unschedulable" in line:
event_line_count += 1
counts["tempUnschedulableEvents"] += 1
if request is not None:
request["tempUnschedulableCount"] += 1
else:
missing_request_ids["tempUnschedulable"] += 1
key = str(account_id) if account_id is not None else "unknown"
if key not in temp_accounts:
temp_accounts[key] = {
"accountId": account_id,
"accountName": account_names.get(str(account_id)),
"count": 0,
}
temp_accounts[key]["count"] += 1
if raw_account_stat is not None:
raw_account_stat["tempUnschedulableEvents"] += 1
if len(temp_samples) < 20:
temp_samples.append({
**base,
"statusCode": item.get("status_code"),
"ruleIndex": item.get("rule_index"),
"matchedKeyword": item.get("matched_keyword"),
"until": item.get("until") or item.get("temp_unschedulable_until"),
"reason": compact_error(item.get("reason") or item.get("error")),
})
elif "openai.upstream_failover_switching" in line:
event_line_count += 1
counts["failoverEvents"] += 1
if request is not None:
request["failoverCount"] += 1
else:
missing_request_ids["failover"] += 1
if account_stat is not None:
account_stat["failoverEvents"] += 1
if isinstance(request_id, str) and request_id:
account_stat["failoverRequestIds"].add(request_id)
status_code = item.get("upstream_status")
if isinstance(status_code, int):
key = str(status_code)
account_stat["upstreamStatusCodes"][key] = account_stat["upstreamStatusCodes"].get(key, 0) + 1
if len(failover_samples) < 20:
failover_samples.append({
**base,
"switchCount": item.get("switch_count"),
"maxSwitches": item.get("max_switches"),
})
elif "openai.account_select_failed" in line:
event_line_count += 1
counts["selectFailedEvents"] += 1
if request is not None:
request["selectFailedCount"] += 1
else:
missing_request_ids["selectFailed"] += 1
if len(select_failed_samples) < 20:
select_failed_samples.append({
**base,
"excludedAccountCount": item.get("excluded_account_count"),
"error": compact_error(item.get("error")),
})
elif "openai.forward_failed" in line:
event_line_count += 1
counts["forwardFailedEvents"] += 1
if request is not None:
request["forwardFailedCount"] += 1
else:
missing_request_ids["forwardFailed"] += 1
if account_stat is not None:
account_stat["forwardFailedEvents"] += 1
if isinstance(request_id, str) and request_id:
account_stat["forwardFailedRequestIds"].add(request_id)
elif "openai.pool_mode_same_account_retry" in line or "gateway.failover_same_account_retry" in line:
event_line_count += 1
counts["sameAccountRetryEvents"] += 1
if request is not None:
request["sameAccountRetryCount"] += 1
else:
missing_request_ids["sameAccountRetry"] += 1
if raw_account_stat is not None:
raw_account_stat["sameAccountRetryEvents"] += 1
elif "http request completed" in line and gateway_request_path(item.get("path")):
event_line_count += 1
counts["gatewayCompletedEvents"] += 1
status_code = item.get("status_code")
if request is not None and isinstance(status_code, int):
request["finalStatus"] = status_code
elif request is None:
missing_request_ids["gatewayCompleted"] += 1
if isinstance(status_code, int):
key = str(status_code)
final_statuses[key] = final_statuses.get(key, 0) + 1
if status_code >= 400:
counts["gatewayFinalErrorEvents"] += 1
correlated = [item for item in requests.values() if item["failoverCount"] > 0]
succeeded = [item for item in correlated if isinstance(item.get("finalStatus"), int) and item["finalStatus"] < 400]
failed = [item for item in correlated if isinstance(item.get("finalStatus"), int) and item["finalStatus"] >= 400]
incomplete = [item for item in correlated if not isinstance(item.get("finalStatus"), int)]
cooled_then_failover = [item for item in correlated if item["tempUnschedulableCount"] > 0]
forward_with_http_success = [
item for item in requests.values()
if item["forwardFailedCount"] > 0 and isinstance(item.get("finalStatus"), int) and item["finalStatus"] < 400
]
forward_with_http_error = [
item for item in requests.values()
if item["forwardFailedCount"] > 0 and isinstance(item.get("finalStatus"), int) and item["finalStatus"] >= 400
]
forward_outcome_missing = [
item for item in requests.values()
if item["forwardFailedCount"] > 0 and not isinstance(item.get("finalStatus"), int)
]
for request in correlated:
request_id = request.get("requestId")
if not isinstance(request_id, str):
continue
for current in account_policy.values():
if request_id not in current["failoverRequestIds"]:
continue
final_status = request.get("finalStatus")
if isinstance(final_status, int) and final_status < 400:
current["failoverRecoveredRequestIds"].add(request_id)
elif isinstance(final_status, int):
current["failoverFailedRequestIds"].add(request_id)
else:
current["failoverOutcomeMissingRequestIds"].add(request_id)
account_policy_rows = []
for current in account_policy.values():
account_policy_rows.append({
"accountId": current["accountId"], "accountName": current["accountName"],
"tempUnschedulableEvents": current["tempUnschedulableEvents"],
"failoverEvents": current["failoverEvents"], "failoverRequests": len(current["failoverRequestIds"]),
"failoverRecovered": len(current["failoverRecoveredRequestIds"]),
"failoverFailed": len(current["failoverFailedRequestIds"]),
"failoverOutcomeMissing": len(current["failoverOutcomeMissingRequestIds"]),
"forwardFailedEvents": current["forwardFailedEvents"], "forwardFailedRequests": len(current["forwardFailedRequestIds"]),
"sameAccountRetryEvents": current["sameAccountRetryEvents"],
"_failureRequestIds": sorted(current["failoverRequestIds"] | current["forwardFailedRequestIds"]),
"upstreamStatusBuckets": [
{"statusCode": int(code), "count": count}
for code, count in sorted(current["upstreamStatusCodes"].items(), key=lambda pair: int(pair[0]))
],
})
sample_limit = 20 if PAYLOAD.get("full") is True else (2 if PAYLOAD.get("account") is not None else 5)
completed = counts["gatewayCompletedEvents"]
final_errors = counts["gatewayFinalErrorEvents"]
result = {
"source": source,
"scope": "group-membership-window",
"attribution": "Events with group_id are exact. Events without group_id are attributed by account membership and request_id correlation; shared-account events can appear in more than one group drilldown.",
"eventLineCount": event_line_count,
**counts,
"tempUnschedulableAccounts": sorted(temp_accounts.values(), key=lambda item: (-item["count"], str(item.get("accountName") or ""))),
"accountPolicyMetrics": sorted(account_policy_rows, key=lambda item: int(item.get("accountId") or 0)),
"accountPolicyAttribution": "Failover and forward-failure request ids enter group scoring only when system-log events carry group_id. Account temp-unschedulable and same-account retry counts are account-wide operational signals and are not added to the failure-rate numerator.",
"failoverRequestCount": len(correlated),
"failoverSucceededRequestCount": len(succeeded),
"failoverFailedRequestCount": len(failed),
"failoverOutcomeMissingRequestCount": len(incomplete),
"cooledThenFailoverRequestCount": len(cooled_then_failover),
"forwardFailureWithHttpSuccessRequestCount": len(forward_with_http_success),
"forwardFailureWithHttpErrorRequestCount": len(forward_with_http_error),
"forwardFailureOutcomeMissingRequestCount": len(forward_outcome_missing),
"requestIdCoverage": {
"missingByEventType": missing_request_ids,
"tempToFailoverCorrelationAvailable": counts["tempUnschedulableEvents"] > missing_request_ids["tempUnschedulable"],
},
"gatewayFinalStatusBuckets": [
{"statusCode": int(code), "count": count}
for code, count in sorted(final_statuses.items(), key=lambda pair: int(pair[0]))
],
"gatewayOutcomeEvidence": {
"completedLogEventCount": completed,
"finalErrorLogEventCount": final_errors,
"observedFinalErrorShare": round(final_errors / completed, 4) if completed > 0 else None,
"denominator": "gateway http request completed log events for Responses, Messages, and Chat Completions paths",
"coverage": "since-window-and-tail-bounded",
"completeTrafficDenominatorGuaranteed": False,
},
"requestDrilldown": {
"succeededAfterFailover": [item["requestId"] for item in succeeded[:sample_limit]],
"failedAfterFailover": [item["requestId"] for item in failed[:sample_limit]],
"outcomeMissingAfterFailover": [item["requestId"] for item in incomplete[:sample_limit]],
"selectFailed": [item["requestId"] for item in requests.values() if item["selectFailedCount"] > 0][:sample_limit],
"forwardFailureWithHttpSuccess": [item["requestId"] for item in forward_with_http_success[:sample_limit]],
"forwardFailureWithHttpError": [item["requestId"] for item in forward_with_http_error[:sample_limit]],
"forwardFailureOutcomeMissing": [item["requestId"] for item in forward_outcome_missing[:sample_limit]],
},
"assessment": (
"failover-observed-with-success" if succeeded else
"failover-observed-without-success" if correlated else
"no-failover-observed"
),
"disclosure": "Rule matches require account_temp_unschedulable evidence; status codes alone are not treated as rule hits. Request outcomes are joined only by request_id. HTTP success can coexist with forward_failed on streaming responses and does not prove the client received a complete terminal event.",
}
if PAYLOAD.get("full") is True:
result["samples"] = {
"tempUnschedulable": temp_samples,
"failovers": failover_samples,
"selectFailures": select_failed_samples,
}
return result
def observed_runtime_errors(token, accounts, group_id, platform=None):
selector = PAYLOAD.get("account")
selected = [item for item in accounts if selector is None or str(item.get("id")) == str(selector) or item.get("name") == selector]
if selector is not None and len(selected) != 1:
raise RuntimeError("account-not-found-or-ambiguous: " + str(selector))
selected_ids = set(str(item.get("id")) for item in selected)
stats = {}
for item in selected:
stats[str(item.get("id"))] = {
"accountName": item.get("name"),
"accountId": item.get("id"),
"upstreamEventCount": 0,
"observedStatusCodes": {},
"forwardFailureCount": 0,
"forwardFailureStatusCodes": {},
"lastSeenAt": None,
"errorSamples": [],
}
native_ops = native_ops_snapshot(token, group_id, platform)
lines, system_log_source = native_system_log_lines(token, platform)
if lines is None:
lines = runtime_log_lines()
system_log_source["fallbackUsed"] = True
else:
marker_queries = [item for item in system_log_source.get("queries", []) if item.get("marker") != "http request completed"]
marker_total = sum(item.get("total", 0) for item in marker_queries if isinstance(item.get("total"), int))
overview_summary = ((native_ops.get("overview") or {}).get("summary") or {})
upstream_count = overview_summary.get("upstream_error_count_excl_429_529") or 0
upstream_count += overview_summary.get("upstream_429_count") or 0
upstream_count += overview_summary.get("upstream_529_count") or 0
if marker_total == 0 and upstream_count > 0:
runtime_event_lines = [
line for line in runtime_log_lines()
if any(marker in line for marker in (
"account_temp_unschedulable", "openai.upstream_failover_switching",
"openai.account_select_failed", "openai.forward_failed",
))
]
lines.extend(runtime_event_lines)
system_log_source["source"] = "sub2api-native-admin-ops-with-runtime-event-fallback"
system_log_source["fallbackUsed"] = True
system_log_source["fallbackReason"] = "native system-log index returned no policy events while the native overview reported upstream errors"
system_log_source["fallbackEventCount"] = len(runtime_event_lines)
effectiveness = policy_effectiveness(lines, selected if selector is not None else accounts, system_log_source.get("source"), group_id)
effectiveness["sourceStatus"] = system_log_source
overview_summary = ((native_ops.get("overview") or {}).get("summary") or {})
native_ops["responseHeaderTimeout"] = runtime_response_header_timeout()
native_ops["requestTiming"] = native_request_timing(token, group_id, platform, overview_summary)
native_ops["customerErrors"] = native_customer_error_profile(token, group_id, platform)
native_ops["accountQuality"] = account_quality_scores(token, selected, group_id, native_ops, effectiveness)
request_total = overview_summary.get("request_count_total")
error_total = overview_summary.get("error_count_total")
if isinstance(request_total, int) and isinstance(error_total, int):
effectiveness["gatewayOutcomeEvidence"] = {
"source": "sub2api-native-admin-ops-dashboard-overview",
"requestCountTotal": request_total,
"errorCountTotal": error_total,
"errorRate": overview_summary.get("error_rate"),
"upstreamErrorRate": overview_summary.get("upstream_error_rate"),
"businessLimitedCount": overview_summary.get("business_limited_count"),
"startTime": overview_summary.get("start_time"),
"endTime": overview_summary.get("end_time"),
"completeTrafficDenominatorGuaranteed": True,
}
scanned = 0
for line in lines:
if "account_upstream_error" not in line and "openai.forward_failed" not in line:
continue
json_start = line.find("{")
if json_start < 0:
continue
try:
item = json.loads(line[json_start:])
except json.JSONDecodeError:
continue
account_id = item.get("account_id") if isinstance(item, dict) else None
account_key = str(account_id)
if account_key not in selected_ids:
continue
scanned += 1
at = line.split("\t", 1)[0].strip() or None
current = stats[account_key]
if at and (current["lastSeenAt"] is None or at > current["lastSeenAt"]):
current["lastSeenAt"] = at
if "account_upstream_error" in line:
code = item.get("status_code")
if isinstance(code, int):
key = str(code)
current["observedStatusCodes"][key] = current["observedStatusCodes"].get(key, 0) + 1
current["upstreamEventCount"] += 1
continue
current["forwardFailureCount"] += 1
error = compact_error(item.get("error"))
match = re.search(r"(?:upstream error:\\s*|status(?:_code)?[=:]\\s*)(\\d{3})", error, re.IGNORECASE)
if match:
key = match.group(1)
current["forwardFailureStatusCodes"][key] = current["forwardFailureStatusCodes"].get(key, 0) + 1
if error and error not in current["errorSamples"] and len(current["errorSamples"]) < 3:
current["errorSamples"].append(error)
rows = []
aggregate = {}
for item in selected:
current = stats[str(item.get("id"))]
observed = [{"statusCode": int(code), "count": count} for code, count in sorted(current["observedStatusCodes"].items(), key=lambda pair: int(pair[0]))]
forwarded = [{"statusCode": int(code), "count": count} for code, count in sorted(current["forwardFailureStatusCodes"].items(), key=lambda pair: int(pair[0]))]
for entry in observed:
key = str(entry["statusCode"])
aggregate[key] = aggregate.get(key, 0) + entry["count"]
current["observedStatusCodes"] = observed
current["forwardFailureStatusCodes"] = forwarded
analysis = ops_error_analysis(token, item, observed)
has_ops_errors = analysis.get("status") == "available" and (analysis.get("opsRecordCount") or 0) > 0
if current["upstreamEventCount"] > 0 or current["forwardFailureCount"] > 0 or has_ops_errors:
if PAYLOAD.get("full") is not True and analysis.get("status") == "available":
analysis = {
"source": analysis.get("source"),
"status": analysis.get("status"),
"opsRecordCount": analysis.get("opsRecordCount"),
"recordsTruncated": analysis.get("recordsTruncated"),
"statusBuckets": analysis.get("statusBuckets"),
"causeBuckets": analysis.get("causeBuckets"),
"rootCause": analysis.get("rootCause"),
"errorRateEvidence": analysis.get("errorRateEvidence"),
"concurrencyRecommendation": analysis.get("concurrencyRecommendation"),
"runtimeConcurrency": analysis.get("runtimeConcurrency"),
"samples": (analysis.get("samples") or [])[:1],
}
current["opsAnalysis"] = analysis
if PAYLOAD.get("full") is not True:
current["errorSamples"] = current["errorSamples"][:1]
if current["upstreamEventCount"] > 0 or current["forwardFailureCount"] > 0 or has_ops_errors or selector is not None:
rows.append(current)
root_causes = {}
account_roots = []
for row in rows:
analysis = row.get("opsAnalysis")
if not isinstance(analysis, dict) or analysis.get("status") != "available":
continue
for bucket in analysis.get("causeBuckets") or []:
cause = bucket.get("cause")
count = bucket.get("count")
if isinstance(cause, str) and isinstance(count, int):
root_causes[cause] = root_causes.get(cause, 0) + count
root = analysis.get("rootCause") or {}
account_roots.append({
"accountId": row.get("accountId"),
"accountName": row.get("accountName"),
"dominantCause": root.get("dominantCause"),
"dominantCount": root.get("dominantCount"),
"dominantShare": root.get("dominantShare"),
})
ranked_causes = sorted(root_causes.items(), key=lambda pair: (-pair[1], pair[0]))
native_overview = native_ops.get("overview") or {}
native_overview_values = native_overview.get("summary") if isinstance(native_overview.get("summary"), dict) else native_overview
native_error_rate_available = isinstance(native_overview_values.get("error_rate") if isinstance(native_overview_values, dict) else None, (int, float))
account_count_with_errors = sum(
1 for item in rows
if item["upstreamEventCount"] > 0
or item["forwardFailureCount"] > 0
or (((item.get("opsAnalysis") or {}).get("opsRecordCount") or 0) > 0)
)
if selector is None and PAYLOAD.get("full") is not True:
for row in rows:
row.pop("opsAnalysis", None)
if PAYLOAD.get("full") is not True:
overview = ((native_ops.get("overview") or {}).get("summary") or {})
availability = ((native_ops.get("accountAvailability") or {}).get("summary") or {})
concurrency = ((native_ops.get("concurrency") or {}).get("summary") or {})
sink = ((native_ops.get("systemLogSink") or {}).get("health") or {})
unavailable_accounts = availability.get("unavailableAccounts") or []
if selector is not None:
unavailable_accounts = [item for item in unavailable_accounts if item.get("accountId") in selected_ids]
native_ops = {
"source": native_ops.get("source"),
"overview": {
"status": (native_ops.get("overview") or {}).get("status"),
"healthScore": overview.get("health_score"),
"requestCount": overview.get("request_count_total"),
"errorCount": overview.get("error_count_total"),
"errorRate": overview.get("error_rate"),
"upstreamErrorRate": overview.get("upstream_error_rate"),
"businessLimitedCount": overview.get("business_limited_count"),
},
"responseHeaderTimeout": native_ops.get("responseHeaderTimeout"),
"requestTiming": native_ops.get("requestTiming"),
"customerErrors": native_ops.get("customerErrors"),
"accountQuality": native_ops.get("accountQuality"),
"accountAvailability": {
"status": (native_ops.get("accountAvailability") or {}).get("status"),
"group": availability.get("group"),
"unavailableAccountCount": availability.get("unavailableAccountCount"),
"unavailableAccounts": unavailable_accounts,
},
"concurrency": {
"status": (native_ops.get("concurrency") or {}).get("status"),
"group": concurrency.get("group"),
},
"systemLogSink": {
"status": (native_ops.get("systemLogSink") or {}).get("status"),
"queueDepth": sink.get("queue_depth"),
"droppedCount": sink.get("dropped_count"),
"writeFailedCount": sink.get("write_failed_count"),
},
}
source_status = effectiveness.get("sourceStatus") or {}
compact_source_status = {
"status": source_status.get("status"),
"source": source_status.get("source"),
"fallbackUsed": source_status.get("fallbackUsed"),
"fallbackReason": source_status.get("fallbackReason"),
"fallbackEventCount": source_status.get("fallbackEventCount"),
"nativeIndexedEventCounts": {
item.get("marker"): item.get("total")
for item in source_status.get("queries", [])
if isinstance(item, dict)
},
}
drilldown = effectiveness.get("requestDrilldown") or {}
effectiveness = {
"source": effectiveness.get("source"),
"scope": effectiveness.get("scope"),
"attribution": effectiveness.get("attribution"),
"assessment": effectiveness.get("assessment"),
"ruleMatches": {
"tempUnschedulableEvents": effectiveness.get("tempUnschedulableEvents"),
"accounts": effectiveness.get("tempUnschedulableAccounts"),
},
"failover": {
"events": effectiveness.get("failoverEvents"),
"requests": effectiveness.get("failoverRequestCount"),
"succeeded": effectiveness.get("failoverSucceededRequestCount"),
"failed": effectiveness.get("failoverFailedRequestCount"),
"outcomeMissing": effectiveness.get("failoverOutcomeMissingRequestCount"),
},
"forwardFailure": {
"events": effectiveness.get("forwardFailedEvents"),
"withHttpSuccess": effectiveness.get("forwardFailureWithHttpSuccessRequestCount"),
"withHttpError": effectiveness.get("forwardFailureWithHttpErrorRequestCount"),
"outcomeMissing": effectiveness.get("forwardFailureOutcomeMissingRequestCount"),
},
"selectFailedEvents": effectiveness.get("selectFailedEvents"),
"requestIdCoverage": effectiveness.get("requestIdCoverage"),
"gatewayOutcomeEvidence": effectiveness.get("gatewayOutcomeEvidence"),
"requestDrilldown": {key: value for key, value in drilldown.items() if value},
"sourceStatus": compact_source_status,
"disclosure": effectiveness.get("disclosure"),
}
if selector is not None:
native_ops["accountAvailability"].pop("group", None)
native_ops.pop("concurrency", None)
native_ops.pop("systemLogSink", None)
effectiveness["sourceStatus"] = {
"source": compact_source_status.get("source"),
"fallbackUsed": compact_source_status.get("fallbackUsed"),
"fallbackReason": compact_source_status.get("fallbackReason"),
}
return {
"window": {"since": PAYLOAD.get("since"), "tail": PAYLOAD.get("tail"), "matchedLogEvents": scanned},
"aggregateObservedStatusCodes": [{"statusCode": int(code), "count": count} for code, count in sorted(aggregate.items(), key=lambda pair: int(pair[0]))],
"accountCountWithErrors": account_count_with_errors,
"rootCauseAnalysis": {
"source": "ops-upstream-errors",
"scope": "account-window-for-group-members",
"attribution": "The upstream-errors endpoint is account scoped. If accounts belong to multiple groups, these cause buckets are shared evidence and must not be added across groups.",
"causeBuckets": [{"cause": cause, "count": count} for cause, count in ranked_causes],
"dominantCause": ranked_causes[0][0] if ranked_causes else None,
"dominantCount": ranked_causes[0][1] if ranked_causes else 0,
"accountRoots": account_roots,
"errorRateAvailable": native_error_rate_available,
},
"nativeOps": native_ops,
"policyEffectiveness": effectiveness,
**(
{"accounts": rows}
if selector is not None or PAYLOAD.get("full") is True
else {"accountDrilldown": "rerun runtime errors with --account <name-or-id> for per-account evidence"}
),
"message": "observedStatusCodes comes only from account_upstream_error; forward failures are supplemental message evidence",
}
def selected_account_details(token, accounts):
selectors = PAYLOAD.get("accounts")
if not isinstance(selectors, list) or not selectors:
selector = PAYLOAD.get("account")
selectors = [selector] if isinstance(selector, str) and selector else []
details = []
selected_ids = set()
for raw_selector in selectors:
selector = str(raw_selector)
detail = find_account(token, accounts, selector)
account_id = detail.get("id")
if account_id in selected_ids:
raise RuntimeError("duplicate-account-selector: " + selector)
if detail.get("type") != "apikey":
raise RuntimeError("runtime temp-unschedulable policy requires an apikey account: " + selector)
selected_ids.add(account_id)
details.append(detail)
return details
def mutation_plan(detail, action, include_rules):
credentials = dict(detail.get("credentials") or {})
before = normalize_policy(credentials)
if action == "apply":
template = PAYLOAD.get("selectedTemplate")
if not isinstance(template, dict):
raise RuntimeError("selected runtime template missing")
desired = dict(credentials)
desired.update(template.get("credentials") or {})
desired_policy = normalize_policy(desired)
exists = "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials
change = "noop" if before == desired_policy else ("update" if exists else "create")
else:
desired = dict(credentials)
desired.pop("temp_unschedulable_enabled", None)
desired.pop("temp_unschedulable_rules", None)
desired_policy = normalize_policy(desired)
change = "delete" if before != desired_policy or "temp_unschedulable_enabled" in credentials or "temp_unschedulable_rules" in credentials else "noop"
return {
"accountName": detail.get("name"),
"accountId": detail.get("id"),
"change": change,
"before": policy_summary(before, include_rules),
"desired": policy_summary(desired_policy, include_rules),
"template": PAYLOAD.get("selectedTemplate", {}).get("id") if isinstance(PAYLOAD.get("selectedTemplate"), dict) else None,
"_desiredCredentials": desired,
"_desiredPolicy": desired_policy,
}
def write_mutation_plan(token, plan):
if plan["change"] == "noop":
return {"attempted": False, "status": "noop", "error": None}
try:
data_of(curl_api("PUT", f"/api/v1/admin/accounts/{plan['accountId']}", bearer=token, payload={"credentials": plan["_desiredCredentials"]}), "update account runtime config")
return {"attempted": True, "status": "succeeded", "error": None}
except Exception as exc:
return {"attempted": True, "status": "failed", "error": compact_error(str(exc))}
def reconcile_mutation_plan(token, plan, write, include_rules):
item = {key: value for key, value in plan.items() if not key.startswith("_")}
item["write"] = write
try:
actual_detail = account_detail(token, plan["accountId"])
actual_policy = normalize_policy(actual_detail.get("credentials"))
item["actual"] = policy_summary(actual_policy, include_rules)
item["reconciled"] = actual_policy == plan["_desiredPolicy"]
item["reconcileError"] = None
except Exception as exc:
item["actual"] = None
item["reconciled"] = False
item["reconcileError"] = compact_error(str(exc))
return item
def runtime_result():
token = login()
action = PAYLOAD["action"]
if action == "errors" and PAYLOAD.get("allGroups") is True:
groups = list_groups(token, PAYLOAD.get("platform"))
overview = all_group_error_overview(token, groups)
return {
"ok": True,
"operation": "errors",
"mutation": False,
"endpoint": "host-docker" if RUNTIME_MODE == "host-docker" else "k3s-service",
"allGroups": overview,
"valuesPrinted": False,
}
group_selector = PAYLOAD.get("group")
group_platform = PAYLOAD.get("platform") if action in ("errors", "infrastructure") else "openai"
if action in ("errors", "infrastructure") and group_selector is None:
group_platform = "openai"
groups = list_groups(token, group_platform)
selector = str(group_selector if group_selector is not None else PAYLOAD["groupName"])
matching_groups = [item for item in groups if str(item.get("id")) == selector or item.get("name") == selector]
if len(matching_groups) > 1:
raise RuntimeError("group-selector-ambiguous: " + selector)
group = matching_groups[0] if matching_groups else None
if not isinstance(group, dict) or group.get("id") is None:
raise RuntimeError("group-not-found: " + selector)
effective_platform = group_platform or group.get("platform")
accounts = list_group_accounts(token, group["id"], effective_platform)
base = {
"group": {"id": group.get("id"), "name": group.get("name"), "platform": effective_platform, "status": group.get("status")},
"templates": [template_summary(item) for item in PAYLOAD["templates"]],
"endpoint": "host-docker" if RUNTIME_MODE == "host-docker" else "k3s-service",
"valuesPrinted": False,
}
if action == "list":
details = [account_detail(token, item["id"]) for item in accounts]
return {**base, "ok": True, "operation": "list", "mutation": False, "accountCount": len(details), "accounts": [account_summary(item) for item in sorted(details, key=lambda value: value.get("name") or "")]}
if action == "errors":
errors = observed_runtime_errors(token, accounts, group["id"], effective_platform)
return {**base, "ok": True, "operation": "errors", "mutation": False, "errors": errors}
detail = find_account(token, accounts, str(PAYLOAD["account"])) if action in ("get", "infrastructure") else None
if action == "get":
return {**base, "ok": True, "operation": "get", "mutation": False, "account": account_summary(detail, True)}
if action == "infrastructure":
return {
**base,
"ok": True,
"operation": "infrastructure",
"mutation": False,
"account": account_summary(detail, False),
"infrastructure": infrastructure_snapshot(token, detail),
}
include_plan_rules = PAYLOAD.get("full") is True
details = selected_account_details(token, accounts)
plans = [mutation_plan(item, action, include_plan_rules) for item in details]
changed = sum(1 for plan in plans if plan["change"] != "noop")
if PAYLOAD.get("confirm") is not True:
items = []
for plan in plans:
item = {key: value for key, value in plan.items() if not key.startswith("_")}
item["write"] = {"attempted": False, "status": "dry-run", "error": None}
item["actual"] = item["before"]
item["reconciled"] = None
item["reconcileError"] = None
items.append(item)
summary = {"selected": len(items), "changed": changed, "writeSucceeded": 0, "writeFailed": 0, "reconciled": 0, "mismatched": 0}
return {**base, "ok": True, "operation": action, "mode": "dry-run", "mutation": False, "partialWrite": False, "summary": summary, "items": items, "credentialsPrinted": False}
writes = [write_mutation_plan(token, plan) for plan in plans]
items = [reconcile_mutation_plan(token, plan, write, include_plan_rules) for plan, write in zip(plans, writes)]
write_succeeded = sum(1 for item in items if item["write"]["status"] == "succeeded")
write_failed = sum(1 for item in items if item["write"]["status"] == "failed")
reconciled = sum(1 for item in items if item["reconciled"] is True)
mismatched = sum(1 for item in items if item["reconciled"] is False)
summary = {"selected": len(items), "changed": changed, "writeSucceeded": write_succeeded, "writeFailed": write_failed, "reconciled": reconciled, "mismatched": mismatched}
ok = write_failed == 0 and mismatched == 0
return {**base, "ok": ok, "operation": action, "mode": "confirmed", "mutation": write_succeeded > 0, "partialWrite": write_succeeded > 0 and write_failed > 0, "summary": summary, "items": items, "credentialsPrinted": False}
try:
result = runtime_result()
except Exception as exc:
result = {"ok": False, "operation": PAYLOAD.get("action"), "mutation": False, "error": str(exc), "valuesPrinted": False}
print(json.dumps(result, ensure_ascii=False, indent=2))
sys.exit(0 if result.get("ok") else 1)
PY
`;
}