mirror of
https://github.com/teamchong/pxpipe.git
synced 2026-07-22 02:02:51 +02:00
fix: bypass header, persisted model scope, named skip reason (#111)
- x-pxpipe-bypass: per-request opt-out for subprocesses that inherited ANTHROPIC_BASE_URL; body forwarded byte-for-byte, header stripped. - Dashboard model-scope toggles persist to the config file's models key (atomic write-then-rename); explicit PXPIPE_MODELS env still wins. - Skip log names the model: skip(unsupported=<model>).
This commit is contained in:
+11
-3
@@ -523,6 +523,7 @@ const STRIP_REQ_HEADERS = new Set([
|
||||
'content-length', // we recompute
|
||||
'expect',
|
||||
'accept-encoding', // let upstream choose
|
||||
'x-pxpipe-bypass', // pxpipe-only opt-out signal; never forwarded upstream
|
||||
]);
|
||||
|
||||
const STRIP_RES_HEADERS = new Set([
|
||||
@@ -788,10 +789,17 @@ export function createProxy(config: ProxyConfig = {}) {
|
||||
};
|
||||
|
||||
// Transform only known shapes; everything else passes through.
|
||||
// Explicit per-request opt-out (#111): a subprocess that merely inherited
|
||||
// ANTHROPIC_BASE_URL (e.g. a plugin's internal `claude` probe) can send
|
||||
// `x-pxpipe-bypass: 1` — via Claude Code's ANTHROPIC_CUSTOM_HEADERS — to
|
||||
// have its traffic forwarded byte-for-byte untouched. Routing and auth
|
||||
// still apply; the header itself is stripped before forwarding.
|
||||
const bypassHeader = req.headers.get('x-pxpipe-bypass');
|
||||
const bypass = bypassHeader !== null && !/^(?:0|false|off|no)$/i.test(bypassHeader.trim());
|
||||
const providerPrefixed = isProviderPrefixedPath(url.pathname);
|
||||
const isMessages = req.method === 'POST' && isAnthropicMessagesPath(url.pathname);
|
||||
const isOpenAIChat = req.method === 'POST' && isOpenAIChatPath(url.pathname);
|
||||
const isOpenAIResponses = req.method === 'POST' && isOpenAIResponsesPath(url.pathname);
|
||||
const isMessages = !bypass && req.method === 'POST' && isAnthropicMessagesPath(url.pathname);
|
||||
const isOpenAIChat = !bypass && req.method === 'POST' && isOpenAIChatPath(url.pathname);
|
||||
const isOpenAIResponses = !bypass && req.method === 'POST' && isOpenAIResponsesPath(url.pathname);
|
||||
const isOpenAIPath = isCanonicalOpenAIPath(
|
||||
url.pathname,
|
||||
req.headers,
|
||||
|
||||
+23
-4
@@ -518,12 +518,20 @@ export class DashboardState {
|
||||
* the developer's actual Claude Code session files. */
|
||||
private readonly ccMapFn: () => Promise<Map<string, ClaudeCodeSessionRef>>;
|
||||
|
||||
/** Host-provided persistence hook for the runtime model scope. The core
|
||||
* override stays in-memory (Edge-safe); a Node host passes a saver that
|
||||
* writes the `models` key of the config file so chip toggles survive a
|
||||
* restart. Best-effort: failures are the hook's problem, never the API's. */
|
||||
private readonly persistModelBases: ((bases: readonly string[]) => void) | undefined;
|
||||
|
||||
constructor(
|
||||
paths?: SessionsPaths,
|
||||
ccMapFn?: () => Promise<Map<string, ClaudeCodeSessionRef>>,
|
||||
persistModelBases?: (bases: readonly string[]) => void,
|
||||
) {
|
||||
this.paths = paths;
|
||||
this.ccMapFn = ccMapFn ?? (() => claudeCodeMap());
|
||||
this.persistModelBases = persistModelBases;
|
||||
}
|
||||
|
||||
private totalsForModel(model: string | undefined): Totals {
|
||||
@@ -1508,25 +1516,36 @@ export class DashboardState {
|
||||
}
|
||||
|
||||
/** POST /fragments/models — add/remove ONE model (Claude or GPT) from the
|
||||
* runtime compress scope. In-memory only; restart resets to the PXPIPE_MODELS
|
||||
* env / built-in default. The model checks read this live. */
|
||||
* runtime compress scope. The model checks read this live. Persisted via
|
||||
* the host's `persistModelBases` hook when provided (Node writes the
|
||||
* config file); otherwise in-memory only and restart resets to the
|
||||
* PXPIPE_MODELS env / built-in default. */
|
||||
handleModelsToggle(model: string, on: boolean): void {
|
||||
const next = new Set(getAllowedModelBases());
|
||||
if (on) next.add(model);
|
||||
else next.delete(model);
|
||||
setAllowedModelBases([...next]);
|
||||
this.applyModelBases([...next]);
|
||||
}
|
||||
|
||||
/** POST /fragments/models with {list} — replace the WHOLE runtime compress
|
||||
* scope from the PXPIPE_MODELS textbox. Same CSV shape as the env var;
|
||||
* empty or off/false/0/no/none = compress nothing. In-memory only. */
|
||||
* empty or off/false/0/no/none = compress nothing. Persistence as above. */
|
||||
handleModelsSet(csv: string): void {
|
||||
const trimmed = csv.trim();
|
||||
const bases =
|
||||
!trimmed || /^(0|false|no|off|none)$/i.test(trimmed)
|
||||
? []
|
||||
: trimmed.split(',').map((s) => s.trim()).filter(Boolean);
|
||||
this.applyModelBases(bases);
|
||||
}
|
||||
|
||||
private applyModelBases(bases: string[]): void {
|
||||
setAllowedModelBases(bases);
|
||||
try {
|
||||
this.persistModelBases?.(bases);
|
||||
} catch {
|
||||
// Persistence is best-effort; the live flip already took effect.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+40
-5
@@ -94,6 +94,35 @@ function applyConfigFileDefaults(): void {
|
||||
}
|
||||
}
|
||||
|
||||
/** Dashboard persistence hook: write the runtime model scope back to the
|
||||
* config file's `models` key so chip toggles survive a restart. Other keys
|
||||
* are preserved; an unreadable file is replaced rather than crashed on.
|
||||
* NOTE: on the next start an explicit PXPIPE_MODELS env still wins over the
|
||||
* persisted value (same precedence as every other config-file default). */
|
||||
function persistModelBasesToConfig(bases: readonly string[]): void {
|
||||
const file = process.env.PXPIPE_CONFIG ?? DEFAULT_CONFIG_FILE;
|
||||
let cfg: Record<string, unknown> = {};
|
||||
try {
|
||||
const parsed = JSON.parse(fs.readFileSync(file, 'utf8')) as unknown;
|
||||
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
|
||||
cfg = parsed as Record<string, unknown>;
|
||||
}
|
||||
} catch {
|
||||
// Missing or invalid file — start fresh with just the models key.
|
||||
}
|
||||
// Empty array round-trips as 'off' via normalizeModelsConfig on load.
|
||||
cfg.models = [...bases];
|
||||
try {
|
||||
fs.mkdirSync(path.dirname(file), { recursive: true });
|
||||
// Write-then-rename so a crash mid-write can't corrupt the config.
|
||||
const tmp = `${file}.tmp-${process.pid}`;
|
||||
fs.writeFileSync(tmp, `${JSON.stringify(cfg, null, 2)}\n`);
|
||||
fs.renameSync(tmp, file);
|
||||
} catch (e) {
|
||||
console.warn(`[pxpipe] could not persist model scope to ${file}: ${(e as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
function parseCli(argv: string[]): RuntimeConfig {
|
||||
// Only flags accepted are --help and --version. Anything else is an
|
||||
// error — there is exactly ONE way to run pxpipe and the dashboard
|
||||
@@ -984,10 +1013,14 @@ async function main(): Promise<void> {
|
||||
// served via the route interception in front of the proxy handler. The
|
||||
// SessionsPaths handle lets the dashboard surface session/disk/stats data
|
||||
// without reaching back into module-scope globals.
|
||||
const dashboard = new DashboardState({
|
||||
eventsFile: opts.eventsFile,
|
||||
sidecarDir: bodySidecarDir,
|
||||
});
|
||||
const dashboard = new DashboardState(
|
||||
{
|
||||
eventsFile: opts.eventsFile,
|
||||
sidecarDir: bodySidecarDir,
|
||||
},
|
||||
undefined,
|
||||
persistModelBasesToConfig,
|
||||
);
|
||||
// Seed the "recent requests" table from the JSONL log so a process restart
|
||||
// doesn't reset what you can see in the UI. Best-effort; ignored on error.
|
||||
await dashboard.replay(opts.eventsFile).catch(() => {});
|
||||
@@ -1049,7 +1082,9 @@ async function main(): Promise<void> {
|
||||
const tag = e.info?.compressed
|
||||
? `compressed ${e.info.origChars}ch → ${e.info.imageCount}img/${e.info.imageBytes}B${extraTag}`
|
||||
: e.info?.reason
|
||||
? `savings:skip(${e.info.reason})`
|
||||
? e.info.reason === 'unsupported_model' && e.model
|
||||
? `skip(unsupported=${e.model})`
|
||||
: `skip(${e.info.reason})`
|
||||
: '';
|
||||
const cacheRead = e.usage?.cache_read_input_tokens ?? 0;
|
||||
const inputTokens = e.usage?.input_tokens ?? 0;
|
||||
|
||||
+3
-1
@@ -149,7 +149,9 @@ export default {
|
||||
const tag = e.info?.compressed
|
||||
? `compressed ${e.info.origChars}ch → ${e.info.imageCount}img/${e.info.imageBytes}B`
|
||||
: e.info?.reason
|
||||
? `savings:skip(${e.info.reason})`
|
||||
? e.info.reason === 'unsupported_model' && e.model
|
||||
? `skip(unsupported=${e.model})`
|
||||
: `skip(${e.info.reason})`
|
||||
: '';
|
||||
const cacheRead = e.usage?.cache_read_input_tokens ?? 0;
|
||||
console.log(`${e.method} ${e.path} → ${e.status} (${e.durationMs}ms) ${tag} cache_read=${cacheRead}`);
|
||||
|
||||
@@ -235,6 +235,43 @@ describe('serveFragment', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('invokes the host persistence hook on scope mutations', () => {
|
||||
const prev = process.env.PXPIPE_MODELS;
|
||||
try {
|
||||
delete process.env.PXPIPE_MODELS;
|
||||
setAllowedModelBases(null);
|
||||
const saved: string[][] = [];
|
||||
const persisting = new DashboardState(tmp, async () => new Map(), (bases) => {
|
||||
saved.push([...bases]);
|
||||
});
|
||||
|
||||
persisting.handleModelsToggle('gpt-5.6-sol', true);
|
||||
expect(saved.at(-1)).toEqual(['claude-fable-5', 'gpt-5.6-sol']);
|
||||
persisting.handleModelsSet('claude-fable-5');
|
||||
expect(saved.at(-1)).toEqual(['claude-fable-5']);
|
||||
// Empty scope persists too (round-trips as 'off' on load).
|
||||
persisting.handleModelsSet('off');
|
||||
expect(saved.at(-1)).toEqual([]);
|
||||
expect(saved).toHaveLength(3);
|
||||
|
||||
// A throwing hook must not break the live flip or the endpoint.
|
||||
const throwing = new DashboardState(tmp, async () => new Map(), () => {
|
||||
throw new Error('disk full');
|
||||
});
|
||||
throwing.handleModelsToggle('gpt-5.5', true);
|
||||
expect(getAllowedModelBases()).toContain('gpt-5.5');
|
||||
|
||||
// No hook (legacy/Worker host) keeps the old in-memory behavior.
|
||||
const bare = new DashboardState(tmp, async () => new Map());
|
||||
bare.handleModelsToggle('grok-4.5', true);
|
||||
expect(getAllowedModelBases()).toContain('grok-4.5');
|
||||
} finally {
|
||||
setAllowedModelBases(null);
|
||||
if (prev === undefined) delete process.env.PXPIPE_MODELS;
|
||||
else process.env.PXPIPE_MODELS = prev;
|
||||
}
|
||||
});
|
||||
|
||||
it('renders header + recent + stats fragments from the same payloads as JSON', async () => {
|
||||
writeEvents(tmp, [
|
||||
ev({ status: 200, model: 'gpt-5.5', compressed: true, orig_chars: 1000, image_bytes: 200 }),
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
import { afterAll, beforeAll, describe, it, expect } from 'vitest';
|
||||
import { createProxy, type ProxyEvent } from '../src/core/proxy.js';
|
||||
|
||||
// Pin the model scope so these contract tests stay independent of the developer shell.
|
||||
let ambientPxpipeModels: string | undefined;
|
||||
beforeAll(() => {
|
||||
ambientPxpipeModels = process.env.PXPIPE_MODELS;
|
||||
process.env.PXPIPE_MODELS = 'claude-fable-5,gpt-5.6-sol';
|
||||
});
|
||||
afterAll(() => {
|
||||
if (ambientPxpipeModels === undefined) delete process.env.PXPIPE_MODELS;
|
||||
else process.env.PXPIPE_MODELS = ambientPxpipeModels;
|
||||
});
|
||||
|
||||
/** Same in-process fetch patch as proxy-usage.test.ts. */
|
||||
function mockUpstream(handler: (req: Request) => Promise<Response> | Response) {
|
||||
const real = globalThis.fetch;
|
||||
globalThis.fetch = ((req: Request | string | URL, init?: RequestInit) => {
|
||||
const r = req instanceof Request ? req : new Request(String(req), init);
|
||||
return Promise.resolve(handler(r));
|
||||
}) as typeof fetch;
|
||||
return () => {
|
||||
globalThis.fetch = real;
|
||||
};
|
||||
}
|
||||
|
||||
// Deliberately OUT of the pinned PXPIPE_MODELS scope: without bypass the
|
||||
// transform path classifies this as `unsupported_model`, giving the tests a
|
||||
// crisp observable for "did classification run at all".
|
||||
const OUT_OF_SCOPE_BODY = JSON.stringify({
|
||||
model: 'claude-3-5-haiku-latest',
|
||||
messages: [{ role: 'user', content: 'hi' }],
|
||||
system: 'short',
|
||||
});
|
||||
|
||||
const UPSTREAM_OK = () =>
|
||||
new Response(
|
||||
JSON.stringify({
|
||||
id: 'msg_b1',
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'ok' }],
|
||||
usage: { input_tokens: 1, output_tokens: 1 },
|
||||
}),
|
||||
{ status: 200, headers: { 'content-type': 'application/json' } },
|
||||
);
|
||||
|
||||
async function roundTrip(headers: Record<string, string>) {
|
||||
const upstreamRequests: Request[] = [];
|
||||
const restore = mockUpstream(async (req) => {
|
||||
upstreamRequests.push(req.clone());
|
||||
return UPSTREAM_OK();
|
||||
});
|
||||
let captured: ProxyEvent | undefined;
|
||||
const proxy = createProxy({
|
||||
transform: {},
|
||||
onRequest: (e) => {
|
||||
captured = e;
|
||||
},
|
||||
});
|
||||
const res = await proxy(
|
||||
new Request('http://localhost/v1/messages', {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json', ...headers },
|
||||
body: OUT_OF_SCOPE_BODY,
|
||||
}),
|
||||
);
|
||||
// Drain the client body so the tee finishes, then give onRequest a tick.
|
||||
await res.text();
|
||||
await new Promise((r) => setTimeout(r, 20));
|
||||
restore();
|
||||
return { upstreamRequests, captured, status: res.status };
|
||||
}
|
||||
|
||||
describe('x-pxpipe-bypass opt-out', () => {
|
||||
it('skips transform classification and forwards the body byte-for-byte', async () => {
|
||||
const { upstreamRequests, captured, status } = await roundTrip({ 'x-pxpipe-bypass': '1' });
|
||||
|
||||
expect(status).toBe(200);
|
||||
expect(upstreamRequests).toHaveLength(1);
|
||||
// Body reaches the upstream untouched.
|
||||
expect(await upstreamRequests[0].text()).toBe(OUT_OF_SCOPE_BODY);
|
||||
// The pxpipe-only signal is never forwarded upstream.
|
||||
expect(upstreamRequests[0].headers.get('x-pxpipe-bypass')).toBeNull();
|
||||
// Classification never ran: no skip reason, despite the out-of-scope model.
|
||||
expect(captured?.info?.reason).toBeUndefined();
|
||||
});
|
||||
|
||||
it('treats any non-falsy value as bypass', async () => {
|
||||
const { upstreamRequests, captured } = await roundTrip({ 'x-pxpipe-bypass': 'true' });
|
||||
expect(await upstreamRequests[0].text()).toBe(OUT_OF_SCOPE_BODY);
|
||||
expect(captured?.info?.reason).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each(['0', 'false', 'off', 'no'])('does not bypass for falsy value %j', async (v) => {
|
||||
const { upstreamRequests, captured } = await roundTrip({ 'x-pxpipe-bypass': v });
|
||||
// Transform path ran and classified the out-of-scope model.
|
||||
expect(captured?.info?.reason).toBe('unsupported_model');
|
||||
// The header is stripped regardless of its value.
|
||||
expect(upstreamRequests[0].headers.get('x-pxpipe-bypass')).toBeNull();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user