import { normalizePlayRunFailure, PLATFORM_DEPLOY_INTERRUPTED_MESSAGE, } from './run-failure'; import { nextPlayRunLedgerTerminalSource, normalizePlayRunLedgerTerminalSource, } from './run-terminal-source'; import { MAX_LEDGER_LOG_LINES_PER_EVENT } from './ledger-safe-payload'; import { normalizePlayDocflowNodeIoState, type PlayDocflowNodeIoState, } from './docflow-node-io'; import { isPlayActivityObservation, type PlayActivityObservation, } from './activity-observation'; import { mergeDatasetBornFrom, normalizeDatasetBornFrom, type PlayDatasetBornFrom, } from './cell-provenance'; export type PlayRunLedgerStatus = | 'queued' | 'running' | 'waiting' | 'completed' | 'failed' | 'cancelled' | 'terminated' | 'timed_out' | 'unknown'; export type PlayRunLedgerStepStatus = | 'running' | 'completed' | 'failed' | 'skipped'; export type PlayRunLedgerEventSource = | 'worker' | 'convex' | 'coordinator' | 'system'; /** * Live progress for a single step/node. * * Verified map-executor producer semantics: * * - `completed` counts SUCCESSFULLY written rows ONLY, never failures/skips. * In-flight it is `min(total, totalRowsWritten + completedExecutedRows)` * where `completedExecutedRows` increments on success only (entry.ts:5635); * failures increment the separate `failedExecutedRows` (entry.ts:5684). * At map finalize it is set to `finalizedRowsWritten` (successes). * - `failed` counts settled-but-failed rows (row-failure isolation persists * these; they re-execute on the next run). * - `total` is the settled-row denominator. In-flight it is the row-count * estimate (`rowCountHint`); at map finalize it is reconciled to * `completed + failed` and CAN SHRINK from the initial estimate when rows are * filtered/skipped. So `total` is monotonic-up during flight but is corrected * down to the settled count at terminal. * * Terminal invariant: once the step is settled, `completed + failed === total`. * Because `completed` is successes-only (NOT all-settled), readers must NOT * assume `completed === total` at terminal — a truthful terminal render is * `{total} rows` when `failed === 0`, else a `completed`/`failed`/`total` * breakdown. See `settleRunningStepsOnTerminal` for why the reducer does not * force `completed = total`. */ export type PlayRunLedgerStepProgress = { completed?: number; total?: number; failed?: number; startedRows?: number; activeRows?: number; waitingRows?: number; completedRows?: number; message?: string; artifactTableNamespace?: string | null; startedAt?: number | null; completedAt?: number | null; updatedAt?: number | null; nodeIo?: PlayDocflowNodeIoState; }; export type PlayRunLedgerStepSnapshot = { stepId: string; label?: string; kind?: string; status: PlayRunLedgerStepStatus; error?: string | null; artifactTableNamespace?: string | null; startedAt?: number | null; completedAt?: number | null; updatedAt?: number | null; progress?: PlayRunLedgerStepProgress | null; }; export type PlayRunLedgerDatasetSnapshot = { datasetId: string; path: string; tableNamespace: string; phase: 'registered' | 'available' | 'failed'; persistedRows: number; succeededRows: number; failedRows: number; complete: boolean; /** * Dataset-grain row birth (ADR 0019). Absent on datasets registered before * the contract and on any registration that did not state one — never * back-filled from row counts, which would assert a lineage nobody observed. */ bornFrom?: PlayDatasetBornFrom | null; updatedAt: number; }; export type PlayRunLedgerSnapshot = { runId: string; playName?: string | null; status: PlayRunLedgerStatus; error?: string | null; createdAt?: number | null; startedAt?: number | null; updatedAt?: number | null; finishedAt?: number | null; durationMs?: number | null; orderedStepIds: string[]; stepsById: Record; orderedDatasetIds: string[]; datasetsById: Record; orderedActivityIds: string[]; activitiesById: Record; /** * Bounded tail of the run's log lines (last {@link LOG_TAIL_LIMIT}). * Full retention lives in the Run Log Stream (Convex `playRunLogChunks`); * the snapshot only carries enough tail for live previews. * `totalLogCount` is the monotonic count of every line ever ingested * (post channel-dedupe), so readers can derive the absolute sequence * number of the first retained tail line: * `totalLogCount - logTail.length + 1`. */ logTail: string[]; totalLogCount: number; /** * True once the Run Log Stream stopped storing line bodies because the run * crossed the retention cap (25k lines / 5MB). `totalLogCount` keeps * counting past the cap; a loud truncation marker is the last stored line. */ logsTruncated: boolean; activeStepId?: string | null; activeActivityId?: string | null; activeArtifactTableNamespace?: string | null; resultTableNamespace?: string | null; resultSummary?: unknown; result?: unknown; /** * Source of the event that made this snapshot terminal. Authoritative * producers ('worker', scheduler-outbox 'system') carry the * play's final result; a 'coordinator'-sourced terminal is a transport * heal whose `result` came from the coordinator's live cache and may be a * partial mid-execution snapshot. Read-side canonical checks treat * 'coordinator'-sourced completeds as provisional (see * `hasCanonicalTerminalRunSnapshot`). Absent on snapshots persisted before * this field existed — those are treated as authoritative. */ terminalSource?: PlayRunLedgerEventSource | null; }; type PlayRunLedgerBaseEvent = { runId: string; seq?: number; occurredAt: number; source: PlayRunLedgerEventSource; }; export type PlayRunLedgerEvent = | (PlayRunLedgerBaseEvent & { type: 'run.created'; playName?: string | null; status?: PlayRunLedgerStatus; runtimeBackend?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'run.started'; playName?: string | null; runtimeBackend?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'run.waiting'; }) | (PlayRunLedgerBaseEvent & { type: 'run.resumed'; }) | (PlayRunLedgerBaseEvent & { type: 'run.completed'; result?: unknown; resultSummary?: unknown; }) | (PlayRunLedgerBaseEvent & { type: 'run.failed'; error?: string | null; result?: unknown; }) | (PlayRunLedgerBaseEvent & { type: 'run.cancelled'; error?: string | null; result?: unknown; }) | (PlayRunLedgerBaseEvent & { type: 'step.started'; stepId: string; label?: string; kind?: string; artifactTableNamespace?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'step.progress'; stepId: string; label?: string; kind?: string; status?: PlayRunLedgerStepStatus; progress: PlayRunLedgerStepProgress; }) | (PlayRunLedgerBaseEvent & { type: 'step.completed'; stepId: string; label?: string; kind?: string; artifactTableNamespace?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'step.failed'; stepId: string; label?: string; kind?: string; error?: string | null; artifactTableNamespace?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'step.skipped'; stepId: string; label?: string; kind?: string; artifactTableNamespace?: string | null; }) | (PlayRunLedgerBaseEvent & { type: 'log.appended'; lines: string[]; /** * Absolute per-run sequence (1-based) of `lines[0]`. Assigned by Run * Log Stream ingestion in Convex only; producers never set it. */ firstSeq?: number; /** * Positional cursor for exactly-once delivery on the worker channel: * the count of lines this producer emitted before `lines[0]`. Ingestion * skips the already-ingested prefix positionally, which preserves * repeated identical lines while absorbing redundant re-sends. */ channelOffset?: number; /** Scheduler attempt that owns this positional log channel. */ producerAttempt?: number; /** * Terminal transport replay normally matches ctx.log lines after * removing timestamps. Detached-attempt recovery uses exact matching so * the same message from a later attempt remains a distinct occurrence. */ transportDedupe?: 'semantic' | 'exact'; /** Set by ingestion when this append crossed the retention cap. */ logsTruncated?: boolean; }) | (PlayRunLedgerBaseEvent & { type: 'dataset.lifecycle'; datasetId: string; path: string; tableNamespace: string; phase: 'registered' | 'available' | 'failed'; persistedRows?: number; succeededRows?: number; failedRows?: number; complete?: boolean; /** Dataset-grain row birth, stated once per dataset (ADR 0019). */ bornFrom?: PlayDatasetBornFrom; }) | (PlayRunLedgerBaseEvent & { type: 'activity.observed'; observation: PlayActivityObservation; }); export type PlayRunLedgerStatusPatch = { runId: string; playName?: string | null; status: string; error?: string | null; runtimeBackend?: string | null; lastCheckpointAt?: number | null; liveLogs?: readonly string[] | null; /** * Positional cursor for `liveLogs[0]` on this producer's log channel (the * count of lines emitted before it). Computed by the producer-side cursor * (see `slicePositionalLogLines`), forwarded onto the `log.appended` event. */ liveLogsChannelOffset?: number | null; liveLogsProducerAttempt?: number | null; liveNodeProgress?: unknown; result?: unknown; }; export const LOG_TAIL_LIMIT = 100; const TERMINAL_STATUSES = new Set([ 'completed', 'failed', 'cancelled', 'terminated', 'timed_out', ]); function isRecord(value: unknown): value is Record { return Boolean(value && typeof value === 'object' && !Array.isArray(value)); } function finiteNumber(value: unknown): number | null { return typeof value === 'number' && Number.isFinite(value) ? value : null; } function optionalFiniteNumber(value: unknown): number | undefined { const normalized = finiteNumber(value); return normalized === null ? undefined : normalized; } function optionalString(value: unknown): string | undefined { return typeof value === 'string' && value.trim() ? value.trim() : undefined; } function optionalNullableString(value: unknown): string | null | undefined { if (value === null) return null; return optionalString(value); } export function normalizePlayRunLedgerStatus( value: unknown, ): PlayRunLedgerStatus { const normalized = String(value ?? '') .trim() .toLowerCase(); switch (normalized) { case 'queued': case 'pending': return 'queued'; case 'running': case 'started': return 'running'; case 'waiting': return 'waiting'; case 'completed': case 'complete': case 'succeeded': return 'completed'; case 'failed': case 'error': return 'failed'; case 'cancelled': case 'canceled': return 'cancelled'; case 'terminated': return 'terminated'; case 'timed_out': case 'timeout': return 'timed_out'; default: return 'unknown'; } } export function isTerminalPlayRunLedgerStatus( status: PlayRunLedgerStatus, ): boolean { return TERMINAL_STATUSES.has(status); } function isTerminalActivityObservation( observation: PlayActivityObservation, ): boolean { return ( observation.state.kind === 'completed' || observation.state.kind === 'failed' ); } /** * Aggregate, row-level outcome of a run derived purely from the settled step * progress. This is an ADDITIVE truthfulness signal, NOT a status: a run can be * `status: 'completed'` (all steps ran to the end) while `failedRows > 0` * because row-failure isolation persists failed rows for retry instead of * failing the whole run. Readers use `hasRowFailures` to render a truthful * "completed with failures" surface without inventing a new public run status. * * `completedRows` here is the reconciled successes-only count summed from the * producer's final step progress (`completed`), which is the durable/persisted * success count — NOT a live in-flight estimate. It is what should be compared * against persisted dataset rows; the two diverging was the 21-vs-15 symptom of * children being lost before their rows settled. */ export type PlayRunRowOutcomeSummary = { completedRows: number; failedRows: number; totalRows: number; hasRowFailures: boolean; }; export function summarizeRunRowOutcomes( snapshot: Pick, ): PlayRunRowOutcomeSummary { let completedRows = 0; let failedRows = 0; let totalRows = 0; for (const step of Object.values(snapshot.stepsById)) { const progress = step.progress; if (!progress) continue; // `completed` is successes-only; `failed` is settled-but-failed rows. Sum // both across every map/step so a multi-map play reports the whole run's // row outcome, not just the last step. completedRows += Math.max(0, finiteNumber(progress.completed) ?? 0); failedRows += Math.max(0, finiteNumber(progress.failed) ?? 0); // Prefer the settled denominator; fall back to completed+failed when a // producer never stamped `total` (per the terminal invariant those agree). const stepTotal = finiteNumber(progress.total); totalRows += stepTotal !== null && stepTotal >= 0 ? stepTotal : Math.max(0, finiteNumber(progress.completed) ?? 0) + Math.max(0, finiteNumber(progress.failed) ?? 0); } return { completedRows, failedRows, totalRows, hasRowFailures: failedRows > 0, }; } export function createEmptyPlayRunLedgerSnapshot(input: { runId: string; playName?: string | null; status?: unknown; error?: string | null; createdAt?: number | null; startedAt?: number | null; updatedAt?: number | null; finishedAt?: number | null; }): PlayRunLedgerSnapshot { const status = normalizePlayRunLedgerStatus(input.status ?? 'unknown'); const startedAt = finiteNumber(input.startedAt) ?? null; const finishedAt = finiteNumber(input.finishedAt) ?? null; return { runId: input.runId, playName: input.playName ?? null, status, error: optionalNullableString(input.error) ?? null, createdAt: finiteNumber(input.createdAt) ?? null, startedAt, updatedAt: finiteNumber(input.updatedAt) ?? finishedAt ?? startedAt ?? finiteNumber(input.createdAt) ?? null, finishedAt, durationMs: startedAt !== null && finishedAt !== null ? Math.max(0, finishedAt - startedAt) : null, orderedStepIds: [], stepsById: {}, orderedDatasetIds: [], datasetsById: {}, orderedActivityIds: [], activitiesById: {}, logTail: [], totalLogCount: 0, logsTruncated: false, activeStepId: null, activeActivityId: null, activeArtifactTableNamespace: null, resultTableNamespace: null, }; } export function normalizePlayRunLedgerSnapshot( value: unknown, fallback: Parameters[0], ): PlayRunLedgerSnapshot { if (!isRecord(value)) { return createEmptyPlayRunLedgerSnapshot(fallback); } const orderedStepIds = Array.isArray(value.orderedStepIds) ? value.orderedStepIds.filter( (entry): entry is string => typeof entry === 'string' && Boolean(entry.trim()), ) : []; const rawSteps = isRecord(value.stepsById) ? value.stepsById : {}; const stepsById: Record = {}; for (const [stepId, rawStep] of Object.entries(rawSteps)) { if (!stepId.trim() || !isRecord(rawStep)) continue; const rawStatus = normalizeStepStatus(rawStep.status); if (!rawStatus) continue; const completedAt = finiteNumber(rawStep.completedAt); const status = rawStatus === 'running' && completedAt !== null ? 'completed' : rawStatus; const rawProgress = isRecord(rawStep.progress) ? rawStep.progress : null; stepsById[stepId] = { stepId, label: optionalString(rawStep.label), kind: optionalString(rawStep.kind), status, error: optionalNullableString(rawStep.error), artifactTableNamespace: optionalNullableString( rawStep.artifactTableNamespace, ), startedAt: finiteNumber(rawStep.startedAt), completedAt, updatedAt: finiteNumber(rawStep.updatedAt), progress: rawProgress ? normalizeStepProgress(rawProgress) : null, }; } const rawDatasets = isRecord(value.datasetsById) ? value.datasetsById : {}; const datasetsById: Record = {}; for (const [datasetId, rawDataset] of Object.entries(rawDatasets)) { if (!datasetId.trim() || !isRecord(rawDataset)) continue; const phase = rawDataset.phase === 'available' || rawDataset.phase === 'failed' ? rawDataset.phase : 'registered'; const datasetBornFrom = normalizeDatasetBornFrom(rawDataset.bornFrom); datasetsById[datasetId] = { datasetId, path: optionalString(rawDataset.path) ?? `datasets.${datasetId}`, tableNamespace: optionalString(rawDataset.tableNamespace) ?? datasetId, phase, persistedRows: Math.max(0, finiteNumber(rawDataset.persistedRows) ?? 0), succeededRows: Math.max(0, finiteNumber(rawDataset.succeededRows) ?? 0), failedRows: Math.max(0, finiteNumber(rawDataset.failedRows) ?? 0), complete: rawDataset.complete === true, // ADR 0019 birth survives the snapshot round trip. A partial record is // dropped rather than half-admitted, so a dataset either states where its // rows came from or says nothing. ...(datasetBornFrom ? { bornFrom: datasetBornFrom } : {}), updatedAt: finiteNumber(rawDataset.updatedAt) ?? 0, }; } const rawActivities = isRecord(value.activitiesById) ? value.activitiesById : {}; const activitiesById: Record = {}; for (const [activityId, rawActivity] of Object.entries(rawActivities)) { if ( activityId.trim() && isPlayActivityObservation(rawActivity) && rawActivity.activityId === activityId ) { activitiesById[activityId] = rawActivity; } } const createdAt = finiteNumber(value.createdAt) ?? fallback.createdAt ?? null; const startedAt = finiteNumber(value.startedAt) ?? fallback.startedAt ?? null; const finishedAt = finiteNumber(value.finishedAt) ?? fallback.finishedAt ?? null; const updatedAt = finiteNumber(value.updatedAt) ?? fallback.updatedAt ?? finishedAt ?? startedAt ?? createdAt ?? null; const error = Object.prototype.hasOwnProperty.call(value, 'error') ? (optionalNullableString(value.error) ?? null) : (fallback.error ?? null); // `logTail` is canonical; `logs` is the pre-Run-Log-Stream persisted name. // Reading both is the deploy-window seam that lets the first post-cutover // append backfill an old snapshot's retained tail into chunk storage. const rawTail = Array.isArray(value.logTail) ? value.logTail : value.logs; const logTail = Array.isArray(rawTail) ? rawTail.filter((line): line is string => typeof line === 'string') : []; return { runId: optionalString(value.runId) ?? fallback.runId, playName: optionalNullableString(value.playName) ?? fallback.playName ?? null, status: normalizePlayRunLedgerStatus(value.status ?? fallback.status), error, createdAt, startedAt, updatedAt, finishedAt, durationMs: startedAt !== null && finishedAt !== null ? Math.max(0, finishedAt - startedAt) : finiteNumber(value.durationMs), orderedStepIds: orderedStepIds.filter((stepId) => stepsById[stepId]), stepsById, orderedDatasetIds: (Array.isArray(value.orderedDatasetIds) ? value.orderedDatasetIds : Object.keys(datasetsById) ).filter( (datasetId): datasetId is string => typeof datasetId === 'string' && Boolean(datasetsById[datasetId]), ), datasetsById, orderedActivityIds: (Array.isArray(value.orderedActivityIds) ? value.orderedActivityIds : Object.keys(activitiesById) ).filter( (activityId): activityId is string => typeof activityId === 'string' && Boolean(activitiesById[activityId]), ), activitiesById, logTail: logTail.slice(-LOG_TAIL_LIMIT), // Snapshots persisted before totalLogCount existed only know the retained // tail, so the best lower bound for the cumulative count is the tail size. totalLogCount: Math.max( finiteNumber(value.totalLogCount) ?? logTail.length, logTail.length, ), logsTruncated: value.logsTruncated === true, activeStepId: optionalNullableString(value.activeStepId), activeActivityId: optionalNullableString(value.activeActivityId), activeArtifactTableNamespace: optionalNullableString( value.activeArtifactTableNamespace, ), resultTableNamespace: optionalNullableString(value.resultTableNamespace), resultSummary: value.resultSummary, result: value.result, terminalSource: normalizePlayRunLedgerTerminalSource(value.terminalSource), }; } /** * Terminal-source transition for a (possibly repeated) terminal event. * A transport heal ('coordinator') never overwrites an already-recorded * authoritative source: when the producer's own terminal has landed, a late * heal replay must not demote the snapshot back to provisional. Any * authoritative source (worker/system/convex) always wins, so the * worker's post-heal `run.completed` flips a healed snapshot to final. */ function nextTerminalSource( base: PlayRunLedgerSnapshot, eventSource: PlayRunLedgerEventSource, ): PlayRunLedgerEventSource { return nextPlayRunLedgerTerminalSource({ currentSource: base.terminalSource, eventSource, }); } function shouldIgnoreNonTerminalAfterStickyTerminal( base: PlayRunLedgerSnapshot, ): boolean { return isTerminalPlayRunLedgerStatus(base.status); } function maxProgressNumber( current: number | undefined, next: number | undefined, ): number | undefined { if (current === undefined) return next; if (next === undefined) return current; return Math.max(current, next); } function mergeMonotonicStepProgress(input: { current?: PlayRunLedgerStepProgress | null; next: PlayRunLedgerStepProgress; }): PlayRunLedgerStepProgress { const current = input.current ?? {}; const next = input.next; return { ...current, ...next, ...(maxProgressNumber(current.completed, next.completed) !== undefined ? { completed: maxProgressNumber(current.completed, next.completed) } : {}), ...(maxProgressNumber(current.total, next.total) !== undefined ? { total: maxProgressNumber(current.total, next.total) } : {}), ...(maxProgressNumber(current.failed, next.failed) !== undefined ? { failed: maxProgressNumber(current.failed, next.failed) } : {}), ...(maxProgressNumber(current.startedRows, next.startedRows) !== undefined ? { startedRows: maxProgressNumber(current.startedRows, next.startedRows), } : {}), ...(maxProgressNumber(current.completedRows, next.completedRows) !== undefined ? { completedRows: maxProgressNumber( current.completedRows, next.completedRows, ), } : {}), }; } /** * Zero the in-flight row accounting on a settled step. `activeRows` (calls in * flight) and `waitingRows` (rows queued behind rate/concurrency limits) are * live-only counters: once a step is terminal nothing is in flight or waiting, * so leaving the last observed values renders a contradiction like * "status: completed · 1 in flight · waiting 29". This preserves the settled * counts (completed/failed/total/completedRows/startedRows) and only clears the * two live accumulators. Idempotent: dropping already-absent fields is a no-op. */ function clearInFlightProgressOnTerminal( progress: PlayRunLedgerStepProgress | null | undefined, ): PlayRunLedgerStepProgress | null | undefined { if (!progress) return progress; if (progress.activeRows === undefined && progress.waitingRows === undefined) { return progress; } const cleared = { ...progress }; delete cleared.activeRows; delete cleared.waitingRows; return cleared; } function isTerminalStepStatus(status: PlayRunLedgerStepStatus): boolean { return status === 'completed' || status === 'failed' || status === 'skipped'; } function normalizeStepStatus(value: unknown): PlayRunLedgerStepStatus | null { const normalized = String(value ?? '') .trim() .toLowerCase(); if ( normalized === 'running' || normalized === 'completed' || normalized === 'failed' || normalized === 'skipped' ) { return normalized; } return null; } function normalizeStepProgress( value: Record, ): PlayRunLedgerStepProgress { return { ...(optionalFiniteNumber(value.completed) !== undefined ? { completed: optionalFiniteNumber(value.completed) } : {}), ...(optionalFiniteNumber(value.total) !== undefined ? { total: optionalFiniteNumber(value.total) } : {}), ...(optionalFiniteNumber(value.failed) !== undefined ? { failed: optionalFiniteNumber(value.failed) } : {}), ...(optionalFiniteNumber(value.startedRows) !== undefined ? { startedRows: optionalFiniteNumber(value.startedRows) } : {}), ...(optionalFiniteNumber(value.activeRows) !== undefined ? { activeRows: optionalFiniteNumber(value.activeRows) } : {}), ...(optionalFiniteNumber(value.waitingRows) !== undefined ? { waitingRows: optionalFiniteNumber(value.waitingRows) } : {}), ...(optionalFiniteNumber(value.completedRows) !== undefined ? { completedRows: optionalFiniteNumber(value.completedRows) } : {}), ...(optionalString(value.message) ? { message: optionalString(value.message) } : {}), ...(optionalNullableString(value.artifactTableNamespace) !== undefined ? { artifactTableNamespace: optionalNullableString( value.artifactTableNamespace, ), } : {}), ...(finiteNumber(value.startedAt) !== null ? { startedAt: finiteNumber(value.startedAt) } : {}), ...(finiteNumber(value.completedAt) !== null ? { completedAt: finiteNumber(value.completedAt) } : {}), ...(finiteNumber(value.updatedAt) !== null ? { updatedAt: finiteNumber(value.updatedAt) } : {}), ...(normalizePlayDocflowNodeIoState(value.nodeIo) ? { nodeIo: normalizePlayDocflowNodeIoState(value.nodeIo) } : {}), }; } function normalizeLiveProgressMap( value: unknown, ): Record { if (!isRecord(value)) { return {}; } const normalized: Record = {}; for (const [stepId, rawProgress] of Object.entries(value)) { if (!stepId.trim() || !isRecord(rawProgress)) continue; normalized[stepId] = { ...normalizeStepProgress(rawProgress), ...(optionalFiniteNumber(rawProgress.startedAt) !== undefined ? { startedAt: optionalFiniteNumber(rawProgress.startedAt) } : {}), ...(optionalFiniteNumber(rawProgress.completedAt) !== undefined ? { completedAt: optionalFiniteNumber(rawProgress.completedAt) } : {}), }; } return normalized; } function appendOrderedStepId( snapshot: PlayRunLedgerSnapshot, stepId: string, ): string[] { return snapshot.orderedStepIds.includes(stepId) ? snapshot.orderedStepIds : [...snapshot.orderedStepIds, stepId]; } function resolveResultTableNamespace( snapshot: PlayRunLedgerSnapshot, ): string | null { const terminalStepNamespace = [...snapshot.orderedStepIds] .reverse() .map((stepId) => snapshot.stepsById[stepId]?.artifactTableNamespace) .find((namespace): namespace is string => Boolean(namespace?.trim())); return snapshot.resultTableNamespace ?? terminalStepNamespace ?? null; } function withTiming(snapshot: PlayRunLedgerSnapshot): PlayRunLedgerSnapshot { const startedAt = snapshot.startedAt ?? null; const finishedAt = snapshot.finishedAt ?? null; return { ...snapshot, durationMs: startedAt !== null && finishedAt !== null ? Math.max(0, finishedAt - startedAt) : null, resultTableNamespace: isTerminalPlayRunLedgerStatus(snapshot.status) ? resolveResultTableNamespace(snapshot) : (snapshot.resultTableNamespace ?? null), }; } const TERMINAL_STATUS_BY_EVENT_TYPE = { 'run.completed': 'completed', 'run.failed': 'failed', 'run.cancelled': 'cancelled', } as const satisfies Partial>; function isRawPlatformDeployResetText( error: string | null | undefined, ): boolean { return Boolean( error && normalizePlayRunFailure(error).code === 'PLATFORM_DEPLOY_INTERRUPTED', ); } function isPlatformDeployFailureText( error: string | null | undefined, ): boolean { return Boolean( error && (isRawPlatformDeployResetText(error) || error.includes(PLATFORM_DEPLOY_INTERRUPTED_MESSAGE)), ); } /** * Append already-deduplicated log lines to the snapshot tail. * * Cross-channel reconciliation (positional worker cursors, bounded * text-window dedupe for recovery channels) happens at Run Log Stream * ingestion in Convex, BEFORE events reach this reducer. The reducer trusts * the event: every line counts, repeated identical lines included. * * Events stamped with `firstSeq` (assigned by ingestion) pin the cumulative * count exactly; unstamped events (reducer-internal anomaly markers, * producer-side snapshot caches) advance it by line count. */ function appendLogLines( snapshot: PlayRunLedgerSnapshot, lines: readonly string[], options?: { firstSeq?: number; logsTruncated?: boolean }, ): PlayRunLedgerSnapshot { const nextLines = lines .map((line) => line.trim()) .filter((line) => line.length > 0); const totalLogCount = typeof options?.firstSeq === 'number' && options.firstSeq > 0 ? Math.max( snapshot.totalLogCount, options.firstSeq + nextLines.length - 1, ) : snapshot.totalLogCount + nextLines.length; return { ...snapshot, logTail: [...snapshot.logTail, ...nextLines].slice(-LOG_TAIL_LIMIT), totalLogCount, logsTruncated: snapshot.logsTruncated || options?.logsTruncated === true, }; } /** * Terminal-status precedence. The first accepted run terminal is immutable: * later events may enrich the same terminal payload, but may never change its * outcome. Attempt retries are not authority to rewrite a terminal run, and * producer timestamps are telemetry rather than an ordering mechanism. */ function conflictingTerminalSnapshot( base: PlayRunLedgerSnapshot, eventType: keyof typeof TERMINAL_STATUS_BY_EVENT_TYPE, occurredAt: number, eventError?: string | null, options?: { allowStaleSameStatusPayloadMerge?: boolean; }, ): PlayRunLedgerSnapshot | null { if (!isTerminalPlayRunLedgerStatus(base.status)) { return null; } const terminalAt = base.finishedAt ?? base.updatedAt ?? 0; if ( base.status === 'completed' && eventType === 'run.failed' && isPlatformDeployFailureText(eventError) ) { return withTiming( appendLogLines(base, [ `[ledger] platform deploy terminal event ${eventType} ignored; status already ${base.status}`, ]), ); } if ( base.status === 'failed' && eventType === 'run.failed' && isPlatformDeployFailureText(eventError) ) { return withTiming(base); } if (TERMINAL_STATUS_BY_EVENT_TYPE[eventType] === base.status) { if ( options?.allowStaleSameStatusPayloadMerge !== true || occurredAt < terminalAt ) { return withTiming(base); } return null; } return withTiming( appendLogLines(base, [ `[ledger] stale conflicting terminal event ${eventType} ignored; status already ${base.status}`, ]), ); } function retryablePlatformDeployFailureSnapshot( base: PlayRunLedgerSnapshot, eventError?: string | null, ): PlayRunLedgerSnapshot | null { if ( isTerminalPlayRunLedgerStatus(base.status) || !eventError || !isRawPlatformDeployResetText(eventError) ) { return null; } return withTiming( appendLogLines(base, [ `[ledger] retryable platform deploy run.failed ignored; status remains ${base.status}`, ]), ); } // Force-settles steps still marked `running` when the RUN reaches a terminal // status (the fallback for steps that never emitted their own step.completed, // e.g. a crash between the last progress bump and the terminal event). This // only flips status and stamps `completedAt`; it deliberately does NOT // reconcile `progress.completed` to `progress.total`. // // Why not reconcile: per PlayRunLedgerStepProgress semantics, `completed` // counts SUCCESSES ONLY, so `completed === total` is NOT a valid invariant at // terminal — a run can legitimately settle with failed/skipped rows (row- // failure isolation). Forcing `completed = total` here would fabricate // successes that never happened and falsify the persisted counts. The truthful // terminal counts are carried by the producer's final reconciled bump, which // the finalize path drains BEFORE emitting step.completed // (entry.ts finalize ordering invariant). This settle path is the last-resort // heal for the missing-final-bump case, where the honest thing is "we know it // stopped, we do not know it fully succeeded" — leave the last observed // counts, just mark it settled so the UI stops rendering a live ticker. function settleRunningStepsOnTerminal( snapshot: PlayRunLedgerSnapshot, status: Extract, occurredAt: number, ): PlayRunLedgerSnapshot { let changed = false; const stepsById = Object.fromEntries( Object.entries(snapshot.stepsById).map(([stepId, step]) => { if (step.status !== 'running') { return [stepId, step]; } changed = true; return [ stepId, { ...step, status, completedAt: step.completedAt ?? occurredAt, updatedAt: occurredAt, progress: step.progress ? clearInFlightProgressOnTerminal({ ...step.progress, completedAt: step.progress.completedAt ?? occurredAt, updatedAt: occurredAt, }) : step.progress, }, ]; }), ) as Record; return changed ? { ...snapshot, stepsById } : snapshot; } function terminalFinishedAt( base: PlayRunLedgerSnapshot, eventType: keyof typeof TERMINAL_STATUS_BY_EVENT_TYPE, occurredAt: number, ): number { const nextStatus = TERMINAL_STATUS_BY_EVENT_TYPE[eventType]; const terminalAt = base.finishedAt ?? base.updatedAt ?? 0; if ( isTerminalPlayRunLedgerStatus(base.status) && nextStatus !== base.status && occurredAt > terminalAt ) { return occurredAt; } return base.finishedAt ?? occurredAt; } function isStepLifecycleEvent(event: PlayRunLedgerEvent): boolean { return ( event.type === 'step.started' || event.type === 'step.progress' || event.type === 'step.completed' || event.type === 'step.failed' || event.type === 'step.skipped' ); } /** * A terminal status patch can arrive twice: first without a step start, then * with the original start timestamp after the runner recovers its progress * buffer. A completed node whose start equals its completion is synthetic, so * this is the only late lifecycle refinement that is safe to admit. It cannot * create a node, change a terminal status, alter completion, or revive active * work. Preserve the Run Snapshot's own timestamps while making the node's * duration truthful. */ function refineTerminalSyntheticStepStart( snapshot: PlayRunLedgerSnapshot, event: Extract, ): PlayRunLedgerSnapshot | null { const current = snapshot.stepsById[event.stepId]; const startedAt = event.progress.startedAt; if ( current?.status !== 'completed' || event.status !== 'completed' || typeof startedAt !== 'number' || current.startedAt == null || current.completedAt == null || current.startedAt !== current.completedAt || startedAt >= current.startedAt ) { return null; } return { ...snapshot, stepsById: { ...snapshot.stepsById, [event.stepId]: { ...current, startedAt, progress: current.progress ? { ...current.progress, startedAt } : { startedAt }, }, }, }; } export function reducePlayRunLedgerEvent( snapshot: PlayRunLedgerSnapshot, event: PlayRunLedgerEvent, ): PlayRunLedgerSnapshot { if (event.runId !== snapshot.runId) { return snapshot; } // A terminal Run Snapshot is immutable with respect to step lifecycle. // Detached runners can race their best-effort progress transport against the // terminal outbox, and gateway processes can restart or split that traffic. // The Play Run Ledger is the durable projection authority, so late // `step.*` facts must be a true no-op here rather than relying on a gateway // process to remember terminal state. if (isTerminalPlayRunLedgerStatus(snapshot.status)) { if (event.type === 'step.progress') { return refineTerminalSyntheticStepStart(snapshot, event) ?? snapshot; } if (isStepLifecycleEvent(event)) return snapshot; } // The scheduler's outbox can replay an older wait transition after its run // reached a terminal state. That transition is no longer meaningful and // must be byte-for-byte inert: even advancing `updatedAt` would re-publish // a stale terminal snapshot to every observer. if ( (event.type === 'run.waiting' || event.type === 'run.resumed') && shouldIgnoreNonTerminalAfterStickyTerminal(snapshot) ) { return snapshot; } const occurredAt = Math.max(0, event.occurredAt); const base: PlayRunLedgerSnapshot = { ...snapshot, updatedAt: Math.max(snapshot.updatedAt ?? 0, occurredAt), }; const ignoreNonTerminalAfterStickyTerminal = (event.type === 'run.created' || event.type === 'run.started') && shouldIgnoreNonTerminalAfterStickyTerminal(base); switch (event.type) { case 'run.created': { if (ignoreNonTerminalAfterStickyTerminal) { return withTiming(base); } // Scheduler wait/resume events own the active lifecycle once a run is // parked. Worker registration can arrive later through an independent // ledger append lane; replaying that older `run.created` fact must not // make an externally-waiting run look active again. const createdStatus = base.status === 'waiting' ? 'waiting' : (event.status ?? base.status); return withTiming({ ...base, playName: event.playName ?? base.playName ?? null, status: createdStatus, createdAt: base.createdAt ?? occurredAt, startedAt: base.startedAt ?? (createdStatus === 'running' ? occurredAt : base.startedAt), }); } case 'run.started': if (ignoreNonTerminalAfterStickyTerminal) { return withTiming(base); } return withTiming({ ...base, playName: event.playName ?? base.playName ?? null, status: isTerminalPlayRunLedgerStatus(base.status) ? base.status : base.status === 'waiting' ? 'waiting' : 'running', startedAt: base.startedAt ?? occurredAt, }); case 'run.waiting': return withTiming({ ...base, status: 'waiting', }); case 'run.resumed': return withTiming({ ...base, status: 'running', }); case 'run.completed': return ( conflictingTerminalSnapshot(base, event.type, occurredAt, null, { allowStaleSameStatusPayloadMerge: (base.result === undefined && event.result !== undefined) || (base.resultSummary === undefined && event.resultSummary !== undefined), }) ?? withTiming({ ...settleRunningStepsOnTerminal(base, 'completed', occurredAt), status: 'completed', error: null, startedAt: base.startedAt ?? occurredAt, finishedAt: terminalFinishedAt(base, event.type, occurredAt), activeStepId: null, activeActivityId: null, activeArtifactTableNamespace: null, result: event.result ?? base.result, resultSummary: event.resultSummary ?? base.resultSummary, terminalSource: nextTerminalSource(base, event.source), }) ); case 'run.failed': return ( retryablePlatformDeployFailureSnapshot(base, event.error) ?? conflictingTerminalSnapshot(base, event.type, occurredAt, event.error, { allowStaleSameStatusPayloadMerge: (base.error == null && event.error != null) || (base.result === undefined && event.result !== undefined), }) ?? withTiming({ ...settleRunningStepsOnTerminal(base, 'failed', occurredAt), status: 'failed', error: base.status === 'failed' ? (base.error ?? event.error ?? null) : (event.error ?? base.error ?? null), startedAt: base.startedAt ?? occurredAt, finishedAt: terminalFinishedAt(base, event.type, occurredAt), activeStepId: null, activeActivityId: null, activeArtifactTableNamespace: null, result: event.result ?? base.result, terminalSource: nextTerminalSource(base, event.source), }) ); case 'run.cancelled': return ( conflictingTerminalSnapshot(base, event.type, occurredAt, event.error, { allowStaleSameStatusPayloadMerge: (base.error == null && event.error != null) || (base.result === undefined && event.result !== undefined), }) ?? withTiming({ ...settleRunningStepsOnTerminal(base, 'failed', occurredAt), status: 'cancelled', error: base.status === 'cancelled' ? (base.error ?? event.error ?? null) : (event.error ?? base.error ?? null), startedAt: base.startedAt ?? occurredAt, finishedAt: terminalFinishedAt(base, event.type, occurredAt), activeStepId: null, activeActivityId: null, activeArtifactTableNamespace: null, result: event.result ?? base.result, terminalSource: nextTerminalSource(base, event.source), }) ); case 'log.appended': return withTiming( appendLogLines(base, event.lines, { ...(typeof event.firstSeq === 'number' ? { firstSeq: event.firstSeq } : {}), ...(event.logsTruncated === true ? { logsTruncated: true } : {}), }), ); case 'step.started': { const current = base.stepsById[event.stepId]; const runTerminal = isTerminalPlayRunLedgerStatus(base.status); const shouldRefineSyntheticStart = current?.startedAt != null && current.completedAt != null && current.startedAt === current.completedAt && occurredAt < current.startedAt; const nextStep: PlayRunLedgerStepSnapshot = { ...(current ?? { stepId: event.stepId, status: 'running' as const }), stepId: event.stepId, label: event.label ?? current?.label, kind: event.kind ?? current?.kind, status: current?.status === 'failed' || current?.status === 'completed' ? current.status : 'running', artifactTableNamespace: event.artifactTableNamespace ?? current?.artifactTableNamespace ?? null, startedAt: shouldRefineSyntheticStart ? occurredAt : (current?.startedAt ?? occurredAt), updatedAt: Math.max(current?.updatedAt ?? 0, occurredAt), }; return withTiming({ ...base, status: isTerminalPlayRunLedgerStatus(base.status) ? base.status : base.status === 'waiting' ? 'waiting' : 'running', startedAt: base.startedAt ?? occurredAt, orderedStepIds: appendOrderedStepId(base, event.stepId), stepsById: { ...base.stepsById, [event.stepId]: nextStep }, activeStepId: runTerminal ? base.activeStepId : nextStep.status === 'running' ? event.stepId : base.activeStepId, activeArtifactTableNamespace: runTerminal ? (base.activeArtifactTableNamespace ?? null) : nextStep.status === 'running' ? (nextStep.artifactTableNamespace ?? null) : (base.activeArtifactTableNamespace ?? null), }); } case 'step.progress': { const current = base.stepsById[event.stepId]; const runTerminal = isTerminalPlayRunLedgerStatus(base.status); const mergedProgress = { ...mergeMonotonicStepProgress({ current: current?.progress, next: event.progress, }), updatedAt: event.progress.updatedAt ?? occurredAt, }; // An explicit `completedAt` on THIS event is the producer's authoritative // completion signal and always settles the step. Absent that, the // `completed >= total` heuristic is a stale-estimate trap while rows are // still in flight (total can shrink below the in-flight count under child // fan-out), so it must NOT infer completion while active rows remain. const eventSignalsCompletion = typeof event.progress.completedAt === 'number'; const hasActiveRows = typeof mergedProgress.activeRows === 'number' && mergedProgress.activeRows > 0; const inferredStatus = eventSignalsCompletion || (!hasActiveRows && typeof mergedProgress.total === 'number' && typeof mergedProgress.completed === 'number' && mergedProgress.total > 0 && mergedProgress.completed >= mergedProgress.total) ? 'completed' : 'running'; const status = current?.status === 'failed' || current?.status === 'skipped' || current?.status === 'completed' ? current.status : event.status === 'running' && inferredStatus === 'completed' ? 'completed' : (event.status ?? current?.status ?? inferredStatus); const completedAt = current?.completedAt ?? event.progress.completedAt ?? (status === 'completed' ? (event.progress.updatedAt ?? occurredAt) : null); const shouldRefineSyntheticStart = current?.startedAt != null && current.completedAt != null && current.startedAt === current.completedAt && event.progress.startedAt != null && event.progress.startedAt < current.startedAt; const startedAt = (shouldRefineSyntheticStart ? event.progress.startedAt : null) ?? current?.startedAt ?? event.progress.startedAt ?? (status === 'completed' || status === 'failed' ? (completedAt ?? event.progress.updatedAt ?? occurredAt) : null); // Once the step is settled, drop the live in-flight/waiting accumulators // so a terminal node never reports active or waiting rows. const progress = isTerminalStepStatus(status) ? clearInFlightProgressOnTerminal(mergedProgress) : mergedProgress; const nextStep: PlayRunLedgerStepSnapshot = { ...(current ?? { stepId: event.stepId, status }), stepId: event.stepId, label: event.label ?? current?.label, kind: event.kind ?? current?.kind, status, error: current?.error ?? null, artifactTableNamespace: progress?.artifactTableNamespace ?? current?.artifactTableNamespace ?? null, startedAt, completedAt, updatedAt: occurredAt, progress, }; return withTiming({ ...base, orderedStepIds: appendOrderedStepId(base, event.stepId), stepsById: { ...base.stepsById, [event.stepId]: nextStep }, activeStepId: runTerminal ? base.activeStepId : status === 'running' ? event.stepId : base.activeStepId === event.stepId ? null : base.activeStepId, activeArtifactTableNamespace: runTerminal ? (base.activeArtifactTableNamespace ?? null) : status === 'running' ? (nextStep.artifactTableNamespace ?? null) : base.activeStepId === event.stepId ? null : (base.activeArtifactTableNamespace ?? null), }); } case 'step.completed': case 'step.failed': case 'step.skipped': { const current = base.stepsById[event.stepId]; const runTerminal = isTerminalPlayRunLedgerStatus(base.status); const preserveFailedStatus = current?.status === 'failed' && event.type !== 'step.failed'; const status = preserveFailedStatus ? 'failed' : event.type === 'step.completed' ? 'completed' : event.type === 'step.failed' ? 'failed' : 'skipped'; const completedAt = status === 'skipped' || preserveFailedStatus ? (current?.completedAt ?? null) : occurredAt; const nextStep: PlayRunLedgerStepSnapshot = { ...(current ?? { stepId: event.stepId, status }), stepId: event.stepId, label: event.label ?? current?.label, kind: event.kind ?? current?.kind, status, error: event.type === 'step.failed' ? (event.error ?? current?.error ?? null) : preserveFailedStatus ? (current?.error ?? null) : null, artifactTableNamespace: event.artifactTableNamespace ?? current?.artifactTableNamespace ?? null, startedAt: current?.startedAt ?? (status === 'completed' || status === 'failed' ? (completedAt ?? occurredAt) : null), completedAt, updatedAt: occurredAt, // Terminal step: clear live in-flight/waiting accounting. progress: clearInFlightProgressOnTerminal(current?.progress) ?? null, }; const runningStepIds = base.orderedStepIds.filter((stepId) => { if (stepId === event.stepId) return false; return base.stepsById[stepId]?.status === 'running'; }); const activeStepId = runningStepIds.at(-1) ?? null; return withTiming({ ...base, orderedStepIds: appendOrderedStepId(base, event.stepId), stepsById: { ...base.stepsById, [event.stepId]: nextStep }, activeStepId: runTerminal ? base.activeStepId : activeStepId, activeArtifactTableNamespace: runTerminal ? (base.activeArtifactTableNamespace ?? null) : activeStepId ? (base.stepsById[activeStepId]?.artifactTableNamespace ?? null) : null, }); } case 'dataset.lifecycle': { const current = base.datasetsById[event.datasetId]; const mergedBornFrom = mergeDatasetBornFrom( current?.bornFrom, event.bornFrom, ); const next: PlayRunLedgerDatasetSnapshot = { datasetId: event.datasetId, path: event.path, tableNamespace: event.tableNamespace, phase: event.phase, persistedRows: Math.max( current?.persistedRows ?? 0, event.persistedRows ?? 0, ), succeededRows: Math.max( current?.succeededRows ?? 0, event.succeededRows ?? 0, ), failedRows: Math.max(current?.failedRows ?? 0, event.failedRows ?? 0), complete: current?.complete === true || event.complete === true, // Birth is stated at registration. A later phase event carries no birth // record and must not erase the one that was witnessed; a paged // dataset re-registers with a larger admitted-row count. ...(mergedBornFrom ? { bornFrom: mergedBornFrom } : {}), updatedAt: occurredAt, }; return withTiming({ ...base, orderedDatasetIds: current ? base.orderedDatasetIds : [...base.orderedDatasetIds, event.datasetId], datasetsById: { ...base.datasetsById, [event.datasetId]: next }, resultTableNamespace: base.resultTableNamespace ?? event.tableNamespace, }); } case 'activity.observed': { const current = base.activitiesById[event.observation.activityId]; if (current && current.observedAt >= event.observation.observedAt) { return withTiming(base); } const terminal = isTerminalActivityObservation(event.observation); return withTiming({ ...base, orderedActivityIds: current ? base.orderedActivityIds : [...base.orderedActivityIds, event.observation.activityId], activitiesById: { ...base.activitiesById, [event.observation.activityId]: event.observation, }, activeActivityId: terminal ? base.activeActivityId === event.observation.activityId ? null : base.activeActivityId : event.observation.activityId, }); } } } export function reducePlayRunLedgerEvents( snapshot: PlayRunLedgerSnapshot, events: readonly PlayRunLedgerEvent[], ): PlayRunLedgerSnapshot { return events.reduce(reducePlayRunLedgerEvent, snapshot); } function progressSignature( progress: PlayRunLedgerStepProgress | null | undefined, ): string { return JSON.stringify({ completed: progress?.completed ?? null, total: progress?.total ?? null, failed: progress?.failed ?? null, startedRows: progress?.startedRows ?? null, activeRows: progress?.activeRows ?? null, waitingRows: progress?.waitingRows ?? null, completedRows: progress?.completedRows ?? null, message: progress?.message ?? null, artifactTableNamespace: progress?.artifactTableNamespace ?? null, startedAt: progress?.startedAt ?? null, completedAt: progress?.completedAt ?? null, nodeIo: progress?.nodeIo ?? null, }); } function terminalEventTypeForStatus( status: PlayRunLedgerStatus, ): 'run.completed' | 'run.failed' | 'run.cancelled' | null { if (status === 'completed') return 'run.completed'; if ( status === 'failed' || status === 'terminated' || status === 'timed_out' ) { return 'run.failed'; } if (status === 'cancelled') return 'run.cancelled'; return null; } /** * Slice the not-yet-sent suffix out of a producer's rotating log buffer, * using positional counts instead of text comparison. * * `bufferTotalCount` is the count of lines the producer ever emitted on this * channel (monotonic); `bufferLines` is its retained tail of those lines. * `sentCount` is the producer-side cursor: lines already handed to the * ledger. Returns the new lines plus the `channelOffset` of the first one, * or null when there is nothing new. Lines that rotated out of the buffer * before they were ever sent surface as a positive gap at ingestion (which * records a loud gap marker) instead of being silently re-numbered. */ export function slicePositionalLogLines(input: { bufferLines: readonly string[]; bufferTotalCount: number; sentCount: number; }): { lines: string[]; channelOffset: number } | null { const total = Math.max(input.bufferTotalCount, input.bufferLines.length); const sent = Math.max(0, input.sentCount); if (total <= sent) { return null; } const bufferFirstOffset = total - input.bufferLines.length; const sendFromOffset = Math.max(sent, bufferFirstOffset); const lines = input.bufferLines.slice(sendFromOffset - bufferFirstOffset); if (lines.length === 0) { return null; } return { lines: [...lines], channelOffset: sendFromOffset }; } function terminalLogReplayChannelOffset(input: { liveLogTotalCount?: number; liveLogsLength: number; }): number | null { if ( typeof input.liveLogTotalCount !== 'number' || !Number.isFinite(input.liveLogTotalCount) || input.liveLogTotalCount < input.liveLogsLength ) { return null; } return input.liveLogTotalCount - input.liveLogsLength; } /** * Forward producer log lines as one `log.appended` event. * * No text dedupe here: redundant-delivery reconciliation moved to Run Log * Stream ingestion (positional `channelOffset` cursors for the worker * channel, bounded tail-window text dedupe for recovery channels). */ export function buildPlayRunLedgerEventsFromLogLines(input: { runId: string; lines: readonly string[]; source?: PlayRunLedgerEventSource; occurredAt?: number; channelOffset?: number | null; producerAttempt?: number | null; transportDedupe?: 'semantic' | 'exact'; includeLogAppend?: boolean; }): PlayRunLedgerEvent[] { const source = input.source ?? 'worker'; const occurredAt = input.occurredAt ?? Date.now(); const positional = typeof input.channelOffset === 'number' && input.channelOffset >= 0 && Number.isFinite(input.channelOffset); // Positional batches must keep one entry per emitted line — dropping a // blank would shift every later line off its channel position. const newLines = positional ? input.lines.map((line) => line.trim() || '(blank log line)') : input.lines.map((line) => line.trim()).filter((line) => line.length > 0); if (newLines.length === 0 || input.includeLogAppend === false) { return []; } const events: PlayRunLedgerEvent[] = []; for ( let index = 0; index < newLines.length; index += MAX_LEDGER_LOG_LINES_PER_EVENT ) { events.push({ type: 'log.appended', runId: input.runId, source, occurredAt, lines: newLines.slice(index, index + MAX_LEDGER_LOG_LINES_PER_EVENT), ...(positional ? { channelOffset: input.channelOffset! + index } : {}), ...(positional && typeof input.producerAttempt === 'number' && Number.isFinite(input.producerAttempt) && input.producerAttempt >= 0 ? { producerAttempt: Math.trunc(input.producerAttempt) } : {}), ...(input.transportDedupe ? { transportDedupe: input.transportDedupe } : {}), }); } return events; } /** * Re-send a terminal retained log tail through terminal transport recovery. * * Live worker flushes use positional channel offsets. Terminal replay is a * different transport: it replays the runner's retained output after terminal * state is known. When the producer knows the retained tail's offset, Convex * uses that offset only to bound occurrence-count reconciliation to the same * tail window; it does not route the replay through the worker cursor. */ export function buildTerminalLogReplayEvents(input: { runId: string; lines: readonly string[]; source?: PlayRunLedgerEventSource; occurredAt?: number; liveLogTotalCount?: number; dedupeMode?: 'semantic' | 'exact'; producerAttempt?: number; }): PlayRunLedgerEvent[] { const channelOffset = terminalLogReplayChannelOffset({ liveLogTotalCount: input.liveLogTotalCount, liveLogsLength: input.lines.length, }); return buildPlayRunLedgerEventsFromLogLines({ runId: input.runId, lines: input.lines, source: 'coordinator', occurredAt: input.occurredAt, channelOffset, producerAttempt: input.producerAttempt, transportDedupe: input.dedupeMode ?? (typeof input.producerAttempt === 'number' ? 'exact' : undefined), }); } export function buildPlayRunLedgerEventsFromStatusPatch(input: { patch: PlayRunLedgerStatusPatch; previousSnapshot: PlayRunLedgerSnapshot; now?: number; source?: PlayRunLedgerEventSource; }): PlayRunLedgerEvent[] { const now = input.now ?? Date.now(); const source = input.source ?? 'worker'; const patch = input.patch; const status = normalizePlayRunLedgerStatus(patch.status); const liveProgress = normalizeLiveProgressMap(patch.liveNodeProgress); const checkpointAt = finiteNumber(patch.lastCheckpointAt) ?? now; const progressStartedAts = Object.values(liveProgress) .map((progress) => progress.startedAt) .filter((value): value is number => typeof value === 'number'); const progressCompletedAts = Object.values(liveProgress) .map((progress) => progress.completedAt) .filter((value): value is number => typeof value === 'number'); const existingStepStartedAts = Object.values(input.previousSnapshot.stepsById) .map((step) => step.startedAt) .filter((value): value is number => typeof value === 'number'); const existingStepCompletedAts = Object.values( input.previousSnapshot.stepsById, ) .map((step) => step.completedAt) .filter((value): value is number => typeof value === 'number'); const logEvents = Array.isArray(patch.liveLogs) ? buildPlayRunLedgerEventsFromLogLines({ runId: patch.runId, lines: patch.liveLogs, source, occurredAt: checkpointAt, channelOffset: patch.liveLogsChannelOffset ?? null, producerAttempt: patch.liveLogsProducerAttempt ?? null, }) : []; const earliestStartedAt = progressStartedAts.length > 0 || existingStepStartedAts.length > 0 ? Math.min(...progressStartedAts, ...existingStepStartedAts) : null; const latestCompletedAt = progressCompletedAts.length > 0 || existingStepCompletedAts.length > 0 ? Math.max(...progressCompletedAts, ...existingStepCompletedAts) : null; const events: PlayRunLedgerEvent[] = []; if ( !input.previousSnapshot.startedAt && (Object.keys(liveProgress).length > 0 || status === 'running' || isTerminalPlayRunLedgerStatus(status)) ) { events.push({ type: 'run.started', runId: patch.runId, playName: patch.playName ?? input.previousSnapshot.playName ?? null, source, occurredAt: earliestStartedAt ?? checkpointAt, runtimeBackend: patch.runtimeBackend ?? null, }); } events.push(...logEvents); for (const [stepId, progress] of Object.entries(liveProgress)) { const previousStep = input.previousSnapshot.stepsById[stepId]; if (progress.startedAt && !previousStep?.startedAt) { events.push({ type: 'step.started', runId: patch.runId, source, occurredAt: progress.startedAt, stepId, artifactTableNamespace: progress.artifactTableNamespace ?? null, }); } const normalizedProgress: PlayRunLedgerStepProgress = { ...(progress.completed !== undefined ? { completed: progress.completed } : {}), ...(progress.total !== undefined ? { total: progress.total } : {}), ...(progress.failed !== undefined ? { failed: progress.failed } : {}), ...(progress.startedRows !== undefined ? { startedRows: progress.startedRows } : {}), ...(progress.activeRows !== undefined ? { activeRows: progress.activeRows } : {}), ...(progress.waitingRows !== undefined ? { waitingRows: progress.waitingRows } : {}), ...(progress.completedRows !== undefined ? { completedRows: progress.completedRows } : {}), ...(progress.message !== undefined ? { message: progress.message } : {}), ...(progress.artifactTableNamespace !== undefined ? { artifactTableNamespace: progress.artifactTableNamespace } : {}), ...(progress.startedAt !== undefined ? { startedAt: progress.startedAt } : {}), ...(progress.completedAt !== undefined ? { completedAt: progress.completedAt } : {}), ...(progress.nodeIo !== undefined ? { nodeIo: progress.nodeIo } : {}), updatedAt: progress.updatedAt ?? checkpointAt, }; if ( progressSignature(normalizedProgress) !== progressSignature(previousStep?.progress) ) { events.push({ type: 'step.progress', runId: patch.runId, source, occurredAt: progress.updatedAt ?? checkpointAt, stepId, status: typeof progress.nodeIo?.error === 'string' ? 'failed' : typeof progress.completedAt === 'number' ? 'completed' : previousStep?.status === 'failed' ? 'failed' : 'running', progress: normalizedProgress, }); } if (progress.completedAt && !previousStep?.completedAt) { events.push({ type: 'step.completed', runId: patch.runId, source, occurredAt: progress.completedAt, stepId, artifactTableNamespace: progress.artifactTableNamespace ?? null, }); } } const terminalType = terminalEventTypeForStatus(status); if ( terminalType && !isTerminalPlayRunLedgerStatus(input.previousSnapshot.status) ) { const occurredAt = latestCompletedAt ?? checkpointAt; if (terminalType === 'run.completed') { events.push({ type: terminalType, runId: patch.runId, source, occurredAt, result: patch.result, }); } else { events.push({ type: terminalType, runId: patch.runId, source, occurredAt, error: patch.error ?? null, result: patch.result, }); } } return events; }