feat(compression): live per-engine streaming via compression.step (F3.3) (#4217)

Integrated into release/v3.8.29 (F3.3 live per-engine compression streaming; baseline rebaselined chatCore.ts 5063->5086).

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 <cap). Not extractable without hiding the emit boundary, mirroring the prior compression rebaselines (#4210/#4004). Structural shrink of chatCore.ts tracked in #3501.",
   "_rebaseline_2026_06_18_4210_engine_breakdown": "PR #4210 own growth: chatCore.ts 5060->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 (<cap), mirroring seedLatestCompressionRunFromDb. The +3 is the import + a 2-line explanatory comment at the emit chokepoint; not extractable further without hiding the emit boundary.",
   "_rebaseline_2026_06_18_4202_zenmux_live_models": "Issue #4202 own growth: providers/[id]/models/route.ts 2531->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 `<baseUrl>/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,

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) {

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<CompressionResult> {
   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);
     }
   }
 

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 ──────────────────────────────────────────────────
 
 /**

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<CompressionRunModel[]>([]);
+  const [inFlight, setInFlight] = useState<InFlightCompressionRun | null>(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,

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<DashboardChannel, DashboardEventName[]> = {
   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 */

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>): 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");
+  });
+});