diff --git a/apps/api/src/index.js b/apps/api/src/index.js index 23fa157..d056a7e 100644 --- a/apps/api/src/index.js +++ b/apps/api/src/index.js @@ -122,6 +122,16 @@ function buildPageInfo(total, page, limit) { }; } +function normalizeOptionalInt(value, { min = null, max = null } = {}) { + if (value === undefined || value === null || value === '') return null; + const parsed = Number(value); + if (!Number.isFinite(parsed)) return null; + const normalized = Math.floor(parsed); + if (min !== null && normalized < min) return null; + if (max !== null && normalized > max) return null; + return normalized; +} + function deterministicUnit(seed, key) { const hex = hashText(`${seed}:${key}`).slice(0, 12); return Number.parseInt(hex, 16) / 0xffffffffffff; @@ -1795,6 +1805,261 @@ app.get('/v1/runs/:id/summary', { preHandler: workspaceGuard({ role: 'viewer', r return runBundle; }); +app.get('/v1/runs/:id/replay-v2/streams', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: 'runParam' }) }, async (request, reply) => { + const run = await fetchRun(request.params.id); + if (!run) return reply.code(404).send({ error: 'not_found' }); + + const { rows } = await query( + `select stream_id, schema_version, started_at, first_seq, last_seq, chunk_count, + event_count, final_received, protocol_version, transport_kind, harbor_root, + fin_seq, ack_seq, ack_received, seek_stride, actionable_command_count, + aligned_command_count, target_resolved_count, orphan_count, target_registry_version, + updated_at + from replay_v2_streams + where run_id = $1 + order by started_at asc nulls last, stream_id asc`, + [request.params.id] + ); + + return { + items: rows, + pageInfo: buildPageInfo(rows.length, 1, Math.max(rows.length, 1)) + }; +}); + +app.get('/v1/runs/:id/replay-v2/events', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: 'runParam' }) }, async (request, reply) => { + const run = await fetchRun(request.params.id); + if (!run) return reply.code(404).send({ error: 'not_found' }); + + const streamId = String(request.query?.streamId || '').trim(); + if (!streamId) return reply.code(400).send({ error: 'stream_id_required' }); + + const fromSeq = normalizeOptionalInt(request.query?.fromSeq, { min: 1 }); + const toSeq = normalizeOptionalInt(request.query?.toSeq, { min: 1 }); + const parsedLimit = normalizeOptionalInt(request.query?.limit, { min: 1, max: 1000 }); + const normalizedLimit = parsedLimit ?? 300; + + if ((request.query?.fromSeq ?? '') !== '' && fromSeq === null) { + return reply.code(400).send({ error: 'invalid_from_seq' }); + } + if ((request.query?.toSeq ?? '') !== '' && toSeq === null) { + return reply.code(400).send({ error: 'invalid_to_seq' }); + } + if ((request.query?.limit ?? '') !== '' && parsedLimit === null) { + return reply.code(400).send({ error: 'invalid_limit' }); + } + if (fromSeq !== null && toSeq !== null && fromSeq > toSeq) { + return reply.code(400).send({ error: 'invalid_seq_range' }); + } + + const streamRes = await query( + `select stream_id + from replay_v2_streams + where run_id = $1 and stream_id = $2`, + [request.params.id, streamId] + ); + if (!streamRes.rows.length) return reply.code(404).send({ error: 'not_found' }); + + const totalRes = await query( + `select count(*)::int as total + from replay_v2_events + where run_id = $1 + and stream_id = $2 + and ($3::int is null or seq >= $3) + and ($4::int is null or seq <= $4)`, + [request.params.id, streamId, fromSeq, toSeq] + ); + + const { rows } = await query( + `select e.seq, e.kind, e.ts, e.monotonic_ms, e.target_id, e.selector_bundle, + e.data_json, e.chunk_id, c.chunk_index, c.final, e.command_id, e.target_ref, + e.payload_json, e.lifecycle_event, e.selector_version, e.dom_signature_hash, e.asset_refs + from replay_v2_events e + left join replay_v2_chunks c on c.id = e.chunk_id + where e.run_id = $1 + and e.stream_id = $2 + and ($3::int is null or e.seq >= $3) + and ($4::int is null or e.seq <= $4) + order by e.seq asc + limit $5`, + [request.params.id, streamId, fromSeq, toSeq, normalizedLimit] + ); + + return { + items: rows, + pageInfo: { + ...buildPageInfo(totalRes.rows[0]?.total, 1, normalizedLimit), + streamId, + fromSeq, + toSeq + } + }; +}); + +app.get('/v1/runs/:id/replay-v2/targets', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: 'runParam' }) }, async (request, reply) => { + const run = await fetchRun(request.params.id); + if (!run) return reply.code(404).send({ error: 'not_found' }); + + const streamId = String(request.query?.streamId || '').trim(); + if (!streamId) return reply.code(400).send({ error: 'stream_id_required' }); + + const seq = normalizeOptionalInt(request.query?.seq, { min: 1 }) ?? 2147483647; + const { rows } = await query( + `select distinct on (target_id) + target_id, selector_version, state, event_seq, lifecycle_event, selector_bundle, metadata_json, dom_signature_hash, created_at + from replay_v2_target_registry + where run_id = $1 + and stream_id = $2 + and event_seq <= $3 + order by target_id, event_seq desc`, + [request.params.id, streamId, seq] + ); + + return { + items: rows, + pageInfo: { + ...buildPageInfo(rows.length, 1, Math.max(rows.length, 1)), + streamId, + seq: Number.isFinite(seq) ? seq : null + } + }; +}); + +app.get('/v1/runs/:id/replay-v2/seek', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: 'runParam' }) }, async (request, reply) => { + const run = await fetchRun(request.params.id); + if (!run) return reply.code(404).send({ error: 'not_found' }); + + const streamId = String(request.query?.streamId || '').trim(); + if (!streamId) return reply.code(400).send({ error: 'stream_id_required' }); + + const seq = normalizeOptionalInt(request.query?.seq, { min: 1 }); + if (seq === null) return reply.code(400).send({ error: 'invalid_seq' }); + + const streamRes = await query( + `select stream_id, first_seq, last_seq, seek_stride + from replay_v2_streams + where run_id = $1 and stream_id = $2`, + [request.params.id, streamId] + ); + const stream = streamRes.rows[0]; + if (!stream) return reply.code(404).send({ error: 'not_found' }); + + const checkpointRes = await query( + `select checkpoint_seq, event_seq, monotonic_ms, target_registry_state_json + from replay_v2_seek_index + where run_id = $1 and stream_id = $2 and checkpoint_seq <= $3 + order by checkpoint_seq desc + limit 1`, + [request.params.id, streamId, seq] + ); + const checkpoint = checkpointRes.rows[0] || null; + const deltaStart = checkpoint ? checkpoint.event_seq + 1 : Math.max(1, stream.first_seq || 1); + + const deltasRes = await query( + `select seq, kind, ts, monotonic_ms, target_id, command_id, target_ref, selector_bundle, + payload_json, lifecycle_event, selector_version, dom_signature_hash, asset_refs + from replay_v2_events + where run_id = $1 + and stream_id = $2 + and seq >= $3 + and seq <= $4 + order by seq asc`, + [request.params.id, streamId, deltaStart, seq] + ); + + const targetsRes = await query( + `select distinct on (target_id) + target_id, selector_version, state, event_seq, lifecycle_event, selector_bundle, metadata_json, dom_signature_hash + from replay_v2_target_registry + where run_id = $1 + and stream_id = $2 + and event_seq <= $3 + order by target_id, event_seq desc`, + [request.params.id, streamId, seq] + ); + + const inspectEventRes = await query( + `select seq, target_id, target_ref, selector_bundle, dom_signature_hash, payload_json + from replay_v2_events + where run_id = $1 + and stream_id = $2 + and seq <= $3 + and target_id is not null + order by seq desc + limit 1`, + [request.params.id, streamId, seq] + ); + const inspectEvent = inspectEventRes.rows[0] || null; + + return { + item: { + streamId, + seq, + checkpoint, + deltas: deltasRes.rows, + resolvedTargets: targetsRes.rows, + liveInspect: inspectEvent ? { + seq: inspectEvent.seq, + targetId: inspectEvent.target_id, + selectorBundle: inspectEvent.selector_bundle, + domSignatureHash: inspectEvent.dom_signature_hash, + payload: inspectEvent.payload_json + } : null + } + }; +}); + +app.get('/v1/runs/:id/replay-v2/metrics', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: 'runParam' }) }, async (request, reply) => { + const run = await fetchRun(request.params.id); + if (!run) return reply.code(404).send({ error: 'not_found' }); + + const streamId = String(request.query?.streamId || '').trim(); + if (!streamId) return reply.code(400).send({ error: 'stream_id_required' }); + + const { rows } = await query( + `select stream_id, first_seq, last_seq, event_count, + actionable_command_count, aligned_command_count, target_resolved_count, + orphan_count, final_received, ack_received, fin_seq, ack_seq, seek_stride, updated_at + from replay_v2_streams + where run_id = $1 and stream_id = $2`, + [request.params.id, streamId] + ); + + const { rows: seqRows } = await query( + `select min(seq) as min_seq, max(seq) as max_seq, count(*) as row_count + from replay_v2_events + where run_id = $1 and stream_id = $2`, + [request.params.id, streamId] + ); + const stream = rows[0]; + if (!stream) return reply.code(404).send({ error: 'not_found' }); + + const actionable = Number(stream.actionable_command_count || 0); + const alignment = actionable > 0 ? Number(stream.aligned_command_count || 0) / actionable : 1; + const targetStability = actionable > 0 ? Number(stream.target_resolved_count || 0) / actionable : 1; + + const seqRow = seqRows?.[0] || {}; + const minSeq = seqRow.min_seq !== null ? Number(seqRow.min_seq) : null; + const maxSeq = seqRow.max_seq !== null ? Number(seqRow.max_seq) : null; + const rowCount = Number(seqRow.row_count || 0); + const spanCount = minSeq === null || maxSeq === null ? 0 : (maxSeq - minSeq + 1); + const gapCount = Math.max(0, spanCount - rowCount); + + return { + item: { + ...stream, + seqContinuity: { + zeroGaps: gapCount === 0, + gapCount + }, + finAckSuccess: Boolean(stream.final_received && stream.ack_received), + commandToDomAlignment: alignment, + targetStability, + orphanSpamRisk: Number(stream.orphan_count || 0) > Math.max(1, Math.floor(Number(stream.event_count || 0) * 0.01)) + } + }; +}); + app.get('/v1/runs/:runId/specs', { preHandler: workspaceGuard({ role: 'viewer', resolveWorkspaceId: async (request) => resolveWorkspaceIdFromRequestPart(request, 'runParam') }) }, async (request, reply) => { const { runId } = request.params; const { status = null, page = 1, limit = 50 } = request.query || {}; diff --git a/apps/ingest/src/index.js b/apps/ingest/src/index.js index 74b278d..febf360 100644 --- a/apps/ingest/src/index.js +++ b/apps/ingest/src/index.js @@ -1,7 +1,17 @@ import crypto from 'node:crypto'; import Fastify from 'fastify'; import pg from 'pg'; -import { INGEST_EVENT_TYPES, assertReplayV2ChunkPayload, isValidIngestType } from '@testharbor/shared'; +import { + INGEST_EVENT_TYPES, + REPLAY_V2_EVENT_KINDS, + REPLAY_V2_LIFECYCLE_EVENTS, + REPLAY_V2_SCHEMA_VERSION, + REPLAY_V2_SEEK_STRIDE, + applyReplayV2EventToTargetRegistry, + assertReplayV2ChunkPayload, + createReplayV2TargetRegistry, + isValidIngestType +} from '@testharbor/shared'; const app = Fastify({ logger: true }); const port = Number(process.env.PORT || 4010); @@ -278,12 +288,152 @@ async function lookupRunContextBySpecRunId(specRunId, db = pool) { return res.rows[0] || null; } +function sha256Hex(value) { + return crypto.createHash('sha256').update(value).digest('hex'); +} + +function jsonClone(value) { + return value == null ? value : JSON.parse(JSON.stringify(value)); +} + +function setDeepValue(target, path, value) { + if (!path.length) return value; + let cursor = target; + for (let i = 0; i < path.length - 1; i += 1) { + cursor = cursor[path[i]]; + } + cursor[path[path.length - 1]] = value; + return target; +} + +function isSensitiveAsset({ sourceUrl, mimeType }) { + const haystack = `${String(sourceUrl || '')} ${String(mimeType || '')}`.toLowerCase(); + return ['token', 'secret', 'credential', 'session', 'cookie', 'authorization'].some((term) => haystack.includes(term)); +} + +function isAllowedAsset({ mimeType, byteSize }) { + const normalizedMime = String(mimeType || '').toLowerCase(); + const allowedMime = !normalizedMime + || normalizedMime.startsWith('image/') + || normalizedMime.startsWith('font/') + || normalizedMime === 'text/css' + || normalizedMime === 'application/javascript' + || normalizedMime === 'text/javascript'; + const allowedSize = byteSize == null || Number(byteSize) <= 10 * 1024 * 1024; + return allowedMime && allowedSize; +} + +function collectAssetCandidates(value, path = [], assets = []) { + if (Array.isArray(value)) { + value.forEach((item, index) => collectAssetCandidates(item, [...path, index], assets)); + return assets; + } + + if (!value || typeof value !== 'object') return assets; + + if (typeof value.url === 'string') { + assets.push({ + path: [...path, 'url'], + sourceUrl: value.url, + mimeType: value.mimeType || value.contentType || null, + byteSize: value.byteSize || value.size || null, + contentBase64: value.contentBase64 || null, + body: typeof value.body === 'string' ? value.body : null + }); + } + + for (const [key, item] of Object.entries(value)) { + if (key === 'url') continue; + collectAssetCandidates(item, [...path, key], assets); + } + return assets; +} + +async function rewritePayloadAssetsToCas(runId, payload, db) { + const rewritten = jsonClone(payload) || {}; + const assetRefs = []; + for (const asset of collectAssetCandidates(rewritten)) { + const sourceMaterial = asset.contentBase64 || asset.body || asset.sourceUrl; + if (!sourceMaterial) continue; + + const sha256 = sha256Hex(Buffer.isBuffer(sourceMaterial) ? sourceMaterial : String(sourceMaterial)); + const casRef = `cas://sha256/${sha256}`; + const blocked = isSensitiveAsset(asset) || !isAllowedAsset(asset); + const blockReason = isSensitiveAsset(asset) + ? 'sensitive_asset_blocklist' + : (!isAllowedAsset(asset) ? 'asset_allowlist_reject' : null); + + await db.query( + `insert into replay_v2_assets_cas ( + run_id, sha256, source_url, cas_ref, mime_type, byte_size, blocked, block_reason, metadata_json + ) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) + on conflict (run_id, sha256) + do update set + source_url = coalesce(replay_v2_assets_cas.source_url, excluded.source_url), + mime_type = coalesce(replay_v2_assets_cas.mime_type, excluded.mime_type), + byte_size = coalesce(replay_v2_assets_cas.byte_size, excluded.byte_size), + blocked = replay_v2_assets_cas.blocked or excluded.blocked, + block_reason = coalesce(replay_v2_assets_cas.block_reason, excluded.block_reason), + metadata_json = coalesce(replay_v2_assets_cas.metadata_json, excluded.metadata_json)`, + [ + runId, + sha256, + asset.sourceUrl, + casRef, + asset.mimeType, + asset.byteSize, + blocked, + blockReason, + JSON.stringify({ source: 'replay-v2', path: asset.path.join('.') }) + ] + ); + + if (!blocked) { + setDeepValue(rewritten, asset.path, casRef); + } + + assetRefs.push({ + sourceUrl: asset.sourceUrl, + casRef, + sha256, + mimeType: asset.mimeType, + byteSize: asset.byteSize, + blocked, + blockReason + }); + } + + return { payload: rewritten, assetRefs }; +} + +async function loadReplayTargetRegistryState(runId, streamId, db) { + const { rows } = await db.query( + `select distinct on (target_id) + target_id, selector_version, state, selector_bundle, metadata_json, dom_signature_hash + from replay_v2_target_registry + where run_id = $1 and stream_id = $2 + order by target_id, event_seq desc`, + [runId, streamId] + ); + + return rows.map((row) => ({ + targetId: row.target_id, + selectorVersion: row.selector_version, + selectorBundle: row.selector_bundle || {}, + metadata: row.metadata_json || null, + state: row.state, + reason: row.state === 'orphaned' ? row.metadata_json?.reason || null : null, + domSignatureHash: row.dom_signature_hash || null + })); +} + async function persistReplayV2Chunk(payload, idempotencyKey) { await withTransaction(async (db) => { await db.query('select pg_advisory_xact_lock(hashtext($1), hashtext($2))', [payload.runId, payload.streamId]); const existingChunk = await db.query( - `select run_id, stream_id, seq_start, seq_end + `select id, run_id, stream_id, seq_start, seq_end from replay_v2_chunks where idempotency_key = $1`, [idempotencyKey] @@ -296,27 +446,34 @@ async function persistReplayV2Chunk(payload, idempotencyKey) { await db.query( `insert into replay_v2_streams ( - run_id, stream_id, schema_version, started_at, metadata_json, - first_seq, last_seq, chunk_count, event_count, final_received, created_at, updated_at + run_id, stream_id, schema_version, started_at, metadata_json, protocol_version, transport_kind, harbor_root, + seek_stride, first_seq, last_seq, chunk_count, event_count, final_received, created_at, updated_at ) - values ($1, $2, $3, $4::timestamptz, $5::jsonb, null, null, 0, 0, false, now(), now()) + values ($1, $2, $3, $4::timestamptz, $5::jsonb, 'v2', $6, $7, $8, null, null, 0, 0, false, now(), now()) on conflict (run_id, stream_id) do update set schema_version = excluded.schema_version, started_at = coalesce(replay_v2_streams.started_at, excluded.started_at), metadata_json = coalesce(replay_v2_streams.metadata_json, excluded.metadata_json), + transport_kind = coalesce(excluded.transport_kind, replay_v2_streams.transport_kind), + harbor_root = coalesce(excluded.harbor_root, replay_v2_streams.harbor_root), + seek_stride = coalesce(excluded.seek_stride, replay_v2_streams.seek_stride), updated_at = now()`, [ payload.runId, payload.streamId, - payload.schemaVersion ?? '2.0', + payload.schemaVersion ?? REPLAY_V2_SCHEMA_VERSION, payload.startedAt ?? null, - JSON.stringify(payload.metadata ?? null) + JSON.stringify(payload.metadata ?? null), + payload.transport?.kind ?? 'ws+msgpack', + payload.transport?.harborRoot ?? null, + payload.seekStride ?? REPLAY_V2_SEEK_STRIDE ] ); const stream = await db.query( - `select first_seq, last_seq, chunk_count, event_count, final_received + `select first_seq, last_seq, chunk_count, event_count, final_received, seek_stride, + actionable_command_count, aligned_command_count, target_resolved_count, orphan_count from replay_v2_streams where run_id = $1 and stream_id = $2 for update`, @@ -351,53 +508,180 @@ async function persistReplayV2Chunk(payload, idempotencyKey) { const chunkResult = await db.query( `insert into replay_v2_chunks ( - run_id, stream_id, idempotency_key, schema_version, - seq_start, seq_end, event_count, chunk_index, final, started_at, payload_json + run_id, stream_id, idempotency_key, schema_version, seq_start, seq_end, event_count, + chunk_index, final, started_at, payload_json, harbor_segment_path, harbor_segment_index, + harbor_byte_offset, harbor_byte_length, frame_codec, acked, ack_meta_json ) - values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::timestamptz, $11::jsonb) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::timestamptz, $11::jsonb, $12, $13, $14, $15, $16, $17, $18::jsonb) returning id`, [ payload.runId, payload.streamId, idempotencyKey, - payload.schemaVersion ?? '2.0', + payload.schemaVersion ?? REPLAY_V2_SCHEMA_VERSION, payload.seqStart, payload.seqEnd, payload.events.length, payload.chunkIndex ?? null, payload.final === true, payload.startedAt ?? null, - JSON.stringify(payload) + JSON.stringify(payload), + payload.transport?.segmentPath ?? null, + payload.transport?.segmentIndex ?? null, + payload.transport?.byteOffset ?? null, + payload.transport?.byteLength ?? null, + payload.transport?.codec ?? 'msgpack', + payload.transport?.ack?.ok === true, + JSON.stringify(payload.transport?.ack ?? null) ] ); const chunkId = chunkResult.rows[0].id; + const registry = createReplayV2TargetRegistry({ + initialState: await loadReplayTargetRegistryState(payload.runId, payload.streamId, db) + }); + const stride = streamState?.seek_stride || payload.seekStride || REPLAY_V2_SEEK_STRIDE; + const seekRows = []; + const registryRows = []; + let actionableCommandCount = 0; + let alignedCommandCount = 0; + let targetResolvedCount = 0; + let orphanCount = 0; + let finSeq = null; + let ackSeq = null; + let lastCheckpointSeq = 0; + const eventValues = []; - const eventPlaceholders = payload.events.map((event, index) => { - const offset = index * 10; + const eventPlaceholders = []; + + for (const [index, inputEvent] of payload.events.entries()) { + const event = inputEvent; + const { payload: rewrittenPayload, assetRefs } = await rewritePayloadAssetsToCas(payload.runId, event.payload || {}, db); + const targetRef = event.targetRef || (event.targetId ? { + targetId: event.targetId, + selectorVersion: event.selectorVersion || event.selectorBundle?.selectorVersion || 1 + } : null); + const lifecycleEvent = event.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE ? (rewrittenPayload.eventType || null) : null; + const domSignatureHash = rewrittenPayload.selectorBundle?.domSignature?.hash + || event.selectorBundle?.domSignature?.hash + || null; + + const offset = index * 17; eventValues.push( payload.runId, payload.streamId, event.seq, event.kind, - event.ts, - event.monotonicMs, - event.targetId ?? null, - JSON.stringify(event.selectorBundle ?? null), - JSON.stringify(event.data ?? null), - chunkId + event.ts || payload.startedAt, + event.monotonicTs, + targetRef?.targetId ?? null, + JSON.stringify(event.selectorBundle ?? rewrittenPayload.selectorBundle ?? null), + JSON.stringify(rewrittenPayload), + chunkId, + event.commandId ?? null, + JSON.stringify(targetRef ?? null), + JSON.stringify(rewrittenPayload), + lifecycleEvent, + targetRef?.selectorVersion ?? null, + domSignatureHash, + JSON.stringify(assetRefs.length ? assetRefs : null) ); - return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}::timestamptz, $${offset + 6}, $${offset + 7}, $${offset + 8}::jsonb, $${offset + 9}::jsonb, $${offset + 10})`; - }); + eventPlaceholders.push(`($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}::timestamptz, $${offset + 6}, $${offset + 7}, $${offset + 8}::jsonb, $${offset + 9}::jsonb, $${offset + 10}, $${offset + 11}, $${offset + 12}::jsonb, $${offset + 13}::jsonb, $${offset + 14}, $${offset + 15}, $${offset + 16}, $${offset + 17}::jsonb)`); + + if (event.kind === REPLAY_V2_EVENT_KINDS.COMMAND && event.commandId) { + actionableCommandCount += 1; + if (targetRef?.targetId || rewrittenPayload.targetSnapshot) alignedCommandCount += 1; + if (!targetRef?.targetId || registry.get(targetRef.targetId)?.state !== 'orphaned') targetResolvedCount += 1; + } + + if (event.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE) { + const nextTarget = applyReplayV2EventToTargetRegistry(registry, { + ...event, + payload: rewrittenPayload, + targetRef + }); + if (nextTarget) { + registryRows.push([ + payload.runId, + payload.streamId, + nextTarget.targetId, + nextTarget.selectorVersion, + nextTarget.state, + event.seq, + lifecycleEvent, + JSON.stringify(nextTarget.selectorBundle ?? null), + JSON.stringify(nextTarget.metadata ?? null), + nextTarget.selectorBundle?.domSignature?.hash || domSignatureHash || null + ]); + if (nextTarget.state === 'orphaned') orphanCount += 1; + } + + if (lifecycleEvent === REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_FIN) finSeq = event.seq; + if (lifecycleEvent === REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_ACK) ackSeq = event.seq; + } + + const shouldCheckpoint = !lastCheckpointSeq + || event.seq - lastCheckpointSeq >= stride + || String(lifecycleEvent || '').startsWith('TARGET_'); + if (shouldCheckpoint) { + lastCheckpointSeq = event.seq; + seekRows.push([ + payload.runId, + payload.streamId, + event.seq, + event.seq, + event.monotonicTs, + JSON.stringify(registry.snapshot()) + ]); + } + } await db.query( `insert into replay_v2_events ( - run_id, stream_id, seq, kind, ts, monotonic_ms, target_id, selector_bundle, data_json, chunk_id + run_id, stream_id, seq, kind, ts, monotonic_ms, target_id, selector_bundle, data_json, chunk_id, + command_id, target_ref, payload_json, lifecycle_event, selector_version, dom_signature_hash, asset_refs ) values ${eventPlaceholders.join(', ')}`, eventValues ); + if (registryRows.length) { + const registryValues = []; + const registryPlaceholders = registryRows.map((row, index) => { + const offset = index * 10; + registryValues.push(...row); + return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}, $${offset + 6}, $${offset + 7}, $${offset + 8}::jsonb, $${offset + 9}::jsonb, $${offset + 10})`; + }); + await db.query( + `insert into replay_v2_target_registry ( + run_id, stream_id, target_id, selector_version, state, event_seq, lifecycle_event, selector_bundle, metadata_json, dom_signature_hash + ) + values ${registryPlaceholders.join(', ')}`, + registryValues + ); + } + + if (seekRows.length) { + const seekValues = []; + const seekPlaceholders = seekRows.map((row, index) => { + const offset = index * 6; + seekValues.push(...row); + return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}, $${offset + 6}::jsonb)`; + }); + await db.query( + `insert into replay_v2_seek_index ( + run_id, stream_id, checkpoint_seq, event_seq, monotonic_ms, target_registry_state_json + ) + values ${seekPlaceholders.join(', ')} + on conflict (run_id, stream_id, checkpoint_seq) + do update set + event_seq = excluded.event_seq, + monotonic_ms = excluded.monotonic_ms, + target_registry_state_json = excluded.target_registry_state_json`, + seekValues + ); + } + await db.query( `update replay_v2_streams set first_seq = coalesce(first_seq, $3), @@ -405,9 +689,34 @@ async function persistReplayV2Chunk(payload, idempotencyKey) { chunk_count = chunk_count + 1, event_count = event_count + $5, final_received = final_received or $6, + fin_seq = coalesce($7, fin_seq), + ack_seq = coalesce($8, ack_seq), + ack_received = ack_received or ($8 is not null) or coalesce($9, false), + fin_ack_meta_json = coalesce($10::jsonb, fin_ack_meta_json), + actionable_command_count = actionable_command_count + $11, + aligned_command_count = aligned_command_count + $12, + target_resolved_count = target_resolved_count + $13, + orphan_count = orphan_count + $14, + target_registry_version = target_registry_version + $15, updated_at = now() where run_id = $1 and stream_id = $2`, - [payload.runId, payload.streamId, payload.seqStart, payload.seqEnd, payload.events.length, payload.final === true] + [ + payload.runId, + payload.streamId, + payload.seqStart, + payload.seqEnd, + payload.events.length, + payload.final === true, + finSeq, + ackSeq, + payload.transport?.ack?.ok === true, + JSON.stringify(payload.transport?.ack ?? null), + actionableCommandCount, + alignedCommandCount, + targetResolvedCount, + orphanCount, + registryRows.length + ] ); }); } diff --git a/apps/web/src/server.js b/apps/web/src/server.js index 027a3c6..fc9ebf5 100644 --- a/apps/web/src/server.js +++ b/apps/web/src/server.js @@ -261,6 +261,11 @@ function metric(label, value) { return `
${escapeHtml(label)}${escapeHtml(value)}
`; } +function formatJsonInline(value) { + if (value === undefined || value === null) return 'n/a'; + return `
view
${escapeHtml(JSON.stringify(value, null, 2))}
`; +} + function renderLayout({ title, shell, currentPath, content }) { const user = shell.session?.user; const workspaceName = shell.selectedWorkspace?.name || 'No workspace'; @@ -828,6 +833,15 @@ function renderRunDetailPage(shell, runDetail) { ${summaryCard('Artifacts', String(summary.artifacts.artifact_count || 0), formatBytes(summary.artifacts.total_artifact_bytes))} +
+
+
+

Replay V2

+

Open the persisted replay viewer for stream summaries and ordered replay events.

+
+ Open Replay V2 +
+

Specs

All spec runs recorded for this run.

@@ -916,6 +930,152 @@ ${test.stacktrace || 'No stacktrace captured'}`)} }); } +function renderReplayV2Page(shell, runId, streamsResp, eventsResp, selectedStreamId, metricsResp, targetsResp, seekResp, seekSeq) { + const streams = streamsResp.items || []; + const events = eventsResp.items || []; + const pageInfo = eventsResp.pageInfo || { total: events.length, limit: events.length }; + const selectedStream = streams.find((stream) => stream.stream_id === selectedStreamId) || streams[0] || null; + const metrics = metricsResp?.item || null; + const targets = targetsResp?.items || []; + const seek = seekResp?.item || null; + const inspect = seek?.liveInspect || null; + const alignmentPct = metrics ? `${Math.round((metrics.commandToDomAlignment || 0) * 100)}%` : 'n/a'; + const targetPct = metrics ? `${Math.round((metrics.targetStability || 0) * 100)}%` : 'n/a'; + const seqContinuityText = metrics + ? (metrics.seqContinuity?.zeroGaps ? 'zero gaps' : `${metrics.seqContinuity?.gapCount || 0} gaps`) + : 'n/a'; + + return renderLayout({ + title: `Replay V2 ${String(runId).slice(0, 8)}`, + shell, + currentPath: '/app/runs', + content: `
+
+

Replay V2

+

Persisted replay streams for run ${escapeHtml(String(runId))}

+

Read model over replay_v2_streams, replay_v2_chunks, and replay_v2_events for basic browser inspection.

+
+
+ ${summaryCard('Streams', String(streams.length), selectedStream ? `Selected: ${selectedStream.stream_id}` : 'No replay streams')} + ${summaryCard('Events shown', String(events.length), `${pageInfo.total || 0} matching rows`)} + ${summaryCard('Selection', selectedStreamId || 'none', selectedStream ? `Seq ${selectedStream.first_seq || 'n/a'}-${selectedStream.last_seq || 'n/a'}` : 'Select a stream')} + ${summaryCard('Seek', seekSeq || 'n/a', inspect?.targetId ? `Inspect ${inspect.targetId}` : 'Nearest checkpoint resolution')} +
+
+
+
+
+

Streams

+

Each card summarizes one persisted replay stream for this run.

+
+ Back to run +
+ ${streams.length ? `
+ ${streams.map((stream) => `
+ ${stream.stream_id === selectedStreamId ? 'Selected stream' : 'Replay stream'} + ${escapeHtml(stream.stream_id)} + Schema ${escapeHtml(stream.schema_version || '2.0')} · started ${escapeHtml(formatDate(stream.started_at))} + Seq ${escapeHtml(stream.first_seq ?? 'n/a')} → ${escapeHtml(stream.last_seq ?? 'n/a')} + ${escapeHtml(stream.event_count)} events · ${escapeHtml(stream.chunk_count)} chunks · final ${stream.final_received ? 'yes' : 'no'} + ${escapeHtml(stream.transport_kind || 'ws+msgpack')} · ACK ${stream.ack_received ? 'yes' : 'no'} · stride ${escapeHtml(stream.seek_stride ?? '50')} + Updated ${escapeHtml(formatDate(stream.updated_at))} +
`).join('')} +
` : '

No replay streams

This run has no persisted Replay V2 stream rows yet.

'} +
+
+
+
+

Gate Metrics

+

Acceptance-gate read model for FIN/ACK, alignment, target stability, and orphan pressure.

+
+ ${metrics ? badge(metrics.finAckSuccess ? 'FIN/ACK ok' : 'FIN/ACK pending', metrics.finAckSuccess ? 'success' : 'warning') : ''} +
+ ${metrics ? `
+ ${summaryCard('FIN/ACK', metrics.finAckSuccess ? '100%' : 'pending', `FIN ${metrics.fin_seq || 'n/a'} · ACK ${metrics.ack_seq || 'n/a'}`)} + ${summaryCard('Seq continuity', seqContinuityText, metrics.seqContinuity?.zeroGaps ? 'No missing sequence numbers' : 'Investigate replay chunk ordering')} + ${summaryCard('Cmd→DOM', alignmentPct, `${metrics.aligned_command_count || 0}/${metrics.actionable_command_count || 0} actionable`)} + ${summaryCard('Target Stability', targetPct, `${metrics.target_resolved_count || 0}/${metrics.actionable_command_count || 0} resolved`)} + ${summaryCard('Orphans', String(metrics.orphan_count || 0), metrics.orphanSpamRisk ? 'Above normal-run threshold' : 'Within normal-run threshold')} +
` : '

No metrics

Select a replay stream to inspect gate metrics.

'} +
+
+
+
+

Seek + Live Inspect

+

Nearest checkpoint plus forward deltas. Target resolution is evaluated at the requested sequence.

+
+ ${selectedStream ? `
+ + + +
` : ''} +
+ ${seek ? `
+ ${summaryCard('Checkpoint', String(seek.checkpoint?.checkpoint_seq || seek.seq), seek.checkpoint ? `${seek.deltas.length} forward deltas` : 'No prior checkpoint')} + ${summaryCard('Resolved targets', String(seek.resolvedTargets.length), inspect?.targetId ? `Inspecting ${inspect.targetId}` : 'No target at seek seq')} + ${summaryCard('Live Inspect', inspect?.domSignatureHash ? inspect.domSignatureHash.slice(0, 12) : 'n/a', inspect?.selectorBundle ? 'Selector bundle ready' : 'No inspect target')} +
+ ${inspect ? `
+ + + + + + + + +
SeqTargetDOM signatureSelector bundlePayload
${escapeHtml(inspect.seq)}${escapeHtml(inspect.targetId)}${escapeHtml(inspect.domSignatureHash || 'n/a')}${formatJsonInline(inspect.selectorBundle)}${formatJsonInline(inspect.payload)}
` : '

No inspect target

No target-backed event exists at or before the selected sequence.

'}` : '

No seek state

Select a stream and sequence to compute synchronized replay state.

'} +
+
+
+
+

Events

+

Ordered replay events for the selected stream. Default selection is the first stream for the run.

+
+ ${selectedStream ? badge(`${pageInfo.total || 0} matching`, 'neutral') : ''} +
+ ${!selectedStream ? '

No stream selected

Select a replay stream to inspect ordered events.

' : events.length ? `
+ + + ${events.map((event) => ` + + + + + + + + + `).join('')} + +
SeqKindTimestampMonotonicCommandTargetPayloadChunk
${escapeHtml(event.seq)}${escapeHtml(event.kind)}${escapeHtml(formatDate(event.ts))}${escapeHtml(`${event.monotonic_ms} ms`)}${escapeHtml(event.command_id || 'n/a')}${event.target_id ? `
${escapeHtml(event.target_id)}
v${escapeHtml(event.selector_version || '1')} · ${escapeHtml(event.lifecycle_event || 'active')}
` : 'n/a'}		${formatJsonInline(event.payload_json || event.data_json)}${event.chunk_id ? `
${escapeHtml(String(event.chunk_id).slice(0, 8))}
index ${escapeHtml(event.chunk_index ?? 'n/a')} · final ${event.final ? 'yes' : 'no'}
` : 'n/a'}
` : '

No replay events

The selected stream has no persisted events in the requested range.

'} +
+
+
+
+

Target Registry

+

Resolved logical targets at the selected sequence with selector bundle versions and DOM signatures.

+
+ ${selectedStream ? badge(`${targets.length} targets`, 'neutral') : ''} +
+ ${!selectedStream ? '

No stream selected

Select a replay stream to inspect target registry state.

' : targets.length ? `
+ + + ${targets.map((target) => ` + + + + + + + + `).join('')} + +
TargetVersionStateLifecycleSeqDOM signatureSelectors
${escapeHtml(target.target_id)}${escapeHtml(target.selector_version)}${escapeHtml(target.state)}${escapeHtml(target.lifecycle_event)}${escapeHtml(target.event_seq)}${escapeHtml(target.dom_signature_hash || 'n/a')}${formatJsonInline(target.selector_bundle)}
` : '

No targets

No target registry rows exist for the selected stream and sequence.

'} +
` + }); +} + function renderArtifactPage(shell, detail) { return renderLayout({ title: 'Artifact Viewer', @@ -1566,6 +1726,57 @@ app.get('/app/runs/:id', async (request, reply) => { return reply.type('text/html').send(renderRunDetailPage(shell, detail)); }); +app.get('/app/runs/:id/replay-v2', async (request, reply) => { + const shell = await loadShellData(request); + if (!shell.session) return requireSession(request, reply); + + const streamsResp = await apiFetch(`/v1/runs/${request.params.id}/replay-v2/streams`, { token: shell.session.token }); + const streams = streamsResp.items || []; + const requestedStreamId = String(request.query?.streamId || '').trim(); + const selectedStreamId = streams.some((stream) => stream.stream_id === requestedStreamId) + ? requestedStreamId + : String(streams[0]?.stream_id || ''); + + let eventsResp = { items: [], pageInfo: { total: 0, limit: 300 } }; + let metricsResp = { item: null }; + let targetsResp = { items: [] }; + let seekResp = { item: null }; + const seekSeq = String(request.query?.seq || '').trim(); + if (selectedStreamId) { + try { + eventsResp = await apiFetch(`/v1/runs/${request.params.id}/replay-v2/events?${new URLSearchParams({ + streamId: selectedStreamId, + limit: '300' + }).toString()}`, { token: shell.session.token }); + metricsResp = await apiFetch(`/v1/runs/${request.params.id}/replay-v2/metrics?${new URLSearchParams({ + streamId: selectedStreamId + }).toString()}`, { token: shell.session.token }); + targetsResp = await apiFetch(`/v1/runs/${request.params.id}/replay-v2/targets?${new URLSearchParams({ + streamId: selectedStreamId, + ...(seekSeq ? { seq: seekSeq } : {}) + }).toString()}`, { token: shell.session.token }); + seekResp = await apiFetch(`/v1/runs/${request.params.id}/replay-v2/seek?${new URLSearchParams({ + streamId: selectedStreamId, + seq: seekSeq || String(streams.find((stream) => stream.stream_id === selectedStreamId)?.last_seq || 1) + }).toString()}`, { token: shell.session.token }); + } catch (error) { + if (error.statusCode !== 404) throw error; + } + } + + return reply.type('text/html').send(renderReplayV2Page( + shell, + request.params.id, + streamsResp, + eventsResp, + selectedStreamId, + metricsResp, + targetsResp, + seekResp, + seekSeq + )); +}); + app.get('/app/tests/:id/history', async (request, reply) => { diff --git a/docs/REPLAY_V2_FULL_ARCHITECTURE_2026-04-03.md b/docs/REPLAY_V2_FULL_ARCHITECTURE_2026-04-03.md new file mode 100644 index 0000000..6b2ef37 --- /dev/null +++ b/docs/REPLAY_V2_FULL_ARCHITECTURE_2026-04-03.md @@ -0,0 +1,80 @@ +# Replay V2 Full Plan Architecture + +Replay V2 now follows a contract-first pipeline with a stable target identity layer and synchronized read models across reporter, ingest, API, and web viewer. + +## Phase A + +- `ReplayEventV2` is normalized around `kind` categories: `command`, `dom`, `network`, `console`, `lifecycle`. +- Target references use `targetRef = { targetId, selectorVersion }`. +- The target registry lifecycle is explicit: + - `TARGET_DECLARE` + - `TARGET_BIND` + - `TARGET_REBIND` + - `TARGET_ORPHAN` +- Selector bundles are stored as bundles, not single selectors: + - primary IDs: `data-cy`, `data-testid`, app IDs + - accessibility fallback: role/name/label/aria path + - structural fallback: CSS path/xpath/nth + - text fallback: text/proximity/near-text + - context anchors: frame/shadow paths and parent/sibling fingerprints + - DOM signature hash +- Resolution order is deterministic and version-bumped on rebind. + +## Phase B + +- `setupNodeEvents` now starts a dedicated WS transport server on port `9223` by default. +- Replay chunks are appended to segmented `.harbor` files beneath `.harbor/replay-v2///` as length-prefixed MessagePack frames. +- Transport metadata is persisted on replay chunks. +- FIN/ACK is represented as lifecycle protocol events and persisted into stream/chunk state. + +## Phase C + +Capture layering is recorded in order at session start: + +1. Cypress command lifecycle with target snapshots at command boundaries +2. rrweb incremental DOM configuration (`recordShadowDom: true`, `inlineStylesheet: true`) +3. CDP auto-attach declaration (`Target.setAutoAttach`) plus network/console/runtime intent +4. screencast explicitly deferred until stability gates pass + +Browser-side capture emission is available through `cy.task()` hooks under the `testharbor:replay:*` namespace. + +## Phase D + +- Replay payload asset URLs are rewritten to `cas://sha256/` when they pass the allowlist. +- CAS metadata is persisted in `replay_v2_assets_cas`. +- Sensitive URL/MIME patterns are blocked and retained with block reasons instead of rewritten. + +## Phase E + +- `replay_v2_seek_index` stores checkpoint snapshots every stride and on target lifecycle boundaries. +- Seek uses nearest checkpoint plus forward deltas from `replay_v2_events`. +- Target resolution at an arbitrary sequence is driven by `replay_v2_target_registry`. +- Live Inspect exposes the resolved selector bundle and DOM signature for the latest target-backed event at or before the requested sequence. + +## Gate Instrumentation + +Persisted stream counters now expose: + +- `actionable_command_count` +- `aligned_command_count` +- `target_resolved_count` +- `orphan_count` +- `final_received` +- `ack_received` + +Derived gates: + +- seq continuity: zero gaps required +- FIN/ACK success: `final_received && ack_received` +- command-to-DOM alignment: `aligned / actionable` +- target stability: `resolved / actionable` +- orphan spam: `orphan_count` within 1% of total event volume + +## Static Verification + +Use: + +- `node scripts/replay-v2-gate-artifacts.mjs` +- `node scripts/replay-v2-fin-ack-check.mjs ` (exits non-zero unless both FIN and ACK are present and correlated) +- `node --check ` +- `git diff --check` diff --git a/docs/REPLAY_V2_PHASE_C.md b/docs/REPLAY_V2_PHASE_C.md new file mode 100644 index 0000000..0aa2b4c --- /dev/null +++ b/docs/REPLAY_V2_PHASE_C.md @@ -0,0 +1,112 @@ +# Replay V2 Phase C + +Replay V2 Phase C adds the first read model and browser viewer on top of the persisted tables from migration `008_replay_v2_storage.sql`. + +## API Endpoints + +### `GET /v1/runs/:id/replay-v2/streams` + +Viewer-guarded through run-based workspace resolution. + +Response shape: + +```json +{ + "items": [ + { + "stream_id": "default", + "schema_version": "2.0", + "started_at": "2026-04-03T12:00:00.000Z", + "first_seq": 1, + "last_seq": 42, + "chunk_count": 3, + "event_count": 42, + "final_received": true, + "updated_at": "2026-04-03T12:00:04.000Z" + } + ], + "pageInfo": { + "page": 1, + "limit": 1, + "total": 1, + "totalPages": 1 + } +} +``` + +### `GET /v1/runs/:id/replay-v2/events` + +Viewer-guarded through run-based workspace resolution. + +Query params: + +- `streamId` required +- `fromSeq` optional +- `toSeq` optional +- `limit` optional, default `300`, max `1000` +- invalid `fromSeq`, `toSeq`, or `limit` values return `400` + +Response shape: + +```json +{ + "items": [ + { + "seq": 1, + "kind": "session.start", + "ts": "2026-04-03T12:00:00.000Z", + "monotonic_ms": 0, + "target_id": null, + "selector_bundle": null, + "data_json": { + "url": "https://example.test" + }, + "chunk_id": "9b4b0b0d-8d8b-40e5-8d95-5ab5f1d0b5e1", + "chunk_index": 0, + "final": false + } + ], + "pageInfo": { + "page": 1, + "limit": 300, + "total": 42, + "totalPages": 1, + "streamId": "default", + "fromSeq": null, + "toSeq": null + } +} +``` + +## Web Viewer + +Route: `GET /app/runs/:id/replay-v2` + +Behavior: + +- fetches replay stream summaries for the run +- selects `?streamId=` when provided, otherwise defaults to the first stream +- fetches ordered events for the selected stream +- renders empty states when the run has no replay streams or the selected stream has no events +- invalid `?streamId=` values gracefully fall back to the first available stream + +The existing run detail page now links to the Replay V2 viewer. + +## Manual Verification + +1. Start the API and web apps against a database with migration `008` applied. +2. Ensure a run exists with persisted Replay V2 rows in `replay_v2_streams`, `replay_v2_chunks`, and `replay_v2_events`. +3. Call `GET /v1/runs/:id/replay-v2/streams` and confirm the stream aggregate fields match the stored rows. +4. Call `GET /v1/runs/:id/replay-v2/events?streamId=` and confirm: + - events are ordered by `seq` ascending + - `chunk_index` and `final` are present when the owning chunk row exists + - `limit`, `fromSeq`, and `toSeq` constrain results as expected +5. Open `/app/runs/:id`, follow the Replay V2 link, and confirm: + - stream summary cards render + - the first stream is selected by default + - changing `?streamId=` changes the event table + - no-stream and no-event cases show explicit empty-state messages +6. Run: + - `node --check apps/api/src/index.js` + - `node --check apps/web/src/server.js` + - `git diff --check` diff --git a/infra/db/migrations/009_replay_v2_full_plan.sql b/infra/db/migrations/009_replay_v2_full_plan.sql new file mode 100644 index 0000000..537fc1f --- /dev/null +++ b/infra/db/migrations/009_replay_v2_full_plan.sql @@ -0,0 +1,89 @@ +ALTER TABLE replay_v2_streams + ADD COLUMN IF NOT EXISTS protocol_version TEXT NOT NULL DEFAULT 'v2', + ADD COLUMN IF NOT EXISTS transport_kind TEXT NOT NULL DEFAULT 'ws+msgpack', + ADD COLUMN IF NOT EXISTS harbor_root TEXT, + ADD COLUMN IF NOT EXISTS fin_seq INT, + ADD COLUMN IF NOT EXISTS ack_seq INT, + ADD COLUMN IF NOT EXISTS ack_received BOOLEAN NOT NULL DEFAULT false, + ADD COLUMN IF NOT EXISTS fin_ack_meta_json JSONB, + ADD COLUMN IF NOT EXISTS seek_stride INT NOT NULL DEFAULT 50, + ADD COLUMN IF NOT EXISTS actionable_command_count INT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS aligned_command_count INT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS target_resolved_count INT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS orphan_count INT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS target_registry_version INT NOT NULL DEFAULT 0; + +ALTER TABLE replay_v2_chunks + ADD COLUMN IF NOT EXISTS harbor_segment_path TEXT, + ADD COLUMN IF NOT EXISTS harbor_segment_index INT, + ADD COLUMN IF NOT EXISTS harbor_byte_offset BIGINT, + ADD COLUMN IF NOT EXISTS harbor_byte_length INT, + ADD COLUMN IF NOT EXISTS frame_codec TEXT NOT NULL DEFAULT 'msgpack', + ADD COLUMN IF NOT EXISTS acked BOOLEAN NOT NULL DEFAULT false, + ADD COLUMN IF NOT EXISTS ack_meta_json JSONB; + +ALTER TABLE replay_v2_events + ADD COLUMN IF NOT EXISTS command_id TEXT, + ADD COLUMN IF NOT EXISTS target_ref JSONB, + ADD COLUMN IF NOT EXISTS payload_json JSONB, + ADD COLUMN IF NOT EXISTS lifecycle_event TEXT, + ADD COLUMN IF NOT EXISTS selector_version INT, + ADD COLUMN IF NOT EXISTS dom_signature_hash TEXT, + ADD COLUMN IF NOT EXISTS asset_refs JSONB; + +CREATE TABLE IF NOT EXISTS replay_v2_target_registry ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL, + stream_id TEXT NOT NULL, + target_id TEXT NOT NULL, + selector_version INT NOT NULL, + state TEXT NOT NULL, + event_seq INT NOT NULL, + lifecycle_event TEXT NOT NULL, + selector_bundle JSONB, + metadata_json JSONB, + dom_signature_hash TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + FOREIGN KEY (run_id, stream_id) REFERENCES replay_v2_streams(run_id, stream_id) ON DELETE CASCADE, + UNIQUE (run_id, stream_id, target_id, event_seq) +); + +CREATE INDEX IF NOT EXISTS idx_replay_v2_target_registry_stream_seq + ON replay_v2_target_registry(run_id, stream_id, event_seq); + +CREATE INDEX IF NOT EXISTS idx_replay_v2_target_registry_target + ON replay_v2_target_registry(run_id, stream_id, target_id, event_seq DESC); + +CREATE TABLE IF NOT EXISTS replay_v2_seek_index ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL, + stream_id TEXT NOT NULL, + checkpoint_seq INT NOT NULL, + event_seq INT NOT NULL, + monotonic_ms INT NOT NULL, + target_registry_state_json JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + FOREIGN KEY (run_id, stream_id) REFERENCES replay_v2_streams(run_id, stream_id) ON DELETE CASCADE, + UNIQUE (run_id, stream_id, checkpoint_seq) +); + +CREATE INDEX IF NOT EXISTS idx_replay_v2_seek_index_stream_seq + ON replay_v2_seek_index(run_id, stream_id, checkpoint_seq); + +CREATE TABLE IF NOT EXISTS replay_v2_assets_cas ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL REFERENCES runs(id) ON DELETE CASCADE, + sha256 TEXT NOT NULL, + source_url TEXT, + cas_ref TEXT NOT NULL, + mime_type TEXT, + byte_size BIGINT, + blocked BOOLEAN NOT NULL DEFAULT false, + block_reason TEXT, + metadata_json JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (run_id, sha256) +); + +CREATE INDEX IF NOT EXISTS idx_replay_v2_assets_cas_run + ON replay_v2_assets_cas(run_id, created_at DESC); diff --git a/packages/cypress-reporter/src/index.js b/packages/cypress-reporter/src/index.js index 294c12c..d5dfa10 100644 --- a/packages/cypress-reporter/src/index.js +++ b/packages/cypress-reporter/src/index.js @@ -1,14 +1,19 @@ import crypto from 'node:crypto'; import fs from 'node:fs'; +import http from 'node:http'; +import path from 'node:path'; import { INGEST_EVENT_TYPES, REPLAY_V2_EVENT_KINDS, + REPLAY_V2_LIFECYCLE_EVENTS, REPLAY_V2_SCHEMA_VERSION, assertReplayV2ChunkPayload, assertReplayV2EventPayload, createReplayV2MonotonicClock, createReplayV2SequenceTracker, createReplayV2TargetRegistry, + encodeMessagePack, + decodeMessagePack, getStableReplayV2TargetId, normalizeReplayV2SelectorBundle } from '@testharbor/shared'; @@ -115,12 +120,210 @@ function extractFailure(attempt) { }; } +function ensureDir(dirPath) { + fs.mkdirSync(dirPath, { recursive: true }); +} + +function normalizeReplayPayload(input) { + return JSON.parse(JSON.stringify(input)); +} + +class HarborSegmentWriter { + constructor({ rootDir, maxBytes = 1024 * 1024 } = {}) { + this.rootDir = rootDir || path.join(process.cwd(), '.harbor', 'replay-v2'); + this.maxBytes = maxBytes; + this.segmentIndex = 0; + this.currentBytes = 0; + ensureDir(this.rootDir); + } + + appendFrame(frame) { + const payload = encodeMessagePack(normalizeReplayPayload(frame)); + const header = Buffer.allocUnsafe(4); + header.writeUInt32BE(payload.length, 0); + const segmentPath = path.join(this.rootDir, `${String(this.segmentIndex).padStart(6, '0')}.harbor`); + if (this.currentBytes + header.length + payload.length > this.maxBytes && this.currentBytes > 0) { + this.segmentIndex += 1; + this.currentBytes = 0; + return this.appendFrame(frame); + } + const byteOffset = this.currentBytes; + fs.appendFileSync(segmentPath, Buffer.concat([header, payload])); + this.currentBytes += header.length + payload.length; + return { + segmentPath, + segmentIndex: this.segmentIndex, + byteOffset, + byteLength: header.length + payload.length + }; + } +} + +function createWebSocketAccept(key) { + return crypto + .createHash('sha1') + .update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`) + .digest('base64'); +} + +function writeWebSocketFrame(socket, data) { + const payload = Buffer.from(data); + let header; + if (payload.length < 126) { + header = Buffer.from([0x82, payload.length]); + } else { + header = Buffer.allocUnsafe(4); + header[0] = 0x82; + header[1] = 126; + header.writeUInt16BE(payload.length, 2); + } + socket.write(Buffer.concat([header, payload])); +} + +function parseWebSocketFrames(buffer, onFrame) { + let offset = 0; + while (offset + 2 <= buffer.length) { + const first = buffer[offset]; + const second = buffer[offset + 1]; + const opcode = first & 0x0f; + const masked = (second & 0x80) === 0x80; + let length = second & 0x7f; + let cursor = offset + 2; + if (length === 126) { + if (cursor + 2 > buffer.length) break; + length = buffer.readUInt16BE(cursor); + cursor += 2; + } + if (masked) { + if (cursor + 4 > buffer.length) break; + } + const mask = masked ? buffer.subarray(cursor, cursor + 4) : null; + if (masked) cursor += 4; + if (cursor + length > buffer.length) break; + const payload = Buffer.from(buffer.subarray(cursor, cursor + length)); + if (mask) { + for (let index = 0; index < payload.length; index += 1) { + payload[index] ^= mask[index % 4]; + } + } + onFrame({ opcode, payload }); + offset = cursor + length; + } + return buffer.subarray(offset); +} + +function decodeTransportMessage(opcode, payload) { + if (!Buffer.isBuffer(payload) || payload.length === 0) return null; + try { + if (opcode === 0x2) return decodeMessagePack(payload); + if (opcode === 0x1) return JSON.parse(payload.toString('utf8')); + } catch { + try { + return JSON.parse(payload.toString('utf8')); + } catch { + return null; + } + } + return null; +} + +class ReplayTransportServer { + constructor({ port = 9223 } = {}) { + this.port = port; + this.server = null; + this.clients = new Set(); + this.pendingFin = new Map(); + } + + start() { + if (this.server) return; + this.server = http.createServer((_req, res) => { + res.writeHead(426, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: false, error: 'upgrade_required' })); + }); + this.server.on('upgrade', (request, socket) => { + const key = request.headers['sec-websocket-key']; + if (!key) { + socket.destroy(); + return; + } + socket.write([ + 'HTTP/1.1 101 Switching Protocols', + 'Upgrade: websocket', + 'Connection: Upgrade', + `Sec-WebSocket-Accept: ${createWebSocketAccept(key)}`, + '', + '' + ].join('\r\n')); + + socket._thBuffer = Buffer.alloc(0); + this.clients.add(socket); + socket.on('data', (chunk) => { + socket._thBuffer = parseWebSocketFrames(Buffer.concat([socket._thBuffer, chunk]), ({ opcode, payload }) => { + if (opcode === 0x8) { + socket.end(); + return; + } + if (opcode !== 0x2 && opcode !== 0x1) return; + const message = decodeTransportMessage(opcode, payload); + if (message?.type === 'TRANSPORT_ACK' && message.finId) { + this.acknowledgeFin(message.finId, { clientAck: true, ts: new Date().toISOString() }); + } + }); + }); + socket.on('close', () => this.clients.delete(socket)); + socket.on('error', () => this.clients.delete(socket)); + }); + this.server.listen(this.port, '0.0.0.0'); + } + + broadcast(frame) { + const payload = encodeMessagePack(frame); + for (const client of this.clients) { + writeWebSocketFrame(client, payload); + } + } + + requestFinAck(finId, meta = {}) { + return new Promise((resolve) => { + const timeout = setTimeout(() => { + const pending = this.pendingFin.get(finId); + if (!pending) return; + clearTimeout(pending.timeout); + this.pendingFin.delete(finId); + pending.resolve({ + ok: false, + finId, + timeoutFallback: true, + timeoutMs: 250, + ...meta + }); + }, 250); + this.pendingFin.set(finId, { resolve, timeout }); + this.broadcast({ type: 'TRANSPORT_FIN', finId, meta }); + }); + } + + acknowledgeFin(finId, meta = {}) { + const pending = this.pendingFin.get(finId); + if (!pending) return; + clearTimeout(pending.timeout); + this.pendingFin.delete(finId); + this.broadcast({ type: 'TRANSPORT_ACK', finId, meta }); + pending.resolve({ ok: true, finId, ...meta }); + } +} + export class TestHarborReporterClient { - constructor({ ingestUrl, token = null, maxRetries = 3, replayChunkSize } = {}) { + constructor({ ingestUrl, token = null, maxRetries = 3, replayChunkSize, replayTransportPort = 9223, harborRoot = null } = {}) { this.ingestUrl = ingestUrl || process.env.TESTHARBOR_INGEST_URL || 'http://localhost:4010/v1/ingest/events'; this.token = token || process.env.TESTHARBOR_INGEST_TOKEN || null; this.maxRetries = maxRetries; this.replayChunkSize = Number(process.env.TESTHARBOR_REPLAY_CHUNK_SIZE || replayChunkSize || 100); + this.replayTransportPort = Number(process.env.TESTHARBOR_REPLAY_WS_PORT || replayTransportPort || 9223); + this.harborRoot = harborRoot || process.env.TESTHARBOR_REPLAY_HARBOR_ROOT || path.join(process.cwd(), '.harbor', 'replay-v2'); + this.transportServer = new ReplayTransportServer({ port: this.replayTransportPort }); + this.transportServer.start(); this.replayV2 = null; } @@ -174,13 +377,31 @@ export class TestHarborReporterClient { eventSequence: createReplayV2SequenceTracker(), chunkSequence: createReplayV2SequenceTracker(), targetRegistry: createReplayV2TargetRegistry(), + harborWriter: new HarborSegmentWriter({ + rootDir: path.join(this.harborRoot, runId, streamId) + }), pendingEvents: [], chunkCount: 0 }; - return this.queueReplayEvent({ - kind: REPLAY_V2_EVENT_KINDS.SESSION_START, - data: { metadata } + this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.SESSION_START, { metadata }, { flushIfNeeded: false }); + this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.CAPTURE_COMMAND, { + order: 1, + targetSnapshotsAtCommandBoundaries: true + }, { flushIfNeeded: false }); + this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.CAPTURE_RRWEB, { + order: 2, + recordShadowDom: true, + inlineStylesheet: true + }, { flushIfNeeded: false }); + this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.CAPTURE_CDP, { + order: 3, + autoAttach: true, + domains: ['Target', 'Network', 'Console', 'Runtime'] + }, { flushIfNeeded: false }); + return this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.CAPTURE_SCREENCAST_DEFERRED, { + order: 4, + deferred: true }); } @@ -188,71 +409,160 @@ export class TestHarborReporterClient { const replay = this.#requireReplayV2Session(); const resolvedTargetId = targetId || getStableReplayV2TargetId({ selectors, framePath, name, kind }); const selectorBundle = normalizeReplayV2SelectorBundle({ ...selectors, framePath }); - replay.targetRegistry.declare({ targetId: resolvedTargetId, selectors: selectorBundle, framePath, metadata }); - return this.queueReplayEvent({ - kind: REPLAY_V2_EVENT_KINDS.TARGET_DECLARED, - targetId: resolvedTargetId, - selectorBundle, - data: { framePath, metadata, name, kind } + replay.targetRegistry.declare({ targetId: resolvedTargetId, selectorBundle, metadata }); + this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE, { + metadata: { framePath, metadata, name, kind }, + selectorBundle + }, { + targetRef: { targetId: resolvedTargetId, selectorVersion: 1 }, + flushIfNeeded: false + }); + replay.targetRegistry.bind({ targetId: resolvedTargetId, selectorBundle, metadata }); + return this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TARGET_BIND, { + metadata: { framePath, metadata, name, kind }, + selectorBundle + }, { + targetRef: { targetId: resolvedTargetId, selectorVersion: 1 } + }); + } + + bindReplayTarget({ targetId, selectors = {}, framePath = null, metadata = null } = {}) { + const replay = this.#requireReplayV2Session(); + const selectorBundle = normalizeReplayV2SelectorBundle({ ...selectors, framePath }); + const bound = replay.targetRegistry.bind({ targetId, selectorBundle, metadata }); + return this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TARGET_BIND, { + metadata: { framePath, metadata }, + selectorBundle + }, { + targetRef: { targetId, selectorVersion: bound.selectorVersion } }); } rebindReplayTarget({ targetId, selectors = {}, framePath = null, metadata = null } = {}) { const replay = this.#requireReplayV2Session(); const selectorBundle = normalizeReplayV2SelectorBundle({ ...selectors, framePath }); - replay.targetRegistry.rebind({ targetId, selectors: selectorBundle, framePath, metadata }); - return this.queueReplayEvent({ - kind: REPLAY_V2_EVENT_KINDS.TARGET_REBOUND, - targetId, - selectorBundle, - data: { framePath, metadata } + const rebound = replay.targetRegistry.rebind({ targetId, selectorBundle, metadata }); + return this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TARGET_REBIND, { + metadata: { framePath, metadata }, + selectorBundle + }, { + targetRef: { targetId, selectorVersion: rebound.selectorVersion } }); } markReplayTargetOrphan({ targetId, reason = null } = {}) { const replay = this.#requireReplayV2Session(); - replay.targetRegistry.orphan({ targetId, reason }); - return this.queueReplayEvent({ - kind: REPLAY_V2_EVENT_KINDS.TARGET_ORPHANED, - targetId, - data: { reason } + const orphaned = replay.targetRegistry.orphan({ targetId, reason }); + return this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TARGET_ORPHAN, { + reason + }, { + targetRef: { targetId, selectorVersion: orphaned.selectorVersion } }); } + queueReplayLifecycle(eventType, payload = {}, options = {}) { + const { flushIfNeeded = true, ...eventOptions } = options; + return this.queueReplayEvent({ + kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, + payload: { + eventType, + ...payload + }, + ...eventOptions + }, { flushIfNeeded }); + } + + queueCommandEvent({ commandId, targetId = null, selectors = null, payload = {} } = {}, options = {}) { + const targetRef = targetId + ? { + targetId, + selectorVersion: this.replayV2?.targetRegistry?.get(targetId)?.selectorVersion || 1 + } + : null; + const targetSnapshot = targetId ? this.replayV2?.targetRegistry?.get(targetId) || null : null; + return this.queueReplayEvent({ + kind: REPLAY_V2_EVENT_KINDS.COMMAND, + commandId: commandId || crypto.randomUUID(), + targetRef, + payload: { + ...payload, + selectorBundle: selectors ? normalizeReplayV2SelectorBundle(selectors) : targetSnapshot?.selectorBundle || null, + targetSnapshot + } + }, options); + } + + queueDomEvent(payload = {}, options = {}) { + return this.queueReplayEvent({ + kind: REPLAY_V2_EVENT_KINDS.DOM, + payload + }, options); + } + + queueNetworkEvent(payload = {}, options = {}) { + return this.queueReplayEvent({ + kind: REPLAY_V2_EVENT_KINDS.NETWORK, + payload + }, options); + } + + queueConsoleEvent(payload = {}, options = {}) { + return this.queueReplayEvent({ + kind: REPLAY_V2_EVENT_KINDS.CONSOLE, + payload + }, options); + } + async queueReplayEvent(event = {}, { flushIfNeeded = true } = {}) { const replay = this.#requireReplayV2Session(); - const monotonicMs = replay.clock.now(); + const monotonicTs = replay.clock.now(); const seq = replay.eventSequence.assign(); - const ts = new Date(Date.parse(replay.startedAt) + monotonicMs).toISOString(); + const ts = new Date(Date.parse(replay.startedAt) + monotonicTs).toISOString(); + const normalizedSelectorBundle = event.payload?.selectorBundle + ? normalizeReplayV2SelectorBundle(event.payload.selectorBundle) + : null; const normalizedEvent = { schemaVersion: REPLAY_V2_SCHEMA_VERSION, runId: replay.runId, streamId: replay.streamId, seq, - monotonicMs, + monotonicTs, ts, - ...event + ...event, + targetRef: event.targetRef || null, + payload: { + ...(event.payload || {}), + ...(normalizedSelectorBundle ? { selectorBundle: normalizedSelectorBundle } : {}) + } }; - if (normalizedEvent.selectorBundle != null) { - normalizedEvent.selectorBundle = normalizeReplayV2SelectorBundle(normalizedEvent.selectorBundle); - } - if ( - normalizedEvent.targetId && - normalizedEvent.kind !== REPLAY_V2_EVENT_KINDS.TARGET_DECLARED && - normalizedEvent.kind !== REPLAY_V2_EVENT_KINDS.TARGET_ORPHANED - ) { - replay.targetRegistry.assertUsable(normalizedEvent.targetId); + if (normalizedEvent.targetRef?.targetId) { + const lifecycleType = normalizedEvent.payload?.eventType; + if ( + normalizedEvent.kind !== REPLAY_V2_EVENT_KINDS.LIFECYCLE + || ![ + REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE, + REPLAY_V2_LIFECYCLE_EVENTS.TARGET_ORPHAN + ].includes(lifecycleType) + ) { + replay.targetRegistry.assertUsable(normalizedEvent.targetRef.targetId); + } } - assertReplayV2EventPayload(normalizedEvent); - replay.pendingEvents.push(normalizedEvent); + const assertedEvent = assertReplayV2EventPayload(normalizedEvent); + replay.pendingEvents.push(assertedEvent); + this.transportServer.broadcast({ + type: 'event', + streamId: replay.streamId, + seq: assertedEvent.seq, + kind: assertedEvent.kind + }); if (flushIfNeeded && replay.pendingEvents.length >= this.#getReplayChunkSize()) { return this.flushReplayV2Chunk(); } - return normalizedEvent; + return assertedEvent; } async flushReplayV2Chunk({ final = false } = {}) { @@ -273,9 +583,29 @@ export class TestHarborReporterClient { final, chunkIndex: replay.chunkCount, startedAt: replay.startedAt, + seekStride: 50, + transport: { + kind: 'ws+msgpack', + codec: 'msgpack', + harborRoot: path.join(this.harborRoot, replay.runId, replay.streamId) + }, events: replay.pendingEvents }; + const segmentMeta = replay.harborWriter.appendFrame({ + type: 'replay.v2.chunk', + runId: replay.runId, + streamId: replay.streamId, + seqStart, + seqEnd, + final, + events: replay.pendingEvents + }); + payload.transport = { + ...payload.transport, + ...segmentMeta + }; + assertReplayV2ChunkPayload(payload); let result; try { @@ -294,13 +624,35 @@ export class TestHarborReporterClient { async endReplayV2({ status = 'completed', metadata = null } = {}) { const replay = this.#requireReplayV2Session(); - await this.queueReplayEvent({ - kind: REPLAY_V2_EVENT_KINDS.SESSION_END, - data: { status, metadata } + const finId = crypto.randomUUID(); + await this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_FIN, { + finId, + status, + metadata + }, { flushIfNeeded: false }); + await this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.SESSION_END, { + status, + metadata }, { flushIfNeeded: false }); const result = await this.flushReplayV2Chunk({ final: true }); + const ack = await this.transportServer.requestFinAck(finId, { + runId: replay.runId, + streamId: replay.streamId + }); + + let ackSegment = null; + let ackResult = null; + if (ack?.ok) { + await this.queueReplayLifecycle(REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_ACK, { + finId, + ack + }, { flushIfNeeded: false }); + ackSegment = replay.harborWriter.appendFrame({ type: 'TRANSPORT_ACK', finId, ack }); + ackResult = await this.flushReplayV2Chunk({ final: true }); + } + this.replayV2 = null; - return result; + return { ...result, ack, ackResult, ackSegment }; } #requireReplayV2Session() { @@ -374,7 +726,9 @@ export function setupTestHarbor(on, config, options = {}) { const client = new TestHarborReporterClient({ ingestUrl, token, - maxRetries: toNumber(options.maxRetries, 3) + maxRetries: toNumber(options.maxRetries, 3), + replayTransportPort: toNumber(options.replayTransportPort || config?.env?.TESTHARBOR_REPLAY_WS_PORT, 9223), + harborRoot: asTrimmedString(options.harborRoot || config?.env?.TESTHARBOR_REPLAY_HARBOR_ROOT) }); const specRunIds = new Map(); @@ -400,6 +754,39 @@ export function setupTestHarbor(on, config, options = {}) { 'testharbor:log'(entry) { // Keep API stable for tests that want to emit custom logs through cy.task(). return entry || null; + }, + async 'testharbor:replay:event'(entry) { + return client.queueReplayEvent(entry || {}); + }, + async 'testharbor:replay:command'(entry) { + return client.queueCommandEvent(entry || {}); + }, + async 'testharbor:replay:dom'(entry) { + return client.queueDomEvent(entry || {}); + }, + async 'testharbor:replay:network'(entry) { + return client.queueNetworkEvent(entry || {}); + }, + async 'testharbor:replay:console'(entry) { + return client.queueConsoleEvent(entry || {}); + }, + async 'testharbor:replay:target:declare'(entry) { + return client.declareReplayTarget(entry || {}); + }, + async 'testharbor:replay:target:bind'(entry) { + return client.bindReplayTarget(entry || {}); + }, + async 'testharbor:replay:target:rebind'(entry) { + return client.rebindReplayTarget(entry || {}); + }, + async 'testharbor:replay:target:orphan'(entry) { + return client.markReplayTargetOrphan(entry || {}); + }, + async 'testharbor:replay:flush'() { + return client.flushReplayV2Chunk(); + }, + async 'testharbor:replay:fin'(entry) { + return client.endReplayV2(entry || {}); } }); @@ -423,6 +810,20 @@ export function setupTestHarbor(on, config, options = {}) { specRunIds.set(specPath, specRunId); runMetrics.totalSpecs += 1; + client.startReplayV2({ + runId, + streamId: specRunId, + metadata: { + specPath, + transportPort: client.replayTransportPort + } + }); + await client.queueDomEvent({ + eventType: 'SPEC_BOUNDARY', + phase: 'before:spec', + specPath + }, { flushIfNeeded: false }); + await sendSafe(INGEST_EVENT_TYPES.SPEC_STARTED, { specRunId, runId, @@ -538,6 +939,22 @@ export function setupTestHarbor(on, config, options = {}) { }); } + if (client.replayV2) { + await client.queueDomEvent({ + eventType: 'SPEC_BOUNDARY', + phase: 'after:spec', + specPath + }, { flushIfNeeded: false }); + await client.endReplayV2({ + status: specStatusFromSummary(results, runMetrics.failCount), + metadata: { + specPath, + screenshots: screenshots.length, + video: Boolean(results?.video) + } + }); + } + await sendSafe(INGEST_EVENT_TYPES.SPEC_FINISHED, { specRunId, status: specStatusFromSummary(results, runMetrics.failCount), @@ -548,6 +965,12 @@ export function setupTestHarbor(on, config, options = {}) { }); on('after:run', async (results) => { + if (client.replayV2) { + await client.endReplayV2({ + status: runStatusFromSummary(results, runMetrics.failCount), + metadata: { forcedClose: true } + }); + } const totalSpecs = toNumber(results?.totalSuites, runMetrics.totalSpecs || specRunIds.size); const totalTests = toNumber(results?.totalTests, runMetrics.totalTests); const passCount = toNumber(results?.totalPassed, runMetrics.passCount); diff --git a/packages/shared/src/index.js b/packages/shared/src/index.js index 62a4460..616ad0d 100644 --- a/packages/shared/src/index.js +++ b/packages/shared/src/index.js @@ -15,12 +15,23 @@ export function isValidIngestType(type) { export { REPLAY_V2_SCHEMA_VERSION, + REPLAY_V2_SCHEMA_VERSION_COMPAT, + REPLAY_V2_SEEK_STRIDE, + REPLAY_V2_TARGET_RESOLUTION_ORDER, REPLAY_V2_EVENT_KINDS, + REPLAY_V2_LIFECYCLE_EVENTS, normalizeReplayV2SelectorBundle, getStableReplayV2TargetId, createReplayV2MonotonicClock, createReplayV2SequenceTracker, createReplayV2TargetRegistry, + normalizeReplayV2EventPayload, assertReplayV2EventPayload, - assertReplayV2ChunkPayload + assertReplayV2ChunkPayload, + applyReplayV2EventToTargetRegistry, + buildReplayV2SeekIndex, + resolveReplayV2TargetStateAtSeq, + evaluateReplayV2GateMetrics, + encodeMessagePack, + decodeMessagePack } from './replay-v2.js'; diff --git a/packages/shared/src/replay-v2.js b/packages/shared/src/replay-v2.js index f9279ba..7b9870a 100644 --- a/packages/shared/src/replay-v2.js +++ b/packages/shared/src/replay-v2.js @@ -1,42 +1,58 @@ import crypto from 'node:crypto'; import { performance } from 'node:perf_hooks'; -export const REPLAY_V2_SCHEMA_VERSION = '2.0'; +export const REPLAY_V2_SCHEMA_VERSION = '2.1'; +export const REPLAY_V2_SCHEMA_VERSION_COMPAT = new Set(['2.0', '2.1']); +export const REPLAY_V2_SEEK_STRIDE = 50; +export const REPLAY_V2_TARGET_RESOLUTION_ORDER = [ + 'test-id', + 'accessibility', + 'structural-css', + 'text-proximity' +]; export const REPLAY_V2_EVENT_KINDS = { - SESSION_START: 'session.start', - SESSION_END: 'session.end', - TARGET_DECLARED: 'target.declared', - TARGET_REBOUND: 'target.rebound', - TARGET_ORPHANED: 'target.orphaned', - DOM_SNAPSHOT: 'dom.snapshot', - DOM_MUTATION: 'dom.mutation', - POINTER: 'pointer', - KEYBOARD: 'keyboard', - INPUT: 'input', - SCROLL: 'scroll', - VIEWPORT: 'viewport', - NAVIGATION: 'navigation', - ASSERTION: 'assertion', - LOG: 'log', - CUSTOM: 'custom' + COMMAND: 'command', + DOM: 'dom', + NETWORK: 'network', + CONSOLE: 'console', + LIFECYCLE: 'lifecycle' +}; + +export const REPLAY_V2_LIFECYCLE_EVENTS = { + SESSION_START: 'SESSION_START', + SESSION_END: 'SESSION_END', + TARGET_DECLARE: 'TARGET_DECLARE', + TARGET_BIND: 'TARGET_BIND', + TARGET_REBIND: 'TARGET_REBIND', + TARGET_ORPHAN: 'TARGET_ORPHAN', + CAPTURE_COMMAND: 'CAPTURE_COMMAND', + CAPTURE_RRWEB: 'CAPTURE_RRWEB', + CAPTURE_CDP: 'CAPTURE_CDP', + CAPTURE_SCREENCAST_DEFERRED: 'CAPTURE_SCREENCAST_DEFERRED', + TRANSPORT_FIN: 'TRANSPORT_FIN', + TRANSPORT_ACK: 'TRANSPORT_ACK' }; const REPLAY_V2_EVENT_KIND_SET = new Set(Object.values(REPLAY_V2_EVENT_KINDS)); -const REPLAY_V2_SELECTOR_KEYS = [ - 'css', - 'xpath', - 'text', - 'testId', - 'role', - 'label', - 'placeholder', - 'altText', - 'title', - 'name', - 'value', - 'framePath' -]; +const LEGACY_EVENT_KIND_TO_V2 = { + 'session.start': { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: REPLAY_V2_LIFECYCLE_EVENTS.SESSION_START }, + 'session.end': { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: REPLAY_V2_LIFECYCLE_EVENTS.SESSION_END }, + 'target.declared': { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE }, + 'target.rebound': { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: REPLAY_V2_LIFECYCLE_EVENTS.TARGET_REBIND }, + 'target.orphaned': { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: REPLAY_V2_LIFECYCLE_EVENTS.TARGET_ORPHAN }, + 'dom.snapshot': { kind: REPLAY_V2_EVENT_KINDS.DOM, eventType: 'SNAPSHOT' }, + 'dom.mutation': { kind: REPLAY_V2_EVENT_KINDS.DOM, eventType: 'MUTATION' }, + pointer: { kind: REPLAY_V2_EVENT_KINDS.COMMAND, eventType: 'POINTER' }, + keyboard: { kind: REPLAY_V2_EVENT_KINDS.COMMAND, eventType: 'KEYBOARD' }, + input: { kind: REPLAY_V2_EVENT_KINDS.COMMAND, eventType: 'INPUT' }, + scroll: { kind: REPLAY_V2_EVENT_KINDS.COMMAND, eventType: 'SCROLL' }, + viewport: { kind: REPLAY_V2_EVENT_KINDS.DOM, eventType: 'VIEWPORT' }, + navigation: { kind: REPLAY_V2_EVENT_KINDS.DOM, eventType: 'NAVIGATION' }, + assertion: { kind: REPLAY_V2_EVENT_KINDS.COMMAND, eventType: 'ASSERTION' }, + log: { kind: REPLAY_V2_EVENT_KINDS.CONSOLE, eventType: 'LOG' }, + custom: { kind: REPLAY_V2_EVENT_KINDS.LIFECYCLE, eventType: 'CUSTOM' } +}; function isPlainObject(value) { return Boolean(value) && typeof value === 'object' && !Array.isArray(value); @@ -51,15 +67,14 @@ function assert(condition, message, details = {}) { } function canonicalizeJson(value) { - if (Array.isArray(value)) { - return value.map((item) => canonicalizeJson(item)); - } + if (Array.isArray(value)) return value.map((item) => canonicalizeJson(item)); if (isPlainObject(value)) { return Object.keys(value) .sort() .reduce((acc, key) => { - acc[key] = canonicalizeJson(value[key]); + const normalized = canonicalizeJson(value[key]); + if (normalized !== undefined) acc[key] = normalized; return acc; }, {}); } @@ -83,32 +98,112 @@ function normalizeSelectorValue(value) { return normalizeScalarSelectorValue(value); } +function normalizeStringArray(value) { + const normalized = normalizeSelectorValue(value); + if (normalized == null) return []; + return Array.isArray(normalized) ? normalized : [normalized]; +} + +function optionalJson(value) { + return value == null ? null : canonicalizeJson(value); +} + +function hashJson(value) { + return crypto.createHash('sha256').update(JSON.stringify(canonicalizeJson(value))).digest('hex'); +} + +function normalizeFrameOrShadowPath(value) { + if (Array.isArray(value)) return normalizeStringArray(value); + return normalizeStringArray(value); +} + +function normalizeDomSignature(bundle = {}) { + const normalized = canonicalizeJson({ + tag: normalizeScalarSelectorValue(bundle.tag || bundle.tagName), + keyAttrs: isPlainObject(bundle.keyAttrs) ? canonicalizeJson(bundle.keyAttrs) : null, + relativePosition: normalizeScalarSelectorValue(bundle.relativePosition), + hash: normalizeScalarSelectorValue(bundle.hash) + }); + + if (normalized.hash) return normalized; + if (!normalized.tag && !normalized.keyAttrs && !normalized.relativePosition) return null; + + return { + ...normalized, + hash: hashJson({ + tag: normalized.tag || null, + keyAttrs: normalized.keyAttrs || null, + relativePosition: normalized.relativePosition || null + }) + }; +} + export function normalizeReplayV2SelectorBundle(bundle = {}) { assert(isPlainObject(bundle), 'replay_v2_selector_bundle_invalid', { bundle }); - const normalized = {}; - for (const key of REPLAY_V2_SELECTOR_KEYS) { - const value = normalizeSelectorValue(bundle[key]); - if (value !== null) normalized[key] = value; - } + const primary = canonicalizeJson({ + dataCy: normalizeSelectorValue(bundle?.primary?.dataCy ?? bundle.dataCy), + dataTestId: normalizeSelectorValue(bundle?.primary?.dataTestId ?? bundle.dataTestId ?? bundle.testId), + appId: normalizeSelectorValue(bundle?.primary?.appId ?? bundle.appId), + stableId: normalizeSelectorValue(bundle?.primary?.stableId ?? bundle.stableId) + }); - if (Number.isInteger(bundle.nth) && bundle.nth >= 0) { - normalized.nth = bundle.nth; - } + const accessibility = canonicalizeJson({ + role: normalizeSelectorValue(bundle?.accessibility?.role ?? bundle.role), + name: normalizeSelectorValue(bundle?.accessibility?.name ?? bundle.name), + label: normalizeSelectorValue(bundle?.accessibility?.label ?? bundle.label), + ariaPath: normalizeSelectorValue(bundle?.accessibility?.ariaPath ?? bundle.ariaPath ?? bundle.xpath) + }); + + const structural = canonicalizeJson({ + cssPath: normalizeSelectorValue(bundle?.structural?.cssPath ?? bundle.cssPath ?? bundle.css), + xpath: normalizeSelectorValue(bundle?.structural?.xpath ?? bundle.xpath), + nth: Number.isInteger(bundle?.structural?.nth ?? bundle.nth) && (bundle?.structural?.nth ?? bundle.nth) >= 0 + ? (bundle?.structural?.nth ?? bundle.nth) + : null + }); + + const text = canonicalizeJson({ + text: normalizeSelectorValue(bundle?.text?.text ?? bundle.text), + proximity: normalizeSelectorValue(bundle?.text?.proximity ?? bundle.proximity), + nearText: normalizeSelectorValue(bundle?.text?.nearText ?? bundle.nearText) + }); + + const context = canonicalizeJson({ + framePath: normalizeFrameOrShadowPath(bundle?.context?.framePath ?? bundle.framePath), + shadowPath: normalizeFrameOrShadowPath(bundle?.context?.shadowPath ?? bundle.shadowPath), + parentFingerprint: normalizeSelectorValue(bundle?.context?.parentFingerprint ?? bundle.parentFingerprint), + siblingFingerprint: normalizeSelectorValue(bundle?.context?.siblingFingerprint ?? bundle.siblingFingerprint) + }); + + const domSignature = normalizeDomSignature(bundle?.domSignature || bundle); + + const normalized = canonicalizeJson({ + resolutionOrder: REPLAY_V2_TARGET_RESOLUTION_ORDER, + primary: Object.values(primary).some(Boolean) ? primary : null, + accessibility: Object.values(accessibility).some(Boolean) ? accessibility : null, + structural: Object.values(structural).some((value) => value != null && value !== '') ? structural : null, + text: Object.values(text).some(Boolean) ? text : null, + context: ( + context.framePath.length + || context.shadowPath.length + || context.parentFingerprint + || context.siblingFingerprint + ) ? context : null, + domSignature + }); - return canonicalizeJson(normalized); + return normalized; } export function getStableReplayV2TargetId(input = {}) { - const normalizedSelectors = normalizeReplayV2SelectorBundle(input.selectors || input.selectorBundle || {}); + const selectorBundle = normalizeReplayV2SelectorBundle(input.selectors || input.selectorBundle || {}); const normalizedIdentity = canonicalizeJson({ - framePath: normalizeSelectorValue(input.framePath) ?? null, - kind: normalizeScalarSelectorValue(input.kind) ?? null, - name: normalizeScalarSelectorValue(input.name) ?? null, - selectors: normalizedSelectors + kind: normalizeScalarSelectorValue(input.kind), + name: normalizeScalarSelectorValue(input.name), + selectorBundle }); - const digest = crypto.createHash('sha256').update(JSON.stringify(normalizedIdentity)).digest('hex'); - return `rv2_tgt_${digest.slice(0, 20)}`; + return `tgt_${hashJson(normalizedIdentity).slice(0, 20)}`; } export function createReplayV2MonotonicClock({ startedAt = new Date().toISOString() } = {}) { @@ -166,8 +261,31 @@ export function createReplayV2SequenceTracker({ initialSeq = 1, previousSeq = 0 }; } -export function createReplayV2TargetRegistry() { +function cloneTargetRecord(record) { + return record ? JSON.parse(JSON.stringify(record)) : null; +} + +function createTargetHistoryEntry(type, seq, record, extra = {}) { + return { + seq, + type, + targetId: record.targetId, + selectorVersion: record.selectorVersion, + state: record.state, + selectorBundle: cloneTargetRecord(record.selectorBundle), + metadata: optionalJson(record.metadata), + reason: record.reason || null, + ...extra + }; +} + +function compareSelectorBundles(a, b) { + return JSON.stringify(canonicalizeJson(a || null)) === JSON.stringify(canonicalizeJson(b || null)); +} + +export function createReplayV2TargetRegistry({ initialState = [] } = {}) { const targets = new Map(); + const history = []; function requireTarget(targetId) { const target = targets.get(targetId); @@ -175,53 +293,127 @@ export function createReplayV2TargetRegistry() { return target; } + function writeRecord(record, historyType, seq, extra = {}) { + targets.set(record.targetId, record); + if (Number.isInteger(seq) && seq > 0) { + history.push(createTargetHistoryEntry(historyType, seq, record, extra)); + } + return cloneTargetRecord(record); + } + + for (const item of initialState) { + if (!item?.targetId) continue; + targets.set(item.targetId, cloneTargetRecord(item)); + } + return { - declare({ targetId, selectors = {}, framePath = null, metadata = null } = {}) { + declare({ targetId, selectorBundle = {}, metadata = null, seq = null } = {}) { assert(typeof targetId === 'string' && targetId.length > 0, 'replay_v2_target_id_invalid', { targetId }); - const normalizedSelectors = normalizeReplayV2SelectorBundle(selectors); + const current = targets.get(targetId); const record = { targetId, - selectors: normalizedSelectors, - framePath: normalizeSelectorValue(framePath), - metadata: metadata ?? null, - state: 'active' + selectorVersion: current?.selectorVersion ?? 1, + selectorBundle: current?.selectorBundle ?? normalizeReplayV2SelectorBundle(selectorBundle), + metadata: metadata ?? current?.metadata ?? null, + state: 'declared', + reason: null }; - targets.set(targetId, record); - return record; + return writeRecord(record, REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE, seq); }, - rebind({ targetId, selectors = {}, framePath = null, metadata = null } = {}) { + bind({ targetId, selectorBundle = {}, metadata = null, seq = null } = {}) { const current = requireTarget(targetId); - const updated = { + const normalizedSelectorBundle = normalizeReplayV2SelectorBundle(selectorBundle); + const selectorVersion = current.selectorVersion || 1; + const record = { ...current, - selectors: normalizeReplayV2SelectorBundle(selectors), - framePath: normalizeSelectorValue(framePath), + selectorVersion, + selectorBundle: normalizedSelectorBundle, metadata: metadata ?? current.metadata ?? null, - state: 'active' + state: 'active', + reason: null }; - targets.set(targetId, updated); - return updated; + return writeRecord(record, REPLAY_V2_LIFECYCLE_EVENTS.TARGET_BIND, seq); }, - orphan({ targetId, reason = null } = {}) { + rebind({ targetId, selectorBundle = {}, metadata = null, seq = null } = {}) { const current = requireTarget(targetId); - const updated = { + const normalizedSelectorBundle = normalizeReplayV2SelectorBundle(selectorBundle); + const selectorVersion = compareSelectorBundles(current.selectorBundle, normalizedSelectorBundle) + ? current.selectorVersion + : (current.selectorVersion || 1) + 1; + const record = { + ...current, + selectorVersion, + selectorBundle: normalizedSelectorBundle, + metadata: metadata ?? current.metadata ?? null, + state: 'active', + reason: null + }; + return writeRecord(record, REPLAY_V2_LIFECYCLE_EVENTS.TARGET_REBIND, seq, { + changed: selectorVersion !== current.selectorVersion + }); + }, + orphan({ targetId, reason = null, seq = null } = {}) { + const current = requireTarget(targetId); + const record = { ...current, state: 'orphaned', - orphanedReason: normalizeScalarSelectorValue(reason) + reason: normalizeScalarSelectorValue(reason) }; - targets.set(targetId, updated); - return updated; + return writeRecord(record, REPLAY_V2_LIFECYCLE_EVENTS.TARGET_ORPHAN, seq); }, assertUsable(targetId) { const current = requireTarget(targetId); - assert(current.state === 'active', 'replay_v2_target_orphaned', { targetId }); - return current; + assert(current.state === 'active' || current.state === 'declared', 'replay_v2_target_orphaned', { targetId }); + return cloneTargetRecord(current); }, get(targetId) { - return targets.get(targetId) || null; + return cloneTargetRecord(targets.get(targetId) || null); + }, + snapshot() { + return [...targets.values()].map((record) => cloneTargetRecord(record)); + }, + history() { + return history.map((entry) => optionalJson(entry)); + }, + resolveAtSeq(seq) { + const state = new Map(); + for (const entry of history) { + if (entry.seq > seq) break; + state.set(entry.targetId, { + targetId: entry.targetId, + selectorVersion: entry.selectorVersion, + selectorBundle: cloneTargetRecord(entry.selectorBundle), + metadata: optionalJson(entry.metadata), + state: entry.state, + reason: entry.reason || null + }); + } + return [...state.values()]; } }; } +function normalizeLegacyEventShape(event) { + const legacy = LEGACY_EVENT_KIND_TO_V2[event.kind]; + if (!legacy) return event; + + const payload = isPlainObject(event.data) ? { ...event.data } : { value: event.data ?? null }; + if (legacy.eventType) payload.eventType = legacy.eventType; + if (event.selectorBundle != null && payload.selectorBundle == null) { + payload.selectorBundle = normalizeReplayV2SelectorBundle(event.selectorBundle); + } + + return { + ...event, + kind: legacy.kind, + payload, + targetRef: event.targetId ? { + targetId: event.targetId, + selectorVersion: Number.isInteger(payload.selectorVersion) && payload.selectorVersion >= 1 ? payload.selectorVersion : 1 + } : event.targetRef + }; +} + function assertString(value, message, details) { assert(typeof value === 'string' && value.length > 0, message, details); } @@ -231,73 +423,134 @@ function assertOptionalString(value, message, details) { assert(typeof value === 'string' && value.length > 0, message, details); } -export function assertReplayV2EventPayload(event) { +export function normalizeReplayV2EventPayload(event = {}) { assert(isPlainObject(event), 'replay_v2_event_invalid', { event }); - assertString(event.kind, 'replay_v2_event_kind_invalid', { kind: event.kind }); - assert(REPLAY_V2_EVENT_KIND_SET.has(event.kind), 'replay_v2_event_kind_unsupported', { kind: event.kind }); - assertString(event.runId, 'replay_v2_event_run_id_missing', { event }); - assertString(event.streamId, 'replay_v2_event_stream_id_missing', { event }); - assert(Number.isInteger(event.seq) && event.seq >= 1, 'replay_v2_event_seq_invalid', { seq: event.seq }); - assert(Number.isInteger(event.monotonicMs) && event.monotonicMs >= 0, 'replay_v2_event_monotonic_invalid', { - monotonicMs: event.monotonicMs + const normalizedInput = normalizeLegacyEventShape({ ...event }); + + const monotonicTs = Number.isInteger(normalizedInput.monotonicTs) + ? normalizedInput.monotonicTs + : normalizedInput.monotonicMs; + const payload = isPlainObject(normalizedInput.payload) + ? { ...normalizedInput.payload } + : isPlainObject(normalizedInput.data) + ? { ...normalizedInput.data } + : {}; + + const selectorVersion = Number.isInteger(normalizedInput.targetRef?.selectorVersion) + ? normalizedInput.targetRef.selectorVersion + : Number.isInteger(payload.selectorVersion) + ? payload.selectorVersion + : 1; + + const selectorBundle = payload.selectorBundle != null + ? normalizeReplayV2SelectorBundle(payload.selectorBundle) + : normalizedInput.selectorBundle != null + ? normalizeReplayV2SelectorBundle(normalizedInput.selectorBundle) + : null; + + if (selectorBundle && !payload.selectorBundle) { + payload.selectorBundle = selectorBundle; + } + + const targetId = normalizedInput.targetRef?.targetId || normalizedInput.targetId || null; + if (targetId && !normalizedInput.targetRef) { + normalizedInput.targetRef = { targetId, selectorVersion }; + } + + const normalizedEvent = canonicalizeJson({ + schemaVersion: normalizedInput.schemaVersion || REPLAY_V2_SCHEMA_VERSION, + runId: normalizedInput.runId, + streamId: normalizedInput.streamId, + seq: normalizedInput.seq, + monotonicTs, + monotonicMs: monotonicTs, + ts: normalizedInput.ts, + kind: normalizedInput.kind, + commandId: normalizeScalarSelectorValue(normalizedInput.commandId), + targetRef: targetId ? { + targetId, + selectorVersion + } : null, + payload }); - assertString(event.ts, 'replay_v2_event_ts_invalid', { ts: event.ts }); - assert(Number.isFinite(Date.parse(event.ts)), 'replay_v2_event_ts_unparseable', { ts: event.ts }); - assertOptionalString(event.targetId, 'replay_v2_event_target_id_invalid', { targetId: event.targetId }); - if (event.selectorBundle != null) { - event.selectorBundle = normalizeReplayV2SelectorBundle(event.selectorBundle); + + if (selectorBundle) normalizedEvent.selectorBundle = selectorBundle; + if (targetId) normalizedEvent.targetId = targetId; + + return normalizedEvent; +} + +export function assertReplayV2EventPayload(event) { + const normalized = normalizeReplayV2EventPayload(event); + if (normalized.schemaVersion != null) { + assert(REPLAY_V2_SCHEMA_VERSION_COMPAT.has(normalized.schemaVersion), 'replay_v2_event_schema_version_invalid', { + schemaVersion: normalized.schemaVersion + }); } - if (event.data != null) { - assert(isPlainObject(event.data) || Array.isArray(event.data), 'replay_v2_event_data_invalid', { data: event.data }); + assertString(normalized.kind, 'replay_v2_event_kind_invalid', { kind: normalized.kind }); + assert(REPLAY_V2_EVENT_KIND_SET.has(normalized.kind), 'replay_v2_event_kind_unsupported', { kind: normalized.kind }); + assertString(normalized.runId, 'replay_v2_event_run_id_missing', { event: normalized }); + assertOptionalString(normalized.streamId, 'replay_v2_event_stream_id_invalid', { streamId: normalized.streamId }); + assert(Number.isInteger(normalized.seq) && normalized.seq >= 1, 'replay_v2_event_seq_invalid', { seq: normalized.seq }); + assert(Number.isInteger(normalized.monotonicTs) && normalized.monotonicTs >= 0, 'replay_v2_event_monotonic_invalid', { + monotonicTs: normalized.monotonicTs + }); + assertOptionalString(normalized.commandId, 'replay_v2_event_command_id_invalid', { commandId: normalized.commandId }); + assert(isPlainObject(normalized.payload), 'replay_v2_event_payload_invalid', { payload: normalized.payload }); + if (normalized.ts != null) { + assertString(normalized.ts, 'replay_v2_event_ts_invalid', { ts: normalized.ts }); + assert(Number.isFinite(Date.parse(normalized.ts)), 'replay_v2_event_ts_unparseable', { ts: normalized.ts }); } - return event; + if (normalized.targetRef != null) { + assertString(normalized.targetRef.targetId, 'replay_v2_target_id_invalid', { targetRef: normalized.targetRef }); + assert(Number.isInteger(normalized.targetRef.selectorVersion) && normalized.targetRef.selectorVersion >= 1, 'replay_v2_selector_version_invalid', { + targetRef: normalized.targetRef + }); + } + return normalized; } export function assertReplayV2ChunkPayload(payload) { assert(isPlainObject(payload), 'replay_v2_chunk_invalid', { payload }); assertString(payload.runId, 'replay_v2_chunk_run_id_missing', { payload }); assertString(payload.streamId, 'replay_v2_chunk_stream_id_missing', { payload }); - assert(Number.isInteger(payload.seqStart) && payload.seqStart >= 1, 'replay_v2_chunk_seq_start_invalid', { - seqStart: payload.seqStart - }); + assert(Number.isInteger(payload.seqStart) && payload.seqStart >= 1, 'replay_v2_chunk_seq_start_invalid', { seqStart: payload.seqStart }); assert(Number.isInteger(payload.seqEnd) && payload.seqEnd >= payload.seqStart, 'replay_v2_chunk_seq_end_invalid', { seqEnd: payload.seqEnd, seqStart: payload.seqStart }); - assert(Array.isArray(payload.events) && payload.events.length > 0, 'replay_v2_chunk_events_invalid', { - events: payload.events - }); - + assert(Array.isArray(payload.events) && payload.events.length > 0, 'replay_v2_chunk_events_invalid', { events: payload.events }); if (payload.schemaVersion != null) { - assert(payload.schemaVersion === REPLAY_V2_SCHEMA_VERSION, 'replay_v2_chunk_schema_version_invalid', { + assert(REPLAY_V2_SCHEMA_VERSION_COMPAT.has(payload.schemaVersion), 'replay_v2_chunk_schema_version_invalid', { schemaVersion: payload.schemaVersion }); } let expectedSeq = payload.seqStart; - let previousMonotonicMs = -1; + let previousMonotonicTs = -1; - for (const event of payload.events) { - assertReplayV2EventPayload(event); - assert(event.runId === payload.runId, 'replay_v2_chunk_run_id_parity_error', { + payload.events = payload.events.map((event) => { + const normalizedEvent = assertReplayV2EventPayload({ ...event, streamId: event.streamId || payload.streamId }); + assert(normalizedEvent.runId === payload.runId, 'replay_v2_chunk_run_id_parity_error', { chunkRunId: payload.runId, - eventRunId: event.runId + eventRunId: normalizedEvent.runId }); - assert(event.streamId === payload.streamId, 'replay_v2_chunk_stream_id_parity_error', { + assert((normalizedEvent.streamId || payload.streamId) === payload.streamId, 'replay_v2_chunk_stream_id_parity_error', { chunkStreamId: payload.streamId, - eventStreamId: event.streamId + eventStreamId: normalizedEvent.streamId }); - assert(event.seq === expectedSeq, 'replay_v2_chunk_sequence_discontinuity', { + assert(normalizedEvent.seq === expectedSeq, 'replay_v2_chunk_sequence_discontinuity', { expectedSeq, - actualSeq: event.seq + actualSeq: normalizedEvent.seq }); - assert(event.monotonicMs >= previousMonotonicMs, 'replay_v2_chunk_monotonic_regression', { - previousMonotonicMs, - monotonicMs: event.monotonicMs + assert(normalizedEvent.monotonicTs >= previousMonotonicTs, 'replay_v2_chunk_monotonic_regression', { + previousMonotonicTs, + monotonicTs: normalizedEvent.monotonicTs }); - previousMonotonicMs = event.monotonicMs; + previousMonotonicTs = normalizedEvent.monotonicTs; expectedSeq += 1; - } + return normalizedEvent; + }); assert(payload.events[0].seq === payload.seqStart, 'replay_v2_chunk_seq_start_parity_error', { seqStart: payload.seqStart, @@ -315,3 +568,327 @@ export function assertReplayV2ChunkPayload(payload) { return payload; } + +export function applyReplayV2EventToTargetRegistry(registry, event) { + const normalizedEvent = assertReplayV2EventPayload(event); + const targetId = normalizedEvent.targetRef?.targetId || null; + const selectorBundle = normalizedEvent.payload?.selectorBundle || normalizedEvent.selectorBundle || {}; + const metadata = normalizedEvent.payload?.metadata ?? null; + const eventType = normalizedEvent.payload?.eventType || null; + + if (normalizedEvent.kind !== REPLAY_V2_EVENT_KINDS.LIFECYCLE || !targetId || !eventType) return null; + if (eventType === REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE) { + return registry.declare({ targetId, selectorBundle, metadata, seq: normalizedEvent.seq }); + } + if (eventType === REPLAY_V2_LIFECYCLE_EVENTS.TARGET_BIND) { + return registry.bind({ targetId, selectorBundle, metadata, seq: normalizedEvent.seq }); + } + if (eventType === REPLAY_V2_LIFECYCLE_EVENTS.TARGET_REBIND) { + return registry.rebind({ targetId, selectorBundle, metadata, seq: normalizedEvent.seq }); + } + if (eventType === REPLAY_V2_LIFECYCLE_EVENTS.TARGET_ORPHAN) { + return registry.orphan({ targetId, reason: normalizedEvent.payload?.reason ?? null, seq: normalizedEvent.seq }); + } + return null; +} + +export function buildReplayV2SeekIndex(events = [], { stride = REPLAY_V2_SEEK_STRIDE } = {}) { + const registry = createReplayV2TargetRegistry(); + const checkpoints = []; + let lastCheckpointSeq = 0; + + for (const rawEvent of events) { + const event = assertReplayV2EventPayload(rawEvent); + applyReplayV2EventToTargetRegistry(registry, event); + const shouldCheckpoint = checkpoints.length === 0 + || event.seq - lastCheckpointSeq >= stride + || (event.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE && String(event.payload?.eventType || '').startsWith('TARGET_')); + + if (!shouldCheckpoint) continue; + lastCheckpointSeq = event.seq; + checkpoints.push({ + checkpointSeq: event.seq, + monotonicTs: event.monotonicTs, + eventSeq: event.seq, + targetRegistryState: registry.snapshot() + }); + } + + return checkpoints; +} + +export function resolveReplayV2TargetStateAtSeq(events = [], seq = Number.MAX_SAFE_INTEGER) { + const registry = createReplayV2TargetRegistry(); + for (const rawEvent of events) { + const event = assertReplayV2EventPayload(rawEvent); + if (event.seq > seq) break; + applyReplayV2EventToTargetRegistry(registry, event); + } + return registry.snapshot(); +} + +function isActionableCommand(event) { + return event.kind === REPLAY_V2_EVENT_KINDS.COMMAND && Boolean(event.commandId); +} + +export function evaluateReplayV2GateMetrics(events = [], { finAckRequired = true } = {}) { + const normalizedEvents = events.map((event) => assertReplayV2EventPayload(event)).sort((a, b) => a.seq - b.seq); + const targetState = new Map(); + let lastSeq = 0; + let seqGapCount = 0; + let actionableCommands = 0; + let alignedCommands = 0; + let stableTargets = 0; + let orphanEvents = 0; + let finSeen = false; + let ackSeen = false; + + for (const event of normalizedEvents) { + if (lastSeq && event.seq !== lastSeq + 1) seqGapCount += 1; + lastSeq = event.seq; + + if (isActionableCommand(event)) { + actionableCommands += 1; + if (event.payload?.targetSnapshot || event.targetRef?.targetId) alignedCommands += 1; + const currentTarget = event.targetRef?.targetId ? targetState.get(event.targetRef.targetId) : null; + if (!event.targetRef?.targetId || (currentTarget && currentTarget.state !== 'orphaned')) { + stableTargets += 1; + } + } + + const registry = createReplayV2TargetRegistry({ initialState: [...targetState.values()] }); + const record = applyReplayV2EventToTargetRegistry(registry, event); + if (record) { + targetState.set(record.targetId, record); + if (record.state === 'orphaned') orphanEvents += 1; + } + + if (event.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE && event.payload?.eventType === REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_FIN) { + finSeen = true; + } + if (event.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE && event.payload?.eventType === REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_ACK) { + ackSeen = true; + } + } + + return { + totals: { + events: normalizedEvents.length, + actionableCommands, + orphanEvents + }, + seqContinuity: { + zeroGaps: seqGapCount === 0, + gapCount: seqGapCount + }, + finAck: { + success: !finAckRequired || (finSeen && ackSeen), + finSeen, + ackSeen + }, + commandToDomAlignment: actionableCommands === 0 ? 1 : alignedCommands / actionableCommands, + targetStability: actionableCommands === 0 ? 1 : stableTargets / actionableCommands, + orphanSpam: orphanEvents <= Math.max(1, Math.floor(normalizedEvents.length * 0.01)) + }; +} + +export function encodeMessagePack(value) { + const chunks = []; + + function pushUInt(valueToWrite, byteLength, prefix8, prefix16, prefix32, prefix64) { + const buffer = Buffer.allocUnsafe(1 + byteLength); + buffer[0] = byteLength === 1 ? prefix8 : byteLength === 2 ? prefix16 : byteLength === 4 ? prefix32 : prefix64; + if (byteLength === 1) buffer.writeUInt8(valueToWrite, 1); + if (byteLength === 2) buffer.writeUInt16BE(valueToWrite, 1); + if (byteLength === 4) buffer.writeUInt32BE(valueToWrite, 1); + if (byteLength === 8) buffer.writeBigUInt64BE(BigInt(valueToWrite), 1); + chunks.push(buffer); + } + + function encodeAny(input) { + if (input == null) { + chunks.push(Buffer.from([0xc0])); + return; + } + if (input === false) { + chunks.push(Buffer.from([0xc2])); + return; + } + if (input === true) { + chunks.push(Buffer.from([0xc3])); + return; + } + if (typeof input === 'number') { + if (Number.isInteger(input) && input >= 0 && input <= 0x7f) { + chunks.push(Buffer.from([input])); + return; + } + if (Number.isInteger(input) && input >= -32 && input < 0) { + chunks.push(Buffer.from([0xe0 | (input + 32)])); + return; + } + if (Number.isInteger(input) && input >= 0 && input <= 0xff) return pushUInt(input, 1, 0xcc, 0xcd, 0xce, 0xcf); + if (Number.isInteger(input) && input >= 0 && input <= 0xffff) return pushUInt(input, 2, 0xcc, 0xcd, 0xce, 0xcf); + if (Number.isInteger(input) && input >= 0 && input <= 0xffffffff) return pushUInt(input, 4, 0xcc, 0xcd, 0xce, 0xcf); + const buffer = Buffer.allocUnsafe(9); + buffer[0] = 0xcb; + buffer.writeDoubleBE(input, 1); + chunks.push(buffer); + return; + } + if (typeof input === 'string') { + const data = Buffer.from(input, 'utf8'); + if (data.length <= 31) { + chunks.push(Buffer.concat([Buffer.from([0xa0 | data.length]), data])); + return; + } + if (data.length <= 0xff) { + chunks.push(Buffer.concat([Buffer.from([0xd9, data.length]), data])); + return; + } + const header = Buffer.allocUnsafe(3); + header[0] = 0xda; + header.writeUInt16BE(data.length, 1); + chunks.push(Buffer.concat([header, data])); + return; + } + if (Buffer.isBuffer(input) || input instanceof Uint8Array) { + const data = Buffer.from(input); + if (data.length <= 0xff) { + chunks.push(Buffer.concat([Buffer.from([0xc4, data.length]), data])); + return; + } + const header = Buffer.allocUnsafe(3); + header[0] = 0xc5; + header.writeUInt16BE(data.length, 1); + chunks.push(Buffer.concat([header, data])); + return; + } + if (Array.isArray(input)) { + const length = input.length; + if (length <= 15) { + chunks.push(Buffer.from([0x90 | length])); + } else { + const header = Buffer.allocUnsafe(3); + header[0] = 0xdc; + header.writeUInt16BE(length, 1); + chunks.push(header); + } + for (const item of input) encodeAny(item); + return; + } + if (isPlainObject(input)) { + const entries = Object.entries(input).filter(([, value]) => value !== undefined); + const length = entries.length; + if (length <= 15) { + chunks.push(Buffer.from([0x80 | length])); + } else { + const header = Buffer.allocUnsafe(3); + header[0] = 0xde; + header.writeUInt16BE(length, 1); + chunks.push(header); + } + for (const [key, value] of entries) { + encodeAny(String(key)); + encodeAny(value); + } + return; + } + + throw new Error(`unsupported_messagepack_type:${typeof input}`); + } + + encodeAny(value); + return Buffer.concat(chunks); +} + +export function decodeMessagePack(buffer) { + const bytes = Buffer.from(buffer); + let offset = 0; + + function read(len) { + const next = bytes.subarray(offset, offset + len); + offset += len; + return next; + } + + function decodeAny() { + const prefix = bytes[offset]; + offset += 1; + + if (prefix <= 0x7f) return prefix; + if (prefix >= 0xe0) return prefix - 0x100; + if ((prefix & 0xf0) === 0x80) { + const size = prefix & 0x0f; + const obj = {}; + for (let i = 0; i < size; i += 1) { + const key = decodeAny(); + obj[key] = decodeAny(); + } + return obj; + } + if ((prefix & 0xf0) === 0x90) { + const size = prefix & 0x0f; + return Array.from({ length: size }, () => decodeAny()); + } + if ((prefix & 0xe0) === 0xa0) { + const size = prefix & 0x1f; + return read(size).toString('utf8'); + } + if (prefix === 0xc0) return null; + if (prefix === 0xc2) return false; + if (prefix === 0xc3) return true; + if (prefix === 0xc4) return read(bytes[offset++]); + if (prefix === 0xc5) { + const size = bytes.readUInt16BE(offset); + offset += 2; + return read(size); + } + if (prefix === 0xcb) { + const value = bytes.readDoubleBE(offset); + offset += 8; + return value; + } + if (prefix === 0xcc) return bytes[offset++]; + if (prefix === 0xcd) { + const value = bytes.readUInt16BE(offset); + offset += 2; + return value; + } + if (prefix === 0xce) { + const value = bytes.readUInt32BE(offset); + offset += 4; + return value; + } + if (prefix === 0xcf) { + const value = Number(bytes.readBigUInt64BE(offset)); + offset += 8; + return value; + } + if (prefix === 0xd9) return read(bytes[offset++]).toString('utf8'); + if (prefix === 0xda) { + const size = bytes.readUInt16BE(offset); + offset += 2; + return read(size).toString('utf8'); + } + if (prefix === 0xdc) { + const size = bytes.readUInt16BE(offset); + offset += 2; + return Array.from({ length: size }, () => decodeAny()); + } + if (prefix === 0xde) { + const size = bytes.readUInt16BE(offset); + offset += 2; + const obj = {}; + for (let i = 0; i < size; i += 1) { + const key = decodeAny(); + obj[key] = decodeAny(); + } + return obj; + } + + throw new Error(`unsupported_messagepack_prefix:${prefix}`); + } + + return decodeAny(); +} diff --git a/scripts/db-migrate-container.mjs b/scripts/db-migrate-container.mjs index 55d9be2..5e89576 100644 --- a/scripts/db-migrate-container.mjs +++ b/scripts/db-migrate-container.mjs @@ -11,7 +11,8 @@ const files = [ "infra/db/migrations/005_batches_11_18.sql", "infra/db/migrations/006_batches_19_26.sql", "infra/db/migrations/007_project_ingest_tokens.sql", - "infra/db/migrations/008_replay_v2_storage.sql" + "infra/db/migrations/008_replay_v2_storage.sql", + "infra/db/migrations/009_replay_v2_full_plan.sql" ]; for (const file of files) { diff --git a/scripts/migrate-in-container.sh b/scripts/migrate-in-container.sh index 4249120..87482fa 100755 --- a/scripts/migrate-in-container.sh +++ b/scripts/migrate-in-container.sh @@ -12,7 +12,8 @@ for file in \ infra/db/migrations/005_batches_11_18.sql \ infra/db/migrations/006_batches_19_26.sql \ infra/db/migrations/007_project_ingest_tokens.sql \ - infra/db/migrations/008_replay_v2_storage.sql + infra/db/migrations/008_replay_v2_storage.sql \ + infra/db/migrations/009_replay_v2_full_plan.sql do echo "Applying $file ..." diff --git a/scripts/phasec-runtime-gate.mjs b/scripts/phasec-runtime-gate.mjs new file mode 100644 index 0000000..8917acd --- /dev/null +++ b/scripts/phasec-runtime-gate.mjs @@ -0,0 +1,393 @@ +import crypto from 'node:crypto'; +import fs from 'node:fs/promises'; + +const API = process.env.API_BASE_URL || 'http://localhost:4000'; +const INGEST = process.env.INGEST_BASE_URL || 'http://localhost:4010'; +const WEB = process.env.WEB_BASE_URL || 'http://localhost:3000'; +const ART_DIR = 'artifacts/phasec-runtime-gate'; + +await fs.mkdir(ART_DIR, { recursive: true }); + +async function req(url, { method = 'GET', headers = {}, body } = {}) { + const res = await fetch(url, { + method, + headers: { + ...headers, + ...(body ? { 'content-type': 'application/json' } : {}) + }, + ...(body ? { body: JSON.stringify(body) } : {}), + redirect: 'manual' + }); + + const text = await res.text(); + let json = null; + try { json = text ? JSON.parse(text) : null; } catch {} + + return { + status: res.status, + text, + json, + headers: Object.fromEntries(res.headers.entries()) + }; +} + +function assert(condition, message) { + if (!condition) throw new Error(message); +} + +function authHeader(token) { + return token ? { authorization: `Bearer ${token}` } : {}; +} + +async function issueProjectToken({ apiToken, projectId, label, ttlDays = 7 }) { + const res = await req(`${API}/v1/projects/${projectId}/ingest-tokens`, { + method: 'POST', + headers: authHeader(apiToken), + body: { label, ttlDays } + }); + assert(res.status === 201 && res.json?.token, `Ingest token issue failed (${projectId}): ${res.status} ${res.text}`); + return res.json.token; +} + +async function ingest({ token, type, payload, idempotencyKey = crypto.randomUUID() }) { + return req(`${INGEST}/v1/ingest/events`, { + method: 'POST', + headers: authHeader(token), + body: { type, idempotencyKey, payload } + }); +} + +const healthApi = await req(`${API}/healthz`); +const healthIngest = await req(`${INGEST}/healthz`); +const healthWeb = await req(`${WEB}/healthz`); +assert(healthApi.status === 200, `API healthz failed: ${healthApi.status}`); +assert(healthIngest.status === 200, `Ingest healthz failed: ${healthIngest.status}`); +assert(healthWeb.status === 200, `Web healthz failed: ${healthWeb.status}`); + +const stamp = Date.now(); +const login = await req(`${API}/v1/auth/login`, { + method: 'POST', + body: { + email: `phasec+${stamp}@local.test`, + name: 'Phase C Gate' + } +}); +assert(login.status === 200 && login.json?.token, `Login failed: ${login.status} ${login.text}`); +const apiToken = login.json.token; + +const workspace = await req(`${API}/v1/workspaces`, { + method: 'POST', + headers: authHeader(apiToken), + body: { + organizationName: 'Phase C Org', + organizationSlug: `phasec-org-${stamp}`, + name: 'Phase C Workspace', + slug: `phasec-ws-${stamp}`, + timezone: 'UTC', + retentionDays: 30 + } +}); +assert(workspace.status === 201 && workspace.json?.item?.id, `Workspace create failed: ${workspace.status} ${workspace.text}`); +const workspaceId = workspace.json.item.id; + +const project = await req(`${API}/v1/projects`, { + method: 'POST', + headers: authHeader(apiToken), + body: { + workspaceId, + name: 'Phase C Project', + slug: `phasec-proj-${stamp}`, + repoUrl: 'https://example.test/repo.git', + defaultBranch: 'main' + } +}); +assert(project.status === 201 && project.json?.item?.id, `Project create failed: ${project.status} ${project.text}`); +const projectId = project.json.item.id; + +const project2 = await req(`${API}/v1/projects`, { + method: 'POST', + headers: authHeader(apiToken), + body: { + workspaceId, + name: 'Phase C Project 2', + slug: `phasec-proj2-${stamp}`, + repoUrl: 'https://example.test/repo2.git', + defaultBranch: 'main' + } +}); +assert(project2.status === 201 && project2.json?.item?.id, `Project2 create failed: ${project2.status} ${project2.text}`); +const project2Id = project2.json.item.id; + +const ingestToken = await issueProjectToken({ apiToken, projectId, label: `phasec-gate-${stamp}` }); +const ingestTokenProject2 = await issueProjectToken({ apiToken, projectId: project2Id, label: `phasec-gate-p2-${stamp}` }); + +// Ingest auth gate checks +const unauth = await ingest({ + token: '', + type: 'heartbeat.signal', + payload: { runId: crypto.randomUUID() } +}); +assert(unauth.status === 401, `Expected 401 for missing ingest auth, got ${unauth.status}`); + +const scopeMismatchRunId = crypto.randomUUID(); +const mismatch = await ingest({ + token: ingestTokenProject2, + type: 'run.started', + payload: { + runId: scopeMismatchRunId, + workspaceId, + projectId, + branch: 'main', + commitSha: 'phasec-gate-mismatch', + ciProvider: 'local' + } +}); +assert(mismatch.status === 403, `Expected 403 token_scope_mismatch, got ${mismatch.status} ${mismatch.text}`); +assert(mismatch.json?.error === 'token_scope_mismatch', `Expected token_scope_mismatch body, got ${mismatch.text}`); + +// Seed replay-v2 data for main run +const runId = crypto.randomUUID(); +const streamDefault = 'default'; +const streamAlt = 'alt'; +const t0 = Date.now(); + +const runStarted = await ingest({ + token: ingestToken, + type: 'run.started', + payload: { + runId, + workspaceId, + projectId, + branch: 'main', + commitSha: 'phasec-gate', + ciProvider: 'local' + } +}); +assert([200, 202].includes(runStarted.status), `run.started failed: ${runStarted.status} ${runStarted.text}`); + +const chunkDefault = await ingest({ + token: ingestToken, + type: 'replay.v2.chunk', + payload: { + runId, + streamId: streamDefault, + schemaVersion: '2.0', + startedAt: new Date(t0).toISOString(), + chunkIndex: 0, + seqStart: 1, + seqEnd: 2, + events: [ + { + kind: 'session.start', + runId, + streamId: streamDefault, + seq: 1, + monotonicMs: 0, + ts: new Date(t0).toISOString(), + data: { url: 'https://example.test/default' } + }, + { + kind: 'log', + runId, + streamId: streamDefault, + seq: 2, + monotonicMs: 25, + ts: new Date(t0 + 25).toISOString(), + data: { level: 'info', message: 'default-stream' } + } + ], + final: true + } +}); +assert([200, 202].includes(chunkDefault.status), `default chunk failed: ${chunkDefault.status} ${chunkDefault.text}`); + +const chunkAlt = await ingest({ + token: ingestToken, + type: 'replay.v2.chunk', + payload: { + runId, + streamId: streamAlt, + schemaVersion: '2.0', + startedAt: new Date(t0 + 1000).toISOString(), + chunkIndex: 0, + seqStart: 1, + seqEnd: 1, + events: [ + { + kind: 'log', + runId, + streamId: streamAlt, + seq: 1, + monotonicMs: 0, + ts: new Date(t0 + 1000).toISOString(), + data: { level: 'info', message: 'alt-stream' } + } + ], + final: false + } +}); +assert([200, 202].includes(chunkAlt.status), `alt chunk failed: ${chunkAlt.status} ${chunkAlt.text}`); + +// Seed run with no replay streams for empty-state web check +const runNoReplay = crypto.randomUUID(); +const runNoReplayStarted = await ingest({ + token: ingestToken, + type: 'run.started', + payload: { + runId: runNoReplay, + workspaceId, + projectId, + branch: 'main', + commitSha: 'phasec-gate-no-replay', + ciProvider: 'local' + } +}); +assert([200, 202].includes(runNoReplayStarted.status), `runNoReplay start failed: ${runNoReplayStarted.status} ${runNoReplayStarted.text}`); + +// API: streams + events +const streams = await req(`${API}/v1/runs/${runId}/replay-v2/streams`, { + headers: authHeader(apiToken) +}); +assert(streams.status === 200, `Streams endpoint failed: ${streams.status} ${streams.text}`); +assert(Array.isArray(streams.json?.items) && streams.json.items.length >= 2, `Expected >=2 streams, got ${streams.json?.items?.length ?? 'n/a'}`); + +const streamSummaryById = Object.fromEntries((streams.json.items || []).map((stream) => [stream.stream_id, stream])); +const expectedStreamSummary = { + [streamDefault]: { + schema_version: '2.0', + first_seq: 1, + last_seq: 2, + chunk_count: 1, + event_count: 2, + final_received: true + }, + [streamAlt]: { + schema_version: '2.0', + first_seq: 1, + last_seq: 1, + chunk_count: 1, + event_count: 1, + final_received: false + } +}; + +for (const [streamId, expected] of Object.entries(expectedStreamSummary)) { + const actual = streamSummaryById[streamId]; + assert(actual, `Missing stream summary for ${streamId}`); + assert(actual.schema_version === expected.schema_version, `schema_version mismatch for ${streamId}: ${actual.schema_version}`); + assert(Number(actual.first_seq) === expected.first_seq, `first_seq mismatch for ${streamId}: ${actual.first_seq}`); + assert(Number(actual.last_seq) === expected.last_seq, `last_seq mismatch for ${streamId}: ${actual.last_seq}`); + assert(Number(actual.chunk_count) === expected.chunk_count, `chunk_count mismatch for ${streamId}: ${actual.chunk_count}`); + assert(Number(actual.event_count) === expected.event_count, `event_count mismatch for ${streamId}: ${actual.event_count}`); + assert(Boolean(actual.final_received) === expected.final_received, `final_received mismatch for ${streamId}: ${actual.final_received}`); +} + + +const defaultEvents = await req(`${API}/v1/runs/${runId}/replay-v2/events?streamId=${encodeURIComponent(streamDefault)}&limit=300`, { + headers: authHeader(apiToken) +}); +assert(defaultEvents.status === 200, `Events endpoint failed: ${defaultEvents.status} ${defaultEvents.text}`); +assert(Array.isArray(defaultEvents.json?.items) && defaultEvents.json.items.length === 2, `Default events count mismatch: ${defaultEvents.json?.items?.length ?? 'n/a'}`); +assert(defaultEvents.json.items[0]?.seq === 1 && defaultEvents.json.items[1]?.seq === 2, 'Default events sequence ordering mismatch'); +assert(defaultEvents.json.items[0]?.chunk_index === 0, `chunk_index missing/mismatch: ${defaultEvents.json.items[0]?.chunk_index}`); +assert(defaultEvents.json.items[0]?.final === true, `final flag mismatch: ${defaultEvents.json.items[0]?.final}`); + +const rangeEvents = await req(`${API}/v1/runs/${runId}/replay-v2/events?streamId=${encodeURIComponent(streamDefault)}&fromSeq=2&toSeq=2&limit=10`, { + headers: authHeader(apiToken) +}); +assert(rangeEvents.status === 200, `Range events failed: ${rangeEvents.status} ${rangeEvents.text}`); +assert(Array.isArray(rangeEvents.json?.items) && rangeEvents.json.items.length === 1 && rangeEvents.json.items[0]?.seq === 2, 'fromSeq/toSeq filter mismatch'); + +const noEvents = await req(`${API}/v1/runs/${runId}/replay-v2/events?streamId=${encodeURIComponent(streamDefault)}&fromSeq=99999&toSeq=99999&limit=50`, { + headers: authHeader(apiToken) +}); +assert(noEvents.status === 200, `No-event range status mismatch: ${noEvents.status} ${noEvents.text}`); +assert(Array.isArray(noEvents.json?.items) && noEvents.json.items.length === 0, `Expected explicit no-event range result, got ${noEvents.json?.items?.length ?? 'n/a'} items`); +assert(noEvents.json.pageInfo?.total === 0, `Expected no-event total=0, got ${noEvents.json.pageInfo?.total}`); + + +const badLimit = await req(`${API}/v1/runs/${runId}/replay-v2/events?streamId=${encodeURIComponent(streamDefault)}&limit=abc`, { + headers: authHeader(apiToken) +}); +assert(badLimit.status === 400, `Invalid limit status mismatch: ${badLimit.status} ${badLimit.text}`); +assert(badLimit.json?.error === 'invalid_limit', `Invalid limit error mismatch: ${badLimit.text}`); + +const badRange = await req(`${API}/v1/runs/${runId}/replay-v2/events?streamId=${encodeURIComponent(streamDefault)}&fromSeq=3&toSeq=2`, { + headers: authHeader(apiToken) +}); +assert(badRange.status === 400, `Invalid seq range status mismatch: ${badRange.status} ${badRange.text}`); +assert(badRange.json?.error === 'invalid_seq_range', `Invalid seq range error mismatch: ${badRange.text}`); + +// Web viewer checks +const replayFallbackPage = await req(`${WEB}/app/runs/${runId}/replay-v2?streamId=does-not-exist`, { + headers: { + cookie: `th_session=${apiToken}` + } +}); +assert(replayFallbackPage.status === 200, `Replay page fallback status mismatch: ${replayFallbackPage.status}`); +assert(replayFallbackPage.text.includes('Replay V2'), 'Replay page missing heading'); +assert(replayFallbackPage.text.includes('streamId=default'), 'Replay page did not fall back to first stream'); +assert(replayFallbackPage.text.includes('default-stream'), 'Replay fallback page did not load default stream events'); + +const replayAltPage = await req(`${WEB}/app/runs/${runId}/replay-v2?streamId=${encodeURIComponent(streamAlt)}`, { + headers: { + cookie: `th_session=${apiToken}` + } +}); +assert(replayAltPage.status === 200, `Replay page stream-switch status mismatch: ${replayAltPage.status}`); +assert(replayAltPage.text.includes('alt-stream'), 'Replay stream switch did not load alternate stream events'); + +const replayNoStreamPage = await req(`${WEB}/app/runs/${runNoReplay}/replay-v2`, { + headers: { + cookie: `th_session=${apiToken}` + } +}); +assert(replayNoStreamPage.status === 200, `Replay no-stream page status mismatch: ${replayNoStreamPage.status}`); +assert(replayNoStreamPage.text.includes('No replay streams'), 'Replay no-stream empty state missing'); + +const runDetailPage = await req(`${WEB}/app/runs/${runId}`, { + headers: { + cookie: `th_session=${apiToken}` + } +}); +assert(runDetailPage.status === 200, `Run detail page status mismatch: ${runDetailPage.status}`); +assert(runDetailPage.text.includes(`href="/app/runs/${runId}/replay-v2"`), 'Run detail page missing Replay V2 link href'); +assert(runDetailPage.text.includes('Open Replay V2'), 'Run detail page missing Open Replay V2 link text'); + + +const summary = { + ok: true, + branch: 'feat/replay-v2-phase-c-20260403', + runId, + runNoReplay, + workspaceId, + projectId, + checks: { + streamsStatus: streams.status, + defaultEventsStatus: defaultEvents.status, + rangeEventsStatus: rangeEvents.status, + noEventsStatus: noEvents.status, + invalidLimitStatus: badLimit.status, + invalidSeqRangeStatus: badRange.status, + streamSummaryParity: 'pass', + webFallbackStatus: replayFallbackPage.status, + webSwitchStatus: replayAltPage.status, + webNoStreamStatus: replayNoStreamPage.status, + runDetailStatus: runDetailPage.status, + }, + artifacts: { + summary: `${ART_DIR}/phasec-runtime-gate.json`, + fallbackHtml: `${ART_DIR}/phasec-runtime-gate-fallback.html`, + altHtml: `${ART_DIR}/phasec-runtime-gate-alt.html`, + noStreamHtml: `${ART_DIR}/phasec-runtime-gate-no-stream.html`, + runDetailHtml: `${ART_DIR}/phasec-runtime-gate-run-detail.html` + } +}; + +await fs.writeFile(`${ART_DIR}/phasec-runtime-gate.json`, JSON.stringify(summary, null, 2)); +await fs.writeFile(`${ART_DIR}/phasec-runtime-gate-fallback.html`, replayFallbackPage.text); +await fs.writeFile(`${ART_DIR}/phasec-runtime-gate-alt.html`, replayAltPage.text); +await fs.writeFile(`${ART_DIR}/phasec-runtime-gate-no-stream.html`, replayNoStreamPage.text); +await fs.writeFile(`${ART_DIR}/phasec-runtime-gate-run-detail.html`, runDetailPage.text); + +console.log(JSON.stringify(summary, null, 2)); diff --git a/scripts/replay-v2-fin-ack-check.mjs b/scripts/replay-v2-fin-ack-check.mjs new file mode 100644 index 0000000..533fe76 --- /dev/null +++ b/scripts/replay-v2-fin-ack-check.mjs @@ -0,0 +1,140 @@ +import fs from 'node:fs/promises'; +import path from 'node:path'; +import { + REPLAY_V2_EVENT_KINDS, + REPLAY_V2_LIFECYCLE_EVENTS, + decodeMessagePack +} from '../packages/shared/src/index.js'; + +function decodeHarborPayload(payload) { + try { + return decodeMessagePack(payload); + } catch { + try { + return JSON.parse(payload.toString('utf8')); + } catch { + return null; + } + } +} + +async function readHarborFrames(segmentDir) { + const entries = (await fs.readdir(segmentDir)) + .filter((entry) => entry.endsWith('.harbor')) + .sort(); + const frames = []; + + for (const entry of entries) { + const filePath = path.join(segmentDir, entry); + const buffer = await fs.readFile(filePath); + let offset = 0; + + while (offset + 4 <= buffer.length) { + const len = buffer.readUInt32BE(offset); + offset += 4; + if (len < 0 || offset + len > buffer.length) { + throw new Error(`invalid_harbor_frame_length:${entry}:${len}`); + } + + const payload = buffer.subarray(offset, offset + len); + offset += len; + + const decoded = decodeHarborPayload(payload); + if (decoded) { + frames.push(decoded); + } + } + } + + return frames; +} + +function findLifecycleEvent(frame, expectedType) { + const events = Array.isArray(frame?.events) ? frame.events : []; + return events.find((event) => ( + event?.kind === REPLAY_V2_EVENT_KINDS.LIFECYCLE + && event?.payload?.eventType === expectedType + )) || null; +} + +function extractFin(frame) { + if (frame?.type === 'TRANSPORT_FIN') { + return { source: 'transport', finId: frame.finId || null }; + } + const lifecycle = findLifecycleEvent(frame, REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_FIN); + if (lifecycle) { + return { source: 'lifecycle', finId: lifecycle?.payload?.finId || null }; + } + return null; +} + +function extractAck(frame) { + if (frame?.type === 'TRANSPORT_ACK') { + return { source: 'transport', finId: frame.finId || null }; + } + const lifecycle = findLifecycleEvent(frame, REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_ACK); + if (lifecycle) { + return { source: 'lifecycle', finId: lifecycle?.payload?.finId || null }; + } + return null; +} + +const segmentDir = process.argv[2]; +if (!segmentDir) { + console.error('usage: node scripts/replay-v2-fin-ack-check.mjs '); + process.exit(1); +} + +const frames = await readHarborFrames(segmentDir); +const finCandidates = frames.map(extractFin).filter(Boolean); +const ackCandidates = frames.map(extractAck).filter(Boolean); + +const finSeen = finCandidates.length > 0; +const ackSeen = ackCandidates.length > 0; + +function findCorrelatedPair(fins, acks) { + for (const fin of fins) { + if (!fin.finId) { + continue; + } + + for (const ack of acks) { + if (ack.finId && fin.finId === ack.finId) { + return { fin, ack, matched: true }; + } + } + } + + return { fin: fins[0] || null, ack: acks[0] || null, matched: false }; +} + +const correlation = findCorrelatedPair(finCandidates, ackCandidates); +const finInfo = correlation.fin; +const ackInfo = correlation.ack; +const finId = finInfo?.finId || null; +const ackFinId = ackInfo?.finId || null; +const finAckMatch = Boolean( + finSeen + && ackSeen + && correlation.matched + && correlation.fin?.finId + && correlation.ack?.finId + && correlation.fin.finId === correlation.ack.finId +); + +const result = { + ok: Boolean(finSeen && ackSeen && finAckMatch), + frameCount: frames.length, + finSeen, + ackSeen, + finId, + ackFinId, + finAckMatch, + finCandidateCount: finCandidates.length, + ackCandidateCount: ackCandidates.length, + finSource: finInfo?.source || null, + ackSource: ackInfo?.source || null +}; + +console.log(JSON.stringify(result, null, 2)); +process.exit(result.ok ? 0 : 2); diff --git a/scripts/replay-v2-gate-artifacts.mjs b/scripts/replay-v2-gate-artifacts.mjs new file mode 100644 index 0000000..6fa2b3c --- /dev/null +++ b/scripts/replay-v2-gate-artifacts.mjs @@ -0,0 +1,74 @@ +import fs from 'node:fs/promises'; +import path from 'node:path'; +import { + REPLAY_V2_EVENT_KINDS, + REPLAY_V2_LIFECYCLE_EVENTS, + buildReplayV2SeekIndex, + evaluateReplayV2GateMetrics +} from '../packages/shared/src/index.js'; + +const outDir = process.argv[2] || path.join(process.cwd(), 'artifacts'); +await fs.mkdir(outDir, { recursive: true }); + +const runId = '00000000-0000-4000-8000-000000000001'; +const streamId = 'gate-sample'; +const targetId = 'tgt_sample_primary'; +const startedAt = new Date('2026-04-03T00:00:00.000Z').getTime(); + +const events = []; +let seq = 1; +function push(kind, payload = {}, extra = {}) { + events.push({ + runId, + streamId, + seq, + monotonicTs: seq * 10, + ts: new Date(startedAt + (seq * 10)).toISOString(), + kind, + payload, + ...extra + }); + seq += 1; +} + +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.SESSION_START }); +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.TARGET_DECLARE, selectorBundle: { primary: { dataTestId: 'checkout-button' } } }, { + targetRef: { targetId, selectorVersion: 1 } +}); +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.TARGET_BIND, selectorBundle: { primary: { dataTestId: 'checkout-button' } } }, { + targetRef: { targetId, selectorVersion: 1 } +}); + +for (let index = 0; index < 294; index += 1) { + push(REPLAY_V2_EVENT_KINDS.COMMAND, { + eventType: 'CLICK', + targetSnapshot: { targetId } + }, { + commandId: `cmd_${index + 1}`, + targetRef: { targetId, selectorVersion: 1 } + }); +} + +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_FIN, finId: 'fin_sample' }); +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.TRANSPORT_ACK, finId: 'fin_sample' }); +push(REPLAY_V2_EVENT_KINDS.LIFECYCLE, { eventType: REPLAY_V2_LIFECYCLE_EVENTS.SESSION_END }); + +const metrics = evaluateReplayV2GateMetrics(events); +const seekIndex = buildReplayV2SeekIndex(events, { stride: 50 }); +const artifact = { + generatedAt: new Date().toISOString(), + runId, + streamId, + eventCount: events.length, + seekCheckpointCount: seekIndex.length, + metrics, + thresholds: { + replayLoadUnder3s: 'deferred-to-runtime', + commandToDomAlignmentMin: 0.95, + targetStabilityMin: 0.98 + } +}; + +const outPath = path.join(outDir, 'replay-v2-gate-artifacts.json'); +await fs.writeFile(outPath, `${JSON.stringify(artifact, null, 2)}\n`); +console.log(JSON.stringify({ ok: true, outPath, eventCount: events.length, checkpointCount: seekIndex.length }, null, 2)); diff --git a/scripts/run-migrations.sh b/scripts/run-migrations.sh index c4ab0a3..6410518 100755 --- a/scripts/run-migrations.sh +++ b/scripts/run-migrations.sh @@ -8,3 +8,4 @@ psql "${DATABASE_URL}" -f infra/db/migrations/005_batches_11_18.sql psql "${DATABASE_URL}" -f infra/db/migrations/006_batches_19_26.sql psql "${DATABASE_URL}" -f infra/db/migrations/007_project_ingest_tokens.sql psql "${DATABASE_URL}" -f infra/db/migrations/008_replay_v2_storage.sql +psql "${DATABASE_URL}" -f infra/db/migrations/009_replay_v2_full_plan.sql