1import { parseFrameEnvelope } from './frame-envelope.mjs';2import {3 isCurrentFrameGeneration,4 normalizeFrameStatsPayload,5} from './frame-stats.mjs';6import {7 FRAME_CHANNEL_CAPACITY,8 activateFrameSession as activatePendingFrameSession,9 isActiveFrameSession,10 newFrameSession,11 stageFrameAcknowledgment,12} from './frame-session.mjs';13import {14 AdaptiveFeedbackPublisher,15 normalizeAdaptiveDecisionEnvelope,16} from './adaptive-feedback.mjs';1718function newTransportTelemetry() {19 return {20 transportDroppedFrames: 0,21 transportObjectDroppedFrames: null,22 transportLateObjectDroppedFrames: null,23 frontendDroppedFrames: 0,24 frontendQueueDroppedFrames: null,25 frontendResyncDroppedFrames: null,26 frontendQueueStats: null,27 frontendResyncStats: null,28 transportIntervalStats: null,29 frontendIpcSendDurationStats: null,30 rustTimingWindow: null,31 networkDiagnostics: null,32 lastTransportFps: 0,33 lastFrontendSendFps: 0,34 streamPathMode: 'unknown',35 streamRttMs: null,36 lastMediaSequence: null,37 };38}3940function assertDependency(value, name) {41 if (typeof value !== 'function') throw new TypeError(`${name} must be a function`);42}4344export function createStreamRuntime({45 invokeCommand,46 videoPipeline,47 isConnected = () => false,48 disconnect = () => {},49 adaptiveFeedbackPublisher = null,50 logger = console,51} = {}) {52 assertDependency(invokeCommand, 'invokeCommand');53 assertDependency(isConnected, 'isConnected');54 assertDependency(disconnect, 'disconnect');55 if (!videoPipeline56 || typeof videoPipeline.processFramePayload !== 'function'57 || typeof videoPipeline.reset !== 'function'58 || typeof videoPipeline.teardown !== 'function'59 || typeof videoPipeline.snapshot !== 'function') {60 throw new TypeError('videoPipeline must implement the stream pipeline interface');61 }62 const feedbackPublisher = adaptiveFeedbackPublisher63 ?? new AdaptiveFeedbackPublisher({ invokeCommand });64 if (!feedbackPublisher65 || typeof feedbackPublisher.start !== 'function'66 || typeof feedbackPublisher.stop !== 'function'67 || typeof feedbackPublisher.publish !== 'function') {68 throw new TypeError('adaptiveFeedbackPublisher must implement start, stop, and publish');69 }7071 let activeFrameSession = null;72 let activeFrameGeneration = null;73 let adaptiveFeedbackAvailable = false;74 let adaptiveFeedbackError = null;75 let adaptiveDecision = null;76 let transport = newTransportTelemetry();7778 function sendFrameAcknowledgment(generation) {79 void invokeCommand('iroh_client_ack_frame', { generation }).catch((error) => {80 logger.warn('frame acknowledgment failed:', error);81 });82 }8384 function acknowledgeFrame(session, generation = null) {85 const readyGeneration = stageFrameAcknowledgment(session, generation);86 if (readyGeneration !== null) sendFrameAcknowledgment(readyGeneration);87 }8889 function failFrameSession(session, error) {90 if (!isActiveFrameSession(session, activeFrameSession)) return false;91 session.failed = true;92 session.failureDetail = `Frame delivery failed: ${error}`;93 logger.error(session.failureDetail);94 if (isConnected()) void disconnect();95 return true;96 }9798 function prepareConnection() {99 if (activeFrameSession) activeFrameSession.closing = true;100 activeFrameSession = null;101 activeFrameGeneration = null;102 feedbackPublisher.stop();103 adaptiveFeedbackAvailable = false;104 adaptiveFeedbackError = null;105 adaptiveDecision = null;106 videoPipeline.reset();107 transport = newTransportTelemetry();108 }109110 function openFrameSession() {111 const session = newFrameSession();112 activeFrameSession = session;113 return session;114 }115116 function activateFrameSession(session, {117 generation,118 adaptiveFeedbackAvailable: feedbackAvailable = false,119 adaptiveFeedbackError: feedbackError = null,120 }) {121 if (!isActiveFrameSession(session, activeFrameSession)) {122 throw new Error('frame session was superseded during connect');123 }124 const activation = activatePendingFrameSession(session, generation);125 activeFrameGeneration = generation;126 for (const pendingGeneration of activation.acknowledgments) {127 sendFrameAcknowledgment(pendingGeneration);128 }129 const pendingError = session.pendingFrameErrors.splice(0).find(130 (payload) => isCurrentFrameGeneration(payload?.generation, generation),131 );132 if (pendingError) throw new Error(pendingError.error || 'Media connection failed');133 adaptiveFeedbackAvailable = feedbackAvailable === true;134 adaptiveFeedbackError = typeof feedbackError === 'string' ? feedbackError : null;135 feedbackPublisher.start(generation, adaptiveFeedbackAvailable);136 }137138 function closeFrameSession(session) {139 const closingSession = session ?? activeFrameSession;140 if (closingSession) closingSession.closing = true;141 if (activeFrameSession === closingSession) activeFrameSession = null;142 activeFrameGeneration = null;143 feedbackPublisher.stop();144 adaptiveFeedbackAvailable = false;145 adaptiveDecision = null;146 }147148 function prepareDisconnect() {149 if (activeFrameSession) activeFrameSession.closing = true;150 activeFrameSession = null;151 activeFrameGeneration = null;152 feedbackPublisher.stop();153 adaptiveFeedbackAvailable = false;154 adaptiveFeedbackError = null;155 adaptiveDecision = null;156 }157158 function teardown() {159 videoPipeline.teardown();160 }161162 function requestKeyframe(reason) {163 if (!Number.isSafeInteger(activeFrameGeneration) || activeFrameGeneration < 1) return false;164 void invokeCommand('iroh_client_request_keyframe', {165 generation: activeFrameGeneration,166 reason,167 }).catch((error) => {168 logger.warn(`keyframe request failed (${reason}):`, error);169 });170 return true;171 }172173 function handleBinaryFrame(message, session) {174 if (!isActiveFrameSession(session, activeFrameSession)) return false;175 let acknowledged = false;176 try {177 const frame = parseFrameEnvelope(message);178 // Parsing establishes the exact envelope and payload bounds. Release the179 // Rust-side delivery permit before codec parsing/decode enqueue so a slow180 // invoke round trip cannot turn one scheduling hiccup into a full-GOP181 // discard. WebCodecs remains independently bounded by decodeQueueSize.182 acknowledgeFrame(session);183 acknowledged = true;184 videoPipeline.processFramePayload({185 width: frame.width,186 height: frame.height,187 codec: frame.codec,188 keyframe: frame.keyframe,189 codecConfig: frame.codecConfig,190 sequence: frame.sequence,191 pts_micros: frame.ptsMicros,192 discontinuity: frame.discontinuity,193 }, frame.data);194 return true;195 } catch (error) {196 failFrameSession(session, error);197 return false;198 } finally {199 if (!acknowledged) {200 // A malformed envelope disconnects above, but must never deadlock the201 // bounded sender while teardown reaches Rust.202 try {203 acknowledgeFrame(session);204 } catch (error) {205 failFrameSession(session, error);206 }207 }208 }209 }210211 function handleFrameStats(payload) {212 if (!isCurrentFrameGeneration(payload?.generation, activeFrameGeneration)) return false;213 const diagnostics = normalizeFrameStatsPayload(payload);214 transport.transportDroppedFrames = diagnostics.transportDroppedFrames215 ?? transport.transportDroppedFrames;216 transport.transportObjectDroppedFrames = diagnostics.objectDroppedFrames;217 transport.transportLateObjectDroppedFrames = diagnostics.lateObjectDroppedFrames;218 transport.frontendDroppedFrames = diagnostics.frontendDroppedFrames219 ?? transport.frontendDroppedFrames;220 transport.frontendQueueDroppedFrames = diagnostics.queueDroppedFrames;221 transport.frontendResyncDroppedFrames = diagnostics.resyncDroppedFrames;222 transport.frontendQueueStats = diagnostics.queue;223 transport.frontendResyncStats = diagnostics.resync;224 transport.transportIntervalStats = diagnostics.transportIntervals;225 transport.frontendIpcSendDurationStats = diagnostics.ipcSendDurations;226 transport.rustTimingWindow = diagnostics.timingWindow;227 transport.networkDiagnostics = diagnostics.networkDiagnostics;228 transport.streamPathMode = typeof payload.path_mode === 'string'229 ? payload.path_mode230 : transport.streamPathMode;231 transport.streamRttMs = Number.isFinite(payload.path_rtt_ms)232 ? payload.path_rtt_ms233 : transport.streamRttMs;234 transport.lastMediaSequence = Number.isSafeInteger(payload.sequence) && payload.sequence >= 0235 ? payload.sequence : transport.lastMediaSequence;236 transport.lastTransportFps = Number.isFinite(payload.transport_receive_fps)237 ? payload.transport_receive_fps238 : Number.isFinite(payload.transport_fps)239 ? payload.transport_fps240 : payload.fps;241 transport.lastFrontendSendFps = Number.isFinite(payload.frontend_send_fps)242 ? payload.frontend_send_fps243 : Number.isFinite(payload.frontend_fps)244 ? payload.frontend_fps245 : payload.fps;246 return true;247 }248249 function handleAdaptiveDecision(payload) {250 try {251 const decision = normalizeAdaptiveDecisionEnvelope(payload, activeFrameGeneration);252 if (decision === null) return false;253 adaptiveDecision = decision;254 return true;255 } catch (error) {256 logger.warn('invalid adaptive bitrate diagnostic:', error);257 return false;258 }259 }260261 function handleAdaptiveFeedbackState(payload) {262 if (!isCurrentFrameGeneration(payload?.generation, activeFrameGeneration)) return false;263 if (payload?.available !== false) return false;264 adaptiveFeedbackAvailable = false;265 adaptiveFeedbackError = typeof payload.error === 'string'266 ? payload.error : 'feedback stream closed';267 feedbackPublisher.stop();268 return true;269 }270271 function handleFrameError(payload) {272 const session = activeFrameSession;273 if (!isActiveFrameSession(session, activeFrameSession)) return false;274 if (session.expectedGeneration === null) {275 if (session.pendingFrameErrors.length < FRAME_CHANNEL_CAPACITY) {276 session.pendingFrameErrors.push(payload);277 return true;278 }279 failFrameSession(session, 'pre-connect frame error capacity exceeded');280 return false;281 }282 if (!isCurrentFrameGeneration(payload?.generation, session.expectedGeneration)) return false;283 return failFrameSession(session, payload?.error || 'Media connection failed');284 }285286 function snapshot(at) {287 return {288 video: at === undefined ? videoPipeline.snapshot() : videoPipeline.snapshot(at),289 generation: activeFrameGeneration,290 ...transport,291 adaptiveFeedbackAvailable,292 adaptiveFeedbackError,293 adaptiveDecision,294 };295 }296297 function publishAdaptiveFeedback(video) {298 return feedbackPublisher.publish({299 lastSequence: transport.lastMediaSequence,300 frontendQueueDepth: transport.frontendQueueStats?.depth ?? 0,301 frontendQueueCapacity: transport.frontendQueueStats?.capacity ?? 4,302 decoderQueueDepth: video.decoderQueueDepth,303 decoderQueueCapacity: video.decoderQueueCapacity,304 presenterQueueDepth: video.presenterQueueDepth,305 presenterQueueCapacity: video.presenterQueueCapacity,306 transportDroppedTotal: transport.transportDroppedFrames,307 frontendDroppedTotal: transport.frontendDroppedFrames,308 decoderDroppedTotal: video.droppedFrames,309 presenterDroppedTotal: video.presentationDroppedFrames,310 // Portal has no clock-synchronized capture-to-delivery measurement yet.311 // IPC send duration is a different boundary and must not be mislabeled.312 transportDeliveryP95Ms: null,313 decodeLatencyP95Ms: video.decodePercentiles.p95,314 presentationLatencyP95Ms: video.presentPercentiles.p95,315 resyncActive: video.recovering || transport.frontendResyncStats?.active === true,316 });317 }318319 return Object.freeze({320 prepareConnection,321 openFrameSession,322 activateFrameSession,323 closeFrameSession,324 prepareDisconnect,325 teardown,326 requestKeyframe,327 handleBinaryFrame,328 handleFrameStats,329 handleAdaptiveDecision,330 handleAdaptiveFeedbackState,331 handleFrameError,332 snapshot,333 publishAdaptiveFeedback,334 get generation() { return activeFrameGeneration; },335 get activeSession() { return activeFrameSession; },336 });337}