From dcf1927f22c93bbc453eb8ad2d5060ffdfa21d63 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elio=20Bonfim=20J=C3=BAnior?= Date: Fri, 10 Jul 2026 16:10:16 +0700 Subject: [PATCH] =?UTF-8?q?feat(pxpipe):=20PXPIPE=20token=20saver=20?= =?UTF-8?q?=E2=80=94=20multimodal=20prompt=20compression=20(#2465)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add pxpipe as an experimental fifth Token Saver: Claude-format request bodies above a configurable size threshold are rendered as dense PNGs via the pxpipe-proxy library API (transformAnthropicMessages) before dispatch, cutting estimated input tokens by ~35-60% on token-dense contexts. Integration follows the Headroom pattern: applied to the final body in chatCore just before dispatch, fail-open on any error/timeout. Managed npm install into DATA_DIR/pxpipe, dynamic loader with per-version cache-bust, JSONL event log with rotation, /api/pxpipe/* endpoints, Token Saver card (marked experimental) + /dashboard/pxpipe page, and per-request Activated/Skipped annotation in Request Details. Disabled by default. --- open-sse/handlers/chatCore.js | 22 +- .../handlers/chatCore/nonStreamingHandler.js | 3 +- open-sse/handlers/chatCore/requestDetail.js | 1 + .../handlers/chatCore/streamingHandler.js | 6 +- open-sse/rtk/pxpipe.js | 104 +++++++ .../dashboard/pxpipe/PxpipeClient.js | 283 ++++++++++++++++++ src/app/(dashboard)/dashboard/pxpipe/page.js | 5 + .../dashboard/token-saver/TokenSaverClient.js | 240 ++++++++++++++- .../usage/components/RequestDetailsTab.js | 44 ++- src/app/api/pxpipe/health/route.js | 16 + src/app/api/pxpipe/install/route.js | 20 ++ src/app/api/pxpipe/logs/route.js | 18 ++ src/app/api/pxpipe/restart/route.js | 16 + src/app/api/pxpipe/start/route.js | 26 ++ src/app/api/pxpipe/stats/route.js | 14 + src/app/api/pxpipe/status/route.js | 21 ++ src/app/api/pxpipe/stop/route.js | 16 + src/lib/db/repos/requestDetailsRepo.js | 1 + src/lib/db/repos/settingsRepo.js | 4 + src/lib/pxpipe/events.js | 125 ++++++++ src/lib/pxpipe/install.js | 123 ++++++++ src/lib/pxpipe/loader.js | 70 +++++ src/lib/pxpipe/service.js | 49 +++ src/shared/components/Sidebar.js | 1 + src/sse/handlers/chat.js | 8 + tests/unit/pxpipe.test.js | 95 ++++++ 26 files changed, 1324 insertions(+), 7 deletions(-) create mode 100644 open-sse/rtk/pxpipe.js create mode 100644 src/app/(dashboard)/dashboard/pxpipe/PxpipeClient.js create mode 100644 src/app/(dashboard)/dashboard/pxpipe/page.js create mode 100644 src/app/api/pxpipe/health/route.js create mode 100644 src/app/api/pxpipe/install/route.js create mode 100644 src/app/api/pxpipe/logs/route.js create mode 100644 src/app/api/pxpipe/restart/route.js create mode 100644 src/app/api/pxpipe/start/route.js create mode 100644 src/app/api/pxpipe/stats/route.js create mode 100644 src/app/api/pxpipe/status/route.js create mode 100644 src/app/api/pxpipe/stop/route.js create mode 100644 src/lib/pxpipe/events.js create mode 100644 src/lib/pxpipe/install.js create mode 100644 src/lib/pxpipe/loader.js create mode 100644 src/lib/pxpipe/service.js create mode 100644 tests/unit/pxpipe.test.js diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index 5305aa04..a1a54d60 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -24,6 +24,7 @@ import { injectCaveman } from "../rtk/caveman.js"; import { injectPonytail } from "../rtk/ponytail.js"; import { compressMessages, formatRtkLog } from "../rtk/index.js"; import { compressWithHeadroom, formatHeadroomLog, formatHeadroomSizeLog, isHeadroomPhantomSavings } from "../rtk/headroom.js"; +import { compressWithPxpipe, formatPxpipeLog } from "../rtk/pxpipe.js"; import { getCapabilitiesForModel } from "../providers/capabilities.js"; import { stripUnsupportedModalities } from "../translator/concerns/modality.js"; import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; @@ -35,7 +36,7 @@ import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; * @param {object} options.credentials - Provider credentials * @param {string} options.sourceFormatOverride - Override detected source format (e.g. "openai-responses") */ -export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, sourceFormatOverride, providerThinking }) { +export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, pxpipeEnabled, pxpipeMinChars, pxpipeTimeoutMs, pxpipeTransform, onPxpipeEvent, sourceFormatOverride, providerThinking }) { const { provider, model } = modelInfo; const requestStartTime = Date.now(); @@ -186,6 +187,21 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred log?.debug?.("PONYTAIL", `${ponytailLevel} | ${finalFormat}`); } + // PXPIPE: image bulky context (Claude-format bodies only), last saver before dispatch + let pxpipeSummary = null; + if (pxpipeEnabled) { + const pxpipeResult = await compressWithPxpipe(translatedBody, { + enabled: true, format: finalFormat, model: upstreamModel, + minChars: pxpipeMinChars, timeoutMs: pxpipeTimeoutMs, transform: pxpipeTransform, + }); + pxpipeSummary = pxpipeResult.summary; + if (pxpipeResult.body) translatedBody = pxpipeResult.body; + const pxpipeLine = formatPxpipeLog(pxpipeSummary); + if (pxpipeLine) log?.info?.("PXPIPE", pxpipeLine); + else log?.debug?.("PXPIPE", `skipped: ${pxpipeSummary.reason}${pxpipeSummary.detail ? ` (${pxpipeSummary.detail})` : ""}`); + try { onPxpipeEvent?.({ provider, model, ...pxpipeSummary }); } catch { /* stats must not break requests */ } + } + const executor = getExecutor(provider); trackPendingRequest(model, provider, connectionId, true); appendRequestLog({ model, provider, connectionId, status: "PENDING" }).catch(() => { }); @@ -254,6 +270,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred request: extractRequestConfig(body, stream), providerRequest: translatedBody || null, response: { error: error.message || String(error), status: error.name === "AbortError" ? 499 : 502, thinking: null }, + pxpipe: pxpipeSummary, status: "error" })).catch(() => { }); @@ -300,6 +317,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred request: extractRequestConfig(body, stream), providerRequest: finalBody || translatedBody || null, response: { error: message, status: statusCode, thinking: null }, + pxpipe: pxpipeSummary, status: "error" })).catch(() => { }); @@ -309,7 +327,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred return createErrorResult(statusCode, errMsg, resetsAtMs); } - const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess }; + const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, pxpipe: pxpipeSummary }; const appendLog = (extra) => appendRequestLog({ model, provider, connectionId, ...extra }).catch(() => { }); const trackDone = () => trackPendingRequest(model, provider, connectionId, false); diff --git a/open-sse/handlers/chatCore/nonStreamingHandler.js b/open-sse/handlers/chatCore/nonStreamingHandler.js index 3996f94f..654ead88 100644 --- a/open-sse/handlers/chatCore/nonStreamingHandler.js +++ b/open-sse/handlers/chatCore/nonStreamingHandler.js @@ -198,7 +198,7 @@ export function translateNonStreamingResponse(responseBody, targetFormat, source /** * Handle non-streaming response from provider. */ -export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog }) { +export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog, pxpipe }) { trackDone(); const contentType = providerResponse.headers.get("content-type") || ""; let responseBody; @@ -296,6 +296,7 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m thinking: translatedResponse?.choices?.[0]?.message?.reasoning_content || translatedResponse?.reasoning_content || null, finish_reason: translatedResponse?.choices?.[0]?.finish_reason || "unknown" }, + pxpipe, status: "success" }, { endpoint: clientRawRequest?.endpoint || null })).catch(err => { console.error("[RequestDetail] Failed to save:", err.message); diff --git a/open-sse/handlers/chatCore/requestDetail.js b/open-sse/handlers/chatCore/requestDetail.js index 451111ad..33122da8 100644 --- a/open-sse/handlers/chatCore/requestDetail.js +++ b/open-sse/handlers/chatCore/requestDetail.js @@ -69,6 +69,7 @@ export function buildRequestDetail(base, overrides = {}) { providerRequest: base.providerRequest || null, providerResponse: base.providerResponse || null, response: base.response || {}, + pxpipe: base.pxpipe || undefined, status: base.status || "success", ...overrides }; diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index fb14bc0b..29ee39f3 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -43,7 +43,7 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, /** * Handle streaming response — pipe provider SSE through transform stream to client. */ -export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete, streamDetailId }) { +export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete, streamDetailId, pxpipe }) { if (onRequestSuccess) { Promise.resolve() .then(onRequestSuccess) @@ -94,6 +94,7 @@ export async function handleStreamingResponse({ providerResponse, provider, mode providerRequest: finalBody || translatedBody || null, providerResponse: "[Streaming - raw response not captured]", response: { content: "[Streaming in progress...]", thinking: null, type: "streaming" }, + pxpipe, status: "success" }, { id: streamDetailId })).catch(err => { console.error("[RequestDetail] Failed to save streaming request:", err.message); @@ -108,7 +109,7 @@ export async function handleStreamingResponse({ providerResponse, provider, mode /** * Build onStreamComplete callback for streaming usage tracking. */ -export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest }) { +export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest, pxpipe }) { const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; const onStreamComplete = (contentObj, usage, ttftAt) => { @@ -127,6 +128,7 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r providerRequest: finalBody || translatedBody || null, providerResponse: safeContent, response: { content: safeContent, thinking: safeThinking, type: "streaming" }, + pxpipe, status: "success" }, { id: streamDetailId })).catch(err => { console.error("[RequestDetail] Failed to update streaming content:", err.message); diff --git a/open-sse/rtk/pxpipe.js b/open-sse/rtk/pxpipe.js new file mode 100644 index 00000000..04ac8f45 --- /dev/null +++ b/open-sse/rtk/pxpipe.js @@ -0,0 +1,104 @@ +// PXPIPE: render bulky Claude-format context as dense PNGs via pxpipe-proxy's +// library API (transformAnthropicMessages). Fail-open like every token saver: +// any error/timeout returns { body: null, summary } and leaves the request untouched. +import { FORMATS } from "../translator/formats.js"; + +const DEFAULT_TIMEOUT_MS = 15000; +const DEFAULT_MIN_CHARS = 25000; +// pxpipe's own profitability gate assumes ~4 chars/token; reuse it for the +// estimated before/after numbers surfaced in stats (marked "estimated" in UI). +const EST_CHARS_PER_TOKEN = 4; + +function bodyChars(body) { + try { + return JSON.stringify(body)?.length || 0; + } catch { + return 0; + } +} + +function estTokens(chars) { + return Math.round(chars / EST_CHARS_PER_TOKEN); +} + +function skipped(reason, extra = {}) { + return { body: null, summary: { applied: false, reason, ...extra } }; +} + +// Transform a Claude-format request body through pxpipe. Returns +// { body: | null, summary } — body is null when nothing changed. +// opts.transform is injected by the host (src side) so open-sse stays free of +// filesystem/install concerns and remains usable standalone. +export async function compressWithPxpipe(body, { enabled, format, model, minChars, timeoutMs, transform } = {}) { + if (!enabled) return skipped("disabled"); + if (typeof transform !== "function") return skipped("not_installed"); + if (!body) return skipped("missing_body"); + if (format !== FORMATS.CLAUDE) return skipped("unsupported_format", { detail: format }); + + const startedAt = Date.now(); + const originalChars = bodyChars(body); + const threshold = Number(minChars) > 0 ? Number(minChars) : DEFAULT_MIN_CHARS; + if (originalChars < threshold) { + return skipped("below_threshold", { originalChars, threshold }); + } + + try { + const encoded = new TextEncoder().encode(JSON.stringify(body)); + const budget = Number(timeoutMs) > 0 ? Number(timeoutMs) : DEFAULT_TIMEOUT_MS; + // transformAnthropicMessages is local CPU work and can't be aborted; race a + // timer and discard the result if it loses (input body is never mutated). + const result = await Promise.race([ + transform({ + body: encoded, + model, + options: { minCompressChars: threshold }, + }), + new Promise((resolve) => setTimeout(() => resolve(null), budget)), + ]); + if (!result) return skipped("timeout", { originalChars, durationMs: Date.now() - startedAt }); + if (!result.applied) { + return skipped(result.reason || "passthrough", { + detail: result.detail, + originalChars, + durationMs: Date.now() - startedAt, + }); + } + + const newBody = JSON.parse(new TextDecoder().decode(result.body)); + const compressedBodyChars = bodyChars(newBody); + const info = result.info || {}; + const imagedChars = info.compressedChars || 0; + // The transformed body is BIGGER in bytes (base64 PNGs) but cheaper in tokens: + // images bill by pixels (Anthropic: pixels/750), not by encoded length. So the + // after-estimate is remaining-text tokens + image tokens — never chars/4 of the + // new body. Provider-billed usage recorded per request stays the ground truth. + const imageTokensEst = info.imageTokens + || (info.imagePixels ? Math.round(info.imagePixels / 750) : (info.imageCount || 0) * 4761); + const summary = { + applied: true, + reason: "applied", + originalChars, + compressedBodyChars, + imagedChars, + imageCount: info.imageCount || 0, + imageBytes: info.imageBytes || 0, + tokensBeforeEst: info.baselineTokens || estTokens(originalChars), + tokensAfterEst: estTokens(Math.max(0, originalChars - imagedChars)) + imageTokensEst, + durationMs: Date.now() - startedAt, + cacheOwnsControl: result.cache?.ownsCacheControl === true, + }; + summary.tokensSavedEst = Math.max(0, summary.tokensBeforeEst - summary.tokensAfterEst); + summary.savedPct = summary.tokensBeforeEst > 0 + ? +((summary.tokensSavedEst / summary.tokensBeforeEst) * 100).toFixed(2) + : 0; + return { body: newBody, summary }; + } catch (e) { + return skipped("transform_error", { detail: e?.message || String(e), originalChars, durationMs: Date.now() - startedAt }); + } +} + +export function formatPxpipeLog(summary) { + if (!summary) return null; + if (!summary.applied) return null; + return `imaged ${summary.imagedChars}ch → ${summary.imageCount} image(s) | est ${summary.tokensBeforeEst}→${summary.tokensAfterEst} tokens (-${summary.savedPct}%) | ${summary.durationMs}ms`; +} diff --git a/src/app/(dashboard)/dashboard/pxpipe/PxpipeClient.js b/src/app/(dashboard)/dashboard/pxpipe/PxpipeClient.js new file mode 100644 index 00000000..b3a42473 --- /dev/null +++ b/src/app/(dashboard)/dashboard/pxpipe/PxpipeClient.js @@ -0,0 +1,283 @@ +"use client"; + +import { useState, useEffect, useCallback } from "react"; +import { + AreaChart, + Area, + XAxis, + YAxis, + CartesianGrid, + Tooltip, + ResponsiveContainer, +} from "recharts"; +import { Card, Button } from "@/shared/components"; + +const fmtTokens = (n) => { + if (n >= 1000000) return `${(n / 1000000).toFixed(2)}M`; + if (n >= 1000) return `${(n / 1000).toFixed(1)}K`; + return String(n || 0); +}; + +const fmtUptime = (ms) => { + if (!ms || ms <= 0) return "—"; + const m = Math.floor(ms / 60000); + const h = Math.floor(m / 60); + return h > 0 ? `${h}h${String(m % 60).padStart(2, "0")}m` : `${m}m`; +}; + +const WINDOW_TABS = [ + { id: "today", label: "Today" }, + { id: "yesterday", label: "Yesterday" }, + { id: "last7d", label: "7 days" }, + { id: "last30d", label: "30 days" }, + { id: "all", label: "All time" }, +]; + +const REASON_LABELS = { + applied: "Prompt exceeded threshold", + below_threshold: "Below size threshold", + not_profitable: "Compression not profitable", + below_min_chars: "Below minimum chars", + below_min_tokens: "Below minimum tokens", + unsupported_model: "Model not in allowlist", + unsupported_format: "Non-Claude request format", + timeout: "Compression timed out", + transform_error: "Transform error", + passthrough: "Passthrough", + disabled: "Disabled", + not_installed: "Not installed", +}; + +function SummaryCard({ label, value, sub, tone }) { + return ( + +

{label}

+

{value}

+ {sub &&

{sub}

} + + ); +} + +export default function PxpipeClient() { + const [status, setStatus] = useState(null); + const [health, setHealth] = useState(null); + const [stats, setStats] = useState(null); + const [logs, setLogs] = useState(null); + const [windowId, setWindowId] = useState("last7d"); + const [loading, setLoading] = useState(true); + + const refresh = useCallback(async () => { + setLoading(true); + try { + const [statusRes, statsRes, logsRes] = await Promise.all([ + fetch("/api/pxpipe/status", { headers: { "Cache-Control": "no-store" } }), + fetch("/api/pxpipe/stats"), + fetch("/api/pxpipe/logs?limit=50"), + ]); + setStatus(await statusRes.json()); + setStats(await statsRes.json()); + setLogs(await logsRes.json()); + const healthRes = await fetch("/api/pxpipe/health", { method: "POST" }); + setHealth(await healthRes.json()); + } catch { + /* sections render placeholders */ + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + refresh(); + }, [refresh]); + + const w = stats?.windows?.[windowId]; + const statusLabel = !status + ? "—" + : !status.installed + ? "Not installed" + : health?.healthy + ? "Healthy" + : status.running + ? "Running" + : "Stopped"; + + return ( +
+
+

+ image + PXPIPE Dashboard +

+
+ + Token Saver settings + + +
+
+ +
+ + + + + + +
+ + +
+

Token savings (estimated)

+
+ {WINDOW_TABS.map((tab) => ( + + ))} +
+
+
+
+

Original tokens

+

{w ? fmtTokens(w.tokensBeforeEst) : "—"}

+
+
+

After PXPIPE

+

{w ? fmtTokens(w.tokensAfterEst) : "—"}

+
+
+

Saved

+

{w ? fmtTokens(w.tokensSavedEst) : "—"}

+
+
+

Reduction

+

{w ? `${w.savedPct}%` : "—"}

+
+
+

+ Estimates from body size before/after imaging; billed usage per request + (recorded on the Usage page) remains the ground truth. Images generated:{" "} + {w ? w.imagesGenerated.toLocaleString() : "—"} · avg compression time:{" "} + {w ? `${w.avgCompressionMs}ms` : "—"} · errors: {w ? w.errors : "—"} +

+ + + +

Tokens saved — last 30 days

+ {stats?.timeline?.some((d) => d.tokensSavedEst > 0) ? ( + + + + + + + + + + d.slice(5)} /> + + [fmtTokens(v), "Tokens saved"]} labelFormatter={(d) => d} /> + + + + ) : ( +
+ No savings recorded yet — enable PXPIPE in the Token Saver and route a large Claude-format request. +
+ )} + + + +

History

+
+ + + + + + + + + + + + + + + {(stats?.recent || []).slice(0, 50).map((ev, i) => ( + + + + + + + + + + + ))} + {(!stats?.recent || stats.recent.length === 0) && ( + + + + )} + +
TimeModelOriginalCompressedSaved%DurationStatus
+ {new Date(ev.ts).toLocaleString()} + {ev.provider ? `${ev.provider}/${ev.model}` : ev.model || "—"} + {ev.applied ? fmtTokens(ev.tokensBeforeEst) : "—"} + + {ev.applied ? fmtTokens(ev.tokensAfterEst) : "—"} + + {ev.applied ? fmtTokens(ev.tokensSavedEst) : "—"} + + {ev.applied ? `${ev.savedPct}%` : "—"} + + {ev.durationMs != null ? `${ev.durationMs}ms` : "—"} + + + {ev.applied ? "Compressed" : REASON_LABELS[ev.reason] || ev.reason} + +
+ No PXPIPE activity yet +
+
+ + + +

PXPIPE Logs

+ {logs?.installLog ? ( + 
+            {logs.installLog}
+
+ ) : ( +

No install log yet.

+ )} + +
+ ); +} diff --git a/src/app/(dashboard)/dashboard/pxpipe/page.js b/src/app/(dashboard)/dashboard/pxpipe/page.js new file mode 100644 index 00000000..e53a44a2 --- /dev/null +++ b/src/app/(dashboard)/dashboard/pxpipe/page.js @@ -0,0 +1,5 @@ +import PxpipeClient from "./PxpipeClient"; + +export default function PxpipePage() { + return ; +} diff --git a/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js b/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js index 74b14d83..fc1f9152 100644 --- a/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js +++ b/src/app/(dashboard)/dashboard/token-saver/TokenSaverClient.js @@ -37,6 +37,19 @@ export default function TokenSaverClient() { const [cavemanLevel, setCavemanLevel] = useState("full"); const [ponytailEnabled, setPonytailEnabled] = useState(false); const [ponytailLevel, setPonytailLevel] = useState("full"); + const [pxpipeEnabled, setPxpipeEnabled] = useState(false); + const [pxpipeMinChars, setPxpipeMinChars] = useState(25000); + const [pxpipeStatus, setPxpipeStatus] = useState({ + installed: false, + installing: false, + running: false, + version: null, + loading: true, + }); + const [pxpipeHealth, setPxpipeHealth] = useState(null); + const [showPxpipeModal, setShowPxpipeModal] = useState(false); + const [pxpipeActionLoading, setPxpipeActionLoading] = useState(false); + const [pxpipeActionError, setPxpipeActionError] = useState(""); const [locale, setLocale] = useState("en"); const { copied, copy } = useCopyToClipboard(); @@ -232,6 +245,59 @@ export default function TokenSaverClient() { patchSetting({ ponytailLevel: level }); }; + const refreshPxpipeStatus = useCallback(async () => { + setPxpipeStatus((s) => ({ ...s, loading: true })); + try { + const res = await fetch("/api/pxpipe/status", { + headers: { "Cache-Control": "no-store" }, + }); + const data = await res.json(); + setPxpipeStatus({ ...data, loading: false }); + if (typeof data.minChars === "number") setPxpipeMinChars(data.minChars); + } catch { + setPxpipeStatus({ installed: false, installing: false, running: false, version: null, loading: false }); + } + }, []); + + const runPxpipeHealth = useCallback(async () => { + try { + const res = await fetch("/api/pxpipe/health", { method: "POST" }); + setPxpipeHealth(await res.json()); + } catch (e) { + setPxpipeHealth({ healthy: false, checks: [], error: e.message }); + } + }, []); + + const pxpipeAction = useCallback( + async (endpoint) => { + setPxpipeActionError(""); + setPxpipeActionLoading(true); + try { + const res = await fetch(`/api/pxpipe/${endpoint}`, { method: "POST" }); + const data = await res.json().catch(() => ({})); + if (!res.ok) throw new Error(data.error || `PXPIPE ${endpoint} failed`); + await refreshPxpipeStatus(); + await runPxpipeHealth(); + } catch (e) { + setPxpipeActionError(e.message); + } finally { + setPxpipeActionLoading(false); + } + }, + [refreshPxpipeStatus, runPxpipeHealth] + ); + + const handlePxpipeEnabled = (value) => { + setPxpipeEnabled(value); + patchSetting({ pxpipeEnabled: value }); + }; + + const handlePxpipeMinCharsBlur = () => { + const next = Math.max(0, Number(pxpipeMinChars) || 25000); + setPxpipeMinChars(next); + patchSetting({ pxpipeMinChars: next }); + }; + useEffect(() => { const loadSettings = async () => { try { @@ -245,12 +311,16 @@ export default function TokenSaverClient() { setCavemanLevel(data.cavemanLevel || "full"); setPonytailEnabled(!!data.ponytailEnabled); setPonytailLevel(data.ponytailLevel || "full"); + setPxpipeEnabled(!!data.pxpipeEnabled); + if (typeof data.pxpipeMinChars === "number") setPxpipeMinChars(data.pxpipeMinChars); refreshHeadroomStatus(); + // PRD: run the PXPIPE health check automatically when the page opens + refreshPxpipeStatus().then(runPxpipeHealth); } } catch {} }; loadSettings(); - }, [refreshHeadroomStatus]); + }, [refreshHeadroomStatus, refreshPxpipeStatus, runPxpipeHealth]); const headroomRunning = !!headroomStatus.running; const headroomStatusLabel = headroomStatus.loading @@ -267,6 +337,23 @@ export default function TokenSaverClient() { const headroomManaged = headroomLocalUrl && !!headroomStatus.managedPid; + const pxpipeHealthy = pxpipeHealth?.healthy === true; + const pxpipeStatusLabel = pxpipeStatus.loading + ? "Checking…" + : pxpipeStatus.installing + ? "Installing…" + : !pxpipeStatus.installed + ? "Not installed" + : pxpipeHealthy + ? "Healthy" + : pxpipeStatus.running + ? "Running" + : "Stopped"; + const pxpipeChipClass = + pxpipeHealthy || pxpipeStatus.running + ? "bg-success/15 text-success" + : "bg-warning/15 text-warning"; + return (
@@ -502,6 +589,49 @@ export default function TokenSaverClient() { />
+
+
+
+

+ Compress prompts as images{" "} + + (PXPIPE) + +

+ + {pxpipeStatusLabel} + + + + Dashboard + +
+

+ Transforms large textual context into optimized images before + sending to the LLM. Ideal for huge prompts, tool outputs and long + conversations. +

+
+ handlePxpipeEnabled(!pxpipeEnabled)} + /> +
+ + setShowPxpipeModal(false)} + > +
+

+ Compress prompts using multimodal encoding. Runs in-process — no + extra server or environment variables required. +

+
+ Status + + {pxpipeStatusLabel} + {pxpipeStatus.version ? ` · v${pxpipeStatus.version}` : ""} + +
+ {pxpipeHealth?.checks?.length > 0 && ( +
+

Health check

+ {pxpipeHealth.checks.map((check) => ( +
+ + {check.ok ? "●" : "○"} {check.label} + + {check.detail && ( + {check.detail} + )} +
+ ))} + {pxpipeHealth.error && ( +

{pxpipeHealth.error}

+ )} +
+ )} + {!pxpipeStatus.installed ? ( +
+

PXPIPE is not installed.

+ +

+ Installs the npm package pxpipe-proxy into + the 9Router data directory. May take a few minutes. +

+
+ ) : ( +
+ {pxpipeStatus.running ? ( + <> + + + + ) : ( + + )} + + + Open Logs + +
+ )} +
+

Minimum prompt size (chars)

+ setPxpipeMinChars(e.target.value)} + onBlur={handlePxpipeMinCharsBlur} + placeholder="25000" + className="font-mono text-sm" + /> +

+ Requests smaller than this bypass PXPIPE and are sent as-is. +

+
+ {pxpipeActionError && ( +

{pxpipeActionError}

+ )} +
+ + +
+
+ ); } diff --git a/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js b/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js index b82b8286..af8eb736 100644 --- a/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js +++ b/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js @@ -412,7 +412,49 @@ export default function RequestDetailsTab() { - + + {selectedDetail.pxpipe && ( +
+
+ image + PXPIPE + + {selectedDetail.pxpipe.applied ? "Activated" : "Skipped"} + +
+ {selectedDetail.pxpipe.applied ? ( +
+
+ Original (est.) + {(selectedDetail.pxpipe.tokensBeforeEst || 0).toLocaleString()} tokens +
+
+ Compressed (est.) + {(selectedDetail.pxpipe.tokensAfterEst || 0).toLocaleString()} tokens +
+
+ Saved + {selectedDetail.pxpipe.savedPct || 0}% +
+
+ Images + {selectedDetail.pxpipe.imageCount || 0} ({selectedDetail.pxpipe.durationMs || 0}ms) +
+
+ ) : ( +

+ Reason: {selectedDetail.pxpipe.reason} + {selectedDetail.pxpipe.detail ? ` — ${selectedDetail.pxpipe.detail}` : ""} +

+ )} +
+ )} + 
diff --git a/src/app/api/pxpipe/health/route.js b/src/app/api/pxpipe/health/route.js
new file mode 100644
index 00000000..f974342b
--- /dev/null
+++ b/src/app/api/pxpipe/health/route.js
@@ -0,0 +1,16 @@
+import { NextResponse } from "next/server";
+import { runHealthCheck } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+
+export async function POST() {
+  try {
+    const result = await runHealthCheck();
+    return NextResponse.json(result);
+  } catch (error) {
+    return NextResponse.json({ healthy: false, checks: [], error: error.message }, { status: 500 });
+  }
+}
+
+// GET mirrors POST so the card can probe on page load without a mutation call.
+export const GET = POST;
diff --git a/src/app/api/pxpipe/install/route.js b/src/app/api/pxpipe/install/route.js
new file mode 100644
index 00000000..c5eedcc0
--- /dev/null
+++ b/src/app/api/pxpipe/install/route.js
@@ -0,0 +1,20 @@
+import { NextResponse } from "next/server";
+import { installPxpipe } from "@/lib/pxpipe/install.js";
+import { unloadPxpipe } from "@/lib/pxpipe/loader.js";
+import { runHealthCheck } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+// npm install can legitimately take minutes on a cold cache.
+export const maxDuration = 300;
+
+// Install (or repair — same operation, reinstalls @latest) then re-run the health check.
+export async function POST() {
+  try {
+    const info = await installPxpipe();
+    unloadPxpipe(); // drop any previously-loaded version so health loads the fresh one
+    const health = await runHealthCheck();
+    return NextResponse.json({ ...info, health });
+  } catch (error) {
+    return NextResponse.json({ error: error.message, code: error.code || null }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/logs/route.js b/src/app/api/pxpipe/logs/route.js
new file mode 100644
index 00000000..051db386
--- /dev/null
+++ b/src/app/api/pxpipe/logs/route.js
@@ -0,0 +1,18 @@
+import { NextResponse } from "next/server";
+import { getInstallLogTail } from "@/lib/pxpipe/install.js";
+import { readPxpipeEvents } from "@/lib/pxpipe/events.js";
+
+export const dynamic = "force-dynamic";
+
+export async function GET(request) {
+  try {
+    const { searchParams } = new URL(request.url);
+    const limit = Math.min(Number(searchParams.get("limit")) || 100, 500);
+    return NextResponse.json({
+      installLog: getInstallLogTail(),
+      events: readPxpipeEvents({ limit }).reverse(),
+    });
+  } catch (error) {
+    return NextResponse.json({ error: error.message }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/restart/route.js b/src/app/api/pxpipe/restart/route.js
new file mode 100644
index 00000000..1aaab39a
--- /dev/null
+++ b/src/app/api/pxpipe/restart/route.js
@@ -0,0 +1,16 @@
+import { NextResponse } from "next/server";
+import { unloadPxpipe, loadPxpipe } from "@/lib/pxpipe/loader.js";
+import { getPxpipeStatus } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+
+// Reload the in-process module (picks up an upgraded install without a server restart).
+export async function POST() {
+  try {
+    unloadPxpipe();
+    await loadPxpipe();
+    return NextResponse.json(getPxpipeStatus());
+  } catch (error) {
+    return NextResponse.json({ error: error.message, code: error.code || null }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/start/route.js b/src/app/api/pxpipe/start/route.js
new file mode 100644
index 00000000..5dad8b2b
--- /dev/null
+++ b/src/app/api/pxpipe/start/route.js
@@ -0,0 +1,26 @@
+import { NextResponse } from "next/server";
+import { getSettings } from "@/lib/localDb";
+import { getInstallInfo, installPxpipe } from "@/lib/pxpipe/install.js";
+import { loadPxpipe } from "@/lib/pxpipe/loader.js";
+import { getPxpipeStatus } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+export const maxDuration = 300;
+
+// "Start" in library mode = warm the in-process transform module.
+// Auto-installs first when the package is missing and pxpipeAutoInstall is on.
+export async function POST() {
+  try {
+    if (!getInstallInfo().installed) {
+      const settings = await getSettings();
+      if (!settings.pxpipeAutoInstall) {
+        return NextResponse.json({ error: "PXPIPE is not installed", code: "NOT_INSTALLED" }, { status: 409 });
+      }
+      await installPxpipe();
+    }
+    await loadPxpipe();
+    return NextResponse.json(getPxpipeStatus());
+  } catch (error) {
+    return NextResponse.json({ error: error.message, code: error.code || null }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/stats/route.js b/src/app/api/pxpipe/stats/route.js
new file mode 100644
index 00000000..c5860b6e
--- /dev/null
+++ b/src/app/api/pxpipe/stats/route.js
@@ -0,0 +1,14 @@
+import { NextResponse } from "next/server";
+import { getPxpipeStats } from "@/lib/pxpipe/events.js";
+
+export const dynamic = "force-dynamic";
+
+export async function GET(request) {
+  try {
+    const { searchParams } = new URL(request.url);
+    const recentLimit = Math.min(Number(searchParams.get("limit")) || 100, 500);
+    return NextResponse.json(getPxpipeStats({ recentLimit }));
+  } catch (error) {
+    return NextResponse.json({ error: error.message }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/status/route.js b/src/app/api/pxpipe/status/route.js
new file mode 100644
index 00000000..663c8398
--- /dev/null
+++ b/src/app/api/pxpipe/status/route.js
@@ -0,0 +1,21 @@
+import { NextResponse } from "next/server";
+import { getSettings } from "@/lib/localDb";
+import { getPxpipeStatus } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+
+export async function GET() {
+  try {
+    const settings = await getSettings();
+    const status = getPxpipeStatus();
+    return NextResponse.json({
+      ...status,
+      enabled: !!settings.pxpipeEnabled,
+      autoInstall: !!settings.pxpipeAutoInstall,
+      minChars: settings.pxpipeMinChars,
+      timeoutMs: settings.pxpipeTimeoutMs,
+    });
+  } catch (error) {
+    return NextResponse.json({ error: error.message }, { status: 500 });
+  }
+}
diff --git a/src/app/api/pxpipe/stop/route.js b/src/app/api/pxpipe/stop/route.js
new file mode 100644
index 00000000..cc675c52
--- /dev/null
+++ b/src/app/api/pxpipe/stop/route.js
@@ -0,0 +1,16 @@
+import { NextResponse } from "next/server";
+import { unloadPxpipe } from "@/lib/pxpipe/loader.js";
+import { getPxpipeStatus } from "@/lib/pxpipe/service.js";
+
+export const dynamic = "force-dynamic";
+
+// "Stop" in library mode = drop the in-process module; requests fail open to
+// uncompressed passthrough until it is started again.
+export async function POST() {
+  try {
+    const wasLoaded = unloadPxpipe();
+    return NextResponse.json({ stopped: wasLoaded, ...getPxpipeStatus() });
+  } catch (error) {
+    return NextResponse.json({ error: error.message }, { status: 500 });
+  }
+}
diff --git a/src/lib/db/repos/requestDetailsRepo.js b/src/lib/db/repos/requestDetailsRepo.js
index 2f308015..defd294b 100644
--- a/src/lib/db/repos/requestDetailsRepo.js
+++ b/src/lib/db/repos/requestDetailsRepo.js
@@ -98,6 +98,7 @@ async function flushToDatabase() {
             providerRequest: truncateField(item.providerRequest, config.maxJsonSize),
             providerResponse: truncateField(item.providerResponse, config.maxJsonSize),
             response: truncateField(item.response, config.maxJsonSize),
+            pxpipe: item.pxpipe || undefined,
           };
 
           db.run(
diff --git a/src/lib/db/repos/settingsRepo.js b/src/lib/db/repos/settingsRepo.js
index 0057cc1c..7b147e1e 100644
--- a/src/lib/db/repos/settingsRepo.js
+++ b/src/lib/db/repos/settingsRepo.js
@@ -42,6 +42,10 @@ const DEFAULT_SETTINGS = {
   cavemanLevel: "full",
   ponytailEnabled: false,
   ponytailLevel: "full",
+  pxpipeEnabled: false,
+  pxpipeAutoInstall: true,
+  pxpipeMinChars: 25000,
+  pxpipeTimeoutMs: 15000,
 };
 
 async function readRaw() {
diff --git a/src/lib/pxpipe/events.js b/src/lib/pxpipe/events.js
new file mode 100644
index 00000000..b1bb880d
--- /dev/null
+++ b/src/lib/pxpipe/events.js
@@ -0,0 +1,125 @@
+import fs from "fs";
+import path from "path";
+import { PXPIPE_DIR } from "./install.js";
+
+const EVENTS_FILE = path.join(PXPIPE_DIR, "events.jsonl");
+const ROTATED_FILE = path.join(PXPIPE_DIR, "events.jsonl.1");
+const MAX_FILE_BYTES = 5 * 1024 * 1024;
+const DAY_MS = 24 * 60 * 60 * 1000;
+
+function ensureDir() {
+  if (!fs.existsSync(PXPIPE_DIR)) fs.mkdirSync(PXPIPE_DIR, { recursive: true });
+}
+
+// Fire-and-forget: stats must never break the request path.
+export function appendPxpipeEvent(event) {
+  try {
+    ensureDir();
+    try {
+      const stat = fs.statSync(EVENTS_FILE);
+      if (stat.size > MAX_FILE_BYTES) fs.renameSync(EVENTS_FILE, ROTATED_FILE);
+    } catch { /* no file yet */ }
+    fs.appendFile(EVENTS_FILE, JSON.stringify({ ts: Date.now(), ...event }) + "\n", () => {});
+  } catch { /* ignore */ }
+}
+
+export function readPxpipeEvents({ sinceMs = null, limit = null } = {}) {
+  const events = [];
+  for (const file of [ROTATED_FILE, EVENTS_FILE]) {
+    try {
+      if (!fs.existsSync(file)) continue;
+      for (const line of fs.readFileSync(file, "utf8").split("\n")) {
+        if (!line) continue;
+        try {
+          const ev = JSON.parse(line);
+          if (sinceMs && ev.ts < sinceMs) continue;
+          events.push(ev);
+        } catch { /* skip corrupt line */ }
+      }
+    } catch { /* ignore */ }
+  }
+  events.sort((a, b) => a.ts - b.ts);
+  return limit ? events.slice(-limit) : events;
+}
+
+function emptyTotals() {
+  return {
+    requests: 0, compressed: 0, bypassed: 0, errors: 0,
+    tokensBeforeEst: 0, tokensAfterEst: 0, tokensSavedEst: 0, savedPct: 0,
+    imagesGenerated: 0, compressionTimeMs: 0, avgCompressionMs: 0,
+  };
+}
+
+function accumulate(totals, ev) {
+  totals.requests++;
+  if (ev.applied) {
+    totals.compressed++;
+    totals.tokensBeforeEst += ev.tokensBeforeEst || 0;
+    totals.tokensAfterEst += ev.tokensAfterEst || 0;
+    totals.tokensSavedEst += ev.tokensSavedEst || 0;
+    totals.imagesGenerated += ev.imageCount || 0;
+    totals.compressionTimeMs += ev.durationMs || 0;
+  } else if (ev.reason === "transform_error" || ev.reason === "timeout") {
+    totals.errors++;
+  } else {
+    totals.bypassed++;
+  }
+}
+
+function finalize(totals) {
+  totals.savedPct = totals.tokensBeforeEst > 0
+    ? +((totals.tokensSavedEst / totals.tokensBeforeEst) * 100).toFixed(2)
+    : 0;
+  totals.avgCompressionMs = totals.compressed > 0
+    ? Math.round(totals.compressionTimeMs / totals.compressed)
+    : 0;
+  return totals;
+}
+
+// Aggregated stats for the dashboard: all-time + windowed totals, a daily
+// tokens-saved timeline (last `timelineDays`), and the most recent events.
+export function getPxpipeStats({ timelineDays = 30, recentLimit = 100 } = {}) {
+  const events = readPxpipeEvents();
+  const now = Date.now();
+  const startOfToday = new Date(new Date(now).setHours(0, 0, 0, 0)).getTime();
+
+  const windows = {
+    all: emptyTotals(),
+    today: emptyTotals(),
+    yesterday: emptyTotals(),
+    last7d: emptyTotals(),
+    last30d: emptyTotals(),
+  };
+
+  const timeline = new Map();
+  for (let i = timelineDays - 1; i >= 0; i--) {
+    const day = new Date(startOfToday - i * DAY_MS);
+    timeline.set(day.toISOString().slice(0, 10), { date: day.toISOString().slice(0, 10), tokensSavedEst: 0, compressed: 0, requests: 0 });
+  }
+
+  for (const ev of events) {
+    accumulate(windows.all, ev);
+    if (ev.ts >= startOfToday) accumulate(windows.today, ev);
+    else if (ev.ts >= startOfToday - DAY_MS) accumulate(windows.yesterday, ev);
+    if (ev.ts >= now - 7 * DAY_MS) accumulate(windows.last7d, ev);
+    if (ev.ts >= now - 30 * DAY_MS) accumulate(windows.last30d, ev);
+
+    const key = new Date(ev.ts).toISOString().slice(0, 10);
+    const bucket = timeline.get(key);
+    if (bucket) {
+      bucket.requests++;
+      if (ev.applied) {
+        bucket.compressed++;
+        bucket.tokensSavedEst += ev.tokensSavedEst || 0;
+      }
+    }
+  }
+
+  for (const w of Object.values(windows)) finalize(w);
+
+  return {
+    windows,
+    timeline: [...timeline.values()],
+    recent: events.slice(-recentLimit).reverse(),
+  };
+}
diff --git a/src/lib/pxpipe/install.js b/src/lib/pxpipe/install.js
new file mode 100644
index 00000000..cdf5a8ce
--- /dev/null
+++ b/src/lib/pxpipe/install.js
@@ -0,0 +1,123 @@
+import fs from "fs";
+import path from "path";
+import { spawn, execSync } from "child_process";
+import { DATA_DIR } from "@/lib/dataDir.js";
+
+export const PXPIPE_DIR = path.join(DATA_DIR, "pxpipe");
+export const PXPIPE_PACKAGE = "pxpipe-proxy";
+const INSTALL_LOG = path.join(PXPIPE_DIR, "install.log");
+const INSTALL_TIMEOUT_MS = 5 * 60 * 1000;
+
+const IS_WIN = process.platform === "win32";
+const NPM_CMD = IS_WIN ? "npm.cmd" : "npm";
+
+// Same PATH extension trick as headroom/detect.js: packaged/launchd environments
+// often miss the Node bin dirs.
+const EXTRA_BINS = IS_WIN
+  ? [`${process.env.ProgramFiles || ""}\\nodejs`, `${process.env.APPDATA || ""}\\npm`]
+  : ["/usr/local/bin", "/opt/homebrew/bin", `${process.env.HOME || ""}/.local/bin`, "/usr/bin", "/bin"];
+const EXTENDED_PATH = [...EXTRA_BINS, process.env.PATH || ""].filter(Boolean).join(path.delimiter);
+
+let installInFlight = null;
+
+function ensureDir() {
+  if (!fs.existsSync(PXPIPE_DIR)) fs.mkdirSync(PXPIPE_DIR, { recursive: true });
+}
+
+export function packageRoot() {
+  return path.join(PXPIPE_DIR, "node_modules", PXPIPE_PACKAGE);
+}
+
+export function libraryEntry() {
+  return path.join(packageRoot(), "dist", "core", "library.js");
+}
+
+export function findNpm() {
+  try {
+    const out = execSync(`${IS_WIN ? "where" : "which"} npm`, {
+      stdio: ["ignore", "pipe", "ignore"],
+      windowsHide: true,
+      env: { ...process.env, PATH: EXTENDED_PATH },
+    }).toString().trim();
+    return out ? out.split(/\r?\n/)[0].trim() : null;
+  } catch {
+    return null;
+  }
+}
+
+// { installed, version, path } — installed means the library entry exists on disk.
+export function getInstallInfo() {
+  try {
+    const pkgJson = path.join(packageRoot(), "package.json");
+    if (!fs.existsSync(pkgJson) || !fs.existsSync(libraryEntry())) {
+      return { installed: false, version: null, path: null };
+    }
+    const pkg = JSON.parse(fs.readFileSync(pkgJson, "utf8"));
+    return { installed: true, version: pkg.version || null, path: packageRoot() };
+  } catch {
+    return { installed: false, version: null, path: null };
+  }
+}
+
+export function isInstalling() {
+  return installInFlight !== null;
+}
+
+// Install (or repair by reinstalling) pxpipe-proxy into DATA_DIR/pxpipe.
+// Serialized: concurrent calls await the same run.
+export function installPxpipe() {
+  if (installInFlight) return installInFlight;
+  installInFlight = runInstall().finally(() => { installInFlight = null; });
+  return installInFlight;
+}
+
+async function runInstall() {
+  const npm = findNpm();
+  if (!npm) {
+    const err = new Error("npm not found on PATH — Node.js/npm is required to install PXPIPE");
+    err.code = "NPM_NOT_FOUND";
+    throw err;
+  }
+
+  ensureDir();
+  const pkgJson = path.join(PXPIPE_DIR, "package.json");
+  if (!fs.existsSync(pkgJson)) {
+    fs.writeFileSync(pkgJson, JSON.stringify({ name: "9router-pxpipe-host", private: true }, null, 2));
+  }
+
+  const outFd = fs.openSync(INSTALL_LOG, "a");
+  fs.writeSync(outFd, `\n[${new Date().toISOString()}] npm install ${PXPIPE_PACKAGE}@latest\n`);
+
+  await new Promise((resolve, reject) => {
+    const child = spawn(npm, ["install", `${PXPIPE_PACKAGE}@latest`, "--no-audit", "--no-fund", "--omit=dev"], {
+      cwd: PXPIPE_DIR,
+      stdio: ["ignore", outFd, outFd],
+      windowsHide: true,
+      env: { ...process.env, PATH: EXTENDED_PATH },
+    });
+    const timer = setTimeout(() => {
+      child.kill("SIGKILL");
+      reject(new Error("npm install timed out after 5 minutes — see install.log"));
+    }, INSTALL_TIMEOUT_MS);
+    child.once("error", (e) => { clearTimeout(timer); reject(e); });
+    child.once("exit", (code) => {
+      clearTimeout(timer);
+      if (code === 0) resolve();
+      else reject(new Error(`npm install exited with code ${code} — see install.log`));
+    });
+  }).finally(() => fs.closeSync(outFd));
+
+  const info = getInstallInfo();
+  if (!info.installed) throw new Error("install finished but package is missing — see install.log");
+  return info;
+}
+
+export function getInstallLogTail(maxLines = 200) {
+  try {
+    if (!fs.existsSync(INSTALL_LOG)) return "";
+    const lines = fs.readFileSync(INSTALL_LOG, "utf8").split(/\r?\n/).filter(Boolean);
+    return lines.slice(-maxLines).join("\n");
+  } catch {
+    return "";
+  }
+}
diff --git a/src/lib/pxpipe/loader.js b/src/lib/pxpipe/loader.js
new file mode 100644
index 00000000..fbd418f5
--- /dev/null
+++ b/src/lib/pxpipe/loader.js
@@ -0,0 +1,70 @@
+import { pathToFileURL } from "url";
+import { getInstallInfo, libraryEntry } from "./install.js";
+
+// Module cache: pxpipe is loaded once per process ("started") and dropped on
+// "stop". In library mode start/stop govern the in-process module, not a daemon.
+let cached = null; // { module, version, loadedAt }
+let loadPromise = null;
+
+export function getLoadedInfo() {
+  return cached ? { loaded: true, version: cached.version, loadedAt: cached.loadedAt } : { loaded: false };
+}
+
+export async function loadPxpipe() {
+  if (cached) return cached;
+  if (loadPromise) return loadPromise;
+  loadPromise = doLoad().finally(() => { loadPromise = null; });
+  return loadPromise;
+}
+
+async function doLoad() {
+  const info = getInstallInfo();
+  if (!info.installed) {
+    const err = new Error("PXPIPE is not installed");
+    err.code = "NOT_INSTALLED";
+    throw err;
+  }
+  // Cache-bust per version so Repair/upgrade takes effect without a server restart.
+  const url = `${pathToFileURL(libraryEntry()).href}?v=${encodeURIComponent(info.version || "0")}`;
+  const mod = await import(/* webpackIgnore: true */ url);
+  if (typeof mod.transformAnthropicMessages !== "function") {
+    throw new Error("installed pxpipe package does not export transformAnthropicMessages");
+  }
+  cached = { module: mod, version: info.version, loadedAt: Date.now() };
+  return cached;
+}
+
+export function unloadPxpipe() {
+  const wasLoaded = !!cached;
+  cached = null;
+  return wasLoaded;
+}
+
+// Transform function for the request pipeline; null when unavailable (fail-open).
+// autoLoad controls whether a cold cache triggers a load (first request warms it).
+export async function getTransform({ autoLoad = true } = {}) {
+  try {
+    if (!cached && !autoLoad) return null;
+    const { module: mod } = await loadPxpipe();
+    return mod.transformAnthropicMessages;
+  } catch {
+    return null;
+  }
+}
+
+// Health self-test: run a tiny synthetic Claude request through the transformer.
+// A healthy module parses it and answers with a machine-readable reason.
+export async function selfTest() {
+  const startedAt = Date.now();
+  const { module: mod } = await loadPxpipe();
+  const body = new TextEncoder().encode(JSON.stringify({
+    model: "claude-fable-5",
+    max_tokens: 16,
+    messages: [{ role: "user", content: "ping" }],
+  }));
+  const result = await mod.transformAnthropicMessages({ body, model: "claude-fable-5" });
+  if (!result || typeof result.applied !== "boolean" || !(result.body instanceof Uint8Array)) {
+    throw new Error("transform returned an unexpected shape");
+  }
+  return { ok: true, reason: result.reason, durationMs: Date.now() - startedAt };
+}
diff --git a/src/lib/pxpipe/service.js b/src/lib/pxpipe/service.js
new file mode 100644
index 00000000..d117c82b
--- /dev/null
+++ b/src/lib/pxpipe/service.js
@@ -0,0 +1,49 @@
+import { getInstallInfo, isInstalling, findNpm } from "./install.js";
+import { getLoadedInfo, loadPxpipe, selfTest } from "./loader.js";
+
+// Aggregate status for the Token Saver card and /api/pxpipe/status.
+// "running" in library mode = module loaded into this process.
+export function getPxpipeStatus() {
+  const install = getInstallInfo();
+  const loaded = getLoadedInfo();
+  return {
+    installed: install.installed,
+    installing: isInstalling(),
+    version: install.version,
+    path: install.path,
+    running: loaded.loaded,
+    loadedAt: loaded.loadedAt || null,
+    uptimeMs: loaded.loaded ? Date.now() - loaded.loadedAt : 0,
+    npmAvailable: !!findNpm(),
+    mode: "library", // in-process transform, not an external proxy
+  };
+}
+
+// PRD health checklist, adapted to library mode: installed? → module loads
+// (the "executable found / port listening" equivalent) → test request transforms.
+export async function runHealthCheck() {
+  const checks = [];
+  const fail = (error) => ({ healthy: false, checks, error });
+
+  const install = getInstallInfo();
+  checks.push({ id: "installed", label: "PXPIPE installed", ok: install.installed, detail: install.version ? `v${install.version}` : null });
+  if (!install.installed) return fail("pxpipe not installed");
+
+  try {
+    await loadPxpipe();
+    checks.push({ id: "module", label: "Transform module loads", ok: true, detail: `v${install.version}` });
+  } catch (e) {
+    checks.push({ id: "module", label: "Transform module loads", ok: false, detail: e.message });
+    return fail(`Cannot load module: ${e.message}`);
+  }
+
+  try {
+    const test = await selfTest();
+    checks.push({ id: "transform", label: "Test request transforms", ok: true, detail: `${test.durationMs}ms (${test.reason})` });
+  } catch (e) {
+    checks.push({ id: "transform", label: "Test request transforms", ok: false, detail: e.message });
+    return fail(`Self-test failed: ${e.message}`);
+  }
+
+  return { healthy: true, checks, error: null };
+}
diff --git a/src/shared/components/Sidebar.js b/src/shared/components/Sidebar.js
index 2c1206df..c8cf8161 100644
--- a/src/shared/components/Sidebar.js
+++ b/src/shared/components/Sidebar.js
@@ -25,6 +25,7 @@ const navItems = [
   { href: "/dashboard/usage", label: "Usage", icon: "bar_chart" },
   { href: "/dashboard/quota", label: "Quota Tracker", icon: "data_usage" },
   { href: "/dashboard/token-saver", label: "Token Saver", icon: "savings" },
+  { href: "/dashboard/pxpipe", label: "PXPIPE", icon: "image" },
   { href: "/dashboard/cli-tools", label: "CLI Tools", icon: "terminal" },
 ];
 
diff --git a/src/sse/handlers/chat.js b/src/sse/handlers/chat.js
index 571930e0..a0c187f5 100644
--- a/src/sse/handlers/chat.js
+++ b/src/sse/handlers/chat.js
@@ -12,6 +12,8 @@ import { getSettings } from "@/lib/localDb";
 import { getModelInfo, getComboModels } from "../services/model.js";
 import { handleChatCore } from "open-sse/handlers/chatCore.js";
 import { DEFAULT_HEADROOM_URL } from "@/lib/headroom/detect";
+import { getTransform as getPxpipeTransform } from "@/lib/pxpipe/loader.js";
+import { appendPxpipeEvent } from "@/lib/pxpipe/events.js";
 import { errorResponse, unavailableResponse } from "open-sse/utils/error.js";
 import { handleComboChat, handleFusionChat } from "open-sse/services/combo.js";
 import { handleBypassRequest } from "open-sse/utils/bypassHandler.js";
@@ -259,6 +261,12 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re
       cavemanLevel: chatSettings.cavemanLevel || "full",
       ponytailEnabled: !!chatSettings.ponytailEnabled,
       ponytailLevel: chatSettings.ponytailLevel || "full",
+      pxpipeEnabled: !!chatSettings.pxpipeEnabled,
+      pxpipeMinChars: chatSettings.pxpipeMinChars,
+      pxpipeTimeoutMs: chatSettings.pxpipeTimeoutMs,
+      // Lazily warms the in-process module on first use; null when not installed (fail-open)
+      pxpipeTransform: chatSettings.pxpipeEnabled ? await getPxpipeTransform() : null,
+      onPxpipeEvent: appendPxpipeEvent,
       providerThinking,
       // Detect source format by endpoint + body
       sourceFormatOverride: request?.url ? detectFormatByEndpoint(new URL(request.url).pathname, body) : null,
diff --git a/tests/unit/pxpipe.test.js b/tests/unit/pxpipe.test.js
new file mode 100644
index 00000000..5c49f126
--- /dev/null
+++ b/tests/unit/pxpipe.test.js
@@ -0,0 +1,95 @@
+import { describe, expect, it, vi } from "vitest";
+import { compressWithPxpipe, formatPxpipeLog } from "../../open-sse/rtk/pxpipe.js";
+
+const bigText = "x".repeat(30000);
+const claudeBody = () => ({
+  model: "claude-fable-5",
+  max_tokens: 100,
+  messages: [{ role: "user", content: bigText }],
+});
+
+// A transform double mimicking pxpipe-proxy/transform's contract.
+const appliedTransform = (outBody) => async () => ({
+  applied: true,
+  reason: "applied",
+  body: new TextEncoder().encode(JSON.stringify(outBody)),
+  info: { compressedChars: 25000, imageCount: 2, imageBytes: 5000, imagePixels: 1500000 },
+  cache: { ownsCacheControl: true, markerCount: 1 },
+});
+
+describe("compressWithPxpipe gates", () => {
+  it("skips when disabled", async () => {
+    const { body, summary } = await compressWithPxpipe(claudeBody(), { enabled: false });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("disabled");
+  });
+
+  it("skips when transform is unavailable (not installed)", async () => {
+    const { body, summary } = await compressWithPxpipe(claudeBody(), { enabled: true, format: "claude", transform: null });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("not_installed");
+  });
+
+  it("skips non-Claude formats", async () => {
+    const transform = vi.fn();
+    const { body, summary } = await compressWithPxpipe(claudeBody(), { enabled: true, format: "openai", transform });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("unsupported_format");
+    expect(transform).not.toHaveBeenCalled();
+  });
+
+  it("bypasses small prompts below minChars", async () => {
+    const transform = vi.fn();
+    const small = { model: "claude-fable-5", messages: [{ role: "user", content: "hi" }] };
+    const { body, summary } = await compressWithPxpipe(small, { enabled: true, format: "claude", minChars: 25000, transform });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("below_threshold");
+    expect(transform).not.toHaveBeenCalled();
+  });
+
+  it("applies the transform and reports savings", async () => {
+    const compressed = { model: "claude-fable-5", messages: [{ role: "user", content: "imaged" }] };
+    const { body, summary } = await compressWithPxpipe(claudeBody(), {
+      enabled: true, format: "claude", minChars: 1000, transform: appliedTransform(compressed),
+    });
+    expect(body).toEqual(compressed);
+    expect(summary.applied).toBe(true);
+    expect(summary.imageCount).toBe(2);
+    expect(summary.tokensBeforeEst).toBeGreaterThan(summary.tokensAfterEst);
+    expect(summary.savedPct).toBeGreaterThan(0);
+    expect(formatPxpipeLog(summary)).toContain("2 image(s)");
+  });
+
+  it("passes through when the transform declines (not_profitable)", async () => {
+    const transform = async () => ({ applied: false, reason: "not_profitable", body: new Uint8Array(), info: {} });
+    const { body, summary } = await compressWithPxpipe(claudeBody(), {
+      enabled: true, format: "claude", minChars: 1000, transform,
+    });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("not_profitable");
+  });
+
+  it("fails open when the transform throws", async () => {
+    const transform = async () => { throw new Error("boom"); };
+    const { body, summary } = await compressWithPxpipe(claudeBody(), {
+      enabled: true, format: "claude", minChars: 1000, transform,
+    });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("transform_error");
+    expect(summary.detail).toBe("boom");
+  });
+
+  it("fails open on timeout", async () => {
+    const transform = () => new Promise(() => {}); // never resolves
+    const { body, summary } = await compressWithPxpipe(claudeBody(), {
+      enabled: true, format: "claude", minChars: 1000, timeoutMs: 50, transform,
+    });
+    expect(body).toBeNull();
+    expect(summary.reason).toBe("timeout");
+  });
+
+  it("does not log skipped requests as savings", () => {
+    expect(formatPxpipeLog({ applied: false, reason: "below_threshold" })).toBeNull();
+    expect(formatPxpipeLog(null)).toBeNull();
+  });
+});