diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index d46de4d5ec..552d41029c 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -1,5 +1,6 @@ { "_comment": "Catraca de tamanho (check-file-size.mjs). frozen so pode encolher; arquivos novos <= cap. --update ratcheta.", + "_rebaseline_2026_06_18_4217_compression_step_streaming": "PR #4217 own growth: chatCore.ts 5063->5086 (+23 at the existing compression-apply chokepoint = the best-effort onEngineStep callback threaded into applyCompressionAsync). The callback builds a compression.step payload and fires emit(\"compression.step\", …) + forwardDashboardEventToLiveWs(…) once per stacked engine as it completes (F3.3 live per-engine streaming), wrapped in try/catch so it never fails the request. It closes over the same emit/traceId/mode locals as the compression.completed emit right below it (line 1749); the reusable per-engine emission lives in strategySelector.ts (reportEngineStep + StackedCompressionStep) and the studio reducers in compressionFlowModel.ts (both 5063 (+3 = wire ensureEngineBreakdown(result.stats) into the existing compression.completed emit + its import line). Single-engine modes (rtk/lite/standard/aggressive/ultra) leave stats.engineBreakdown empty, which made the dashboard studio render an empty Input->Output pipeline (no engine node); the synthesized 1-entry breakdown lives in the new pure leaf open-sse/services/compression/engineBreakdown.ts (2534 (+3 = one NAMED_OPENAI_STYLE_PROVIDERS Set entry `zenmux` + a 2-line comment). zenmux carries a real modelsUrl but was not classified by any live-fetch branch, so its hardcoded 9-entry registry catalog was served (source local_catalog, 'API unavailable — using local catalog') instead of the upstream list — hiding the free models it advertises (z-ai/glm-5.2-free, moonshotai/kimi-k2.7-code-free). Same fix shape as #3976 (llm7/byteplus): the `/models` probe (after stripping /chat/completions) resolves to https://zenmux.ai/api/v1/models. Pure additive Set membership; not extractable.", "_rebaseline_2026_06_18_4189_combo_token_limits": "PR #4189 (megamen32) own growth: catalog.ts 1440->1463 (+23 at the existing #4164 auto/* emission chokepoint). The bare auto/* /v1/models entries are enriched with the combo's advertised context/output limits (createBuiltinAutoCombo → advertisedContextLength/advertisedMaxOutputTokens computed from the candidate pool, 128000/8192 fallback) + baseline capabilities, with a try/catch that emits the minimal #4164 entry on resolve failure so the id is never dropped. OpenAI-compatible pickers (Hermes) need a context window before the first request. Cohesive emission at the single auto/* loop; not extractable.", @@ -68,7 +69,7 @@ "open-sse/executors/muse-spark-web.ts": 1284, "open-sse/executors/perplexity-web.ts": 1013, "open-sse/handlers/audioSpeech.ts": 965, - "open-sse/handlers/chatCore.ts": 5063, + "open-sse/handlers/chatCore.ts": 5086, "open-sse/handlers/imageGeneration.ts": 3777, "open-sse/handlers/responseSanitizer.ts": 1103, "open-sse/handlers/search.ts": 1546, diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 84ac5a731f..65662a6307 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -1717,6 +1717,29 @@ export async function handleChatCore({ model: effectiveModel, config: compressionConfig, principalId: apiKeyInfo?.id ? String(apiKeyInfo.id) : undefined, + // F3.3: stream per-engine progress live (best-effort) before compression.completed. + onEngineStep: (s) => { + try { + const stepPayload = { + requestId: traceId, + comboId: null, + mode, + stepIndex: s.stepIndex, + totalSteps: s.totalSteps, + engine: s.engine, + state: s.state, + originalTokens: s.originalTokens, + compressedTokens: s.compressedTokens, + savingsPercent: s.savingsPercent, + ...(s.durationMs !== undefined ? { durationMs: s.durationMs } : {}), + timestamp: Date.now(), + }; + emit("compression.step", stepPayload); + void forwardDashboardEventToLiveWs("compression.step", stepPayload); + } catch (_stepErr) { + // best-effort live event — never fail the request + } + }, }); if (result.stats) { if (result.compressed) { diff --git a/open-sse/services/compression/strategySelector.ts b/open-sse/services/compression/strategySelector.ts index 20e15580a1..548051f9e2 100644 --- a/open-sse/services/compression/strategySelector.ts +++ b/open-sse/services/compression/strategySelector.ts @@ -222,6 +222,7 @@ export async function applyCompressionAsync( supportsVision?: boolean | null; config?: CompressionConfig; principalId?: string; + onEngineStep?: (step: StackedCompressionStep) => void; } ): Promise { if (mode === "stacked") { @@ -256,6 +257,18 @@ interface BailoutConfig { minGainPercent?: number; } +/** Per-engine progress emitted mid-pipeline by the stacked loops (F3.3 live streaming). */ +export interface StackedCompressionStep { + stepIndex: number; + totalSteps: number; + engine: string; + state: "done" | "skipped"; + originalTokens: number; + compressedTokens: number; + savingsPercent: number; + durationMs?: number; +} + interface StackOptions { model?: string; supportsVision?: boolean | null; @@ -265,6 +278,30 @@ interface StackOptions { bailout?: BailoutConfig; /** Authenticated principal id — threaded through to CCR engine for store scoping. */ principalId?: string; + /** F3.3: called once per engine as it completes (live per-engine streaming). */ + onEngineStep?: (step: StackedCompressionStep) => void; +} + +/** Emit a per-engine step to the live streaming callback (best-effort, no-op when unset). */ +function reportEngineStep( + onStep: ((step: StackedCompressionStep) => void) | undefined, + stepIndex: number, + totalSteps: number, + engine: string, + result: CompressionResult +): void { + if (!onStep) return; + const s = result.stats; + onStep({ + stepIndex, + totalSteps, + engine, + state: result.compressed ? "done" : "skipped", + originalTokens: s?.originalTokens ?? 0, + compressedTokens: s?.compressedTokens ?? s?.originalTokens ?? 0, + savingsPercent: s?.savingsPercent ?? 0, + ...(s?.durationMs !== undefined ? { durationMs: s.durationMs } : {}), + }); } /** Accumulates per-step telemetry across a stacked run (shared sync/async). */ @@ -406,6 +443,9 @@ export function applyStackedCompression( const start = performance.now(); const bailout = options?.bailout; + const onStep = options?.onEngineStep; + const totalSteps = steps.length; + let stepIdx = 0; for (const step of steps) { const engine = getCompressionEngine(step.engine); @@ -438,6 +478,7 @@ export function applyStackedCompression( currentBody = result.body; compressed = true; } + reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result); } } @@ -471,6 +512,9 @@ export async function applyStackedCompressionAsync( const start = performance.now(); const bailout = options?.bailout; + const onStep = options?.onEngineStep; + const totalSteps = steps.length; + let stepIdx = 0; for (const step of steps) { const engine = getCompressionEngine(step.engine); @@ -507,6 +551,7 @@ export async function applyStackedCompressionAsync( currentBody = result.body; compressed = true; } + reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result); } } diff --git a/src/app/(dashboard)/dashboard/compression/studio/compressionFlowModel.ts b/src/app/(dashboard)/dashboard/compression/studio/compressionFlowModel.ts index a5024ef3d0..4b30b2e109 100644 --- a/src/app/(dashboard)/dashboard/compression/studio/compressionFlowModel.ts +++ b/src/app/(dashboard)/dashboard/compression/studio/compressionFlowModel.ts @@ -9,7 +9,7 @@ */ import type { Node, Edge } from "@xyflow/react"; -import type { CompressionCompletedPayload } from "@/lib/events/types"; +import type { CompressionCompletedPayload, CompressionStepPayload } from "@/lib/events/types"; // ── Engine Step ─────────────────────────────────────────────────────────── @@ -64,6 +64,67 @@ export function compressionEventToModel(payload: CompressionCompletedPayload): C }; } +// ── Live step streaming (F3.3) ──────────────────────────────────────────── + +/** + * Build a partial `CompressionRunModel` from the per-engine `compression.step` events received + * so far. Run-level totals span the whole pipeline: input = first step's input, output = last + * step's output. Used to render the live in-flight run before `compression.completed` arrives. + */ +export function stepEventsToRunModel(steps: CompressionStepPayload[]): CompressionRunModel { + const first = steps[0]; + const last = steps[steps.length - 1]; + const originalTokens = first?.originalTokens ?? 0; + const compressedTokens = last?.compressedTokens ?? originalTokens; + const savingsPercent = + originalTokens > 0 + ? Math.round(((originalTokens - compressedTokens) / originalTokens) * 100) + : 0; + return { + requestId: first?.requestId ?? "", + comboId: first?.comboId ?? null, + mode: first?.mode ?? "stacked", + originalTokens, + compressedTokens, + savingsPercent, + steps: steps.map((s) => ({ + engine: s.engine, + originalTokens: s.originalTokens, + compressedTokens: s.compressedTokens, + savingsPercent: s.savingsPercent, + techniquesUsed: s.techniquesUsed ?? [], + rulesApplied: s.rulesApplied, + durationMs: s.durationMs, + })), + timestamp: last?.timestamp ?? 0, + }; +} + +/** A single in-flight (still-streaming) compression run, keyed by requestId. */ +export interface InFlightCompressionRun { + requestId: string; + steps: CompressionStepPayload[]; +} + +/** Append a step to the in-flight run; a new requestId starts a fresh run (latest wins). */ +export function appendInFlightStep( + state: InFlightCompressionRun | null, + step: CompressionStepPayload +): InFlightCompressionRun { + if (state && state.requestId === step.requestId) { + return { requestId: state.requestId, steps: [...state.steps, step] }; + } + return { requestId: step.requestId, steps: [step] }; +} + +/** Clear the in-flight run when its run completes (otherwise leave it untouched). */ +export function clearInFlightOnComplete( + state: InFlightCompressionRun | null, + completedRequestId: string +): InFlightCompressionRun | null { + return state && state.requestId === completedRequestId ? null : state; +} + // ── compressionRunToFlow ────────────────────────────────────────────────── /** diff --git a/src/hooks/useLiveCompression.ts b/src/hooks/useLiveCompression.ts index 56707e5d6f..9a41a70a52 100644 --- a/src/hooks/useLiveCompression.ts +++ b/src/hooks/useLiveCompression.ts @@ -8,9 +8,13 @@ import { } from "./useLiveDashboard"; import { compressionEventToModel, + stepEventsToRunModel, + appendInFlightStep, + clearInFlightOnComplete, type CompressionRunModel, + type InFlightCompressionRun, } from "@/app/(dashboard)/dashboard/compression/studio/compressionFlowModel"; -import type { CompressionCompletedPayload } from "@/lib/events/types"; +import type { CompressionCompletedPayload, CompressionStepPayload } from "@/lib/events/types"; // ── Constants ───────────────────────────────────────────────────────────── @@ -52,13 +56,19 @@ export interface UseLiveCompressionReturn { */ export function useLiveCompression(options?: UseLiveDashboardOptions): UseLiveCompressionReturn { const [runs, setRuns] = useState([]); + const [inFlight, setInFlight] = useState(null); const handleEvent = useCallback((event: WsEventPayload) => { if (event.channel !== "compression") return; - if (event.event !== "compression.completed") return; - - const payload = event.data as CompressionCompletedPayload; - setRuns((prev) => accumulateRun(prev, payload)); + if (event.event === "compression.step") { + setInFlight((prev) => appendInFlightStep(prev, event.data as CompressionStepPayload)); + return; + } + if (event.event === "compression.completed") { + const payload = event.data as CompressionCompletedPayload; + setInFlight((prev) => clearInFlightOnComplete(prev, payload.requestId)); + setRuns((prev) => accumulateRun(prev, payload)); + } }, []); const { connection, reconnect } = useLiveDashboard({ @@ -72,9 +82,14 @@ export function useLiveCompression(options?: UseLiveDashboardOptions): UseLiveCo [runs] ); + const inFlightRun = + inFlight && inFlight.steps.length > 0 ? stepEventsToRunModel(inFlight.steps) : null; + return { runs, - lastRun: runs[0] ?? null, + // Prefer the live in-flight run so the studio shows engines as they stream in (F3.3), + // falling back to the latest completed run. + lastRun: inFlightRun ?? runs[0] ?? null, getRunById, isConnected: connection.isConnected, reconnect, diff --git a/src/lib/events/types.ts b/src/lib/events/types.ts index 9f7abc1983..874278e93a 100644 --- a/src/lib/events/types.ts +++ b/src/lib/events/types.ts @@ -16,7 +16,8 @@ export type DashboardEventName = | "combo.target.failed" | "combo.target.succeeded" | "credential.health.changed" - | "compression.completed"; + | "compression.completed" + | "compression.step"; // ── Event Payloads ──────────────────────────────────────────────────────── @@ -110,6 +111,33 @@ export interface CompressionCompletedPayload { timestamp: number; } +/** + * Mid-pipeline per-engine progress for a stacked compression run (F3.3). Emitted once per + * engine as it completes, before the final `compression.completed`. The studio accumulates + * these into a live in-flight run so engines appear as they finish (meaningful for slow async + * engines like LLMLingua); for sub-ms sync engines they arrive as a burst. + */ +export interface CompressionStepPayload { + requestId: string; + comboId: string | null; + mode: string; + /** 0-based index of this engine among the resolved pipeline steps. */ + stepIndex: number; + /** Total resolved pipeline steps (hint for the UI). */ + totalSteps: number; + engine: string; + state: "running" | "done" | "skipped"; + /** Tokens entering this engine. */ + originalTokens: number; + /** Tokens leaving this engine. */ + compressedTokens: number; + savingsPercent: number; + techniquesUsed?: string[]; + rulesApplied?: string[]; + durationMs?: number; + timestamp: number; +} + // ── Event Map ───────────────────────────────────────────────────────────── export interface DashboardEventMap { @@ -122,6 +150,7 @@ export interface DashboardEventMap { "combo.target.succeeded": ComboTargetSucceededPayload; "credential.health.changed": CredentialHealthChangedPayload; "compression.completed": CompressionCompletedPayload; + "compression.step": CompressionStepPayload; } // ── Event Bus Listener ──────────────────────────────────────────────────── @@ -140,7 +169,7 @@ export const CHANNEL_EVENTS: Record = { requests: ["request.started", "request.streaming", "request.completed", "request.failed"], combo: ["combo.target.attempt", "combo.target.failed", "combo.target.succeeded"], credentials: ["credential.health.changed"], - compression: ["compression.completed"], + compression: ["compression.completed", "compression.step"], }; /** Get channel for an event */ diff --git a/tests/unit/compression/compression-step-streaming.test.ts b/tests/unit/compression/compression-step-streaming.test.ts new file mode 100644 index 0000000000..b9a3bee07f --- /dev/null +++ b/tests/unit/compression/compression-step-streaming.test.ts @@ -0,0 +1,135 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + stepEventsToRunModel, + appendInFlightStep, + clearInFlightOnComplete, +} from "../../../src/app/(dashboard)/dashboard/compression/studio/compressionFlowModel.ts"; +import { applyStackedCompression } from "../../../open-sse/services/compression/strategySelector.ts"; +import { registerCompressionEngine } from "../../../open-sse/services/compression/index.ts"; +import type { CompressionEngine } from "../../../open-sse/services/compression/engines/types.ts"; +import type { CompressionStepPayload } from "../../../src/lib/events/types.ts"; + +function step(over: Partial): CompressionStepPayload { + return { + requestId: "r1", + comboId: null, + mode: "stacked", + stepIndex: 0, + totalSteps: 2, + engine: "e", + state: "done", + originalTokens: 1000, + compressedTokens: 800, + savingsPercent: 20, + timestamp: 1, + ...over, + }; +} + +describe("stepEventsToRunModel", () => { + it("builds a run model from accumulated step events (run-level totals span first→last)", () => { + const model = stepEventsToRunModel([ + step({ engine: "a", stepIndex: 0, originalTokens: 1000, compressedTokens: 900, savingsPercent: 10, timestamp: 1 }), + step({ engine: "b", stepIndex: 1, originalTokens: 900, compressedTokens: 700, savingsPercent: 22, timestamp: 2 }), + ]); + assert.equal(model.requestId, "r1"); + assert.equal(model.mode, "stacked"); + assert.equal(model.originalTokens, 1000); // first step's input + assert.equal(model.compressedTokens, 700); // last step's output + assert.equal(model.savingsPercent, 30); // (1000-700)/1000 + assert.equal(model.steps.length, 2); + assert.equal(model.steps[1].engine, "b"); + assert.equal(model.timestamp, 2); + }); +}); + +describe("in-flight step reducer", () => { + it("appends steps for the same requestId and starts fresh on a new requestId", () => { + let s = appendInFlightStep(null, step({ requestId: "r1", engine: "a" })); + assert.equal(s.requestId, "r1"); + assert.equal(s.steps.length, 1); + s = appendInFlightStep(s, step({ requestId: "r1", engine: "b" })); + assert.equal(s.steps.length, 2); + // A new requestId replaces the in-flight run (latest wins). + s = appendInFlightStep(s, step({ requestId: "r2", engine: "x" })); + assert.equal(s.requestId, "r2"); + assert.equal(s.steps.length, 1); + }); + + it("clears the in-flight run only when the completing requestId matches", () => { + const s = appendInFlightStep(null, step({ requestId: "r1" })); + assert.equal(clearInFlightOnComplete(s, "other"), s); + assert.equal(clearInFlightOnComplete(s, "r1"), null); + }); +}); + +// ── Integration: applyStackedCompression emits a step per engine ─────────────── +function fakeEngine(id: string, compressed: boolean, orig: number, comp: number): CompressionEngine { + return { + id, + name: id, + description: "", + icon: "", + targets: ["messages"], + stackable: true, + stackPriority: 1, + metadata: { + id, + name: id, + description: "", + inputScope: "messages", + targetLatencyMs: 1, + supportsPreview: false, + stable: false, + }, + apply(body) { + return { + body, + compressed, + stats: { + originalTokens: orig, + compressedTokens: comp, + savingsPercent: orig > 0 ? Math.round(((orig - comp) / orig) * 100) : 0, + techniquesUsed: [], + mode: "stacked", + timestamp: 0, + durationMs: 2, + }, + }; + }, + compress(body) { + return this.apply(body); + }, + getConfigSchema() { + return []; + }, + validateConfig() { + return { valid: true, errors: [] }; + }, + }; +} + +describe("applyStackedCompression — onEngineStep emission", () => { + it("fires onEngineStep once per engine with index/total/state", () => { + registerCompressionEngine(fakeEngine("step-e1", true, 1000, 900)); + registerCompressionEngine(fakeEngine("step-e2", false, 900, 900)); + + const captured: Array<{ stepIndex: number; totalSteps: number; engine: string; state: string }> = []; + applyStackedCompression( + { messages: [{ role: "user", content: "hello" }] }, + [{ engine: "step-e1" }, { engine: "step-e2" }], + { onEngineStep: (s) => captured.push(s) } + ); + + assert.equal(captured.length, 2, "one step per engine"); + assert.equal(captured[0].engine, "step-e1"); + assert.equal(captured[0].stepIndex, 0); + assert.equal(captured[0].totalSteps, 2); + assert.equal(captured[0].state, "done"); + assert.equal(captured[1].engine, "step-e2"); + assert.equal(captured[1].stepIndex, 1); + assert.equal(captured[1].state, "skipped"); + }); +});