mirror of
https://github.com/tiennm99/ccs.git
synced 2026-07-16 04:18:05 +00:00
feat(proxy,cli): emit lifecycle stages with x-ccs-request-id propagation
Wrap proxy server entry edge in withRequestContext so every inbound request gets a requestId (reused from x-ccs-request-id header when valid UUID-ish, freshly minted otherwise). messages-route emits 7 stages: intake / auth / transform / route / dispatch / upstream / respond, each with latencyMs and structured error metadata on failure. CCS CLI entry (ccs.ts) wraps main() in runWithRequestId and emits cli.command.start / complete / failed stages so command lifecycle is correlatable end-to-end. Refs #1141, #1138
This commit is contained in:
+33
-3
@@ -84,7 +84,7 @@ import { execClaude, stripAnthropicRoutingEnv, stripBrowserEnv } from './utils/s
|
||||
import { isDeprecatedGlmtProfileName, normalizeDeprecatedGlmtEnv } from './utils/glmt-deprecation';
|
||||
import { createOpenAICompatLaunchSettings } from './utils/openai-compat-launch-settings';
|
||||
import { maybeWarnAboutResumeLaneMismatch } from './auth/resume-lane-warning';
|
||||
import { createLogger } from './services/logging';
|
||||
import { createLogger, runWithRequestId } from './services/logging';
|
||||
import { buildCodexBrowserMcpOverrides } from './utils/browser-codex-overrides';
|
||||
import type { ProfileDetectionResult } from './auth/profile-detector';
|
||||
import type { BrowserLaunchOverride } from './utils/browser';
|
||||
@@ -1683,5 +1683,35 @@ process.on('SIGINT', () => {
|
||||
}
|
||||
});
|
||||
|
||||
// Run main
|
||||
main().catch(handleError);
|
||||
// Run main inside a per-invocation request context so all backend logging
|
||||
// emitted during this CLI run shares a single requestId. CLI text output
|
||||
// (stdout/stderr) is unaffected — the requestId lives in logs only.
|
||||
const cliEntryStartedAt = Date.now();
|
||||
const cliEntryLogger = createLogger('cli:entry');
|
||||
runWithRequestId(() => {
|
||||
cliEntryLogger.stage('intake', 'cli.command.start', 'CLI invocation started', {
|
||||
argv: process.argv.slice(2),
|
||||
});
|
||||
return main()
|
||||
.then(() => {
|
||||
cliEntryLogger.stage(
|
||||
'respond',
|
||||
'cli.command.complete',
|
||||
'CLI invocation completed',
|
||||
{ exitCode: process.exitCode ?? 0 },
|
||||
{ latencyMs: Date.now() - cliEntryStartedAt }
|
||||
);
|
||||
})
|
||||
.catch((err) => {
|
||||
const error =
|
||||
err instanceof Error
|
||||
? { name: err.name, message: err.message, stack: err.stack }
|
||||
: { name: 'Error', message: String(err) };
|
||||
cliEntryLogger.stage('cleanup', 'cli.command.failed', 'CLI invocation failed', undefined, {
|
||||
level: 'error',
|
||||
latencyMs: Date.now() - cliEntryStartedAt,
|
||||
error,
|
||||
});
|
||||
handleError(err);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -207,11 +207,23 @@ export async function handleProxyMessagesRequest(
|
||||
insecureDispatcher?: Dispatcher
|
||||
): Promise<void> {
|
||||
const transformer = new ProxySseStreamTransformer();
|
||||
const startedAt = Date.now();
|
||||
|
||||
logger.stage('intake', 'request.received', 'Proxy /v1/messages request received', {
|
||||
method: req.method || 'POST',
|
||||
remoteAddress: req.socket.remoteAddress || null,
|
||||
});
|
||||
|
||||
if (!validateIncomingProxyAuth(req.headers, expectedAuthToken)) {
|
||||
logger.warn('auth.invalid', 'Rejected proxy message request with invalid auth token', {
|
||||
remoteAddress: req.socket.remoteAddress || null,
|
||||
});
|
||||
logger.stage(
|
||||
'auth',
|
||||
'auth.invalid',
|
||||
'Rejected proxy message request with invalid auth token',
|
||||
{
|
||||
remoteAddress: req.socket.remoteAddress || null,
|
||||
},
|
||||
{ level: 'warn' }
|
||||
);
|
||||
await pipeWebResponseToNode(
|
||||
transformer.error(401, 'authentication_error', 'Missing or invalid local proxy token'),
|
||||
res
|
||||
@@ -219,11 +231,14 @@ export async function handleProxyMessagesRequest(
|
||||
return;
|
||||
}
|
||||
|
||||
logger.stage('auth', 'auth.ok', 'Proxy auth validated');
|
||||
|
||||
let timeoutMs = REQUEST_TIMEOUT_MS;
|
||||
try {
|
||||
const rawBody = await readJsonBody(req);
|
||||
logger.stage('transform', 'request.transform.start', 'Transforming inbound proxy body');
|
||||
const upstream = buildUpstreamRequest(profile, rawBody);
|
||||
logger.info('request.forward', 'Forwarding Anthropic request to OpenAI-compatible upstream', {
|
||||
logger.stage('route', 'request.routed', 'Resolved proxy upstream route', {
|
||||
profileName: upstream.route.profile.profileName,
|
||||
provider: upstream.route.profile.provider,
|
||||
baseUrl: upstream.route.profile.baseUrl,
|
||||
@@ -240,7 +255,8 @@ export async function handleProxyMessagesRequest(
|
||||
res,
|
||||
controller,
|
||||
(source) => {
|
||||
logger.info(
|
||||
logger.stage(
|
||||
'cleanup',
|
||||
'request.disconnect',
|
||||
'Aborting upstream request after local client disconnect',
|
||||
{
|
||||
@@ -252,28 +268,47 @@ export async function handleProxyMessagesRequest(
|
||||
);
|
||||
|
||||
try {
|
||||
logger.stage('dispatch', 'upstream.dispatch', 'Dispatching upstream fetch', {
|
||||
profileName: profile.profileName,
|
||||
});
|
||||
const upstreamResponse = await fetch(
|
||||
resolveOpenAIChatCompletionsUrl(upstream.route.profile.baseUrl),
|
||||
buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher)
|
||||
);
|
||||
logger.info('response.received', 'Received upstream response', {
|
||||
logger.stage('upstream', 'upstream.response', 'Received upstream response', {
|
||||
profileName: profile.profileName,
|
||||
routedProfileName: upstream.route.profile.profileName,
|
||||
status: upstreamResponse.status,
|
||||
});
|
||||
const response = await transformer.transform(upstreamResponse);
|
||||
await pipeWebResponseToNode(response, res);
|
||||
logger.stage('respond', 'request.respond', 'Proxy response written', undefined, {
|
||||
latencyMs: Date.now() - startedAt,
|
||||
});
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
cleanupDisconnectHandlers();
|
||||
}
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : 'Unknown proxy error';
|
||||
logger.error('request.failed', 'Proxy message request failed', {
|
||||
profileName: profile.profileName,
|
||||
error: message,
|
||||
abort: error instanceof Error && error.name === 'AbortError',
|
||||
});
|
||||
const errInfo = {
|
||||
name: error instanceof Error ? error.name : 'Error',
|
||||
message,
|
||||
};
|
||||
logger.stage(
|
||||
'cleanup',
|
||||
'request.failed',
|
||||
'Proxy message request failed',
|
||||
{
|
||||
profileName: profile.profileName,
|
||||
abort: error instanceof Error && error.name === 'AbortError',
|
||||
},
|
||||
{
|
||||
level: 'error',
|
||||
latencyMs: Date.now() - startedAt,
|
||||
error: errInfo,
|
||||
}
|
||||
);
|
||||
const status =
|
||||
error instanceof Error && error.name === 'AbortError'
|
||||
? 502
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import * as http from 'http';
|
||||
import { randomUUID } from 'crypto';
|
||||
import { Agent } from 'undici';
|
||||
import type { OpenAICompatProfileConfig } from '../profile-router';
|
||||
import { OPENAI_COMPAT_PROXY_SERVICE_NAME } from '../proxy-daemon-paths';
|
||||
import { createLogger } from '../../services/logging';
|
||||
import { createLogger, withRequestContext } from '../../services/logging';
|
||||
import {
|
||||
handleProxyMessagesRequest,
|
||||
handleProxyModelsRequest,
|
||||
@@ -10,6 +11,18 @@ import {
|
||||
} from './messages-route';
|
||||
import { writeJson } from './http-helpers';
|
||||
|
||||
const REQUEST_ID_HEADER = 'x-ccs-request-id';
|
||||
// Loose UUID-ish guard: accepts UUIDs and similar opaque ids; rejects empty / control chars.
|
||||
const REQUEST_ID_PATTERN = /^[A-Za-z0-9._-]{8,128}$/;
|
||||
|
||||
function resolveInboundRequestId(headers: http.IncomingHttpHeaders): string {
|
||||
const raw = headers[REQUEST_ID_HEADER];
|
||||
if (typeof raw === 'string' && REQUEST_ID_PATTERN.test(raw.trim())) {
|
||||
return raw.trim();
|
||||
}
|
||||
return randomUUID();
|
||||
}
|
||||
|
||||
export interface OpenAICompatProxyServerOptions {
|
||||
profile: OpenAICompatProfileConfig;
|
||||
host?: string;
|
||||
@@ -28,13 +41,25 @@ export function startOpenAICompatProxyServer(options: OpenAICompatProxyServerOpt
|
||||
const insecureDispatcher = options.insecure
|
||||
? new Agent({ connect: { rejectUnauthorized: false } })
|
||||
: undefined;
|
||||
const server = http.createServer(async (req, res) => {
|
||||
const method = req.method || 'GET';
|
||||
const requestUrl = req.url || '/';
|
||||
const parsedUrl = new URL(requestUrl, 'http://127.0.0.1');
|
||||
const pathname =
|
||||
parsedUrl.pathname.length > 1 ? parsedUrl.pathname.replace(/\/+$/, '') : parsedUrl.pathname;
|
||||
const server = http.createServer((req, res) => {
|
||||
const requestId = resolveInboundRequestId(req.headers);
|
||||
res.setHeader(REQUEST_ID_HEADER, requestId);
|
||||
void withRequestContext({ requestId }, async () => {
|
||||
const method = req.method || 'GET';
|
||||
const requestUrl = req.url || '/';
|
||||
const parsedUrl = new URL(requestUrl, 'http://127.0.0.1');
|
||||
const pathname =
|
||||
parsedUrl.pathname.length > 1 ? parsedUrl.pathname.replace(/\/+$/, '') : parsedUrl.pathname;
|
||||
await handleProxyRequest(req, res, method, pathname);
|
||||
});
|
||||
});
|
||||
|
||||
async function handleProxyRequest(
|
||||
req: http.IncomingMessage,
|
||||
res: http.ServerResponse,
|
||||
method: string,
|
||||
pathname: string
|
||||
): Promise<void> {
|
||||
if ((method === 'GET' || method === 'HEAD') && pathname === '/health') {
|
||||
if (method === 'HEAD') {
|
||||
res.writeHead(200, { 'Content-Type': 'application/json' });
|
||||
@@ -105,7 +130,7 @@ export function startOpenAICompatProxyServer(options: OpenAICompatProxyServerOpt
|
||||
pathname,
|
||||
});
|
||||
writeJson(res, 404, { error: 'Not found' });
|
||||
});
|
||||
}
|
||||
|
||||
logger.info('server.start', 'OpenAI-compatible proxy server listening', {
|
||||
baseUrl: `http://${host}:${options.port}`,
|
||||
|
||||
Reference in New Issue
Block a user