diff --git a/nodejs/src/fusionProgress.ts b/nodejs/src/fusionProgress.ts new file mode 100644 index 0000000000..a179d8e0d9 --- /dev/null +++ b/nodejs/src/fusionProgress.ts @@ -0,0 +1,761 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +/** + * Presentation-neutral HydraFusion progress projection. + * + * `reduceFusionProgress` folds session events into {@link FusionProgressState}, a deterministic + * snapshot of *what the runtime is doing right now* for a HydraFusion turn. It contains no prose, + * colors, timers, layout, or CLI-specific concepts: consumers (terminal UIs, web UIs, logs, + * dashboards) decide entirely how to render it. + * + * Design constraints: + * + * - **Pure and event-driven.** The reducer never performs I/O and never issues an RPC. Feed it the + * events a session already emits; a resumed session can be rebuilt by replaying its durable + * event log. + * - **Tolerant.** Duplicate events are idempotent, out-of-order events never regress a terminal + * phase state, missed ephemeral events are recovered from the durable events that follow, and + * every field added by newer runtimes is optional — against an older runtime the projection + * simply degrades to what the existing phase events carry. + * - **Single-turn and authoritative about turn identity.** The projection tracks one HydraFusion + * turn at a time. Evidence that arrives before the turn resolves (a phase, activity, tool, or + * permission event) is adopted only while *no* turn identity has been established yet, and it + * promotes the projection to `"running"`. Once an identity is established — or once the turn + * reaches a terminal `"completed"`/`"fallback"` state — only the durable `session.fusion_resolved` + * event may replace it. Late ephemerals from a previous turn, and the first ephemerals of a next + * turn that has not resolved yet, are intentionally ignored until that resolution arrives. + * - **Privacy-preserving.** Only event discriminants, phase kinds/roles/scopes, phase-plan + * metadata, safe response-byte counts, tool call IDs, permission attribution, commit IDs, and + * the stable terminal outcome are retained. Phase content, verdicts, prompts, reasoning, + * critiques, provider error messages, and concrete model identities are never read or stored. + * + * The reducer accepts a structurally typed event ({@link FusionProgressEventInput}) rather than the + * generated `SessionEvent` union, so it works with generated events, hand-built events, and raw + * JSON-RPC payloads alike, and can land ahead of the generated declarations for the newest + * experimental fields (`session.fusion_resolved.data.phasePlan`, + * `assistant.fusion_phase_activity`, and `fusion` attribution on permission events). + * + * @experimental The underlying HydraFusion event contract is experimental and may change. + */ + +/** Known HydraFusion phase kinds. Unrecognized kinds from newer runtimes are preserved verbatim. */ +export type FusionProgressPhaseKind = + | "primary" + | "judge" + | "repair" + | "draft" + | "critic" + | "revision" + | "follow_up" + | (string & {}); + +/** Conversation scope a phase executes in. Unrecognized scopes are preserved verbatim. */ +export type FusionProgressScope = "root" | "review" | (string & {}); + +/** Orchestration pattern selected for the turn. Unrecognized patterns are preserved verbatim. */ +export type FusionProgressPattern = "single" | "cascade" | "critique" | (string & {}); + +/** Kind of turn HydraFusion routing ran for. */ +export type FusionProgressTurnKind = "user" | "compaction" | (string & {}); + +/** Lifecycle of the HydraFusion turn currently represented by the state. */ +export type FusionProgressStatus = + /** No HydraFusion activity has been observed. */ + | "inactive" + /** Routing started and has not yet resolved. */ + | "routing" + /** Routing failed; the turn runs on a deterministic concrete fallback instead. */ + | "fallback" + /** + * The turn's phases are executing — either because the route resolved, or because phase + * evidence was observed before the resolution and no other turn identity existed yet. + */ + | "running" + /** The turn reached its aggregate outcome. */ + | "completed"; + +/** Observed lifecycle of a single phase. */ +export type FusionProgressPhaseStatus = "running" | "succeeded" | "failed" | "cancelled"; + +/** Safe activity discriminant most recently observed for a phase. */ +export type FusionProgressActivityKind = + | "model_output" + | "tool_started" + | "tool_completed" + | (string & {}); + +/** Observed lifecycle of a tool call attributed to a phase. */ +export type FusionProgressToolCallStatus = "started" | "completed"; + +/** One entry of the expected phase plan published with the resolved route. */ +export interface FusionPlannedPhase { + /** Phase kind the plan expects to run. */ + readonly kind: FusionProgressPhaseKind; + /** Semantic role assigned to the planned phase. Never a concrete model identity. */ + readonly role: string; + /** Conversation scope the planned phase executes in. */ + readonly scope: FusionProgressScope; + /** Whether the planned phase only runs when an earlier phase requires it. */ + readonly conditional: boolean; +} + +/** A tool call attributed to a phase, tracked by its identifier only. */ +export interface FusionProgressToolCall { + /** Runtime-assigned tool call identifier. Carries no tool arguments or results. */ + readonly toolCallId: string; + /** Whether the call has been observed completing. */ + readonly status: FusionProgressToolCallStatus; +} + +/** Most recently observed safe activity for a phase. */ +export interface FusionProgressActivity { + /** Activity discriminant reported by the runtime. */ + readonly kind: FusionProgressActivityKind; + /** Tool call the activity refers to, when the activity is tool-scoped. */ + readonly toolCallId?: string; + /** Cumulative response size in bytes reported with the activity, when available. */ + readonly totalResponseSizeBytes?: number; +} + +/** Observed state of one concrete phase of the turn. */ +export interface FusionProgressPhase { + /** Stable identifier of the concrete phase. */ + readonly phaseId: string; + /** Phase kind, when any observed event carried it. */ + readonly kind?: FusionProgressPhaseKind; + /** Semantic role assigned to the phase, when observed. Never a concrete model identity. */ + readonly role?: string; + /** Conversation scope the phase executes in, when observed. */ + readonly scope?: FusionProgressScope; + /** Observed lifecycle state. Terminal states are never downgraded by later events. */ + readonly status: FusionProgressPhaseStatus; + /** Whether the phase failed and the turn degraded to another phase. */ + readonly degraded: boolean; + /** Phase the turn degraded to after this phase failed, when reported. */ + readonly degradedToPhaseId?: string; + /** Most recent safe activity observed for the phase. */ + readonly activity?: FusionProgressActivity; + /** Highest cumulative response size in bytes observed for the phase. */ + readonly totalResponseSizeBytes?: number; + /** Tool calls attributed to the phase, in first-observation order. */ + readonly toolCalls: readonly FusionProgressToolCall[]; + /** Index of the matching {@link FusionProgressState.plan} entry, when the plan is known. */ + readonly planIndex?: number; +} + +/** A permission request attributed to a HydraFusion phase and still awaiting a decision. */ +export interface FusionProgressPermission { + /** Identifier used to respond to the request. */ + readonly requestId: string; + /** Turn the request belongs to. */ + readonly fusionId?: string; + /** Phase that raised the request, when attributed. */ + readonly phaseId?: string; + /** Kind of the phase that raised the request, when attributed. */ + readonly phaseKind?: FusionProgressPhaseKind; + /** Semantic role of the phase that raised the request, when attributed. */ + readonly role?: string; + /** Conversation scope of the phase that raised the request, when attributed. */ + readonly scope?: FusionProgressScope; +} + +/** Aggregate terminal outcome of the turn. */ +export interface FusionProgressCompletion { + /** Stable machine-readable aggregate outcome reported by the runtime. */ + readonly outcome: string; + /** Whether the turn reported using a degraded route. The reason string is deliberately dropped. */ + readonly degraded: boolean; + /** Idempotency identifier of the authoritative final commit, when reported. */ + readonly commitId?: string; + /** Phase whose output supplied the authoritative final content, when reported. */ + readonly finalSourcePhaseId?: string; +} + +/** Deterministic, presentation-neutral projection of HydraFusion progress. */ +export interface FusionProgressState { + /** Lifecycle of the turn currently represented. */ + readonly status: FusionProgressStatus; + /** Stable identifier of the turn, once any event carries it. */ + readonly fusionId?: string; + /** Session turn the route belongs to, when reported. */ + readonly turnId?: string; + /** Kind of turn routing ran for, when reported. */ + readonly turnKind?: FusionProgressTurnKind; + /** Orchestration pattern selected for the turn, when reported. */ + readonly pattern?: FusionProgressPattern; + /** Routing policy used for the turn, when reported. */ + readonly policy?: string; + /** Expected phase plan. Empty when the runtime does not publish one. */ + readonly plan: readonly FusionPlannedPhase[]; + /** Whether {@link FusionProgressState.plan} came from the runtime or is unavailable. */ + readonly planSource: "runtime" | "unavailable"; + /** Observed phases in first-observation order. Active phases have status `"running"`. */ + readonly phases: readonly FusionProgressPhase[]; + /** Attributed permission requests still awaiting a decision, in request order. */ + readonly pendingPermissions: readonly FusionProgressPermission[]; + /** Commit identifier observed on published authoritative output for this turn. */ + readonly publishedCommitId?: string; + /** Aggregate terminal outcome, once the turn completes. */ + readonly completion?: FusionProgressCompletion; +} + +/** + * Minimal structural shape the reducer needs from a session event. + * + * Generated `SessionEvent` values satisfy this shape, as do raw decoded JSON-RPC payloads. + */ +export interface FusionProgressEventInput { + /** Event type discriminator. */ + readonly type: string; + /** Event payload. Read defensively; unknown or malformed payloads are ignored. */ + readonly data?: unknown; +} + +const EMPTY_STATE: FusionProgressState = { + status: "inactive", + plan: [], + planSource: "unavailable", + phases: [], + pendingPermissions: [], +}; + +/** Returns the empty projection used as the seed for {@link reduceFusionProgress}. */ +export function initialFusionProgressState(): FusionProgressState { + return EMPTY_STATE; +} + +/** + * Folds one session event into the HydraFusion progress projection. + * + * Events that are not HydraFusion-related, that carry no usable payload, or that belong to a + * different turn are returned unchanged — by reference, so consumers can cheaply detect that + * nothing moved — and the reducer can be applied to an entire event stream: + * + * ```ts + * const state = events.reduce(reduceFusionProgress, initialFusionProgressState()); + * ``` + * + * Turn identity is authoritative: unresolved early evidence is adopted only when no turn identity + * exists yet, and an established or terminal projection is replaced only by a durable + * `session.fusion_resolved`. Late ephemerals from a previous turn — and the first ephemerals of a + * next turn — are ignored until that turn resolves. + * + * @experimental + */ +export function reduceFusionProgress( + state: FusionProgressState, + event: FusionProgressEventInput +): FusionProgressState { + const data = asRecord(event?.data); + switch (event?.type) { + case "session.fusion_route_started": + return beginTurn(state, { + ...EMPTY_STATE, + status: "routing", + ...definedEntry("turnKind", optionalString(data?.turnKind)), + ...definedEntry("policy", optionalString(data?.policy)), + }); + case "session.fusion_route_failed": + return beginTurn(state, { + ...EMPTY_STATE, + status: "fallback", + ...definedEntry("turnKind", state.turnKind), + ...definedEntry("policy", optionalString(data?.policy) ?? state.policy), + }); + case "session.fusion_resolved": + return reduceResolved(state, data); + case "assistant.fusion_phase_started": + return reducePhaseSignal(state, data, "running"); + case "assistant.fusion_phase_completed": + return reducePhaseSignal(state, data, phaseStatusOf(data, "succeeded")); + case "assistant.fusion_phase_failed": + return reducePhaseSignal(state, data, phaseStatusOf(data, "failed")); + case "assistant.fusion_phase_activity": + return reduceActivity(state, data); + case "assistant.message": + return reducePublishedCommit(state, asRecord(data?.fusion)); + case "tool.execution_start": + return reduceAttributedTool(state, data, "started"); + case "tool.execution_complete": + return reduceAttributedTool(state, data, "completed"); + case "permission.requested": + return reducePermissionRequested(state, data); + case "permission.completed": + return reducePermissionCompleted(state, data); + case "session.fusion_completed": + return reduceCompleted(state, data); + default: + return state; + } +} + +function reduceResolved( + state: FusionProgressState, + data: Record | undefined +): FusionProgressState { + const fusionId = optionalString(data?.fusionId); + if (fusionId === undefined) { + return state; + } + const base = alignTurn(state, fusionId, true); + if (base === undefined) { + return state; + } + const plan = readPhasePlan(data?.phasePlan); + const next: FusionProgressState = { + ...base, + status: base.status === "completed" ? "completed" : "running", + fusionId, + ...definedEntry("turnId", optionalString(data?.turnId) ?? base.turnId), + ...definedEntry("pattern", optionalString(data?.pattern) ?? base.pattern), + ...definedEntry("policy", optionalString(data?.policy) ?? base.policy), + plan: plan ?? base.plan, + planSource: plan !== undefined ? "runtime" : base.planSource, + }; + return settle(state, withPlanIndexes(next)); +} + +function reducePhaseSignal( + state: FusionProgressState, + data: Record | undefined, + status: FusionProgressPhaseStatus +): FusionProgressState { + const phaseId = optionalString(data?.phaseId); + if (phaseId === undefined) { + return state; + } + const base = alignTurn(state, optionalString(data?.fusionId), false); + if (base === undefined) { + return state; + } + const degradedToPhaseId = optionalString(data?.degradedToPhaseId); + return upsertPhase(base, phaseId, (phase) => ({ + ...phase, + kind: optionalString(data?.phaseKind) ?? phase.kind, + role: optionalString(data?.role) ?? phase.role, + scope: optionalString(data?.conversationScope) ?? phase.scope, + status: mergePhaseStatus(phase.status, status), + degraded: phase.degraded || degradedToPhaseId !== undefined, + degradedToPhaseId: degradedToPhaseId ?? phase.degradedToPhaseId, + })); +} + +function reduceActivity( + state: FusionProgressState, + data: Record | undefined +): FusionProgressState { + const phaseId = optionalString(data?.phaseId); + const activityKind = optionalString(data?.activity); + if (phaseId === undefined || activityKind === undefined) { + return state; + } + const base = alignTurn(state, optionalString(data?.fusionId), false); + if (base === undefined) { + return state; + } + const toolCallId = optionalString(data?.toolCallId); + const bytes = optionalNonNegativeInteger(data?.totalResponseSizeBytes); + return upsertPhase(base, phaseId, (phase) => ({ + ...phase, + kind: optionalString(data?.phaseKind) ?? phase.kind, + role: optionalString(data?.role) ?? phase.role, + scope: optionalString(data?.conversationScope) ?? phase.scope, + activity: { + kind: activityKind, + ...(toolCallId !== undefined ? { toolCallId } : {}), + ...(bytes !== undefined ? { totalResponseSizeBytes: bytes } : {}), + }, + totalResponseSizeBytes: maxDefined(phase.totalResponseSizeBytes, bytes), + toolCalls: + toolCallId !== undefined && activityKind !== "model_output" + ? upsertToolCall( + phase.toolCalls, + toolCallId, + activityKind === "tool_completed" ? "completed" : "started" + ) + : phase.toolCalls, + })); +} + +function reduceAttributedTool( + state: FusionProgressState, + data: Record | undefined, + status: FusionProgressToolCallStatus +): FusionProgressState { + const fusion = asRecord(data?.fusion); + const phaseId = optionalString(fusion?.phaseId); + const toolCallId = optionalString(data?.toolCallId); + if (phaseId === undefined || toolCallId === undefined) { + return state; + } + const base = alignTurn(state, optionalString(fusion?.fusionId), false); + if (base === undefined) { + return state; + } + return upsertPhase(base, phaseId, (phase) => ({ + ...phase, + kind: optionalString(fusion?.phaseKind) ?? phase.kind, + role: optionalString(fusion?.role) ?? phase.role, + scope: optionalString(fusion?.conversationScope) ?? phase.scope, + toolCalls: upsertToolCall(phase.toolCalls, toolCallId, status), + })); +} + +function reducePublishedCommit( + state: FusionProgressState, + fusion: Record | undefined +): FusionProgressState { + const commitId = optionalString(fusion?.commitId); + if (commitId === undefined) { + return state; + } + const base = alignTurn(state, optionalString(fusion?.fusionId), false); + if (base === undefined || base.publishedCommitId === commitId) { + return base ?? state; + } + return { ...base, publishedCommitId: commitId }; +} + +function reducePermissionRequested( + state: FusionProgressState, + data: Record | undefined +): FusionProgressState { + const fusion = asRecord(data?.fusion); + const requestId = optionalString(data?.requestId); + if (fusion === undefined || requestId === undefined) { + return state; + } + const base = alignTurn(state, optionalString(fusion.fusionId), false); + if (base === undefined) { + return state; + } + const pending: FusionProgressPermission = { + requestId, + ...definedEntry("fusionId", optionalString(fusion.fusionId)), + ...definedEntry("phaseId", optionalString(fusion.phaseId)), + ...definedEntry("phaseKind", optionalString(fusion.phaseKind)), + ...definedEntry("role", optionalString(fusion.role)), + ...definedEntry("scope", optionalString(fusion.conversationScope)), + }; + const existingIndex = base.pendingPermissions.findIndex( + (candidate) => candidate.requestId === requestId + ); + if (existingIndex >= 0) { + const existing = base.pendingPermissions[existingIndex]; + if (shallowEqual(existing, pending)) { + return base; + } + const pendingPermissions = [...base.pendingPermissions]; + pendingPermissions[existingIndex] = pending; + return { ...base, pendingPermissions }; + } + return { ...base, pendingPermissions: [...base.pendingPermissions, pending] }; +} + +function reducePermissionCompleted( + state: FusionProgressState, + data: Record | undefined +): FusionProgressState { + const requestId = optionalString(data?.requestId); + if (requestId === undefined) { + return state; + } + const pendingPermissions = state.pendingPermissions.filter( + (pending) => pending.requestId !== requestId + ); + if (pendingPermissions.length === state.pendingPermissions.length) { + return state; + } + return { ...state, pendingPermissions }; +} + +function reduceCompleted( + state: FusionProgressState, + data: Record | undefined +): FusionProgressState { + const outcome = optionalString(data?.outcome); + if (outcome === undefined) { + return state; + } + const base = alignTurn(state, optionalString(data?.fusionId), false); + if (base === undefined) { + return state; + } + const commitId = optionalString(data?.commitId); + const completion: FusionProgressCompletion = { + outcome, + degraded: data?.degradedReason !== undefined && data?.degradedReason !== null, + ...definedEntry("commitId", commitId), + ...definedEntry("finalSourcePhaseId", optionalString(data?.finalSourcePhaseId)), + }; + return settle(state, { + ...base, + status: "completed", + ...definedEntry("pattern", optionalString(data?.pattern) ?? base.pattern), + ...definedEntry("turnId", optionalString(data?.turnId) ?? base.turnId), + completion, + }); +} + +/** Preserves referential identity when an event produced no observable change. */ +function settle(state: FusionProgressState, next: FusionProgressState): FusionProgressState { + return shallowEqual(state, next) ? state : next; +} + +/** + * Applies a pre-turn routing signal. + * + * Routing signals carry no `fusionId`, so a duplicated one cannot be distinguished from the start + * of another turn. A resolved turn is therefore only ever replaced by a later + * `session.fusion_resolved`, which is durable and carries an explicit identity. + */ +function beginTurn(state: FusionProgressState, next: FusionProgressState): FusionProgressState { + if (state.status === "running") { + return state; + } + return shallowEqual(state, next) ? state : next; +} + +/** + * Reconciles the incoming turn identity with the projected one. + * + * Two rules keep the projection stable: + * + * - **Early evidence is adopted only when no turn identity exists yet.** A phase, activity, tool, + * or permission event that arrives before `session.fusion_resolved` establishes the turn and + * promotes the projection to `"running"`, because that evidence can only come from a turn that is + * already executing. + * - **Only an authoritative turn-starting event replaces an established or terminal projection.** + * Once a turn identity is known, a mismatching `fusionId` is accepted only from the durable + * `session.fusion_resolved` path, and a completed or fallback projection is likewise only ever + * replaced there. Late ephemerals from a previous turn — and the first ephemerals of a next turn + * that has not resolved yet — are ignored instead of overwriting or resurrecting it. + * + * Returns the state to apply the event to, or `undefined` when the event must not disturb the + * current projection. + */ +function alignTurn( + state: FusionProgressState, + fusionId: string | undefined, + startsTurn: boolean +): FusionProgressState | undefined { + if (fusionId === undefined) { + return state; + } + if (state.fusionId === undefined) { + if (startsTurn) { + return { ...state, fusionId }; + } + if (isTerminalStatus(state.status)) { + return undefined; + } + // Evidence of an executing phase implies the turn resolved, even if that event was missed. + return { ...state, fusionId, status: "running" }; + } + if (state.fusionId === fusionId) { + return state; + } + return startsTurn ? { ...EMPTY_STATE, fusionId } : undefined; +} + +/** Terminal projections are never resurrected by anything but an authoritative new route. */ +function isTerminalStatus(status: FusionProgressStatus): boolean { + return status === "completed" || status === "fallback"; +} + +function upsertPhase( + state: FusionProgressState, + phaseId: string, + update: (phase: FusionProgressPhase) => FusionProgressPhase +): FusionProgressState { + const index = state.phases.findIndex((phase) => phase.phaseId === phaseId); + const existing = + index >= 0 + ? state.phases[index] + : { + phaseId, + status: "running" as FusionProgressPhaseStatus, + degraded: false, + toolCalls: [], + }; + const updated = compact(update(existing)); + if (index >= 0 && shallowEqual(existing, updated)) { + return state; + } + const phases = index >= 0 ? [...state.phases] : [...state.phases, updated]; + if (index >= 0) { + phases[index] = updated; + } + return withPlanIndexes({ ...state, phases }); +} + +/** Drops explicitly-undefined keys so equality checks and serialization stay stable. */ +function compact(value: T): T { + const entries = Object.entries(value).filter(([, entry]) => entry !== undefined); + return Object.fromEntries(entries) as T; +} + +function upsertToolCall( + toolCalls: readonly FusionProgressToolCall[], + toolCallId: string, + status: FusionProgressToolCallStatus +): readonly FusionProgressToolCall[] { + const index = toolCalls.findIndex((call) => call.toolCallId === toolCallId); + if (index < 0) { + return [...toolCalls, { toolCallId, status }]; + } + // "completed" is terminal: a duplicated or out-of-order "started" never regresses it. + if (toolCalls[index].status === "completed" || status === "started") { + return toolCalls; + } + const next = [...toolCalls]; + next[index] = { toolCallId, status }; + return next; +} + +/** + * Matches observed phases against the published plan, in order, consuming each plan entry once. + * + * Matching prefers an entry with the same kind *and* scope, then falls back to kind alone so a + * runtime that reports a scope the plan did not anticipate still yields a usable projection. + */ +function withPlanIndexes(state: FusionProgressState): FusionProgressState { + if (state.plan.length === 0) { + return state.phases.some((phase) => phase.planIndex !== undefined) + ? { ...state, phases: state.phases.map(({ planIndex: _planIndex, ...rest }) => rest) } + : state; + } + const used = new Set(); + const matched = new Map(); + for (const phase of state.phases) { + const index = state.plan.findIndex( + (planned, candidate) => + !used.has(candidate) && planned.kind === phase.kind && planned.scope === phase.scope + ); + if (index >= 0) { + used.add(index); + matched.set(phase.phaseId, index); + } + } + for (const phase of state.phases) { + if (matched.has(phase.phaseId)) { + continue; + } + const index = state.plan.findIndex( + (planned, candidate) => !used.has(candidate) && planned.kind === phase.kind + ); + if (index >= 0) { + used.add(index); + matched.set(phase.phaseId, index); + } + } + let changed = false; + const phases = state.phases.map((phase) => { + const planIndex = matched.get(phase.phaseId); + if (planIndex === undefined) { + if (phase.planIndex === undefined) { + return phase; + } + changed = true; + const { planIndex: _planIndex, ...rest } = phase; + return rest; + } + if (phase.planIndex === planIndex) { + return phase; + } + changed = true; + return { ...phase, planIndex }; + }); + return changed ? { ...state, phases } : state; +} + +function readPhasePlan(value: unknown): readonly FusionPlannedPhase[] | undefined { + if (!Array.isArray(value)) { + return undefined; + } + const plan: FusionPlannedPhase[] = []; + for (const entry of value) { + const record = asRecord(entry); + const kind = optionalString(record?.kind); + if (kind === undefined) { + continue; + } + plan.push({ + kind, + role: optionalString(record?.role) ?? "", + scope: optionalString(record?.scope) ?? "root", + conditional: record?.conditional === true, + }); + } + return plan; +} + +function phaseStatusOf( + data: Record | undefined, + fallback: FusionProgressPhaseStatus +): FusionProgressPhaseStatus { + const status = optionalString(data?.status); + return status === "succeeded" || status === "failed" || status === "cancelled" + ? status + : fallback; +} + +/** Terminal phase states are sticky, so duplicated or late "started" events cannot regress them. */ +function mergePhaseStatus( + current: FusionProgressPhaseStatus, + incoming: FusionProgressPhaseStatus +): FusionProgressPhaseStatus { + if (incoming === "running") { + return current; + } + return current === "running" ? incoming : current; +} + +function asRecord(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +function optionalString(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function optionalNonNegativeInteger(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; +} + +function maxDefined(current: number | undefined, incoming: number | undefined): number | undefined { + if (incoming === undefined) { + return current; + } + return current === undefined ? incoming : Math.max(current, incoming); +} + +function definedEntry(key: K, value: V | undefined): Record | undefined { + return value === undefined ? undefined : ({ [key]: value } as Record); +} + +function shallowEqual(left: object, right: object): boolean { + const leftRecord = left as Record; + const rightRecord = right as Record; + const leftKeys = Object.keys(leftRecord); + if (leftKeys.length !== Object.keys(rightRecord).length) { + return false; + } + return leftKeys.every((key) => valueEqual(leftRecord[key], rightRecord[key])); +} + +function valueEqual(left: unknown, right: unknown): boolean { + if (Array.isArray(left) && Array.isArray(right)) { + return left.length === right.length && left.every((item, i) => valueEqual(item, right[i])); + } + const leftRecord = asRecord(left); + const rightRecord = asRecord(right); + if (leftRecord !== undefined && rightRecord !== undefined) { + return shallowEqual(leftRecord, rightRecord); + } + return left === right; +} diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts index 9d55ab1d10..84fcce31f4 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -13,6 +13,26 @@ export { DisableBypassPermissionsModes, RuntimeConnection } from "./types.js"; export { BuiltInTools, ToolSet } from "./toolSet.js"; export { CopilotSession, type AssistantMessageEvent } from "./session.js"; export { defineFactory, FactoryResumeError, isFactoryRunTerminal } from "./factory.js"; +export { + initialFusionProgressState, + reduceFusionProgress, + type FusionPlannedPhase, + type FusionProgressActivity, + type FusionProgressActivityKind, + type FusionProgressCompletion, + type FusionProgressEventInput, + type FusionProgressPattern, + type FusionProgressPermission, + type FusionProgressPhase, + type FusionProgressPhaseKind, + type FusionProgressPhaseStatus, + type FusionProgressScope, + type FusionProgressState, + type FusionProgressStatus, + type FusionProgressToolCall, + type FusionProgressToolCallStatus, + type FusionProgressTurnKind, +} from "./fusionProgress.js"; export { Canvas, CanvasError, diff --git a/nodejs/test/fusion-progress.test.ts b/nodejs/test/fusion-progress.test.ts new file mode 100644 index 0000000000..feb1620e4b --- /dev/null +++ b/nodejs/test/fusion-progress.test.ts @@ -0,0 +1,761 @@ +import { describe, expect, it } from "vitest"; +import { + initialFusionProgressState, + reduceFusionProgress, + type FusionProgressEventInput, + type FusionProgressState, +} from "../src/fusionProgress.js"; +import type { SessionEvent } from "../src/types.js"; + +// Compile-time proof that the generated session-event union satisfies the structural input the +// reducer accepts, so consumers can feed session events directly without adapters. +const _generatedEventsAreAcceptedInput: FusionProgressEventInput = {} as SessionEvent; + +const FUSION_ID = "fusion-1"; + +function apply(events: readonly FusionProgressEventInput[]): FusionProgressState { + return events.reduce(reduceFusionProgress, initialFusionProgressState()); +} + +function routeStarted(turnKind = "user"): FusionProgressEventInput { + return { + type: "session.fusion_route_started", + data: { attemptId: "attempt-1", turnKind, policy: "balanced" }, + }; +} + +function resolved( + pattern: string, + phasePlan?: readonly Record[], + fusionId = FUSION_ID +): FusionProgressEventInput { + return { + type: "session.fusion_resolved", + data: { + contractVersion: 1, + fusionId, + turnId: "turn-1", + pattern, + policy: "balanced", + // Model identities are part of the real payload and must never surface in the state. + primaryModel: "secret-primary-model", + secondaryModel: "secret-secondary-model", + fallbackModel: "secret-fallback-model", + followUpModel: "secret-follow-up-model", + syntheticModel: "secret-synthetic-model", + ...(phasePlan !== undefined ? { phasePlan } : {}), + }, + }; +} + +function phaseStarted( + phaseId: string, + phaseKind: string, + role: string, + conversationScope = "root", + fusionId = FUSION_ID +): FusionProgressEventInput { + return { + type: "assistant.fusion_phase_started", + data: { + fusionId, + phaseId, + phaseKind, + role, + conversationScope, + pattern: "cascade", + model: "secret-phase-model", + }, + }; +} + +function phaseCompleted( + phaseId: string, + phaseKind: string, + status = "succeeded", + role = "solver", + conversationScope = "root", + fusionId = FUSION_ID +): FusionProgressEventInput { + return { + type: "assistant.fusion_phase_completed", + data: { + fusionId, + phaseId, + phaseKind, + role, + conversationScope, + status, + content: "secret phase content that must never be projected", + verdict: "secret verdict", + model: "secret-phase-model", + durationMs: 1234, + usage: { inputTokens: 10, outputTokens: 20 }, + }, + }; +} + +function activity( + phaseId: string, + kind: string, + extra: Record = {}, + fusionId = FUSION_ID +): FusionProgressEventInput { + return { + type: "assistant.fusion_phase_activity", + data: { + fusionId, + phaseId, + phaseKind: "primary", + pattern: "single", + role: "solver", + conversationScope: "root", + activity: kind, + ...extra, + }, + }; +} + +function completed(overrides: Record = {}): FusionProgressEventInput { + return { + type: "session.fusion_completed", + data: { + fusionId: FUSION_ID, + turnId: "turn-1", + outcome: "succeeded", + commitId: "commit-1", + degradedReason: null, + finalSourcePhaseId: "phase-primary", + pattern: "single", + phaseCount: 1, + durationMs: 4321, + finalSourceModel: "secret-final-model", + syntheticModel: "secret-synthetic-model", + followUpModel: "secret-follow-up-model", + ...overrides, + }, + }; +} + +const CASCADE_PLAN = [ + { kind: "primary", role: "solver", scope: "root", conditional: false }, + { kind: "judge", role: "judge", scope: "review", conditional: false }, + { kind: "repair", role: "repairer", scope: "root", conditional: true }, +]; + +describe("reduceFusionProgress", () => { + it("starts empty and ignores unrelated events", () => { + const state = initialFusionProgressState(); + expect(state.status).toBe("inactive"); + expect(state.plan).toEqual([]); + expect(state.planSource).toBe("unavailable"); + expect(reduceFusionProgress(state, { type: "user.message", data: { text: "hi" } })).toBe( + state + ); + expect(reduceFusionProgress(state, { type: "assistant.message", data: undefined })).toBe( + state + ); + }); + + it("projects a single-pattern turn from routing to completion", () => { + const state = apply([ + routeStarted(), + resolved("single", [{ kind: "primary", role: "solver", scope: "root" }]), + phaseStarted("phase-primary", "primary", "solver"), + activity("phase-primary", "model_output", { totalResponseSizeBytes: 512 }), + phaseCompleted("phase-primary", "primary"), + { + type: "assistant.message", + data: { + content: "final answer", + fusion: { + fusionId: FUSION_ID, + pattern: "single", + policy: "balanced", + commitId: "commit-1", + syntheticModel: "secret", + }, + }, + }, + completed(), + ]); + + expect(state.status).toBe("completed"); + expect(state.fusionId).toBe(FUSION_ID); + expect(state.turnId).toBe("turn-1"); + expect(state.turnKind).toBe("user"); + expect(state.pattern).toBe("single"); + expect(state.planSource).toBe("runtime"); + expect(state.plan).toEqual([ + { kind: "primary", role: "solver", scope: "root", conditional: false }, + ]); + expect(state.phases).toHaveLength(1); + expect(state.phases[0]).toMatchObject({ + phaseId: "phase-primary", + kind: "primary", + role: "solver", + scope: "root", + status: "succeeded", + planIndex: 0, + totalResponseSizeBytes: 512, + }); + expect(state.publishedCommitId).toBe("commit-1"); + expect(state.completion).toEqual({ + outcome: "succeeded", + degraded: false, + commitId: "commit-1", + finalSourcePhaseId: "phase-primary", + }); + }); + + it("tracks cascade phases against the published plan, including a conditional repair", () => { + const state = apply([ + routeStarted(), + resolved("cascade", CASCADE_PLAN), + phaseStarted("p1", "primary", "solver"), + phaseCompleted("p1", "primary"), + phaseStarted("p2", "judge", "judge", "review"), + phaseCompleted("p2", "judge", "succeeded", "judge", "review"), + phaseStarted("p3", "repair", "repairer"), + ]); + + expect(state.status).toBe("running"); + expect(state.plan).toHaveLength(3); + expect(state.plan[2]).toMatchObject({ kind: "repair", conditional: true }); + expect(state.phases.map((phase) => [phase.phaseId, phase.status, phase.planIndex])).toEqual( + [ + ["p1", "succeeded", 0], + ["p2", "succeeded", 1], + ["p3", "running", 2], + ] + ); + }); + + it("tracks a critique turn including a review-scoped critic and a failed phase", () => { + const state = apply([ + resolved("critique", [ + { kind: "draft", role: "drafter", scope: "root", conditional: false }, + { kind: "critic", role: "critic", scope: "review", conditional: false }, + { kind: "revision", role: "reviser", scope: "root", conditional: true }, + ]), + phaseStarted("d1", "draft", "drafter"), + phaseCompleted("d1", "draft", "succeeded", "drafter"), + phaseStarted("c1", "critic", "critic", "review"), + { + type: "assistant.fusion_phase_failed", + data: { + fusionId: FUSION_ID, + phaseId: "c1", + phaseKind: "critic", + role: "critic", + conversationScope: "review", + status: "failed", + reason: "provider_error", + errorMessage: "secret provider error detail", + model: "secret-critic-model", + degradedToPhaseId: "d1", + durationMs: 12, + usage: { inputTokens: 1, outputTokens: 0 }, + }, + }, + ]); + + expect(state.phases[0]).toMatchObject({ + phaseId: "d1", + scope: "root", + status: "succeeded", + }); + expect(state.phases[1]).toMatchObject({ + phaseId: "c1", + kind: "critic", + scope: "review", + status: "failed", + degraded: true, + degradedToPhaseId: "d1", + }); + }); + + it("degrades to phase events when an older runtime publishes no plan", () => { + const state = apply([ + routeStarted(), + resolved("cascade"), + phaseStarted("p1", "primary", "solver"), + phaseCompleted("p1", "primary"), + ]); + + expect(state.planSource).toBe("unavailable"); + expect(state.plan).toEqual([]); + expect(state.phases[0].planIndex).toBeUndefined(); + expect(state.phases[0]).toMatchObject({ kind: "primary", status: "succeeded" }); + expect(state.status).toBe("running"); + }); + + it("ignores a malformed phase plan payload without losing the route", () => { + const state = apply([ + resolved("cascade", undefined), + { + type: "session.fusion_resolved", + data: { fusionId: FUSION_ID, pattern: "cascade", phasePlan: "not-an-array" }, + }, + ]); + + expect(state.status).toBe("running"); + expect(state.plan).toEqual([]); + expect(state.planSource).toBe("unavailable"); + }); + + it("recovers a phase that never emitted its ephemeral started event", () => { + const state = apply([resolved("single"), phaseCompleted("p1", "primary")]); + + expect(state.phases).toHaveLength(1); + expect(state.phases[0]).toMatchObject({ + phaseId: "p1", + kind: "primary", + status: "succeeded", + }); + }); + + it("is idempotent for duplicated events", () => { + const events = [ + routeStarted(), + resolved("cascade", CASCADE_PLAN), + phaseStarted("p1", "primary", "solver"), + activity("p1", "tool_started", { toolCallId: "call-1" }), + activity("p1", "tool_completed", { toolCallId: "call-1" }), + phaseCompleted("p1", "primary"), + completed({ pattern: "cascade" }), + ]; + + const once = apply(events); + const twice = apply(events.flatMap((event) => [event, event])); + + expect(twice).toEqual(once); + expect(reduceFusionProgress(once, completed({ pattern: "cascade" }))).toBe(once); + expect(reduceFusionProgress(once, resolved("cascade", CASCADE_PLAN))).toBe(once); + + // Routing signals carry no turn identity, so a repeat while still routing is a no-op. + const routing = apply([routeStarted()]); + expect(reduceFusionProgress(routing, routeStarted())).toBe(routing); + // A routing signal after a resolved turn never discards the resolved projection. + const running = apply([routeStarted(), resolved("single")]); + expect(reduceFusionProgress(running, routeStarted())).toBe(running); + // Repeating a phase signal leaves the projection untouched. + const phase = apply([resolved("single"), phaseStarted("p1", "primary", "solver")]); + expect(reduceFusionProgress(phase, phaseStarted("p1", "primary", "solver"))).toBe(phase); + }); + + it("does not regress terminal phase state on out-of-order events", () => { + const state = apply([ + resolved("single"), + phaseCompleted("p1", "primary"), + phaseStarted("p1", "primary", "solver"), + activity("p1", "model_output", { totalResponseSizeBytes: 64 }), + ]); + + expect(state.phases[0].status).toBe("succeeded"); + expect(state.phases[0].totalResponseSizeBytes).toBe(64); + }); + + it("keeps the largest observed response size across out-of-order activity", () => { + const state = apply([ + resolved("single"), + activity("p1", "model_output", { totalResponseSizeBytes: 900 }), + activity("p1", "model_output", { totalResponseSizeBytes: 300 }), + ]); + + expect(state.phases[0].totalResponseSizeBytes).toBe(900); + expect(state.phases[0].activity).toEqual({ + kind: "model_output", + totalResponseSizeBytes: 300, + }); + }); + + it("records phase activity and tool calls, keeping completion terminal", () => { + const state = apply([ + resolved("single"), + phaseStarted("p1", "primary", "solver"), + activity("p1", "tool_started", { toolCallId: "call-1" }), + activity("p1", "tool_completed", { toolCallId: "call-1" }), + activity("p1", "tool_started", { toolCallId: "call-1" }), + activity("p1", "tool_started", { toolCallId: "call-2" }), + ]); + + expect(state.phases[0].toolCalls).toEqual([ + { toolCallId: "call-1", status: "completed" }, + { toolCallId: "call-2", status: "started" }, + ]); + expect(state.phases[0].activity).toEqual({ kind: "tool_started", toolCallId: "call-2" }); + }); + + it("attributes tool executions to phases on runtimes without activity events", () => { + const fusion = { + fusionId: FUSION_ID, + pattern: "single", + policy: "balanced", + syntheticModel: "secret-synthetic-model", + phaseId: "p1", + phaseKind: "primary", + role: "solver", + conversationScope: "root", + sourceModel: "secret-source-model", + }; + const state = apply([ + resolved("single"), + { + type: "tool.execution_start", + data: { toolCallId: "call-9", toolName: "bash", arguments: "secret args", fusion }, + }, + { + type: "tool.execution_complete", + data: { toolCallId: "call-9", result: "secret tool output", fusion }, + }, + ]); + + expect(state.phases[0]).toMatchObject({ phaseId: "p1", kind: "primary", role: "solver" }); + expect(state.phases[0].toolCalls).toEqual([{ toolCallId: "call-9", status: "completed" }]); + }); + + it("tracks attributed permission requests until they are resolved", () => { + const requested: FusionProgressEventInput = { + type: "permission.requested", + data: { + requestId: "req-1", + permissionRequest: { kind: "shell", command: "rm -rf secret" }, + fusion: { + fusionId: FUSION_ID, + pattern: "cascade", + policy: "balanced", + syntheticModel: "secret-synthetic-model", + phaseId: "p1", + phaseKind: "primary", + role: "solver", + conversationScope: "root", + }, + }, + }; + + const pending = apply([resolved("cascade"), requested, requested]); + expect(pending.pendingPermissions).toEqual([ + { + requestId: "req-1", + fusionId: FUSION_ID, + phaseId: "p1", + phaseKind: "primary", + role: "solver", + scope: "root", + }, + ]); + + const resolvedPermission = reduceFusionProgress(pending, { + type: "permission.completed", + data: { requestId: "req-1", result: { kind: "approved" } }, + }); + expect(resolvedPermission.pendingPermissions).toEqual([]); + expect( + reduceFusionProgress(resolvedPermission, { + type: "permission.completed", + data: { requestId: "req-1", result: { kind: "approved" } }, + }) + ).toBe(resolvedPermission); + }); + + it("ignores permission requests that carry no fusion attribution", () => { + const state = apply([ + resolved("single"), + { + type: "permission.requested", + data: { requestId: "req-2", permissionRequest: { kind: "shell" } }, + }, + ]); + + expect(state.pendingPermissions).toEqual([]); + }); + + it("records the published commit and the authoritative completion commit", () => { + const state = apply([ + resolved("cascade", CASCADE_PLAN), + phaseStarted("p1", "primary", "solver"), + { + type: "assistant.message", + data: { + content: "answer", + fusion: { + fusionId: FUSION_ID, + pattern: "cascade", + policy: "balanced", + syntheticModel: "secret-synthetic-model", + commitId: "commit-7", + sourcePhaseId: "p1", + }, + }, + }, + completed({ commitId: "commit-7", degradedReason: "judge_unavailable" }), + ]); + + expect(state.publishedCommitId).toBe("commit-7"); + expect(state.completion).toEqual({ + outcome: "succeeded", + degraded: true, + commitId: "commit-7", + finalSourcePhaseId: "phase-primary", + }); + expect(state.status).toBe("completed"); + }); + + it("reports a deterministic fallback when routing fails", () => { + const state = apply([ + routeStarted("compaction"), + { + type: "session.fusion_route_failed", + data: { + attemptId: "attempt-1", + policy: "balanced", + reason: "router_unavailable", + errorMessage: "secret router error", + fallbackModel: "secret-fallback-model", + syntheticModel: "secret-synthetic-model", + }, + }, + ]); + + expect(state.status).toBe("fallback"); + expect(state.policy).toBe("balanced"); + expect(state.turnKind).toBe("compaction"); + expect(state.phases).toEqual([]); + expect(JSON.stringify(state)).not.toContain("secret"); + }); + + it("starts a new turn on a new resolved route and ignores late events from the previous turn", () => { + const firstTurn = apply([ + resolved("single"), + phaseStarted("p1", "primary", "solver"), + completed(), + ]); + const secondTurn = apply([ + resolved("single"), + phaseStarted("p1", "primary", "solver"), + completed(), + resolved("cascade", CASCADE_PLAN, "fusion-2"), + phaseStarted("q1", "primary", "solver", "root", "fusion-2"), + // Late ephemeral from the previous turn must not disturb the current projection. + activity("p1", "model_output", { totalResponseSizeBytes: 10 }), + phaseCompleted("p1", "primary", "succeeded"), + ]); + + expect(firstTurn.status).toBe("completed"); + expect(secondTurn.fusionId).toBe("fusion-2"); + expect(secondTurn.status).toBe("running"); + expect(secondTurn.completion).toBeUndefined(); + expect(secondTurn.publishedCommitId).toBeUndefined(); + expect(secondTurn.phases.map((phase) => phase.phaseId)).toEqual(["q1"]); + }); + + it("begins a fresh projection when routing starts after a completed turn", () => { + const state = apply([ + resolved("single", [{ kind: "primary", role: "solver", scope: "root" }]), + phaseStarted("p1", "primary", "solver"), + completed(), + routeStarted(), + ]); + + expect(state.status).toBe("routing"); + expect(state.fusionId).toBeUndefined(); + expect(state.phases).toEqual([]); + expect(state.plan).toEqual([]); + expect(state.completion).toBeUndefined(); + }); + + it("adopts a turn identity discovered from a phase event before the route resolves", () => { + const beforeResolved = apply([phaseStarted("p1", "primary", "solver")]); + + // Unresolved early evidence must never leave the projection looking inactive. + expect(beforeResolved.status).toBe("running"); + expect(beforeResolved.fusionId).toBe(FUSION_ID); + expect(beforeResolved.phases.map((phase) => phase.phaseId)).toEqual(["p1"]); + expect(beforeResolved.plan).toEqual([]); + expect(beforeResolved.planSource).toBe("unavailable"); + + const state = reduceFusionProgress( + beforeResolved, + resolved("single", [{ kind: "primary", role: "solver", scope: "root" }]) + ); + + expect(state.fusionId).toBe(FUSION_ID); + expect(state.status).toBe("running"); + expect(state.phases.map((phase) => phase.phaseId)).toEqual(["p1"]); + expect(state.phases[0].planIndex).toBe(0); + }); + + it("promotes unresolved activity, tool, and permission evidence to a running turn", () => { + const fromActivity = apply([activity("p1", "model_output", { totalResponseSizeBytes: 8 })]); + expect(fromActivity.status).toBe("running"); + expect(fromActivity.fusionId).toBe(FUSION_ID); + + const fromTool = apply([ + { + type: "tool.execution_start", + data: { + toolCallId: "call-1", + fusion: { fusionId: FUSION_ID, pattern: "single", phaseId: "p1" }, + }, + }, + ]); + expect(fromTool.status).toBe("running"); + expect(fromTool.phases.map((phase) => phase.phaseId)).toEqual(["p1"]); + + const fromPermission = apply([ + { + type: "permission.requested", + data: { + requestId: "req-1", + fusion: { fusionId: FUSION_ID, pattern: "single", phaseId: "p1" }, + }, + }, + ]); + expect(fromPermission.status).toBe("running"); + expect(fromPermission.pendingPermissions).toHaveLength(1); + }); + + it("never lets a stale non-resolved event replace a terminal turn", () => { + const turnB = apply([ + // Turn A runs and completes. + resolved("single", [{ kind: "primary", role: "solver", scope: "root" }]), + phaseStarted("p1", "primary", "solver"), + completed(), + // Turn B resolves authoritatively, runs, and completes. + resolved("cascade", CASCADE_PLAN, "fusion-2"), + phaseStarted("q1", "primary", "solver", "root", "fusion-2"), + { + type: "session.fusion_completed", + data: { + fusionId: "fusion-2", + turnId: "turn-2", + outcome: "succeeded", + commitId: "commit-2", + degradedReason: null, + finalSourcePhaseId: "q1", + pattern: "cascade", + }, + }, + ]); + + expect(turnB.status).toBe("completed"); + expect(turnB.completion?.commitId).toBe("commit-2"); + + // Late ephemerals and durable stragglers from turn A must all be inert. + const staleEvents: FusionProgressEventInput[] = [ + activity("p1", "model_output", { totalResponseSizeBytes: 4096 }), + activity("p1", "tool_started", { toolCallId: "stale-call" }), + phaseStarted("p1", "primary", "solver"), + phaseCompleted("p1", "primary"), + { + type: "tool.execution_complete", + data: { + toolCallId: "stale-call", + fusion: { fusionId: FUSION_ID, pattern: "single", phaseId: "p1" }, + }, + }, + { + type: "permission.requested", + data: { + requestId: "stale-req", + fusion: { fusionId: FUSION_ID, pattern: "single", phaseId: "p1" }, + }, + }, + { + type: "assistant.message", + data: { fusion: { fusionId: FUSION_ID, pattern: "single", commitId: "commit-1" } }, + }, + completed(), + ]; + + for (const stale of staleEvents) { + expect(reduceFusionProgress(turnB, stale)).toBe(turnB); + } + + const afterStale = staleEvents.reduce(reduceFusionProgress, turnB); + expect(afterStale).toBe(turnB); + expect(afterStale.fusionId).toBe("fusion-2"); + expect(afterStale.completion).toEqual({ + outcome: "succeeded", + degraded: false, + commitId: "commit-2", + finalSourcePhaseId: "q1", + }); + expect(afterStale.publishedCommitId).toBeUndefined(); + expect(afterStale.plan).toBe(turnB.plan); + expect(afterStale.phases).toBe(turnB.phases); + expect(afterStale.phases.map((phase) => phase.phaseId)).toEqual(["q1"]); + }); + + it("never lets phase evidence resurrect a routing-fallback projection", () => { + const fallback = apply([ + routeStarted(), + { + type: "session.fusion_route_failed", + data: { attemptId: "attempt-1", policy: "balanced", reason: "router_unavailable" }, + }, + ]); + + expect(fallback.status).toBe("fallback"); + expect(reduceFusionProgress(fallback, phaseStarted("p1", "primary", "solver"))).toBe( + fallback + ); + expect(reduceFusionProgress(fallback, activity("p1", "model_output"))).toBe(fallback); + expect(reduceFusionProgress(fallback, completed())).toBe(fallback); + + // Only an authoritative resolved route starts the next turn. + const nextTurn = reduceFusionProgress(fallback, resolved("single")); + expect(nextTurn.status).toBe("running"); + expect(nextTurn.fusionId).toBe(FUSION_ID); + }); + + it("never projects content, verdicts, error detail, or concrete model identities", () => { + const state = apply([ + routeStarted(), + resolved("cascade", CASCADE_PLAN), + phaseStarted("p1", "primary", "solver"), + activity("p1", "model_output", { totalResponseSizeBytes: 128 }), + phaseCompleted("p1", "primary"), + { + type: "assistant.fusion_phase_failed", + data: { + fusionId: FUSION_ID, + phaseId: "p2", + phaseKind: "judge", + role: "judge", + conversationScope: "review", + status: "failed", + reason: "provider_error", + errorMessage: "secret provider error", + model: "secret-judge-model", + }, + }, + { + type: "assistant.message", + data: { + content: "secret assistant content", + reasoningText: "secret reasoning", + fusion: { + fusionId: FUSION_ID, + pattern: "cascade", + policy: "balanced", + commitId: "commit-1", + sourceModel: "secret-source-model", + syntheticModel: "secret-synthetic-model", + }, + }, + }, + completed({ degradedReason: "judge_unavailable" }), + ]); + + const serialized = JSON.stringify(state); + expect(serialized).not.toContain("secret"); + expect(serialized).not.toContain("provider_error"); + expect(serialized).not.toContain("judge_unavailable"); + expect(serialized).not.toContain("Model"); + expect(serialized).not.toContain("content"); + expect(serialized).not.toContain("verdict"); + expect(state.completion?.degraded).toBe(true); + }); +});