From 1a89ef41f7d9d7aee1186d5846af84d7e86d9fba Mon Sep 17 00:00:00 2001 From: Carlos Castro Date: Sun, 30 Aug 2026 18:55:03 -0700 Subject: [PATCH 1/2] Add presentation-neutral HydraFusion progress reducer Projects HydraFusion session events into a deterministic, presentation-neutral snapshot so any consumer (terminal UI, web UI, logs) can render turn progress without reimplementing the event bookkeeping. The reducer is pure and event-driven: it tolerates duplicate and out-of-order events, recovers phases whose ephemeral signals were missed, and degrades to the existing phase events on runtimes that do not yet publish the new phase plan, phase activity, or permission attribution. It never issues an RPC. It retains only event discriminants, phase kinds/roles/scopes, phase plan metadata, response byte counts, tool call IDs, permission attribution, commit IDs, and the stable terminal outcome. Phase content, verdicts, prompts, reasoning, provider error detail, and concrete model identities are never read. The newest experimental producer fields (session.fusion_resolved.data.phasePlan, assistant.fusion_phase_activity, and fusion attribution on permission events) do not have generated declarations in this repo yet, so the reducer accepts a structurally typed event input that the generated SessionEvent union satisfies. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b51c537e-a447-4a75-af56-0528910afdce --- nodejs/src/fusionProgress.ts | 724 ++++++++++++++++++++++++++++ nodejs/src/index.ts | 20 + nodejs/test/fusion-progress.test.ts | 628 ++++++++++++++++++++++++ 3 files changed, 1372 insertions(+) create mode 100644 nodejs/src/fusionProgress.ts create mode 100644 nodejs/test/fusion-progress.test.ts diff --git a/nodejs/src/fusionProgress.ts b/nodejs/src/fusionProgress.ts new file mode 100644 index 0000000000..456add98af --- /dev/null +++ b/nodejs/src/fusionProgress.ts @@ -0,0 +1,724 @@ +/*--------------------------------------------------------------------------------------------- + * 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. + * - **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" + /** A route resolved and the turn's phases are executing. */ + | "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, so the reducer can be applied to an entire event stream: + * + * ```ts + * const state = events.reduce(reduceFusionProgress, initialFusionProgressState()); + * ``` + * + * @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. + * + * Returns the state to apply the event to, or `undefined` when the event belongs to a different + * turn that must not disturb the current projection (a late ephemeral from a previous turn). + */ +function alignTurn( + state: FusionProgressState, + fusionId: string | undefined, + startsTurn: boolean +): FusionProgressState | undefined { + if (fusionId === undefined) { + return state; + } + if (state.fusionId === undefined) { + return { ...state, fusionId }; + } + if (state.fusionId === fusionId) { + return state; + } + if (startsTurn || state.status === "completed" || state.status === "fallback") { + return { ...EMPTY_STATE, fusionId }; + } + return undefined; +} + +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..490aac6f79 --- /dev/null +++ b/nodejs/test/fusion-progress.test.ts @@ -0,0 +1,628 @@ +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 state = apply([ + phaseStarted("p1", "primary", "solver"), + 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("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); + }); +}); From 870a1d6f9619fc2c962a7f8d0b79a374bd096014 Mon Sep 17 00:00:00 2001 From: Carlos Castro Date: Sun, 30 Aug 2026 19:18:37 -0700 Subject: [PATCH 2/2] Harden HydraFusion progress turn-identity rules Review follow-up on two turn-identity defects in the progress reducer. alignTurn adopted a turn identity from unresolved phase, activity, tool, or permission evidence without moving the projection out of "inactive", so a consumer folding those events before session.fusion_resolved saw a turn with phases but no lifecycle. Such evidence can only come from a turn that is already executing, so adopting it now promotes the projection to "running". alignTurn also let any mismatching fusionId replace a completed or fallback projection, so a late ephemeral from a previous turn could wipe the current turn's completion, commit, plan, and phases. Only the durable session.fusion_resolved path may now establish or replace a turn identity, and a terminal projection is never resurrected by non-resolved evidence. Late prior-turn events and the first ephemerals of a next turn are ignored until it resolves. Adds regressions asserting the intermediate pre-resolution state, promotion from each unresolved evidence kind, reference-stable completion/commit/plan/phases across a full sweep of stale prior-turn events after a later turn completed, and that a fallback projection is only replaced by an authoritative resolved route. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b51c537e-a447-4a75-af56-0528910afdce --- nodejs/src/fusionProgress.ts | 55 +++++++++-- nodejs/test/fusion-progress.test.ts | 141 +++++++++++++++++++++++++++- 2 files changed, 183 insertions(+), 13 deletions(-) diff --git a/nodejs/src/fusionProgress.ts b/nodejs/src/fusionProgress.ts index 456add98af..a179d8e0d9 100644 --- a/nodejs/src/fusionProgress.ts +++ b/nodejs/src/fusionProgress.ts @@ -19,6 +19,13 @@ * 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, @@ -61,7 +68,10 @@ export type FusionProgressStatus = | "routing" /** Routing failed; the turn runs on a deterministic concrete fallback instead. */ | "fallback" - /** A route resolved and the turn's phases are executing. */ + /** + * 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"; @@ -220,12 +230,18 @@ export function initialFusionProgressState(): FusionProgressState { * 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, so the reducer can be applied to an entire event stream: + * 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( @@ -505,8 +521,20 @@ function beginTurn(state: FusionProgressState, next: FusionProgressState): Fusio /** * Reconciles the incoming turn identity with the projected one. * - * Returns the state to apply the event to, or `undefined` when the event belongs to a different - * turn that must not disturb the current projection (a late ephemeral from a previous turn). + * 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, @@ -517,15 +545,24 @@ function alignTurn( return state; } if (state.fusionId === undefined) { - return { ...state, fusionId }; + 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; } - if (startsTurn || state.status === "completed" || state.status === "fallback") { - return { ...EMPTY_STATE, fusionId }; - } - return undefined; + 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( diff --git a/nodejs/test/fusion-progress.test.ts b/nodejs/test/fusion-progress.test.ts index 490aac6f79..feb1620e4b 100644 --- a/nodejs/test/fusion-progress.test.ts +++ b/nodejs/test/fusion-progress.test.ts @@ -566,10 +566,19 @@ describe("reduceFusionProgress", () => { }); it("adopts a turn identity discovered from a phase event before the route resolves", () => { - const state = apply([ - phaseStarted("p1", "primary", "solver"), - resolved("single", [{ kind: "primary", role: "solver", scope: "root" }]), - ]); + 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"); @@ -577,6 +586,130 @@ describe("reduceFusionProgress", () => { 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(),