Rp Telemetry

lib/telemetry-recorder.js

'use strict';

// Telemetry recorder for RePlatform migration runs.
//
// The recorder owns everything mechanical and error-prone: schema validation
// (enums, subtype-per-event-type), timestamping, per-class event aggregation,
// the run/attempt/session resume model, the mechanical privacy scrub, atomic
// journal appends, and the well-formedness gate at finalize. The rp-telemetry
// companion skill carries judgment only (what to record, which subtype,
// observation-only prose) and always persists through this module — the signal
// layer is never hand-written.
//
// Storage layout, per migration project:
//   telemetry/events.jsonl                          append-only journal of the
//                                                   current (unfinalized) run
//   run-telemetry.json                              latest finalized signal doc
//   telemetry/runs/run-<attempt>-<run_id>.json      archived finalized signal docs
//   telemetry/runs/run-<attempt>-<run_id>.jsonl     archived raw journals
//
// The journal is the source of truth during a run; the finalized document is a
// pure function of the journal (assembleDocument), so a crash between the
// finalize append and the document write is recoverable on the next call.

const fs = require('node:fs');
const path = require('node:path');
const os = require('node:os');
const crypto = require('node:crypto');
const { execFileSync } = require('node:child_process');
const biSink = require('./bi-sink.js');

// 1.1.0 — added `skills_commit` to the run-start record and rollup (additive).
// 1.2.0 — BI sink: added `telemetry_health.bi_push_failures`, the
//         `bi_push_failed`/`bi_push_truncated` journal record types (additive;
//         readers per the external reader contract skip unknown record types).
// 1.3.0 — added the `discovered_entity_types` dimension and its finalize
//         cross-check flags (`discovered_entity_types_null_after_discovery`,
//         `volume_row_missing:<type>`), so a discovered entity class silently
//         absent from `volumes` is mechanically visible (additive; local-only —
//         not a registered BI column).
// 1.4.0 — BI backlog healing: `bi_push_ok` watermark + `bi_push_skipped`
//         journal record types, `telemetry_health.bi_push_skipped`, and a full
//         idempotent re-push whenever a session boundary (resume, identity
//         arrival) or an event push finds journal rows with no confirming
//         watermark. Before this, mid-run push failures were healed only by
//         finalize's full re-push — a run that halted and never finalized
//         stranded its tail in the local journal forever.
//         Additive; readers per the external
//         reader contract skip unknown record types.
// 1.5.0 — `transcript_digest` record type — a deterministic parse
//         of the Claude Code session transcript (cost, turn counts, tool
//         failures, retry-loop signatures, hook errors, per-stage wall time),
//         computed by a script rather than self-reported. Carried by
//         `replay()`/`assembleDocument()` into `rollup.transcript_digest`, so
//         `finalize`'s existing BI push carries it too — no new BI call, no
//         new free text, no scrub work (every field is counts/versions/tool
//         names). Additive; readers per the external reader contract skip
//         unknown record types.
// 1.6.0 — bounded knowledge observations; existing actualSkill BI column, no raw custom keys.
// 1.7.0 — a run with no operator id sends under a per-run placeholder subject: the optional
//         `run_start.provisional_subject` field, the `provisional_subject` journal record type
//         (for journals that predate it) and `subject` on `bi_push_ok` markers. Additive;
//         readers per the external reader contract skip unknown record types and fields.
const TELEMETRY_SCHEMA_VERSION = '1.7.0';
const FREE_TEXT_MAX = 400;
const SHAPES_KEPT_PER_CLASS = 3;
const SHAPES_MAX_PER_CALL = 10;
const EVIDENCE_REFS_MAX = 5;
// Class-token lists (plugins, entity types). A real WordPress site can run well over 64 plugins.
const CLASS_LIST_MAX = 256;
// A resume gap below this is treated as in-session noise, not a wait interval.
const IMPLICIT_WAIT_MIN_MS = 60 * 1000;
// Automatic resume of a stale, unfinalized journal (resumeIfStale). Every value here is an
// unverified engineering setting — confirm at the first live runs, never report as measured.
// The manual route script cannot know which sitting it belongs to, so it treats this much
// journal silence as an execution boundary (a heuristic, not a detected session): one quiet
// sitting of >= 15 minutes is counted as two sessions, and a real break shorter than this is
// left to the agent's own `start`. Deliberately NOT IMPLICIT_WAIT_MIN_MS — stages routinely run
// longer than a minute without a journal write.
const ROUTE_RESUME_MIN_GAP_MS = 15 * 60 * 1000;
// A session opened less than this long ago belongs to the same execution: the supervisor joins
// it instead of opening another, and an explicit `start` joins an automatic one. Separate from
// IMPLICIT_WAIT_MIN_MS even though the value matches.
const SESSION_ADOPT_WINDOW_MS = 60 * 1000;
// Transport cap on the automatic heal; at the cap the request is cancelled, not abandoned.
const HEAL_DEADLINE_MS = 3000;
// The closing push is a single send of the whole run and nothing heals it afterwards (the journal is
// archived), so a placeholder-subject finalize gets a longer cap than the per-event sends.
const FINALIZE_DEADLINE_MS = 20000;
// After a send ends in a timeout or an abort (a BI that hangs rather than refuses), further
// ordinary sends, under the placeholder or the operator's id, are skipped for this long on the run's
// own clock, so a blocked network costs a run one deadline per minute and not one per command.
// Never applies to `finalize` or the `identity` repair.
const SEND_PAUSE_MS = 60 * 1000;
// Journal-lock waits: automatic resume gives up (busy no-op), marker appends are skipped
// (backlog stays for the next heal), explicit commands throw (non-gating in the CLI).
const LOCK_WAIT_MS = 2000;
const MARKER_LOCK_WAIT_MS = 1000;
const COMMAND_LOCK_WAIT_MS = 5000;
const PLACEHOLDER_MARKER_LOCK_WAIT_MS = 500;
const LEGACY_SUBJECT = '__legacy__';
// A wait at least this long with no halt_needs_user event in its stage means the
// capture lost *why* the run stalled — flagged at finalize, never silent.
const WAIT_WITHOUT_HALT_FLAG_MS = 5 * 60 * 1000;
// Metered durations may legitimately overlap slightly with elapsed time (clock granularity,
// concurrent API calls inside one stage), so allow modest slack before calling it a contradiction.
const OVER_ATTRIBUTION_TOLERANCE = 1.25;
// Absolute slack on top of the ratio, so a very short stage is not flagged over a few hundred ms.
const OVER_ATTRIBUTION_FLOOR_MS = 5 * 1000;
const EVIDENCE_SIZE_FLAG_BYTES = 2 * 1024 * 1024;

const STAGES = [
  'config', 'discovery', 'mcp_gate', 'mapping', 'mapping_review', 'setup_discovery',
  'codegen', 'approval_gate', 'setup_provisioning', 'storefront_build', 'extract',
  'import', 'finish',
];
const STAGE_OUTCOMES = ['passed', 'halted', 'failed', 'skipped'];
const TERMINAL_STATES = ['completed', 'halted_needs_user', 'failed', 'abandoned_by_user'];
const SEVERITIES = ['blocking', 'degraded', 'cosmetic', 'info'];
const SOURCE_PLATFORMS = ['wordpress', 'woocommerce', 'shopify', 'csv', 'other'];
const DELIVERY_MODES = ['management', 'website', 'management_and_website'];
const DESTINATION_STRATEGIES = ['new_site', 'existing_site'];
const OPERATOR_ACCEPTANCE = ['accepted', 'rework_needed', 'rejected', 'unknown'];
const VOLUME_TARGETS = ['native', 'cms', 'none'];
const VERIFICATION_METHODS = ['query_back', 'route_check', 'manual_inspection'];

const EVENT_SUBTYPES = {
  halt_needs_user: ['missing_input', 'manual_only', 'systemic_failure'],
  manual_action_required: ['plan_or_billing', 'dashboard_only', 'external_dependency', 'other'],
  error: null,
  fidelity_loss: ['dropped_field', 'unverified_enum', 'no_target', 'coerced_value'],
  api_gap: ['missing_api', 'missing_capability', 'internal_only', 'other'],
  skill_coverage_gap: ['guessed_value', 'undocumented_workaround', 'ambiguous_instruction', 'path_not_covered'],
  user_decision: ['accepted', 'declined', 'deferred', 'amended'],
  pipeline_defect: ['state_inconsistency', 'ordering_violation', 'record_defect', 'other'],
};
const EVENT_TYPES = Object.keys(EVENT_SUBTYPES);

// Privacy tiers travel with the record so no sink can ignore them. Floors, not
// defaults: identifying:* fields are never transmitted raw without the
// pseudonymization floor or the locus-appropriate authority.
const FIELD_TIERS = {
  'rollup.source_site_url': 'identifying:client',
  'rollup.wix_user_id': 'identifying:operator',
  '*': 'behavioral',
};

const OUTCOME_FOR_TERMINAL = {
  completed: 'passed',
  failed: 'failed',
  halted_needs_user: 'halted',
  abandoned_by_user: 'halted',
};

// Last-line-of-defense scrub over free text, shape field names, and locators.
// Order matters: broader assignment/JWT shapes before generic opaque strings.
const SCRUB_PATTERNS = [
  ['secret_assignment', /\b(?:token|secret|password|passwd|api[_-]?key|authorization|bearer)\b\s*[:=]\s*\S+/gi],
  ['jwt', /\beyJ[A-Za-z0-9_-]{8,}(?:\.[A-Za-z0-9_-]{4,}){1,4}\b/g],
  ['url', /\b(?:https?|ftp):\/\/\S+/gi],
  ['email', /[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)+/g],
  ['opaque', /(?=[A-Za-z0-9+/_-]*[0-9])[A-Za-z0-9+/_-]{32,}={0,2}/g],
];

class ValidationError extends Error {
  constructor(errors, hint) {
    super(Array.isArray(errors) ? errors.join('; ') : String(errors));
    this.name = 'ValidationError';
    this.errors = Array.isArray(errors) ? errors : [String(errors)];
    this.hint = hint || null;
  }
}

function nowIso(now) {
  if (now instanceof Date) return now.toISOString();
  if (typeof now === 'string') return new Date(now).toISOString();
  return new Date().toISOString();
}

function tsMs(iso) {
  return Date.parse(iso);
}

function telemetryDir(projectDir) {
  return path.join(projectDir, 'telemetry');
}

function journalPath(projectDir) {
  return path.join(telemetryDir(projectDir), 'events.jsonl');
}

function runsDir(projectDir) {
  return path.join(telemetryDir(projectDir), 'runs');
}

function signalPath(projectDir) {
  return path.join(projectDir, 'run-telemetry.json');
}

function scrubText(value) {
  let hits = 0;
  let text = String(value);
  for (const [name, re] of SCRUB_PATTERNS) {
    text = text.replace(re, () => {
      hits += 1;
      return `[scrubbed:${name}]`;
    });
  }
  return { text, hits };
}

function normalizeToken(value) {
  return String(value)
    .toLowerCase()
    .trim()
    .replace(/[^a-z0-9]+/g, '_')
    .replace(/^_+|_+$/g, '');
}

function stableStringify(value) {
  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(',')}]`;
  }
  if (value && typeof value === 'object') {
    const keys = Object.keys(value).sort();
    return `{${keys.map((k) => `${JSON.stringify(k)}:${stableStringify(value[k])}`).join(',')}}`;
  }
  return JSON.stringify(value);
}

function checkEnum(errors, field, value, allowed) {
  if (!allowed.includes(value)) {
    errors.push(`${field} must be one of: ${allowed.join(', ')} (got: ${JSON.stringify(value)})`);
    return false;
  }
  return true;
}

function checkBoundedText(errors, field, value, { required = false, max = FREE_TEXT_MAX } = {}) {
  if (value === undefined || value === null) {
    if (required) errors.push(`${field} is required (a sentence or two, observation-only)`);
    return null;
  }
  if (typeof value !== 'string' || value.trim() === '') {
    errors.push(`${field} must be a non-empty string`);
    return null;
  }
  if (value.length > max) {
    errors.push(`${field} exceeds ${max} chars (got ${value.length}) — keep it to a sentence or two; the coded fields carry the structure`);
    return null;
  }
  return value.trim();
}

function checkApiSurface(errors, field, value) {
  if (value === undefined || value === null) return null;
  const str = String(value);
  if (/:\/\//.test(str) || /[?\s]/.test(str)) {
    errors.push(`${field} must be an API/endpoint class like "stores/v3" or "wp/v2/posts", never a URL (URLs can embed auth)`);
    return null;
  }
  if (str.length > 64 || !/^[A-Za-z0-9/_.:-]+$/.test(str)) {
    errors.push(`${field} must be a short endpoint class matching [A-Za-z0-9/_.:-], max 64 chars`);
    return null;
  }
  return str;
}

function checkOpaqueId(errors, field, value, { max = 64 } = {}) {
  if (value === undefined || value === null) return null;
  const str = String(value);
  if (/[\s]/.test(str) || /:\/\//.test(str) || str.length > max) {
    errors.push(`${field} must be an opaque identifier (no whitespace, no URL), max ${max} chars`);
    return null;
  }
  return str;
}

function checkCount(errors, field, value, { defaultValue = 0, min = 0 } = {}) {
  if (value === undefined || value === null) return defaultValue;
  if (!Number.isInteger(value) || value < min) {
    errors.push(`${field} must be an integer >= ${min}`);
    return defaultValue;
  }
  return value;
}

function normalizeOrigin(errors, value) {
  try {
    const url = new URL(String(value));
    if (url.protocol !== 'http:' && url.protocol !== 'https:') {
      errors.push('source_site_url must be an http(s) URL');
      return null;
    }
    // Bare origin only — drops path, query, and any embedded credentials.
    return `${url.protocol}//${url.host}`;
  } catch {
    errors.push(`source_site_url is not a parseable URL: ${JSON.stringify(String(value))}`);
    return null;
  }
}

function sanitizeShapeValue(value, errors, keyPath, depth, counter) {
  if (value === null || typeof value === 'number' || typeof value === 'boolean') {
    return value;
  }
  if (typeof value === 'string') {
    if (value.length > 64) {
      errors.push(`observed_shapes${keyPath} string value exceeds 64 chars — shapes carry field names and types, never record values`);
      return null;
    }
    const { text, hits } = scrubText(value);
    counter.hits += hits;
    return text;
  }
  if (depth >= 4) {
    errors.push(`observed_shapes${keyPath} nests deeper than 4 levels`);
    return null;
  }
  if (Array.isArray(value)) {
    return value.map((item, i) => sanitizeShapeValue(item, errors, `${keyPath}[${i}]`, depth + 1, counter));
  }
  if (typeof value === 'object') {
    const out = {};
    for (const [key, item] of Object.entries(value)) {
      // Custom field names on a client site can themselves be identifying.
      const { text, hits } = scrubText(key);
      counter.hits += hits;
      out[text] = sanitizeShapeValue(item, errors, `${keyPath}.${key}`, depth + 1, counter);
    }
    return out;
  }
  errors.push(`observed_shapes${keyPath} has unsupported value type ${typeof value}`);
  return null;
}

function validateEvidenceRefs(errors, value) {
  if (value === undefined || value === null) return null;
  if (!Array.isArray(value) || value.length === 0) {
    errors.push('evidence_refs must be a non-empty array of {artifact, locator?} or omitted');
    return null;
  }
  if (value.length > EVIDENCE_REFS_MAX) {
    errors.push(`evidence_refs is capped at ${EVIDENCE_REFS_MAX} sample refs per event`);
    return null;
  }
  const refs = [];
  for (const [i, ref] of value.entries()) {
    if (!ref || typeof ref !== 'object' || typeof ref.artifact !== 'string') {
      errors.push(`evidence_refs[${i}] must be {artifact: <project-relative path>, locator?: <heading/anchor>}`);
      continue;
    }
    const artifact = ref.artifact.replace(/\\/g, '/');
    if (path.isAbsolute(artifact) || artifact.split('/').includes('..')) {
      errors.push(`evidence_refs[${i}].artifact must be a project-root-relative path (no absolute paths, no ..)`);
      continue;
    }
    if (/(^|\/)config\//.test(artifact) || artifact.endsWith('.env')) {
      errors.push(`evidence_refs[${i}].artifact must never point at secret-bearing config files`);
      continue;
    }
    let locator = null;
    if (ref.locator !== undefined && ref.locator !== null) {
      if (typeof ref.locator !== 'string' || ref.locator.length > 120) {
        errors.push(`evidence_refs[${i}].locator must be a string of at most 120 chars (prefer a section heading — line ranges break on regeneration)`);
        continue;
      }
      locator = scrubText(ref.locator).text;
    }
    refs.push({ artifact, locator });
  }
  return refs;
}

// --- dimension (rollup identity) fields -------------------------------------

// --- cost/latency metering -------------------------------------------------
// Duration fields split `active_ms` into its three real components. Before this existed,
// active_ms was a single wall-clock number fusing agent reasoning, subprocess execution, remote
// API latency and defect-repair time — which made it impossible to localize a bottleneck.
//
// Anything NOT attributed by a meter call stays visible as `unattributed_ms`. That is deliberate:
// a partially-metered stage must not look fully explained.
const METER_DURATION_KEYS = ['model_ms', 'api_ms', 'script_ms'];
const METER_COUNT_KEYS = ['input_tokens', 'output_tokens', 'cache_read_tokens', 'cache_write_tokens', 'api_calls', 'api_retries'];
const METER_ALLOWED_KEYS = new Set([...METER_DURATION_KEYS, ...METER_COUNT_KEYS, 'stage']);

function validateMeter(input, { openStage, lastStage }) {
  const errors = [];
  if (!input || typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['meter payload must be a JSON object']);
  }
  for (const key of Object.keys(input)) {
    if (!METER_ALLOWED_KEYS.has(key)) {
      errors.push(`unknown meter field: ${key} (allowed: ${[...METER_ALLOWED_KEYS].join(', ')})`);
    }
  }
  const stageName = input.stage || openStage || lastStage;
  if (!stageName) errors.push('no stage is open and none was named; pass --stage <stage>');
  else checkEnum(errors, 'stage', stageName, STAGES);

  const out = { stage: stageName };
  let any = false;
  for (const key of [...METER_DURATION_KEYS, ...METER_COUNT_KEYS]) {
    if (input[key] === undefined || input[key] === null) continue;
    const value = input[key];
    if (!Number.isFinite(value) || value < 0 || Math.floor(value) !== value) {
      errors.push(`${key} must be a non-negative integer`);
      continue;
    }
    out[key] = value;
    any = true;
  }
  if (!any) errors.push(`meter needs at least one measurement (${[...METER_DURATION_KEYS, ...METER_COUNT_KEYS].join(', ')})`);
  if (errors.length > 0) {
    throw new ValidationError(errors, 'meter records measured numbers only — never estimate them by hand');
  }
  return out;
}

// --- source key/value reads ----------------------------------------------
//
// Per-run accounting for the ONE capability that widened what this pipeline can read out of a
// source site: key-scoped access to key/value ("EAV") tables. Four counts and a table list.
//
// KEY NAMES ARE NOT TELEMETRY. A run's own artifacts hold those; what is worth aggregating
// ACROSS runs is `redacted_values` -- a non-zero total means some site somewhere keeps a
// credential where nobody expected one, and that is a fleet-level signal no single run can
// see. Table names are bare WordPress table names (`postmeta`), which are structural, not
// customer data. Anything finer-grained stays local.
const SOURCE_READ_COUNT_KEYS = ['discovery_queries', 'scoped_queries', 'tier_refusals', 'redacted_values'];
const SOURCE_READ_ALLOWED_KEYS = new Set([...SOURCE_READ_COUNT_KEYS, 'discovery_tables', 'scoped_tables']);
const SOURCE_READ_TABLE_KEYS = ['discovery_tables', 'scoped_tables'];
const SOURCE_READ_MAX_TABLES = 50;

function validateSourceRead(input) {
  const errors = [];
  if (!input || typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['source-read payload must be a JSON object']);
  }
  for (const key of Object.keys(input)) {
    if (!SOURCE_READ_ALLOWED_KEYS.has(key)) {
      errors.push(`unknown source-read field: ${key} (allowed: ${[...SOURCE_READ_ALLOWED_KEYS].join(', ')})`);
    }
  }
  const out = {};
  let any = false;
  for (const key of SOURCE_READ_COUNT_KEYS) {
    if (input[key] === undefined || input[key] === null) continue;
    const value = input[key];
    if (!Number.isFinite(value) || value < 0 || Math.floor(value) !== value) {
      errors.push(`${key} must be a non-negative integer`);
      continue;
    }
    out[key] = value;
    any = true;
  }
  for (const key of SOURCE_READ_TABLE_KEYS) {
    if (input[key] === undefined || input[key] === null) continue;
    const value = input[key];
    if (!Array.isArray(value) || value.some((t) => typeof t !== 'string' || !/^[A-Za-z0-9_]+$/.test(t))) {
      errors.push(`${key} must be an array of bare table names (^[A-Za-z0-9_]+$)`);
      continue;
    }
    if (value.length > SOURCE_READ_MAX_TABLES) {
      errors.push(`${key} lists ${value.length} tables, more than the ${SOURCE_READ_MAX_TABLES} this field carries`);
      continue;
    }
    out[key] = [...new Set(value)];
    any = true;
  }
  if (!any) errors.push(`source-read needs at least one measurement (${[...SOURCE_READ_ALLOWED_KEYS].join(', ')})`);
  if (errors.length > 0) {
    throw new ValidationError(errors, 'source-read records counted reads only — never key names, and never a value');
  }
  return out;
}

// A pricing snapshot makes historical cost recomputable when list prices change. Rates are per
// MILLION tokens, matching how model pricing is published.
function validatePricingSnapshot(errors, value) {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    errors.push('model_pricing_snapshot must be an object of { <model>: { input_per_mtok, output_per_mtok, ... } }');
    return undefined;
  }
  const out = {};
  const rateKeys = ['input_per_mtok', 'output_per_mtok', 'cache_read_per_mtok', 'cache_write_per_mtok'];
  for (const [model, rates] of Object.entries(value)) {
    if (typeof model !== 'string' || model.length > 64) {
      errors.push('model_pricing_snapshot keys must be short model ids');
      continue;
    }
    if (!rates || typeof rates !== 'object' || Array.isArray(rates)) {
      errors.push(`model_pricing_snapshot.${model} must be an object of numeric rates`);
      continue;
    }
    const entry = {};
    for (const [key, rate] of Object.entries(rates)) {
      if (!rateKeys.includes(key)) {
        errors.push(`model_pricing_snapshot.${model}.${key} is not a known rate (allowed: ${rateKeys.join(', ')})`);
        continue;
      }
      if (!Number.isFinite(rate) || rate < 0) {
        errors.push(`model_pricing_snapshot.${model}.${key} must be a non-negative number`);
        continue;
      }
      entry[key] = rate;
    }
    out[model] = entry;
  }
  return out;
}

const DIM_VALIDATORS = {
  model_pricing_snapshot: (errors, v) => validatePricingSnapshot(errors, v),
  source_platform: (errors, v) => (checkEnum(errors, 'source_platform', v, SOURCE_PLATFORMS) ? v : undefined),
  source_platform_version: (errors, v) => {
    if (typeof v !== 'string' || v.length > 32) {
      errors.push('source_platform_version must be a short version string');
      return undefined;
    }
    return v;
  },
  source_site_url: (errors, v) => {
    const origin = normalizeOrigin(errors, v);
    return origin === null ? undefined : origin;
  },
  source_extensions: (errors, v) => {
    if (!Array.isArray(v) || v.length > CLASS_LIST_MAX || v.some((item) => typeof item !== 'string' || item.length > 48)) {
      errors.push('source_extensions must be an array of short extension/plugin class names');
      return undefined;
    }
    return [...new Set(v.map(normalizeToken).filter(Boolean))];
  },
  discovered_entity_types: (errors, v) => {
    if (!Array.isArray(v) || v.length > CLASS_LIST_MAX || v.some((item) => typeof item !== 'string' || item.length > 48)) {
      errors.push('discovered_entity_types must be an array of short entity class tokens (e.g. product, post, event)');
      return undefined;
    }
    return [...new Set(v.map(normalizeToken).filter(Boolean))];
  },
  source_acquisition: (errors, v) => {
    const token = normalizeToken(v);
    if (!token || token.length > 32) {
      errors.push('source_acquisition must be a short class token, e.g. admin_api | public_storefront | file_export');
      return undefined;
    }
    return token;
  },
  delivery_mode: (errors, v) => (checkEnum(errors, 'delivery_mode', v, DELIVERY_MODES) ? v : undefined),
  destination_strategy: (errors, v) => (checkEnum(errors, 'destination_strategy', v, DESTINATION_STRATEGIES) ? v : undefined),
  site_id: (errors, v) => {
    const id = checkOpaqueId(errors, 'site_id', v);
    return id === null ? undefined : id;
  },
  wix_user_id: (errors, v) => {
    const id = checkOpaqueId(errors, 'wix_user_id', v);
    return id === null ? undefined : id;
  },
  skills_version: (errors, v) => {
    if (typeof v !== 'string' || v.length > 32) {
      errors.push('skills_version must be a short version string');
      return undefined;
    }
    return v.trim();
  },
  skills_commit: (errors, v) => {
    if (typeof v !== 'string' || !/^[0-9a-f]{7,40}$/.test(v.trim())) {
      errors.push('skills_commit must be a git commit SHA (7-40 hex chars)');
      return undefined;
    }
    return v.trim();
  },
  runtime_env: (errors, v) => {
    if (!v || typeof v !== 'object' || Array.isArray(v)) {
      errors.push('runtime_env must be an object with {os?, node_version?, agent_runtime?, model?}');
      return undefined;
    }
    const out = {};
    for (const key of ['os', 'node_version', 'agent_runtime', 'model']) {
      if (v[key] === undefined || v[key] === null) continue;
      if (typeof v[key] !== 'string' || v[key].length > 64 || /[/\\]/.test(v[key])) {
        errors.push(`runtime_env.${key} must be a coarse platform fact (short string, no paths)`);
        continue;
      }
      out[key] = v[key];
    }
    const unknown = Object.keys(v).filter((k) => !['os', 'node_version', 'agent_runtime', 'model'].includes(k));
    if (unknown.length > 0) {
      errors.push(`runtime_env has unknown keys: ${unknown.join(', ')}`);
      return undefined;
    }
    return out;
  },
};

function validateDims(input) {
  if (input === undefined || input === null) return {};
  if (typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['dimensions must be a JSON object']);
  }
  const errors = [];
  const dims = {};
  for (const [key, value] of Object.entries(input)) {
    if (value === undefined || value === null) continue;
    const validator = DIM_VALIDATORS[key];
    if (!validator) {
      errors.push(`unknown dimension field: ${key} (allowed: ${Object.keys(DIM_VALIDATORS).join(', ')})`);
      continue;
    }
    const normalized = validator(errors, value);
    if (normalized !== undefined) dims[key] = normalized;
  }
  if (errors.length > 0) throw new ValidationError(errors);
  return dims;
}

// --- event validation --------------------------------------------------------

const EVENT_ALLOWED_KEYS = new Set([
  'event_type', 'subtype', 'stage', 'skill', 'entity_type', 'severity',
  'wix_api_surface', 'source_api_surface', 'wix_app_id', 'error_code',
  'what_happened', 'expected', 'actual', 'observed_shapes', 'evidence_refs',
  'count', 'retry_count', 'recovered', 'decision_point', 'knowledge_observation',
]);

function validateEvent(input, context) {
  if (!input || typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['event must be a JSON object']);
  }
  const errors = [];
  const scrub = { hits: 0 };

  for (const key of Object.keys(input)) {
    if (!EVENT_ALLOWED_KEYS.has(key)) {
      errors.push(`unknown event field: ${key} — there is deliberately no fix/root_cause/recommendation field; record the observation only`);
    }
  }

  const eventType = input.event_type;
  checkEnum(errors, 'event_type', eventType, EVENT_TYPES);

  let subtype = null;
  if (EVENT_SUBTYPES[eventType] === null) {
    if (input.subtype !== undefined && input.subtype !== null) {
      errors.push(`event_type ${eventType} defines no subtypes (error_code is its discriminator)`);
    }
  } else if (EVENT_SUBTYPES[eventType]) {
    if (checkEnum(errors, `subtype (for ${eventType})`, input.subtype, EVENT_SUBTYPES[eventType])) {
      subtype = input.subtype;
    }
  }

  let stage = input.stage;
  if (stage === undefined || stage === null) {
    stage = context.openStage || context.lastStage || null;
    if (!stage) {
      errors.push('stage is required (no stage has been started yet to default to)');
    }
  } else {
    checkEnum(errors, 'stage', stage, STAGES);
  }

  if (typeof input.skill !== 'string' || !/^[a-z][a-z0-9-]*$/.test(input.skill)) {
    errors.push('skill is required — the active rp-* resource (e.g. "rp-mapper") or "wix-replatform"');
  }

  checkEnum(errors, 'severity', input.severity, SEVERITIES);

  let entityType = null;
  if (input.entity_type !== undefined && input.entity_type !== null) {
    entityType = normalizeToken(input.entity_type);
    if (!entityType || entityType.length > 32) {
      errors.push('entity_type must be a short entity class token (e.g. product, post, media)');
      entityType = null;
    }
  }

  const wixApiSurface = checkApiSurface(errors, 'wix_api_surface', input.wix_api_surface);
  const sourceApiSurface = checkApiSurface(errors, 'source_api_surface', input.source_api_surface);
  const wixAppId = checkOpaqueId(errors, 'wix_app_id', input.wix_app_id);

  let errorCode = null;
  if (input.error_code !== undefined && input.error_code !== null) {
    const str = String(input.error_code);
    if (str.length > 48 || /\s/.test(str)) {
      errors.push('error_code must be a short machine code (e.g. 428, WDE0110)');
    } else {
      errorCode = str;
    }
  }
  if (eventType === 'error' && errorCode === null) {
    errors.push('error events require error_code — it is the discriminator for this type');
  }
  if (eventType === 'api_gap' && wixApiSurface === null) {
    errors.push('api_gap events require wix_api_surface — it is part of the stable signature');
  }

  const whatHappened = checkBoundedText(errors, 'what_happened', input.what_happened, { required: true });
  const expected = checkBoundedText(errors, 'expected', input.expected);
  const actual = checkBoundedText(errors, 'actual', input.actual);

  const count = checkCount(errors, 'count', input.count, { defaultValue: 1, min: 1 });

  let retryCount;
  let recovered;
  if (eventType === 'error') {
    retryCount = checkCount(errors, 'retry_count', input.retry_count, { defaultValue: 0, min: 0 });
    recovered = input.recovered === undefined || input.recovered === null ? false : input.recovered;
    if (typeof recovered !== 'boolean') {
      errors.push('recovered must be a boolean');
      recovered = false;
    }
  } else {
    if (input.retry_count !== undefined || input.recovered !== undefined) {
      errors.push('retry_count/recovered are only valid on error events');
    }
  }

  let decisionPoint;
  if (eventType === 'user_decision') {
    decisionPoint = typeof input.decision_point === 'string' ? normalizeToken(input.decision_point) : '';
    if (!decisionPoint || decisionPoint.length > 48) {
      errors.push('user_decision events require decision_point — which checkpoint or fork (e.g. mapping_review, approval_gate)');
    }
  } else if (input.decision_point !== undefined) {
    errors.push('decision_point is only valid on user_decision events');
  }

  let observedShapes = null;
  if (input.observed_shapes !== undefined && input.observed_shapes !== null) {
    if (!Array.isArray(input.observed_shapes) || input.observed_shapes.length === 0) {
      errors.push('observed_shapes must be a non-empty array of sanitized shape objects or omitted');
    } else if (input.observed_shapes.length > SHAPES_MAX_PER_CALL) {
      errors.push(`observed_shapes is capped at ${SHAPES_MAX_PER_CALL} shapes per call (folding keeps ${SHAPES_KEPT_PER_CLASS} distinct per class)`);
    } else {
      observedShapes = input.observed_shapes.map((shape, i) => {
        if (!shape || typeof shape !== 'object' || Array.isArray(shape)) {
          errors.push(`observed_shapes[${i}] must be an object of field names and types`);
          return null;
        }
        return sanitizeShapeValue(shape, errors, `[${i}]`, 0, scrub);
      });
    }
  }

  const evidenceRefs = validateEvidenceRefs(errors, input.evidence_refs);
  let knowledgeObservation;
  if (input.knowledge_observation !== undefined) {
    try { knowledgeObservation = require('./knowledge-observation').validate(input.knowledge_observation); }
    catch { errors.push('knowledge_observation requires bounded public identifiers and measured or unknown counts'); }
    if (input.actual !== undefined || !['skill_coverage_gap','api_gap','fidelity_loss'].includes(eventType)) errors.push('knowledge_observation owns actual and requires a coverage, API or fidelity event');
  }

  if (errors.length > 0) {
    throw new ValidationError(errors, classKeyFor({
      event_type: eventType, stage, entity_type: entityType, wix_api_surface: wixApiSurface,
      source_api_surface: sourceApiSurface, wix_app_id: wixAppId, error_code: errorCode,
      subtype, decision_point: decisionPoint,
    }));
  }

  const scrubbedWhat = scrubText(whatHappened);
  const scrubbedExpected = expected === null ? null : scrubText(expected);
  const scrubbedActual = actual === null ? null : scrubText(actual);
  scrub.hits += scrubbedWhat.hits + (scrubbedExpected ? scrubbedExpected.hits : 0) + (scrubbedActual ? scrubbedActual.hits : 0);

  const event = {
    event_type: eventType,
    subtype,
    stage,
    skill: input.skill,
    entity_type: entityType,
    severity: input.severity,
    wix_api_surface: wixApiSurface,
    source_api_surface: sourceApiSurface,
    wix_app_id: wixAppId,
    error_code: errorCode,
    what_happened: scrubbedWhat.text,
    expected: scrubbedExpected ? scrubbedExpected.text : null,
    actual: knowledgeObservation ? JSON.stringify({knowledgeObservation}) : scrubbedActual ? scrubbedActual.text : null,
    ...(knowledgeObservation ? {knowledge_observation:knowledgeObservation} : {}),
    observed_shapes: observedShapes,
    evidence_refs: evidenceRefs,
    count,
  };
  if (eventType === 'error') {
    event.retry_count = retryCount;
    event.recovered = recovered;
  }
  if (eventType === 'user_decision') {
    event.decision_point = decisionPoint;
  }
  return { event, classKey: classKeyFor(event), scrubHits: scrub.hits };
}

function classKeyFor(event) {
  const parts = [
    event.event_type, event.stage, event.entity_type, event.wix_api_surface,
    event.source_api_surface, event.wix_app_id, event.error_code, event.subtype,
  ];
  if (event.event_type === 'user_decision') parts.push(event.decision_point);
  if (event.knowledge_observation) parts.push(event.knowledge_observation.signature,event.knowledge_observation.toolkitVersion,event.knowledge_observation.outcome,event.knowledge_observation.verificationLevel,event.knowledge_observation.affectedCount);
  return parts.map((p) => (p === null || p === undefined ? '' : String(p))).join('|');
}

// --- rollup (finalize) validation --------------------------------------------

function validateVolumes(errors, input) {
  if (input === undefined || input === null) return [];
  if (!Array.isArray(input)) {
    errors.push('volumes must be an array of per-entity-type volume records');
    return [];
  }
  const volumes = [];
  for (const [i, raw] of input.entries()) {
    if (!raw || typeof raw !== 'object') {
      errors.push(`volumes[${i}] must be an object`);
      continue;
    }
    const entityType = typeof raw.entity_type === 'string' ? normalizeToken(raw.entity_type) : '';
    if (!entityType) {
      errors.push(`volumes[${i}].entity_type is required`);
      continue;
    }
    let target = null;
    if (raw.target !== undefined && raw.target !== null) {
      if (checkEnum(errors, `volumes[${i}].target`, raw.target, VOLUME_TARGETS)) target = raw.target;
    }
    const targetSurface = checkApiSurface(errors, `volumes[${i}].target_surface`, raw.target_surface);
    const volume = {
      entity_type: entityType,
      target,
      target_surface: targetSurface,
      discovered: checkCount(errors, `volumes[${i}].discovered`, raw.discovered),
      planned: raw.planned === undefined || raw.planned === null
        ? null
        : checkCount(errors, `volumes[${i}].planned`, raw.planned),
      attempted: checkCount(errors, `volumes[${i}].attempted`, raw.attempted),
      succeeded: checkCount(errors, `volumes[${i}].succeeded`, raw.succeeded),
      failed: checkCount(errors, `volumes[${i}].failed`, raw.failed),
      skipped: checkCount(errors, `volumes[${i}].skipped`, raw.skipped),
      already_imported: checkCount(errors, `volumes[${i}].already_imported`, raw.already_imported),
    };
    if (volume.attempted !== volume.succeeded + volume.failed) {
      errors.push(`volumes[${i}] (${entityType}): attempted must equal succeeded + failed (${volume.attempted} != ${volume.succeeded} + ${volume.failed})`);
    }
    volumes.push(volume);
  }
  return volumes;
}

function validateVerification(errors, input) {
  if (input === undefined || input === null) return [];
  if (!Array.isArray(input)) {
    errors.push('verification must be an array of verification records');
    return [];
  }
  const records = [];
  for (const [i, raw] of input.entries()) {
    if (!raw || typeof raw !== 'object') {
      errors.push(`verification[${i}] must be an object`);
      continue;
    }
    const subject = typeof raw.subject === 'string' ? normalizeToken(raw.subject) : '';
    if (!subject) {
      errors.push(`verification[${i}].subject is required (an entity_type, or "routes")`);
      continue;
    }
    if (!checkEnum(errors, `verification[${i}].method`, raw.method, VERIFICATION_METHODS)) continue;
    const record = {
      subject,
      method: raw.method,
      checked: checkCount(errors, `verification[${i}].checked`, raw.checked),
      passed: checkCount(errors, `verification[${i}].passed`, raw.passed),
      failed: checkCount(errors, `verification[${i}].failed`, raw.failed),
    };
    if (record.passed + record.failed > record.checked) {
      errors.push(`verification[${i}] (${subject}): passed + failed must not exceed checked`);
    }
    records.push(record);
  }
  return records;
}

function validateFinalizeInput(input, context) {
  if (!input || typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['finalize payload must be a JSON object']);
  }
  const errors = [];
  const dimKeys = {};
  const known = new Set(['terminal_state', 'stopped_at_stage', 'operator_acceptance', 'volumes', 'verification']);
  for (const [key, value] of Object.entries(input)) {
    if (known.has(key)) continue;
    if (DIM_VALIDATORS[key]) {
      dimKeys[key] = value;
      continue;
    }
    errors.push(`unknown finalize field: ${key}`);
  }

  checkEnum(errors, 'terminal_state', input.terminal_state, TERMINAL_STATES);

  let stoppedAtStage = null;
  if (input.terminal_state === 'completed') {
    if (input.stopped_at_stage !== undefined && input.stopped_at_stage !== null) {
      errors.push('stopped_at_stage must be null when terminal_state is completed');
    }
  } else if (input.stopped_at_stage !== undefined && input.stopped_at_stage !== null) {
    if (checkEnum(errors, 'stopped_at_stage', input.stopped_at_stage, STAGES)) {
      stoppedAtStage = input.stopped_at_stage;
    }
  } else {
    stoppedAtStage = context.openStage || context.lastStage || 'config';
  }

  let operatorAcceptance = 'unknown';
  if (input.operator_acceptance !== undefined && input.operator_acceptance !== null) {
    if (checkEnum(errors, 'operator_acceptance', input.operator_acceptance, OPERATOR_ACCEPTANCE)) {
      operatorAcceptance = input.operator_acceptance;
    }
  }

  const volumes = validateVolumes(errors, input.volumes);
  const verification = validateVerification(errors, input.verification);

  let dims = {};
  try {
    dims = validateDims(dimKeys);
  } catch (error) {
    errors.push(...error.errors);
  }

  if (errors.length > 0) throw new ValidationError(errors);
  return {
    terminal_state: input.terminal_state,
    stopped_at_stage: stoppedAtStage,
    operator_acceptance: operatorAcceptance,
    volumes,
    verification,
    dims,
  };
}

// --- transcript_digest validation ---------------------------------
// Every field is derived, machine-computed data (transcript-digest.js), never
// agent free text — so the gate here checks shape and bounds, not prose rules.

const TOOL_NAME_MAX = 80;

function checkToolName(errors, field, value) {
  if (typeof value !== 'string' || value.length === 0 || value.length > TOOL_NAME_MAX || /\s/.test(value)) {
    errors.push(`${field} must be a short tool name (no whitespace, max ${TOOL_NAME_MAX} chars)`);
    return null;
  }
  return value;
}

function validateNamedCountList(errors, field, value, { max = 10 } = {}) {
  if (value === undefined || value === null) return [];
  if (!Array.isArray(value) || value.length > max) {
    errors.push(`${field} must be an array of at most ${max} entries`);
    return [];
  }
  return value.map((entry, i) => {
    if (!entry || typeof entry !== 'object') {
      errors.push(`${field}[${i}] must be an object`);
      return null;
    }
    return entry;
  }).filter(Boolean);
}

function validateTranscriptDigest(input) {
  if (!input || typeof input !== 'object' || Array.isArray(input)) {
    throw new ValidationError(['transcript_digest payload must be a JSON object']);
  }
  const errors = [];
  const known = new Set(['source', 'cost', 'interaction', 'time', 'tools', 'signals']);
  for (const key of Object.keys(input)) {
    if (!known.has(key)) errors.push(`unknown transcript_digest field: ${key}`);
  }

  const source = input.source || {};
  const sourceOut = {
    sessions: checkCount(errors, 'source.sessions', source.sessions, { defaultValue: 1, min: 1 }),
    transcript_turns: checkCount(errors, 'source.transcript_turns', source.transcript_turns),
    cc_version: null,
    tail_complete: typeof source.tail_complete === 'boolean' ? source.tail_complete : null,
  };
  if (sourceOut.tail_complete === null) errors.push('source.tail_complete must be a boolean');
  if (source.cc_version !== undefined && source.cc_version !== null) {
    if (typeof source.cc_version !== 'string' || source.cc_version.length > 32) {
      errors.push('source.cc_version must be a short version string');
    } else {
      sourceOut.cc_version = source.cc_version;
    }
  }

  const cost = input.cost || {};
  const costOut = {
    input_tokens: checkCount(errors, 'cost.input_tokens', cost.input_tokens),
    output_tokens: checkCount(errors, 'cost.output_tokens', cost.output_tokens),
    cache_read_input_tokens: checkCount(errors, 'cost.cache_read_input_tokens', cost.cache_read_input_tokens),
    cache_creation_input_tokens: checkCount(errors, 'cost.cache_creation_input_tokens', cost.cache_creation_input_tokens),
  };

  const interaction = input.interaction || {};
  const interactionOut = {
    human_turns: checkCount(errors, 'interaction.human_turns', interaction.human_turns),
    agent_turns: checkCount(errors, 'interaction.agent_turns', interaction.agent_turns),
    sidechain_turns: checkCount(errors, 'interaction.sidechain_turns', interaction.sidechain_turns),
  };

  const time = input.time || {};
  const timeOut = {
    wall_ms: checkCount(errors, 'time.wall_ms', time.wall_ms),
    active_ms: checkCount(errors, 'time.active_ms', time.active_ms),
    idle_ms: checkCount(errors, 'time.idle_ms', time.idle_ms),
    longest_gap_ms: checkCount(errors, 'time.longest_gap_ms', time.longest_gap_ms),
  };

  const tools = input.tools || {};
  const topFailing = validateNamedCountList(errors, 'tools.top_failing', tools.top_failing).map((entry, i) => ({
    tool: checkToolName(errors, `tools.top_failing[${i}].tool`, entry.tool),
    n: checkCount(errors, `tools.top_failing[${i}].n`, entry.n, { defaultValue: 1, min: 1 }),
  }));
  const toolsOut = {
    calls: checkCount(errors, 'tools.calls', tools.calls),
    failures: checkCount(errors, 'tools.failures', tools.failures),
    top_failing: topFailing,
  };

  const signals = input.signals || {};
  const retryLoops = validateNamedCountList(errors, 'signals.retry_loops', signals.retry_loops).map((entry, i) => {
    let stage = null;
    if (entry.stage !== undefined && entry.stage !== null) {
      if (checkEnum(errors, `signals.retry_loops[${i}].stage`, entry.stage, STAGES)) stage = entry.stage;
    }
    return {
      tool: checkToolName(errors, `signals.retry_loops[${i}].tool`, entry.tool),
      repeats: checkCount(errors, `signals.retry_loops[${i}].repeats`, entry.repeats, { defaultValue: 1, min: 1 }),
      stage,
    };
  });
  const signalsOut = {
    retry_loops: retryLoops,
    hook_blocks: checkCount(errors, 'signals.hook_blocks', signals.hook_blocks),
    hook_errors: checkCount(errors, 'signals.hook_errors', signals.hook_errors),
    files_edited_mid_run: checkCount(errors, 'signals.files_edited_mid_run', signals.files_edited_mid_run),
  };

  if (errors.length > 0) throw new ValidationError(errors);
  return {
    source: sourceOut, cost: costOut, interaction: interactionOut, time: timeOut, tools: toolsOut, signals: signalsOut,
  };
}

// --- journal ------------------------------------------------------------------

function readJournalFile(file) {
  const records = [];
  let malformed = 0;
  for (const line of fs.readFileSync(file, 'utf8').split('\n')) {
    if (line.trim() === '') continue;
    try {
      records.push(JSON.parse(line));
    } catch {
      malformed += 1;
    }
  }
  return { records, malformed };
}

function readJournal(projectDir) {
  const file = journalPath(projectDir);
  if (!fs.existsSync(file)) return null;
  return readJournalFile(file);
}

function appendJournal(projectDir, record) {
  const file = journalPath(projectDir);
  fs.mkdirSync(path.dirname(file), { recursive: true });
  fs.appendFileSync(file, `${JSON.stringify(record)}\n`, 'utf8');
}

// Several records in ONE appendFileSync, so a transaction is all-or-a-torn-suffix.
function appendJournalRecords(projectDir, records) {
  if (records.length === 0) return;
  const file = journalPath(projectDir);
  fs.mkdirSync(path.dirname(file), { recursive: true });
  fs.appendFileSync(file, records.map((r) => `${JSON.stringify(r)}\n`).join(''), 'utf8');
}

// --- journal lock -----------------------------------------------------------------

const LOCK_CLAIM_RE = /^([^.]+)\.(\d+)\.[a-f0-9-]+\.owner$/;

function sleepSync(ms) {
  if (ms > 0) Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms);
}

function lockBusyError(lockPath, detail) {
  const error = new ValidationError(
    [`telemetry journal is locked by another process (${detail}): ${lockPath}`],
    'another telemetry writer is active on this project; retry the call',
  );
  error.code = 'ELOCKBUSY';
  return error;
}

// Test-only hook: when RP_TELEMETRY_TEST_LOCK_HOLD_UNTIL names a file, a lock holder waits
// (bounded) until that file exists before running its critical section, so a test can force
// the lock acquisition order between two processes. Unset in every real run.
function testHoldHook() {
  const until = process.env.RP_TELEMETRY_TEST_LOCK_HOLD_UNTIL;
  if (!until) return;
  const deadline = Date.now() + 10000;
  while (!fs.existsSync(until) && Date.now() < deadline) sleepSync(20);
}

// One writer per project journal. Same claim-file algorithm as local-state's withStateLock
// (a unique <host>.<pid>.<uuid>.owner claim published before scanning; a same-host dead-PID
// claim is removed; the container directory is never deleted), in `telemetry/.lock/` — never
// the importer's `state/.lock`. Synchronous on purpose: the critical section is local I/O
// only (a fresh read, seq computation and one append), never a network call, so the lock is
// never held across an await and the command API keeps its sync functions sync. Contention
// retries every ~50 ms up to `waitMs`; a claim from another host cannot be checked and is
// treated as busy at once. Busy → ValidationError with code ELOCKBUSY.
function withJournalLock(projectDir, fn, { waitMs = COMMAND_LOCK_WAIT_MS } = {}) {
  const lockPath = path.join(telemetryDir(projectDir), '.lock');
  fs.mkdirSync(lockPath, { recursive: true });
  const host = Buffer.from(os.hostname()).toString('base64url');
  const deadline = Date.now() + waitMs;
  for (;;) {
    const claim = `${host}.${process.pid}.${crypto.randomUUID()}.owner`;
    const claimPath = path.join(lockPath, claim);
    fs.writeFileSync(claimPath, JSON.stringify({ pid: process.pid, host, acquiredAt: new Date().toISOString() }), { flag: 'wx', mode: 0o600 });
    let blocker = null;
    try {
      for (const entry of fs.readdirSync(lockPath)) {
        if (entry === claim) continue;
        const match = LOCK_CLAIM_RE.exec(entry);
        if (!match || match[1] !== host) { blocker = { foreign: true, detail: `claim ${entry} cannot be checked locally` }; break; }
        try {
          process.kill(Number(match[2]), 0);
        } catch (error) {
          if (error.code === 'ESRCH') { fs.rmSync(path.join(lockPath, entry), { force: true }); continue; }
          if (error.code !== 'EPERM') throw error;
        }
        blocker = { foreign: false, detail: `held by PID ${match[2]}` };
        break;
      }
    } catch (error) {
      fs.rmSync(claimPath, { force: true });
      throw error;
    }
    if (!blocker) {
      try {
        testHoldHook();
        return fn();
      } finally {
        fs.rmSync(claimPath, { force: true });
      }
    }
    fs.rmSync(claimPath, { force: true });
    const remaining = deadline - Date.now();
    if (blocker.foreign || remaining <= 0) throw lockBusyError(lockPath, blocker.detail);
    sleepSync(Math.min(remaining, 25 + Math.floor(Math.random() * 50)));
  }
}

function replay(records) {
  const state = {
    runStart: null,
    dims: {},
    sessionCount: 1,
    seq: 0,
    events: [],
    rejected: [],
    entries: [],
    waits: [],
    meters: [],
    sourceReads: [],
    openStage: null,
    openStageStart: null,
    lastStage: null,
    openWait: null,
    scrubHits: 0,
    biPushFailures: 0,
    biPushTruncated: 0,
    biPushSkipped: 0,
    lastBiPushOkSeq: 0,
    // Highest confirmed `through_seq` per subject the rows were sent under; a marker written
    // before subjects were recorded is under LEGACY_SUBJECT (it was always the operator's id).
    biOkBySubject: {},
    // The run's placeholder subject (random GUID) while the operator's id is unknown.
    provisionalSubject: null,
    lastEventSeq: 0,
    finalize: null,
    lastTs: null,
    // The activity clock: the MAXIMUM parseable ts over every record that is not a transport
    // marker (bi_push_*). Markers are bookkeeping about delivery and may be appended late with
    // the initiating call's older ts, so they never move it, and a maximum never goes backwards.
    lastActivityTs: null,
    // The current attempt's latest session opener (run_start or session_start), in file order.
    lastOpener: null,
    transcriptDigest: null,
  };
  let lastActivityMs = -Infinity;
  for (const record of records) {
    state.seq = Math.max(state.seq, record.seq || 0);
    state.lastTs = record.ts || state.lastTs;
    if (typeof record.type === 'string' && !record.type.startsWith('bi_push_') && typeof record.ts === 'string') {
      const ms = tsMs(record.ts);
      if (Number.isFinite(ms) && ms > lastActivityMs) {
        lastActivityMs = ms;
        state.lastActivityTs = record.ts;
      }
    }
    switch (record.type) {
      case 'run_start':
        state.runStart = record;
        state.lastOpener = record;
        Object.assign(state.dims, record.dims || {});
        if (biSink.isGuid(record.provisional_subject)) state.provisionalSubject = record.provisional_subject;
        break;
      case 'session_start':
        state.sessionCount = record.session;
        state.lastOpener = record;
        break;
      case 'dims':
        Object.assign(state.dims, record.dims || {});
        break;
      case 'stage_start':
        state.openStage = record.stage;
        state.openStageStart = record.ts;
        state.lastStage = record.stage;
        break;
      case 'stage_end':
        if (state.openStage === record.stage) {
          state.entries.push({ stage: record.stage, start: state.openStageStart, end: record.ts, outcome: record.outcome });
          state.openStage = null;
          state.openStageStart = null;
        }
        break;
      case 'wait_start':
        state.openWait = { stage: record.stage, start: record.ts };
        break;
      case 'wait_end':
        if (state.openWait) {
          state.waits.push({ ...state.openWait, end: record.ts });
          state.openWait = null;
        }
        break;
      case 'meter':
        // Accumulate: a stage may be entered more than once, and each generated script reports its
        // own invocation. Summing is the only reading that survives a resume.
        state.meters.push(record.meter);
        break;
      case 'source_read':
        // Same accumulate-don't-overwrite reasoning: a run reads a source site across several
        // stages and possibly across a resume, and only the sum is the run's real total.
        state.sourceReads.push(record.source_read);
        break;
      case 'event':
        state.events.push(record);
        state.scrubHits += record.scrub_hits || 0;
        state.lastEventSeq = Math.max(state.lastEventSeq, record.seq || 0);
        break;
      case 'rejected':
        state.rejected.push(record);
        break;
      case 'bi_push_failed':
        state.biPushFailures += 1;
        break;
      case 'bi_push_truncated':
        state.biPushTruncated += record.rows || 1;
        break;
      case 'bi_push_skipped':
        state.biPushSkipped += record.rows || 1;
        break;
      case 'bi_push_ok': {
        const through = record.through_seq || 0;
        state.lastBiPushOkSeq = Math.max(state.lastBiPushOkSeq, through);
        const key = typeof record.subject === 'string' ? record.subject : LEGACY_SUBJECT;
        state.biOkBySubject[key] = Math.max(state.biOkBySubject[key] || 0, through);
        break;
      }
      case 'provisional_subject':
        if (!state.provisionalSubject && biSink.isGuid(record.subject)) state.provisionalSubject = record.subject;
        break;
      case 'finalize':
        state.finalize = record;
        break;
      case 'transcript_digest':
        state.transcriptDigest = {
          source: record.source,
          cost: record.cost,
          interaction: record.interaction,
          time: record.time,
          tools: record.tools,
          signals: record.signals,
        };
        break;
      default:
        break;
    }
  }
  return state;
}

function requireActiveRun(projectDir) {
  const journal = readJournal(projectDir);
  if (!journal) {
    throw new ValidationError(
      ['no active telemetry run in this project'],
      "call `rp-telemetry.js start '<dims-json>'` first",
    );
  }
  const state = replay(journal.records);
  if (!state.runStart) {
    throw new ValidationError(['telemetry journal is corrupt: missing run-start record']);
  }
  if (state.finalize) {
    // A crash after the finalize append but before archival: finish the archival
    // now, then report no active run.
    archiveFinalizedJournal(projectDir, journal, state);
    throw new ValidationError(
      ['the previous run is already finalized'],
      "call `rp-telemetry.js start '<dims-json>'` to open the next run",
    );
  }
  return { journal, state };
}

// --- fold + document assembly ---------------------------------------------------

function foldEvents(eventRecords, runId) {
  const byClass = new Map();
  for (const record of eventRecords) {
    const e = record.event;
    let folded = byClass.get(record.class_key);
    if (!folded) {
      folded = {
        run_id: runId,
        seq: record.seq,
        count: 0,
        stage: e.stage,
        skill: e.skill,
        event_type: e.event_type,
        subtype: e.subtype,
        entity_type: e.entity_type,
        severity: e.severity,
        wix_api_surface: e.wix_api_surface,
        source_api_surface: e.source_api_surface,
        wix_app_id: e.wix_app_id,
        error_code: e.error_code,
        what_happened: e.what_happened,
        expected: e.expected,
        actual: e.actual,
        ...(e.knowledge_observation ? {knowledge_observation:e.knowledge_observation} : {}),
        observed_shapes: null,
        evidence_refs: null,
        _shapes: new Map(),
        _refs: new Map(),
      };
      if (e.event_type === 'error') {
        folded.retry_count = 0;
        folded.recovered = false;
      }
      if (e.event_type === 'user_decision') folded.decision_point = e.decision_point;
      byClass.set(record.class_key, folded);
    }
    folded.count += e.count;
    if (e.event_type === 'error') {
      folded.retry_count += e.retry_count || 0;
      // A retry chain is often recorded per attempt. The latest occurrence knows
      // whether the operation ultimately succeeded, and a recovering occurrence's
      // `actual` carries the change that made it succeed — the first (failing)
      // attempt's fields must never erase that half of the observation.
      folded.recovered = Boolean(e.recovered);
      if (e.recovered && e.actual) folded.actual = e.actual;
    }
    for (const shape of e.observed_shapes || []) {
      const key = stableStringify(shape);
      if (!folded._shapes.has(key)) folded._shapes.set(key, shape);
    }
    for (const ref of e.evidence_refs || []) {
      const key = `${ref.artifact}#${ref.locator || ''}`;
      if (!folded._refs.has(key)) folded._refs.set(key, ref);
    }
  }
  const events = [...byClass.values()].sort((a, b) => a.seq - b.seq);
  for (const event of events) {
    const shapes = [...event._shapes.values()];
    if (shapes.length > 0) {
      event.observed_shapes = shapes.slice(0, SHAPES_KEPT_PER_CLASS);
      // Surfaced bound, never a silent one.
      if (shapes.length > SHAPES_KEPT_PER_CLASS) event.shapes_seen = shapes.length;
    }
    const refs = [...event._refs.values()];
    if (refs.length > 0) event.evidence_refs = refs.slice(0, EVIDENCE_REFS_MAX);
    delete event._shapes;
    delete event._refs;
  }
  return events;
}

// Event types that mean the stage did not run clean. `contained_recovery` is DERIVED from these
// rather than self-reported, because a stage that spent its time debugging is exactly the stage an
// agent is least likely to remember to flag.
const RECOVERY_EVENT_TYPES = new Set(['error', 'pipeline_defect']);

function computeStages(state, finalizeTs) {
  const entries = [...state.entries];
  const waits = [...state.waits];
  if (state.openWait) {
    waits.push({ ...state.openWait, end: finalizeTs });
  }
  const byStage = new Map();
  const order = [];
  const stageRecord = (stage) => {
    if (!byStage.has(stage)) {
      byStage.set(stage, { stage, outcome: 'halted', active_ms: 0, waiting_ms: 0, _entries: [] });
      order.push(stage);
    }
    return byStage.get(stage);
  };
  for (const entry of entries) {
    const record = stageRecord(entry.stage);
    record.outcome = entry.outcome;
    record._entries.push(entry);
  }
  for (const wait of waits) {
    const record = stageRecord(wait.stage || 'config');
    record.waiting_ms += Math.max(0, tsMs(wait.end) - tsMs(wait.start));
  }
  // Metered measurements, summed per stage. A stage named by a meter but never opened still gets a
  // record, so a mis-attributed meter is visible instead of silently discarded.
  for (const meter of state.meters || []) {
    const record = stageRecord(meter.stage);
    for (const key of [...METER_DURATION_KEYS, ...METER_COUNT_KEYS]) {
      if (meter[key] === undefined) continue;
      record[key] = (record[key] || 0) + meter[key];
    }
  }

  // Stages that contained an error or a pipeline defect. Without this, a clean run and a thrash are
  // indistinguishable in the rollup, and every per-stage duration silently includes repair time.
  const recoveryStages = new Set(
    (state.events || [])
      .filter((e) => RECOVERY_EVENT_TYPES.has(e.event && e.event.event_type ? e.event.event_type : e.event_type))
      .map((e) => (e.event && e.event.stage ? e.event.stage : e.stage))
      .filter(Boolean),
  );

  for (const stage of order) {
    const record = byStage.get(stage);
    let active = 0;
    for (const entry of record._entries) {
      active += Math.max(0, tsMs(entry.end) - tsMs(entry.start));
      for (const wait of waits) {
        if ((wait.stage || 'config') !== stage) continue;
        const overlap = Math.min(tsMs(entry.end), tsMs(wait.end)) - Math.max(tsMs(entry.start), tsMs(wait.start));
        if (overlap > 0) active -= overlap;
      }
    }
    record.active_ms = Math.max(0, active);
    record.contained_recovery = recoveryStages.has(stage);
    // The remainder of active_ms that no meter explained. This is the honesty field: it is how much
    // of the stage is still a black box, and it must never be silently folded into a component.
    const attributed = METER_DURATION_KEYS.reduce((sum, key) => sum + (record[key] || 0), 0);
    record.unattributed_ms = Math.max(0, record.active_ms - attributed);
    delete record._entries;
  }
  return order.map((stage) => byStage.get(stage));
}

// Cost is DERIVED from token counts and the pricing snapshot, never recorded as a figure — a stored
// dollar amount silently goes wrong the moment list prices change, while tokens are a fact.
function computeCost(stages, pricingSnapshot, model) {
  const totals = { input_tokens: 0, output_tokens: 0, cache_read_tokens: 0, cache_write_tokens: 0 };
  for (const stage of stages) {
    for (const key of Object.keys(totals)) totals[key] += stage[key] || 0;
  }
  const anyTokens = Object.values(totals).some((v) => v > 0);
  const rates = pricingSnapshot && model ? pricingSnapshot[model] : null;
  if (!anyTokens || !rates) {
    return { tokens: totals, estimated_cost_usd: null, cost_basis: rates ? 'no_tokens_recorded' : 'no_pricing_snapshot' };
  }
  const perM = (count, rate) => (rate === undefined ? 0 : (count / 1e6) * rate);
  const usd = perM(totals.input_tokens, rates.input_per_mtok)
    + perM(totals.output_tokens, rates.output_per_mtok)
    + perM(totals.cache_read_tokens, rates.cache_read_per_mtok)
    + perM(totals.cache_write_tokens, rates.cache_write_per_mtok);
  return { tokens: totals, estimated_cost_usd: Number(usd.toFixed(6)), cost_basis: 'derived_from_snapshot' };
}

const MANIFEST_EXCLUDE_DIRS = new Set(['config', 'data', 'preview', 'node_modules', 'frontend', 'telemetry']);
// .ndjson: the pipeline's audit logs (logs/audit-*.ndjson) are named
// evidence — a pipeline_defect deep-dive depends on them being in the bundle.
const MANIFEST_EXTENSIONS = new Set(['.md', '.json', '.jsonl', '.ndjson', '.js', '.mjs', '.cjs', '.ts', '.sh', '.yaml', '.yml']);

function walkEvidence(projectDir) {
  const files = [];
  const walk = (rel) => {
    const abs = rel === '' ? projectDir : path.join(projectDir, rel);
    const entries = fs.readdirSync(abs, { withFileTypes: true }).sort((a, b) => a.name.localeCompare(b.name));
    for (const entry of entries) {
      if (entry.name.startsWith('.')) continue;
      const relPath = rel === '' ? entry.name : `${rel}/${entry.name}`;
      if (entry.isDirectory()) {
        if (MANIFEST_EXCLUDE_DIRS.has(entry.name)) continue;
        walk(relPath);
        continue;
      }
      if (!entry.isFile()) continue;
      if (entry.name.endsWith('.env')) continue;
      if (relPath === 'run-telemetry.json') continue; // added explicitly, first
      if (!MANIFEST_EXTENSIONS.has(path.extname(entry.name))) continue;
      files.push({ path: relPath, bytes: fs.statSync(path.join(projectDir, relPath)).size });
    }
  };
  walk('');
  return files;
}

function assembleDocument(projectDir, records, malformedLines = 0) {
  const state = replay(records);
  state.malformedLines = malformedLines;
  const runStart = state.runStart;
  const payload = state.finalize.payload;
  const finalizeTs = state.finalize.ts;
  const dims = { ...state.dims, ...payload.dims };

  const events = foldEvents(state.events, runStart.run_id);
  const stages = computeStages(state, finalizeTs);
  const activeMs = stages.reduce((sum, s) => sum + s.active_ms, 0);
  const waitingMs = stages.reduce((sum, s) => sum + s.waiting_ms, 0);

  // The headline question this schema exists to answer: how much of the run was deterministic code
  // versus model reasoning. `unattributed_ms` is reported alongside so the split is never mistaken
  // for a complete accounting of active_ms.
  const sumOver = (key) => stages.reduce((sum, s) => sum + (s[key] || 0), 0);
  const agenticMs = sumOver('model_ms');
  const deterministicMs = sumOver('script_ms') + sumOver('api_ms');
  const unattributedMs = sumOver('unattributed_ms');
  const recoveryStages = stages.filter((s) => s.contained_recovery).map((s) => s.stage);
  const cost = computeCost(
    stages,
    dims.model_pricing_snapshot || null,
    (runStart.runtime_env && runStart.runtime_env.model) || (dims.runtime_env && dims.runtime_env.model) || null,
  );
  const timing = {
    active_ms: activeMs,
    waiting_ms: waitingMs,
    agentic_ms: agenticMs,
    deterministic_ms: deterministicMs,
    unattributed_ms: unattributedMs,
    // What fraction of the EXPLAINED time was the model. Null when nothing was metered, rather
    // than a misleading 0.
    agentic_share: agenticMs + deterministicMs > 0
      ? Number((agenticMs / (agenticMs + deterministicMs)).toFixed(4))
      : null,
    api_calls: sumOver('api_calls'),
    api_retries: sumOver('api_retries'),
    stages_with_recovery: recoveryStages,
  };

  const eventCountByType = {};
  for (const event of events) {
    const entry = eventCountByType[event.event_type] || { classes: 0, occurrences: 0 };
    entry.classes += 1;
    entry.occurrences += event.count;
    eventCountByType[event.event_type] = entry;
  }

  const acceptedClasses = new Set(state.events.map((r) => r.class_key));
  const dropped = state.rejected.filter((r) => !r.class_key || !acceptedClasses.has(r.class_key)).length;

  const flags = [];
  const hasEventType = (...types) => events.some((e) => types.includes(e.event_type));
  if (payload.volumes.some((v) => v.failed > 0) && !hasEventType('error')) {
    flags.push('volumes_report_failures_but_no_error_events');
  }
  for (const volume of payload.volumes) {
    if (volume.planned !== null && volume.attempted > volume.planned + volume.already_imported) {
      // Execution diverging from the approved plan must surface mechanically.
      flags.push(`attempted_exceeds_planned:${volume.entity_type}`);
    }
    if (volume.attempted + volume.skipped + volume.already_imported > volume.discovered) {
      flags.push(`volumes_exceed_discovered:${volume.entity_type}`);
    }
  }
  // Metering flags cover data that is WRONG, never data that is merely ABSENT.
  //
  // A run that meters nothing is not unhealthy — it is a runtime that does not report usage, which
  // describes every run recorded before metering existed. Flagging those would fire on ~all
  // historical runs and train readers to ignore the flag list. Absence is already visible, and
  // queryable, in `timing`: `agentic_share` is null and `unattributed_ms` carries the whole stage.
  // So only contradictions are flagged here, and each one implies metering was in use.
  for (const stage of stages) {
    const attributed = METER_DURATION_KEYS.reduce((sum, key) => sum + (stage[key] || 0), 0);
    if (attributed === 0) continue;
    // Partially metered is more dangerous than not metered at all: it looks explained.
    if (stage.unattributed_ms > stage.active_ms * 0.5) {
      flags.push(`stage_mostly_unattributed:${stage.stage}`);
    }
    // Attributed time exceeding elapsed time is impossible, so it means the measurements are wrong:
    // usually the same interval metered twice across a resume, or concurrent work summed serially.
    // unattributed_ms clamps at 0, so without this flag the contradiction would be invisible.
    if (attributed > stage.active_ms * OVER_ATTRIBUTION_TOLERANCE + OVER_ATTRIBUTION_FLOOR_MS) {
      flags.push(`stage_over_attributed:${stage.stage}`);
    }
  }
  if (cost.cost_basis === 'no_pricing_snapshot' && (cost.tokens.input_tokens > 0 || cost.tokens.output_tokens > 0)) {
    flags.push('tokens_recorded_without_pricing_snapshot');
  }
  if (payload.verification.some((v) => v.failed > 0) && !hasEventType('error', 'fidelity_loss')) {
    flags.push('verification_failures_but_no_matching_events');
  }
  if (state.malformedLines > 0) {
    flags.push(`journal_malformed_lines:${state.malformedLines}`);
  }
  if (state.biPushTruncated > 0) {
    flags.push(`bi_push_truncated:${state.biPushTruncated}`);
  }
  // Rows swallowed by RP_TELEMETRY_BI_DISABLED while operator identity was
  // known. Legitimate only for tests/offline dev — on a production run this
  // flag means the kill-switch leaked into the session.
  if (state.biPushSkipped > 0) {
    flags.push('bi_sink_disabled_during_run');
  }
  // No operator identity ever resolved: the run was sent to BI under its placeholder
  // subject, with no operator on it. It is complete and correct on disk and counted in BI
  // as unattributed; this flag is how a reader tells "never identified" from "identified,
  // copy not yet delivered".
  if (!biSink.isGuid(dims.wix_user_id)) {
    flags.push('operator_identity_unresolved');
  }
  // A stalled wait whose stage has no halt_needs_user event lost the "why" of
  // the stall — the single most actionable signal for where runs stall.
  const haltStages = new Set(events.filter((e) => e.event_type === 'halt_needs_user').map((e) => e.stage));
  const allWaits = [...state.waits, ...(state.openWait ? [{ ...state.openWait, end: finalizeTs }] : [])];
  const flaggedWaitStages = new Set();
  for (const w of allWaits) {
    const waitStage = w.stage || 'config';
    if (tsMs(w.end) - tsMs(w.start) < WAIT_WITHOUT_HALT_FLAG_MS) continue;
    if (haltStages.has(waitStage) || flaggedWaitStages.has(waitStage)) continue;
    flaggedWaitStages.add(waitStage);
    flags.push(`wait_without_halt_event:${waitStage}`);
  }
  // Discovery is where extension classes become known; a null left after a
  // passed discovery usually means the dims call was skipped, not that the
  // source has no extensions (stamp [] explicitly for that).
  const discoveryPassed = state.entries.some((e) => e.stage === 'discovery' && e.outcome === 'passed');
  if (discoveryPassed && (dims.source_extensions === undefined || dims.source_extensions === null)
    && dims.source_platform !== 'csv') {
    flags.push('source_extensions_null_after_discovery');
  }
  // Same rationale for entity classes — and they exist for every source platform,
  // csv included, so no exemption. Stamp [] explicitly for an empty source.
  if (discoveryPassed && (dims.discovered_entity_types === undefined || dims.discovered_entity_types === null)) {
    flags.push('discovered_entity_types_null_after_discovery');
  }
  // The demand signal for unsupported verticals is the volume rows (target
  // "none", discovered/skipped counts). A discovered class with no volume row
  // under-reports that demand invisibly, so cross-check on a completed run —
  // earlier terminal states legitimately finalize before volumes exist.
  if (payload.terminal_state === 'completed' && Array.isArray(dims.discovered_entity_types)) {
    const volumeTypes = new Set(payload.volumes.map((v) => v.entity_type));
    for (const entityType of dims.discovered_entity_types) {
      if (!volumeTypes.has(entityType)) flags.push(`volume_row_missing:${entityType}`);
    }
  }

  const evidence = walkEvidence(projectDir);
  const evidenceBytes = evidence.reduce((sum, f) => sum + f.bytes, 0);
  if (evidenceBytes > EVIDENCE_SIZE_FLAG_BYTES) {
    flags.push(`evidence_unusually_large:${Math.round(evidenceBytes / 1024)}kb`);
  }

  const rollup = {
    run_id: runStart.run_id,
    project_id: runStart.project_id,
    attempt: runStart.attempt,
    session_count: state.sessionCount,
    schema_version: TELEMETRY_SCHEMA_VERSION,
    skills_version: dims.skills_version || runStart.skills_version || null,
    skills_commit: dims.skills_commit || runStart.skills_commit || null,
    runtime_env: runStart.runtime_env || dims.runtime_env
      ? { ...(runStart.runtime_env || {}), ...(dims.runtime_env || {}) }
      : null,
    site_id: dims.site_id ?? null,
    wix_user_id: dims.wix_user_id ?? null,
    source_platform: dims.source_platform ?? null,
    source_platform_version: dims.source_platform_version ?? null,
    source_site_url: dims.source_site_url ?? null,
    source_extensions: dims.source_extensions ?? null,
    discovered_entity_types: dims.discovered_entity_types ?? null,
    source_acquisition: dims.source_acquisition ?? null,
    delivery_mode: dims.delivery_mode ?? null,
    destination_strategy: dims.destination_strategy ?? null,
    terminal_state: payload.terminal_state,
    stopped_at_stage: payload.stopped_at_stage,
    stages,
    volumes: payload.volumes,
    verification: payload.verification,
    operator_acceptance: payload.operator_acceptance,
    telemetry_health: {
      rejected_calls: state.rejected.length,
      dropped_events: dropped,
      scrub_hits: state.scrubHits,
      bi_push_failures: state.biPushFailures,
      bi_push_skipped: state.biPushSkipped,
      flags,
    },
    event_count_by_type: eventCountByType,
    active_ms: activeMs,
    waiting_ms: waitingMs,
    timing,
    cost,
    transcript_digest: state.transcriptDigest,
    // ALWAYS present, even when the run never touched a key/value table -- an
    // all-zero object says "this run read no key/value data", which is a different and useful
    // fact from a missing field, and only one of the two can be aggregated across runs.
    source_reads: summarizeSourceReads(state),
    model_pricing_snapshot: dims.model_pricing_snapshot ?? null,
    run_started: runStart.ts,
    run_ended: finalizeTs,
    bundle_manifest: ['run-telemetry.json', ...evidence.map((f) => f.path)],
  };

  return {
    document: {
      schema_version: TELEMETRY_SCHEMA_VERSION,
      field_tiers: FIELD_TIERS,
      rollup,
      events,
    },
    evidenceBytes,
    evidenceFiles: evidence,
  };
}

function writeJsonAtomic(filePath, data) {
  fs.mkdirSync(path.dirname(filePath), { recursive: true });
  const tempPath = `${filePath}.${process.pid}.tmp`;
  fs.writeFileSync(tempPath, `${JSON.stringify(data, null, 2)}\n`, 'utf8');
  fs.renameSync(tempPath, filePath);
}

function archiveName(state) {
  return `run-${state.runStart.attempt}-${state.runStart.run_id}`;
}

function archiveFinalizedJournal(projectDir, journal, state) {
  // Deterministic completion of a finalize: rebuild the document from the
  // journal (idempotent), write signal + archive copies, move the journal.
  const assembled = assembleDocument(projectDir, journal.records, journal.malformed);
  const base = archiveName(state);
  fs.mkdirSync(runsDir(projectDir), { recursive: true });
  writeJsonAtomic(signalPath(projectDir), assembled.document);
  writeJsonAtomic(path.join(runsDir(projectDir), `${base}.json`), assembled.document);
  fs.renameSync(journalPath(projectDir), path.join(runsDir(projectDir), `${base}.jsonl`));
  return assembled;
}

function lastFinalizedAttempt(projectDir) {
  let last = 0;
  const signal = signalPath(projectDir);
  if (fs.existsSync(signal)) {
    try {
      const doc = JSON.parse(fs.readFileSync(signal, 'utf8'));
      if (doc && doc.rollup && Number.isInteger(doc.rollup.attempt)) last = Math.max(last, doc.rollup.attempt);
    } catch {
      // unreadable prior signal file — fall through to the archive scan
    }
  }
  const dir = runsDir(projectDir);
  if (fs.existsSync(dir)) {
    for (const name of fs.readdirSync(dir)) {
      const match = /^run-(\d+)-/.exec(name);
      if (match) last = Math.max(last, Number(match[1]));
    }
  }
  return last;
}

function resolveSkillsVersion() {
  let dir = __dirname;
  for (let depth = 0; depth < 7; depth += 1) {
    const candidate = path.join(dir, 'VERSION');
    if (fs.existsSync(candidate)) {
      const version = fs.readFileSync(candidate, 'utf8').trim();
      if (/^\d+\.\d+\.\d+$/.test(version)) return version;
    }
    // A published bundle also carries its version in the manifest publish-skills-to-wix.sh stamps.
    const manifest = path.join(dir, '.publish-manifest.json');
    if (fs.existsSync(manifest)) {
      try {
        const { version } = JSON.parse(fs.readFileSync(manifest, 'utf8'));
        if (/^\d+\.\d+\.\d+$/.test(version)) return version;
      } catch {
        // unreadable manifest — keep walking
      }
    }
    const parent = path.dirname(dir);
    if (parent === dir) break;
    dir = parent;
  }
  return null;
}

function resolveSkillsCommit() {
  // The exact source commit behind this bundle — the drill-down provenance the
  // coarse, hand-bumped `skills_version` (semver) cannot give. Two sources, in
  // order of authority:
  //
  // 1. `.publish-manifest.json`, stamped by publish-skills-to-wix.sh next to
  //    VERSION at publish time. This is the only source that works in a partner
  //    runtime (no git there) AND the only correct one once the bundle lives
  //    inside the consuming wix/skills repo — where a git probe would return
  //    that repo's HEAD, not the replatform source commit. So it wins.
  // 2. Dev-mode git fallback: runs straight from the replatform checkout, where
  //    the recorder's own directory is inside the source repo, have no manifest
  //    but do have git — so `git -C __dirname rev-parse --short HEAD` is the source
  //    commit. Never reached once a manifest is present (case 1).
  let dir = __dirname;
  for (let depth = 0; depth < 7; depth += 1) {
    const candidate = path.join(dir, '.publish-manifest.json');
    if (fs.existsSync(candidate)) {
      try {
        const manifest = JSON.parse(fs.readFileSync(candidate, 'utf8'));
        if (typeof manifest.sourceCommit === 'string' && /^[0-9a-f]{7,40}$/.test(manifest.sourceCommit)) {
          return manifest.sourceCommit;
        }
      } catch {
        // unreadable manifest — fall through to the dev-mode git probe
      }
    }
    const parent = path.dirname(dir);
    if (parent === dir) break;
    dir = parent;
  }
  try {
    const sha = execFileSync('git', ['-C', __dirname, 'rev-parse', '--short', 'HEAD'], {
      encoding: 'utf8',
      stdio: ['ignore', 'pipe', 'ignore'],
    }).trim();
    if (/^[0-9a-f]{7,40}$/.test(sha)) return sha;
  } catch {
    // no git, not a repo, or git unavailable — commit provenance stays null
  }
  return null;
}

// The operator's Wix user id — the routing key the BI sink refuses to push
// without (bi-sink rule 3). Auto-stamped like skills_version/skills_commit
// above, and for the same reason: a value nobody thinks to pass is a value that
// never arrives. Runs from 2026-07 to 2026-08 shipped with no resolution step
// at all, so any run whose caller didn't happen to hand over a GUID pushed
// exactly zero rows and left no trace in BI.
//
// Source: `~/.wix/auth/account.json`, written by `wix login`. The migration
// flow already requires that login (rp-execute-setup mints its site token from
// the same CLI session), so the file is present by the time telemetry starts,
// and it is the operator's own artifact — this works identically for a Wix
// employee and a customer. `userInfo.userId` there is the same GUID that lands
// in `events.dbo.users_10` as `logged_user_id` (verified 2026-09-01).
//
// Deliberately NOT used, for the record:
//   - `wix whoami` returns only an email ("Logged in as <email>"), and BI
//     routing needs a GUID. No public endpoint resolves email -> user id.
//   - the site-mute response's `mutedBy.wixUserId` would work but makes
//     identity depend on a side-effecting call that only happens after
//     provisioning; identity is needed at `start`.
//   - `account.id` / `accountOwner` from the account APIs are the ACCOUNT
//     guid, not the user's — one user can own several accounts, so those
//     would silently attribute runs to the wrong id.
// The right long-term fix is a public "who am I" endpoint; the capability
// exists internally (`Oauth2Ng/UserInfo` returns `sub`) but is not exposed.
//
// Fail-soft by contract: the path is CLI-internal and undocumented, so any
// miss (no login, moved file, malformed JSON, non-GUID value) returns null and
// the run proceeds unattributed but never blocked. Telemetry must never fail a
// migration.
//
// A run with no operator identity is not silent: it sends under a per-run placeholder
// subject (a random GUID; BI ingests a GUID no account owns, but did not ingest the
// all-zeros GUID, which must never be used) and carries the `operator_identity_unresolved`
// health flag until the operator's id is known.
function resolveOperatorIdentity() {
  return require('../../../lib/operator-identity').resolveOperatorIdentity();
}

// --- BI sink ----------------------------------------------------------

// Push rows to BI and journal any failures/truncation/skips so telemetry loss
// is visible in telemetry_health, never silent. Never throws — BI unreachable,
// slow, or rejecting is a telemetry failure, not a run failure. Returns the
// last journal seq used plus sent/failed/skipped row counts.
//
// `throughSeq` is the healing watermark: when every row of a push lands, a
// `bi_push_ok` record asserts "all pushable journal content with seq <=
// through_seq is in BI". Anything journaled past the watermark — because a
// push failed, the sink was disabled, or the journal predates the watermark
// record type — is backlog, and the next push opportunity re-sends it in full
// (idempotent: the reviewer dedupes on the natural keys, latest row wins).
//
// The transport runs with no journal lock held. Its markers are appended afterwards under a
// fresh lock with a fresh `seq` from a fresh read (the `seq` argument is only the fallback
// return value when nothing is appended), keeping the initiating call's `ts` — safe because
// the activity clock ignores bi_push_* records. `signal` (optional) cancels the transport;
// `markerLockWaitMs` bounds the marker lock, and a busy marker lock skips the markers (no
// bi_push_ok → the backlog stays and the next heal re-sends). Returns `aborted` and
// `markersSkipped` besides the counts.
async function pushToBi(projectDir, ts, seq, rows, loggedUserId, target, {
  throughSeq, signal, markerLockWaitMs = COMMAND_LOCK_WAIT_MS,
} = {}) {
  const result = await biSink.pushRows(rows, loggedUserId, signal ? { signal } : undefined);
  const markers = [];
  if (result.skipped > 0) {
    markers.push({ type: 'bi_push_skipped', ts, target, rows: result.skipped });
  }
  if (result.truncated > 0) {
    markers.push({ type: 'bi_push_truncated', ts, target, rows: result.truncated });
  }
  for (const failure of result.failures) {
    markers.push({ type: 'bi_push_failed', ts, target, rows: failure.rows, detail: failure.detail });
  }
  if (throughSeq !== undefined && result.sent > 0 && result.failures.length === 0 && result.skipped === 0) {
    markers.push({ type: 'bi_push_ok', ts, target, through_seq: throughSeq, subject: loggedUserId });
  }
  let markersSkipped = false;
  if (markers.length > 0) {
    try {
      seq = withJournalLock(projectDir, () => {
        const fresh = readJournal(projectDir);
        // The journal was archived meanwhile: never create a marker-only journal.
        if (!fresh) return null;
        let next = replay(fresh.records).seq;
        const stamped = markers.map((marker) => {
          next += 1;
          const { type, ...rest } = marker;
          return { type, ts: rest.ts, seq: next, ...rest };
        });
        appendJournalRecords(projectDir, stamped);
        return next;
      }, { waitMs: markerLockWaitMs }) ?? seq;
    } catch (error) {
      if (error && error.code !== 'ELOCKBUSY') throw error;
      markersSkipped = true;
    }
  }
  return {
    seq,
    sent: result.sent,
    failed: result.failures.reduce((sum, f) => sum + f.rows, 0),
    skipped: result.skipped,
    aborted: result.failures.some((f) => f.detail === 'aborted'),
    markersSkipped,
  };
}

// A send under the run's placeholder (no operator id yet) is bounded by the heal deadline. Before the
// Wix login exists the machine's egress is the likeliest to be blocked, and an unbounded send per
// recorded event would slow the run on every call; a bounded one just leaves the run in backlog,
// which the next send retries. Sends under the operator's id keep their existing behaviour.
// True when the newest send outcome in the journal is a hang (timeout or abort) less than
// SEND_PAUSE_MS before `ts`. A later confirmed send clears it. Reads the journal's own
// markers; no extra state.
function sendPauseActive(projectDir, ts) {
  try {
    const journal = readJournal(projectDir);
    if (!journal) return false;
    const now = tsMs(ts);
    for (let i = journal.records.length - 1; i >= 0; i -= 1) {
      const r = journal.records[i];
      if (r.type === 'bi_push_ok') return false;
      if (r.type === 'bi_push_failed') {
        if (r.detail !== 'timeout' && r.detail !== 'aborted') return false;
        const at = tsMs(r.ts);
        return Number.isFinite(at) && Number.isFinite(now) && now >= at && now - at < SEND_PAUSE_MS;
      }
    }
    return false;
  } catch {
    return false;
  }
}

async function pushToBiForSubject(operatorId, projectDir, ts, seq, rows, subject, target, options = {}) {
  // A hung BI pauses ordinary sends for a minute under either subject: the operator's id gives no
  // reason to wait 8 s per batch on every `record`. `force` (finalize, the repair) always sends.
  if (!options.force && sendPauseActive(projectDir, ts)) {
    return { seq, sent: 0, failed: 0, skipped: 0, aborted: false, markersSkipped: false, deferred: true };
  }
  if (options.signal) return pushToBi(projectDir, ts, seq, rows, subject, target, options);
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), options.deadlineMs || HEAL_DEADLINE_MS);
  try {
    // A busy marker lock is waited on briefly before the operator's id is known, as before.
    return await pushToBi(projectDir, ts, seq, rows, subject, target, {
      ...(operatorId ? {} : { markerLockWaitMs: PLACEHOLDER_MARKER_LOCK_WAIT_MS }), ...options, signal: controller.signal,
    });
  } finally {
    clearTimeout(timer);
  }
}

// The subject a run's rows are sent under: the operator's Wix user id once it is known, else the
// run's placeholder (a random GUID made for this run alone). This is the only place a subject is
// chosen, so once the operator's id is known nothing can be sent under the placeholder again.
function operatorIdOf(state) {
  return biSink.isGuid(state.dims.wix_user_id) ? state.dims.wix_user_id : null;
}

function subjectOf(state) {
  return operatorIdOf(state) || state.provisionalSubject || null;
}

// Same as `subjectOf`, but a run that has neither (a journal written before placeholders
// existed) gets its placeholder written now. Call inside the journal lock.
function ensureSubjectLocked(projectDir, state, ts) {
  const existing = subjectOf(state);
  if (existing) return existing;
  const subject = crypto.randomUUID();
  appendJournal(projectDir, { type: 'provisional_subject', ts, seq: state.seq + 1, subject });
  state.provisionalSubject = subject;
  state.seq += 1;
  return subject;
}

// {subject, operatorId, view} for a send made after the journal lock was released. The caller's
// `state` is a snapshot from before it released the lock, and another process may have bound
// the operator's id since, so the subject (and the send markers behind the backlog decision) are
// taken from a fresh read of the journal; a legacy journal gets its placeholder here, once. `view`
// is the caller's snapshot with the fresh subject facts, for `biBacklog` and `healRows`.
function resolveSendSubject(projectDir, state, ts) {
  const journal = readJournal(projectDir);
  const fresh = journal ? replay(journal.records) : null;
  const base = fresh && fresh.runStart && state.runStart && fresh.runStart.run_id === state.runStart.run_id ? fresh : state;
  const view = {
    ...state, dims: base.dims, provisionalSubject: base.provisionalSubject, biOkBySubject: base.biOkBySubject,
  };
  let subject = subjectOf(view);
  if (!subject) {
    // The event (or other journal content) is already on disk by now, so a busy lock must not turn
    // into a failed command: with no subject the caller skips the send and the run stays in backlog.
    try {
      subject = withJournalLock(projectDir, () => {
        const current = readJournal(projectDir);
        if (!current) return null;
        return ensureSubjectLocked(projectDir, replay(current.records), ts);
      });
    } catch {
      subject = null;
    }
    view.provisionalSubject = subject;
  }
  return { subject, operatorId: operatorIdOf(view), view };
}

// True when journal content that belongs in BI has no confirming watermark FOR THE CURRENT
// SUBJECT: a prior send failed or was skipped, the journal predates the watermark, or the
// operator's id arrived after rows went out under the placeholder (so nothing is confirmed under
// the operator's id yet). A marker written before subjects were recorded counts for the
// operator's id and for nothing else.
function currentSubjectOkSeq(state) {
  const subject = subjectOf(state);
  if (!subject) return 0;
  return operatorIdOf(state)
    ? Math.max(state.biOkBySubject[subject] || 0, state.biOkBySubject[LEGACY_SUBJECT] || 0)
    : (state.biOkBySubject[subject] || 0);
}

function biBacklog(state) {
  const subject = subjectOf(state);
  if (!subject) return true;
  const confirmed = operatorIdOf(state)
    ? Math.max(state.biOkBySubject[subject] || 0, state.biOkBySubject[LEGACY_SUBJECT] || 0)
    : (state.biOkBySubject[subject] || 0);
  return confirmed === 0 || confirmed < state.lastEventSeq;
}

// The full idempotent re-push set: the started row plus every folded event
// class (the finalized rollup only exists at finalize, which builds its own).
function healRows(state, wixUserId, extraEventRecords = []) {
  const folded = foldEvents([...state.events, ...extraEventRecords], state.runStart.run_id);
  // `wixUserId` is the operator's id or null; the placeholder is never written as the operator.
  return [biSink.startedRunRow(state.runStart, wixUserId), ...folded.map(biSink.eventRow)];
}

// The folded BI row for one event class: prior journal records of the class plus
// the just-appended one. Re-pushes of a growing class are safe — the reviewer
// dedupes on (run_id, seq), latest row wins with the cumulative count.
function foldedRowForClass(state, eventRecord) {
  const classRecords = [
    ...state.events.filter((r) => r.class_key === eventRecord.class_key),
    eventRecord,
  ];
  return biSink.eventRow(foldEvents(classRecords, state.runStart.run_id)[0]);
}

// --- public command API ---------------------------------------------------------

// Late identity adoption. A run normally starts before the operator's Wix CLI login exists
// (the login happens in the destination stage), so `start` opens it without a wix_user_id and
// every row goes out under the run's placeholder. Any later command calls this first: when the
// active run still has no GUID identity and the CLI login record now resolves one, journal it
// as a dims record and re-send the whole run under the operator's id -- no operator action, no
// explicit `start` resume. Best-effort by design: it never throws, never blocks the command
// that called it (the push is bounded by HEAL_DEADLINE_MS), and a still-unresolved identity is
// a silent no-op (the run keeps sending under its placeholder).
const ADOPT_LOCK_WAIT_MS = 1500;
const ADOPT_MARKER_LOCK_WAIT_MS = 500;
const IDENTITY_SETTLED_FILE = '.identity-bound';

// The run id and whether the journal already holds the operator's GUID, from journal TEXT the
// caller already read (so the answer and the run id come from the same bytes). Only lines that
// mention `wix_user_id` are parsed, so a long journal costs a substring scan, not a replay.
function journalIdentityFacts(text) {
  const newline = text.indexOf('\n');
  let runId = null;
  try { runId = JSON.parse(newline === -1 ? text : text.slice(0, newline)).run_id || null; } catch { /* unreadable first line */ }
  let hasId = false;
  for (const line of text.split('\n')) {
    if (!line.includes('"wix_user_id"')) continue;
    try {
      const record = JSON.parse(line);
      if ((record.type === 'run_start' || record.type === 'dims') && biSink.isGuid(record.dims && record.dims.wix_user_id)) {
        hasId = true;
        break;
      }
    } catch { /* a torn or malformed line is not an answer */ }
  }
  return { runId, hasId };
}

// The run id on the journal's first line without reading the rest: reads in chunks up to the
// first newline, however long that line is (a run_start can carry large dimensions).
function firstRunId(file) {
  const fd = fs.openSync(file, 'r');
  try {
    const chunk = Buffer.alloc(65536);
    let text = '';
    for (let position = 0; position < 16 * 1024 * 1024;) {
      const read = fs.readSync(fd, chunk, 0, chunk.length, position);
      if (read === 0) break;
      text += chunk.toString('utf8', 0, read);
      position += read;
      if (text.includes('\n')) break;
    }
    return JSON.parse(text.split('\n')[0]).run_id || null;
  } catch {
    return null;
  } finally {
    fs.closeSync(fd);
  }
}

// Once a run has its operator id it never loses it, so the answer is remembered next to the
// journal, keyed by run id (a later run in the same project has a different run id). A settled
// run then costs one small file read and one first-line read per command.
function identitySettled(projectDir, file) {
  try {
    const settled = JSON.parse(fs.readFileSync(path.join(telemetryDir(projectDir), IDENTITY_SETTLED_FILE), 'utf8'));
    return Boolean(settled.run_id) && settled.run_id === firstRunId(file);
  } catch {
    return false;
  }
}

function markIdentitySettled(projectDir, runId) {
  try {
    if (runId) fs.writeFileSync(path.join(telemetryDir(projectDir), IDENTITY_SETTLED_FILE), JSON.stringify({ run_id: runId }));
  } catch { /* a missing marker only costs a re-check */ }
}

async function adoptIdentity(projectDir, { now, resolveIdentity = resolveOperatorIdentity } = {}) {
  try {
    // Cheapest checks first, because this runs on every command: no login yet (one small file
    // read) means nothing to do; a run already known to have its id is settled by the marker;
    // otherwise a text match over the journal settles "id already there" without a replay.
    const userId = resolveIdentity();
    if (!biSink.isGuid(userId)) return { adopted: false };
    const file = journalPath(projectDir);
    if (!fs.existsSync(file)) return { adopted: false };
    if (identitySettled(projectDir, file)) return { adopted: false };
    const facts = journalIdentityFacts(fs.readFileSync(file, 'utf8'));
    if (facts.hasId) {
      markIdentitySettled(projectDir, facts.runId);
      return { adopted: false };
    }
    const ts = nowIso(now);
    const plan = withJournalLock(projectDir, () => {
      const journal = readJournal(projectDir);
      if (!journal) return null;
      const state = replay(journal.records);
      if (!state.runStart || state.finalize) return null;
      if (biSink.isGuid(state.dims.wix_user_id)) {
        markIdentitySettled(projectDir, state.runStart.run_id);
        return null;
      }
      const seq = state.seq + 1;
      appendJournal(projectDir, { type: 'dims', ts, seq, dims: { wix_user_id: userId } });
      markIdentitySettled(projectDir, state.runStart.run_id);
      return { state, seq, userId };
    }, { waitMs: ADOPT_LOCK_WAIT_MS });
    if (!plan) return { adopted: false };
    // Bounded like every hook-driven send: an unreachable BI never holds a command up for long,
    // neither in the transport nor in writing the markers afterwards.
    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), HEAL_DEADLINE_MS);
    try {
      await pushToBi(projectDir, ts, plan.seq, healRows(plan.state, plan.userId), plan.userId,
        'identity_heal', { throughSeq: plan.seq, signal: controller.signal, markerLockWaitMs: ADOPT_MARKER_LOCK_WAIT_MS });
    } finally {
      clearTimeout(timer);
    }
    return { adopted: true };
  } catch {
    return { adopted: false };
  }
}

// Re-send whatever BI has not confirmed for the run, under its current subject. This is the
// repair the agent is told to run when `attention` reports a backlog: unlike `record`, which only
// heals as a side effect of the next event, it sends now. Best effort and bounded like every
// other send; a run with nothing pending, no active run, or sending switched off does nothing.
async function flush(projectDir, { now } = {}) {
  try {
    if (biSink.disabled()) return { flushed: false };
    const ts = nowIso(now);
    const journal = readJournal(projectDir);
    if (!journal) return { flushed: false };
    const state = replay(journal.records);
    if (!state.runStart || state.finalize) return { flushed: false };
    const { subject, operatorId, view } = resolveSendSubject(projectDir, state, ts);
    if (!subject || !biBacklog(view)) return { flushed: false };
    await pushToBiForSubject(operatorId, projectDir, ts, state.seq, healRows(state, operatorId), subject, 'flush',
      { throughSeq: state.seq, force: true });
    return { flushed: true };
  } catch {
    return { flushed: false };
  }
}

// What the agent must be told about the run's telemetry, or null when nothing is wrong. Worked
// out from the journal on every call, with no stored state. Trouble is: (a) a send backlog under
// the run's current subject while sending is switched on, or (b) a Wix login that exists while the
// run still has no operator id. A run with no login yet is not in trouble: it sends under its
// placeholder. The result is for the agent only; it never carries a URL, a response body or an id.
// `last_failure` is the short code of the newest failed send. Never throws.
function attention(projectDir, { resolveIdentity = resolveOperatorIdentity } = {}) {
  try {
    const journal = readJournal(projectDir);
    if (!journal) return null;
    const state = replay(journal.records);
    if (!state.runStart || state.finalize) return null;
    const issues = [];
    if (!biSink.disabled() && biBacklog(state)) issues.push('bi_backlog');
    let userId = null;
    try { userId = resolveIdentity(); } catch { userId = null; }
    if (biSink.isGuid(userId) && !operatorIdOf(state)) issues.push('identity_not_adopted');
    if (issues.length === 0) return null;
    let lastFailure = null;
    for (let i = journal.records.length - 1; i >= 0; i -= 1) {
      if (journal.records[i].type === 'bi_push_failed') {
        lastFailure = typeof journal.records[i].detail === 'string' ? journal.records[i].detail.slice(0, 40) : null;
        break;
      }
    }
    return { issues, last_failure: lastFailure, repair: 'rp-telemetry.js identity' };
  } catch {
    return null;
  }
}

// The same notice for a closing push (`finalize`, `rebuild --push`) that left rows unsent. The
// run is archived by then, so the repair is the explicit re-send. Sending switched off is not trouble.
function closingAttention(biPush) {
  if (!biPush || biSink.disabled()) return null;
  if (!(biPush.failed > 0) && !(biPush.skipped > 0)) return null;
  // A journal with no subject to send under cannot be re-sent by any command: say nothing rather
  // than name a repair that can never work.
  if (Array.isArray(biPush.failure_details) && biPush.failure_details.includes('no_operator_identity')) return null;
  const detail = Array.isArray(biPush.failure_details) && typeof biPush.failure_details[0] === 'string'
    ? biPush.failure_details[0].slice(0, 40) : null;
  return { issues: ['bi_push_failed'], last_failure: detail, repair: 'rp-telemetry.js rebuild --push' };
}

// --- supervisor liveness -------------------------------------------------------

// The command-line supervisor's liveness record (run-controller.js `activePath`). Read directly
// rather than required, so the recorder never depends on the orchestrator's modules.
function supervisorActivePath(projectDir) {
  return path.join(projectDir, 'state', 'runner', 'active.json');
}

function pidAlive(pid) {
  if (!Number.isInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return Boolean(error && error.code === 'EPERM');
  }
}

// The live supervisor record, or null: `active.json` exists, has `status: 'running'`, and its
// `pid` is alive on this host. Missing, unreadable, non-running or dead-PID → null.
function liveSupervisor(projectDir) {
  try {
    const active = JSON.parse(fs.readFileSync(supervisorActivePath(projectDir), 'utf8'));
    if (!active || active.status !== 'running' || !pidAlive(active.pid)) return null;
    return active;
  } catch {
    return null;
  }
}

function defaultIsSupervised(projectDir) {
  return liveSupervisor(projectDir) !== null;
}

// `startedAt` of the live supervisor run, or null when there is no live supervisor or its
// record carries no parseable `startedAt`.
function defaultSupervisorStartedAt(projectDir) {
  const active = liveSupervisor(projectDir);
  if (!active || typeof active.startedAt !== 'string' || !Number.isFinite(tsMs(active.startedAt))) return null;
  return active.startedAt;
}

// --- shared resume records --------------------------------------------------------

// The one place the resume records are built, for both explicit `start()` and automatic
// `resumeIfStale()`: close an open wait, or impute the dead interval as waiting when the gap
// since the last ACTIVITY is at least IMPLICIT_WAIT_MIN_MS; then one `session_start` (carrying
// `trigger` only when automatic); then a `dims` record when there is anything to add. Identity
// is resolved exactly as `start()` always has when neither the call nor the journal has it.
// Pure: returns the records with their seqs; the caller appends them in one write.
function buildResumeRecords(state, ts, dims, { trigger = null, resolveIdentity = resolveOperatorIdentity } = {}) {
  const records = [];
  let seq = state.seq;
  let closedWait = false;
  let imputedWaitMs = 0;
  if (state.openWait) {
    seq += 1;
    records.push({ type: 'wait_end', ts, seq });
    closedWait = true;
  } else if (state.lastActivityTs && tsMs(ts) - tsMs(state.lastActivityTs) >= IMPLICIT_WAIT_MIN_MS) {
    // The run stopped without an open wait (e.g. a crash); the dead interval
    // is waiting time, not active time.
    const stage = state.openStage || state.lastStage || 'config';
    seq += 1;
    records.push({ type: 'wait_start', ts: state.lastActivityTs, seq, stage, imputed: true });
    seq += 1;
    records.push({ type: 'wait_end', ts, seq, imputed: true });
    closedWait = true;
    imputedWaitMs = tsMs(ts) - tsMs(state.lastActivityTs);
  }
  seq += 1;
  records.push(trigger
    ? { type: 'session_start', ts, seq, session: state.sessionCount + 1, trigger }
    : { type: 'session_start', ts, seq, session: state.sessionCount + 1 });
  // A run that opened without identity — begun before the recorder resolved
  // it, or with the CLI not yet logged in — gets it here, so the heal has a
  // route. Journaled as dims like any other late-arriving dimension.
  const identity = resolveMissingIdentity(state, dims, resolveIdentity);
  if (Object.keys(dims).length > 0) {
    seq += 1;
    records.push({ type: 'dims', ts, seq, dims });
  }
  return { records, seq, closedWait, imputedWaitMs, userId: identity };
}

// Adds a resolved wix_user_id to `dims` when neither the call nor the journal carries one, and
// returns the identity the heal routes under (call dims first, then the journal), or null.
function resolveMissingIdentity(state, dims, resolveIdentity) {
  if (!biSink.isGuid(dims.wix_user_id) && !biSink.isGuid(state.dims.wix_user_id)) {
    const resolvedUserId = resolveIdentity();
    if (resolvedUserId) dims.wix_user_id = resolvedUserId;
  }
  if (biSink.isGuid(dims.wix_user_id)) return dims.wix_user_id;
  return biSink.isGuid(state.dims.wix_user_id) ? state.dims.wix_user_id : null;
}

// Does an explicit `start()` join the current attempt's latest session opener instead of
// opening another session? Measured from the opener itself, never from the latest activity:
// recent activity after an old automatic opener does not make that session "just opened".
//   - window: the opener is an automatic (`trigger`-bearing) session_start and
//     0 <= now - opener.ts < SESSION_ADOPT_WINDOW_MS;
//   - live supervisor run: a live supervisor owns the project with a parseable startedAt, and
//     startedAt - opener.ts < SESSION_ADOPT_WINDOW_MS (one supervisor process is one session).
function startAdopts(state, ts, supervisorStartedAt) {
  const opener = state.lastOpener;
  if (!opener || typeof opener.ts !== 'string') return false;
  const openerMs = tsMs(opener.ts);
  if (!Number.isFinite(openerMs)) return false;
  const age = tsMs(ts) - openerMs;
  if (opener.type === 'session_start' && opener.trigger && age >= 0 && age < SESSION_ADOPT_WINDOW_MS) return true;
  const startedAt = supervisorStartedAt();
  const startedMs = typeof startedAt === 'string' ? tsMs(startedAt) : NaN;
  // Strictly inside the window, like every other adoption boundary: an opener exactly a
  // window before startedAt is an earlier execution's session, not this run's.
  return Number.isFinite(startedMs) && openerMs > startedMs - SESSION_ADOPT_WINDOW_MS;
}

// --- public command API ---------------------------------------------------------

// `resolveIdentity` is injectable for one reason: tests need to construct a run
// with no operator identity, and on any real machine with a logged-in CLI that
// state is otherwise unreachable. Production callers (the CLI) never pass it.
// `supervisorStartedAt` is injectable for the same reason (tests need no real supervisor);
// the default reads the live supervisor's `active.json`.
async function start(projectDir, dimsInput, {
  now, resolveIdentity = resolveOperatorIdentity, supervisorStartedAt,
} = {}) {
  const ts = nowIso(now);
  const dims = validateDims(dimsInput);
  const readStartedAt = supervisorStartedAt || (() => defaultSupervisorStartedAt(projectDir));

  // Everything local happens under the journal lock from a fresh read, so a concurrent
  // automatic resume and this call can never both open a session or reuse a seq. The BI push
  // runs after the lock is released.
  const plan = withJournalLock(projectDir, () => {
    const journal = readJournal(projectDir);
    if (journal) {
      const state = replay(journal.records);
      if (!state.runStart) {
        throw new ValidationError(['telemetry journal is corrupt: missing run-start record']);
      }
      if (state.finalize) {
        archiveFinalizedJournal(projectDir, journal, state);
      } else if (startAdopts(state, ts, readStartedAt)) {
        // Join the session automatic recovery (or this supervisor run) already opened: no
        // second session_start, no imputed wait. The caller's dims (and resolved identity)
        // still land; the heal runs only when there is backlog.
        resolveMissingIdentity(state, dims, resolveIdentity);
        let seq = state.seq;
        if (Object.keys(dims).length > 0) {
          seq += 1;
          appendJournal(projectDir, { type: 'dims', ts, seq, dims });
        }
        // The run as it stands after this call's dims: the subject it sends under, and whether
        // the operator's id has just arrived (rows already sent under the placeholder do not
        // count as sent under the operator's id).
        const next = { ...state, dims: { ...state.dims, ...dims }, seq };
        const bound = Boolean(operatorIdOf(next)) && !operatorIdOf(state);
        const subject = ensureSubjectLocked(projectDir, next, ts);
        return {
          kind: 'adopted', state, seq: next.seq, subject, operatorId: operatorIdOf(next),
          heal: bound || biBacklog(next),
        };
      } else {
        // Resume: same run_id, same attempt, one more session. Cross-session user
        // latency lands in waiting_ms, never in a phantom second run.
        const built = buildResumeRecords(state, ts, dims, { resolveIdentity });
        appendJournalRecords(projectDir, built.records);
        const next = { ...state, dims: { ...state.dims, ...dims }, seq: built.seq };
        const subject = ensureSubjectLocked(projectDir, next, ts);
        return {
          kind: 'resumed', state, seq: next.seq, subject, operatorId: operatorIdOf(next), heal: true,
        };
      }
    }

    const attempt = lastFinalizedAttempt(projectDir) + 1;
    const runtimeEnv = {
      os: `${os.platform()} ${os.release()}`,
      node_version: process.version,
      agent_runtime: (dims.runtime_env && dims.runtime_env.agent_runtime) || null,
      model: (dims.runtime_env && dims.runtime_env.model) || null,
      ...(dims.runtime_env || {}),
    };
    delete dims.runtime_env;
    const skillsVersion = dims.skills_version || resolveSkillsVersion();
    delete dims.skills_version;
    const skillsCommit = dims.skills_commit || resolveSkillsCommit();
    delete dims.skills_commit;
    // Same precedence as the provenance stamps above: an explicitly passed
    // wix_user_id wins (a runtime with better provenance than the CLI session),
    // otherwise resolve it ourselves. Unlike those, it stays in `dims` — it is a
    // run dimension the rollup and every BI push key off.
    if (!biSink.isGuid(dims.wix_user_id)) {
      const resolvedUserId = resolveIdentity();
      if (resolvedUserId) dims.wix_user_id = resolvedUserId;
    }

    // No operator id yet: the run sends under a random placeholder made for it alone, until the
    // operator's id is known (never the all-zeros GUID, never derived from anything).
    const provisionalSubject = biSink.isGuid(dims.wix_user_id) ? null : crypto.randomUUID();

    const record = {
      type: 'run_start',
      ts,
      seq: 1,
      run_id: crypto.randomUUID(),
      project_id: path.basename(path.resolve(projectDir)),
      attempt,
      session: 1,
      schema_version: TELEMETRY_SCHEMA_VERSION,
      skills_version: skillsVersion,
      skills_commit: skillsCommit,
      runtime_env: runtimeEnv,
      dims,
      ...(provisionalSubject ? { provisional_subject: provisionalSubject } : {}),
    };
    appendJournal(projectDir, record);
    return { kind: 'new', record, attempt, skillsVersion, skillsCommit };
  });

  if (plan.kind === 'adopted' || plan.kind === 'resumed') {
    const { state } = plan;
    // Session-boundary heal: re-push the started row and every folded event
    // class whenever operator identity is known — from this call's dims or
    // the journal. A halted run may never reach finalize's full re-push, so
    // the resume is the one reliable moment to recover rows the previous
    // session's transport lost (re-sends are harmless: reviewer-side dedup).
    if (plan.heal) {
      await pushToBiForSubject(plan.operatorId, projectDir, ts, plan.seq, healRows(state, plan.operatorId),
        plan.subject, 'resume_heal', { throughSeq: plan.seq });
    }
    return {
      resumed: true,
      ...(plan.kind === 'adopted' ? { adopted: true } : {}),
      run_id: state.runStart.run_id,
      attempt: state.runStart.attempt,
      session_count: plan.kind === 'adopted' ? state.sessionCount : state.sessionCount + 1,
      open_stage: state.openStage,
    };
  }

  const { record, attempt, skillsVersion, skillsCommit } = plan;
  // Emit `replatform_run` phase:started. Without a GUID operator identity it goes out under
  // the run's placeholder subject (never the all-zeros GUID: BI did not ingest it); the
  // operator's copy follows when the id is known.
  {
    const operatorId = biSink.isGuid(record.dims.wix_user_id) ? record.dims.wix_user_id : null;
    const subject = operatorId || record.provisional_subject;
    await pushToBiForSubject(operatorId, projectDir, ts, record.seq,
      [biSink.startedRunRow(record, operatorId)], subject, 'run_started',
      { throughSeq: record.seq });
  }
  return {
    resumed: false,
    run_id: record.run_id,
    project_id: record.project_id,
    attempt,
    session_count: 1,
    skills_version: skillsVersion,
    skills_commit: skillsCommit,
  };
}

// --- automatic resume of a stale, unfinalized journal ------------------------------

const RESUME_TRIGGERS = ['route', 'supervisor'];

function validateResumeOptions(options) {
  const errors = [];
  if (!RESUME_TRIGGERS.includes(options.trigger)) {
    errors.push(`trigger must be one of ${RESUME_TRIGGERS.join(', ')} (got: ${options.trigger})`);
  }
  for (const key of ['minGapMs', 'healDeadlineMs', 'lockWaitMs', 'markerLockWaitMs']) {
    const value = options[key];
    if (value !== undefined && !(Number.isInteger(value) && value >= 0)) {
      errors.push(`${key} must be a finite integer >= 0 (got: ${value})`);
    }
  }
  if (options.now !== undefined) {
    const parsed = options.now instanceof Date ? options.now.getTime() : (typeof options.now === 'string' ? Date.parse(options.now) : NaN);
    if (!Number.isFinite(parsed)) errors.push(`now must parse to a finite time (got: ${options.now})`);
  }
  if (options.isSupervised !== undefined && typeof options.isSupervised !== 'function') {
    errors.push('isSupervised must be a function');
  }
  if (errors.length > 0) throw new ValidationError(errors);
}

// Classification rows 2–8 (row 1, `supervised`, is decided before any read). Returns the
// no-op reason, or null when the journal should be resumed.
function classifyForResume(journal, ts, trigger, minGapMs) {
  if (!journal) return { reason: 'absent' };
  const state = replay(journal.records);
  if (!state.runStart) return { reason: 'unreadable' };
  if (state.finalize) return { reason: 'finalized' };
  const lastActivityMs = state.lastActivityTs ? tsMs(state.lastActivityTs) : NaN;
  if (!Number.isFinite(lastActivityMs)) return { reason: 'invalid_timestamp' };
  const nowMs = tsMs(ts);
  if (nowMs < lastActivityMs) return { reason: 'clock_skew' };
  if (trigger === 'supervisor' && state.lastOpener) {
    const openerMs = tsMs(state.lastOpener.ts);
    if (Number.isFinite(openerMs) && nowMs - openerMs >= 0 && nowMs - openerMs < SESSION_ADOPT_WINDOW_MS) {
      return { reason: 'adopted' };
    }
  }
  if (nowMs - lastActivityMs < minGapMs) return { reason: 'recent' };
  return { reason: null, state };
}

function readJournalOrNull(projectDir) {
  const file = journalPath(projectDir);
  if (!fs.existsSync(file)) return null;
  return readJournalFile(file);
}

// Called by the two fixed places that notice a resumed execution — the start of a
// supervisor run (`trigger: 'supervisor'`, once per process) and the manual route script
// (`trigger: 'route'`, before every stage). Records "a new session started" on an OPEN,
// readable journal that nobody told; everything else (no journal, only finished ones, a
// damaged file) belongs to preflight recovery and is left untouched.
//
// Only invalid options throw (ValidationError). Every journal, lock, clock and transport
// condition returns a result, and a no-op performs local reads only: no lock, no write, no
// network. A resume takes the journal lock, re-classifies from a fresh read, appends its
// records in one write, releases the lock, then heals BI under a hard, cancelling deadline.
async function resumeIfStale(projectDir, options = {}) {
  validateResumeOptions(options);
  const {
    trigger,
    now,
    minGapMs = trigger === 'supervisor' ? 0 : ROUTE_RESUME_MIN_GAP_MS,
    healDeadlineMs = HEAL_DEADLINE_MS,
    lockWaitMs = LOCK_WAIT_MS,
    markerLockWaitMs = MARKER_LOCK_WAIT_MS,
    isSupervised = () => defaultIsSupervised(projectDir),
    resolveIdentity = resolveOperatorIdentity,
  } = options;
  const ts = nowIso(now);
  const noop = (reason) => ({ resumed: false, reason, trigger });

  // Row 1: while a supervisor run is live, it already recorded this execution.
  if (trigger === 'route' && isSupervised()) return noop('supervised');

  // Unlocked pre-read: a no-op stays read-only. Only a candidate for resume takes the lock.
  const pre = classifyForResume(readJournalOrNull(projectDir), ts, trigger, minGapMs);
  if (pre.reason) return noop(pre.reason);

  let tx;
  try {
    tx = withJournalLock(projectDir, () => {
      const fresh = classifyForResume(readJournalOrNull(projectDir), ts, trigger, minGapMs);
      if (fresh.reason) return { reason: fresh.reason };
      const built = buildResumeRecords(fresh.state, ts, {}, { trigger, resolveIdentity });
      appendJournalRecords(projectDir, built.records);
      const next = { ...fresh.state, seq: built.seq };
      if (biSink.isGuid(built.userId)) next.dims = { ...fresh.state.dims, wix_user_id: built.userId };
      const subject = ensureSubjectLocked(projectDir, next, ts);
      return {
        reason: null, state: fresh.state, built, subject, operatorId: operatorIdOf(next), seq: next.seq,
      };
    }, { waitMs: lockWaitMs });
  } catch (error) {
    if (error && error.code === 'ELOCKBUSY') return noop('busy');
    throw error;
  }
  if (tx.reason) return noop(tx.reason);

  const { state, built } = tx;
  let heal = 'failed';
  {
    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), healDeadlineMs);
    try {
      const pushed = await pushToBi(projectDir, ts, tx.seq, healRows(state, tx.operatorId),
        tx.subject, 'resume_heal', { throughSeq: tx.seq, signal: controller.signal, markerLockWaitMs });
      if (pushed.aborted) heal = 'deadline';
      else if (pushed.skipped > 0) heal = 'disabled';
      else if (pushed.failed === 0) heal = 'ok';
      else if (pushed.sent > 0) heal = 'partial';
      else heal = 'failed';
    } catch {
      heal = controller.signal.aborted ? 'deadline' : 'failed';
    } finally {
      clearTimeout(timer);
    }
  }
  return {
    resumed: true,
    reason: 'resumed',
    trigger,
    run_id: state.runStart.run_id,
    attempt: state.runStart.attempt,
    session_count: state.sessionCount + 1,
    closed_wait: built.closedWait,
    imputed_wait_ms: built.imputedWaitMs,
    heal,
  };
}

async function dims(projectDir, dimsInput, { now, resolveIdentity } = {}) {
  await adoptIdentity(projectDir, { now, resolveIdentity });
  const ts = nowIso(now);
  const { state, validated } = withJournalLock(projectDir, () => {
    const { state: fresh } = requireActiveRun(projectDir);
    const checked = validateDims(dimsInput);
    if (Object.keys(checked).length === 0) {
      throw new ValidationError(['dims payload contains no known dimension fields']);
    }
    appendJournal(projectDir, { type: 'dims', ts, seq: fresh.seq + 1, dims: checked });
    return { state: fresh, validated: checked };
  });
  // Identity becoming known mid-run unlocks the BI route: heal the whole
  // backlog (started row + every folded event class), not just the started
  // row — held-back event rows must not wait on a finalize that may never come.
  if (biSink.isGuid(validated.wix_user_id)) {
    await pushToBi(projectDir, ts, state.seq + 1,
      healRows(state, validated.wix_user_id), validated.wix_user_id, 'identity_heal',
      { throughSeq: state.seq + 1 });
  }
  return { recorded: Object.keys(validated) };
}

async function record(projectDir, eventInput, { now, resolveIdentity } = {}) {
  await adoptIdentity(projectDir, { now, resolveIdentity });
  const ts = nowIso(now);
  const { state, validated, eventRecord } = withJournalLock(projectDir, () => {
    const { state: fresh } = requireActiveRun(projectDir);
    let checked;
    try {
      checked = validateEvent(eventInput, fresh);
    } catch (error) {
      if (error instanceof ValidationError) {
        // The schema gate refused the call: count it so telemetry loss is never
        // silent, then surface the errors for a retry.
        appendJournal(projectDir, {
          type: 'rejected',
          ts,
          seq: fresh.seq + 1,
          class_key: error.hint || null,
          reasons: error.errors,
        });
        error.hint = 'fix the listed fields and retry the record call';
      }
      throw error;
    }
    const built = {
      type: 'event',
      ts,
      seq: fresh.seq + 1,
      class_key: checked.classKey,
      scrub_hits: checked.scrubHits,
      event: checked.event,
    };
    appendJournal(projectDir, built);
    return { state: fresh, validated: checked, eventRecord: built };
  });
  // Emit the folded `replatform_run_event` row, under the run's subject (the operator's id
  // once known, else its placeholder). With backlog behind the watermark, this push carries
  // the full heal set instead of just the new class — an event push is often
  // the first working transport after a failure.
  {
    const { subject, operatorId, view } = resolveSendSubject(projectDir, state, ts);
    if (subject) {
      const rows = biBacklog(view)
        ? healRows(state, operatorId, [eventRecord])
        : [foldedRowForClass(state, eventRecord)];
      await pushToBiForSubject(operatorId, projectDir, ts, eventRecord.seq, rows, subject, 'run_event',
        { throughSeq: eventRecord.seq });
    }
  }
  const priorInClass = state.events.filter((r) => r.class_key === validated.classKey).length;
  return {
    recorded: true,
    seq: state.seq + 1,
    event_type: validated.event.event_type,
    folded_into_existing_class: priorInClass > 0,
    scrub_hits: validated.scrubHits,
  };
}

// A stage whose own handoff still lists open work did not pass, whatever the caller says.
// Only stages that write a handoff with these fields are checked; a missing file checks nothing.
const STAGE_HANDOFFS = { setup_discovery: 'setup/llm-handoff.json', codegen: 'execution/llm-handoff.json' };
function openHandoffItems(projectDir, stageName) {
  const relative = STAGE_HANDOFFS[stageName];
  if (!relative) return null;
  let handoff;
  try { handoff = JSON.parse(fs.readFileSync(path.join(projectDir, relative), 'utf8')); } catch { return null; }
  // Only fields that mean "this stage did not finish"; open decisions or notes are not blockers.
  const open = ['blockers', 'unresolved', 'openItems'].flatMap((key) => (Array.isArray(handoff?.[key]) ? handoff[key] : []));
  if (handoff?.needsLlm !== true && open.length === 0) return null;
  return { relative, codes: open.map((item) => (item && typeof item === 'object' ? item.code ?? item.id : item)).filter(Boolean) };
}

function stage(projectDir, action, stageName, { outcome, now } = {}) {
  return withJournalLock(projectDir, () => stageLocked(projectDir, action, stageName, { outcome, now }));
}

function stageLocked(projectDir, action, stageName, { outcome, now } = {}) {
  const ts = nowIso(now);
  const { state } = requireActiveRun(projectDir);
  const errors = [];
  checkEnum(errors, 'stage', stageName, STAGES);
  if (errors.length > 0) throw new ValidationError(errors);
  if (state.openWait) {
    throw new ValidationError(
      [`a wait interval is open (stage ${state.openWait.stage})`],
      'call `wait end` before stage boundaries',
    );
  }
  if (action === 'start') {
    if (state.openStage) {
      throw new ValidationError(
        [`stage ${state.openStage} is still open`],
        `call \`stage end ${state.openStage} --outcome <passed|halted|failed|skipped>\` first`,
      );
    }
    appendJournal(projectDir, { type: 'stage_start', ts, seq: state.seq + 1, stage: stageName });
    return { stage: stageName, started: true };
  }
  if (action === 'end') {
    if (state.openStage !== stageName) {
      throw new ValidationError([
        state.openStage
          ? `open stage is ${state.openStage}, not ${stageName}`
          : `no stage is open (last stage: ${state.lastStage || 'none'})`,
      ]);
    }
    const resolvedOutcome = outcome === undefined || outcome === null ? 'passed' : outcome;
    const outcomeErrors = [];
    checkEnum(outcomeErrors, 'outcome', resolvedOutcome, STAGE_OUTCOMES);
    if (outcomeErrors.length > 0) throw new ValidationError(outcomeErrors);
    const open = resolvedOutcome === 'passed' ? openHandoffItems(projectDir, stageName) : null;
    if (open) {
      throw new ValidationError(
        [`${stageName} cannot end as passed: ${open.relative} still has open items${open.codes.length ? ` (${open.codes.join(', ')})` : ' (needsLlm: true)'}`],
        `resolve them and rewrite ${open.relative}, or call \`stage end ${stageName} --outcome halted\``,
      );
    }
    appendJournal(projectDir, { type: 'stage_end', ts, seq: state.seq + 1, stage: stageName, outcome: resolvedOutcome });
    return { stage: stageName, ended: true, outcome: resolvedOutcome };
  }
  throw new ValidationError([`stage action must be start or end (got: ${action})`]);
}

// Attach measured latency/token counts to a stage. Separate from `stage end` on purpose: a stage may
// be metered several times (each generated script reports its own invocation), and token counts are
// known by the agent runtime rather than by whatever closed the stage.
function meter(projectDir, input, { now } = {}) {
  const ts = nowIso(now);
  const payload = withJournalLock(projectDir, () => {
    const { state } = requireActiveRun(projectDir);
    const checked = validateMeter(input, { openStage: state.openStage, lastStage: state.lastStage });
    appendJournal(projectDir, { type: 'meter', ts, seq: state.seq + 1, stage: checked.stage, meter: checked });
    return checked;
  });
  const recorded = Object.keys(payload).filter((k) => k !== 'stage');
  return { metered: true, stage: payload.stage, fields: recorded };
}

// Counts key/value source reads for the run. Accumulates across calls, so a
// caller may report per-batch rather than having to hold a running total itself.
function sourceRead(projectDir, input, { now } = {}) {
  const ts = nowIso(now);
  const payload = withJournalLock(projectDir, () => {
    const { state } = requireActiveRun(projectDir);
    const checked = validateSourceRead(input);
    appendJournal(projectDir, { type: 'source_read', ts, seq: state.seq + 1, source_read: checked });
    return checked;
  });
  return { recorded: true, fields: Object.keys(payload) };
}

/**
 * The run's totals, as the source-read accounting defines them. `redacted_values` is summed from EVERY
 * response's redactionMetadata, not only key/value reads -- the guard applies to every read,
 * and an earlier draft of the response contract made that total impossible to compute
 * correctly by nesting the counter inside a key/value-only object.
 */
function summarizeSourceReads(state) {
  const totals = { discovery_queries: 0, scoped_queries: 0, tier_refusals: 0, redacted_values: 0 };
  const discoveryTables = new Set();
  const scopedTables = new Set();
  for (const read of state.sourceReads || []) {
    for (const key of SOURCE_READ_COUNT_KEYS) totals[key] += read[key] || 0;
    for (const table of read.discovery_tables || []) discoveryTables.add(table);
    for (const table of read.scoped_tables || []) scopedTables.add(table);
  }
  return {
    ...totals,
    discovery_tables: [...discoveryTables].sort(),
    scoped_tables: [...scopedTables].sort(),
  };
}

// Records the one `transcript_digest` per run: a script-computed
// parse of the Claude Code session transcript, never an agent self-report. No
// BI push here — the fields ride to BI inside `finalize`'s existing push, once
// `rollup.transcript_digest` carries them. A second call in the same run
// overwrites the prior one (replay keeps the latest), which is what an offline
// re-run against a more complete transcript is for.
async function transcriptDigest(projectDir, payload, { now } = {}) {
  const ts = nowIso(now);
  const seq = withJournalLock(projectDir, () => {
    const { state } = requireActiveRun(projectDir);
    const validated = validateTranscriptDigest(payload);
    const next = state.seq + 1;
    appendJournal(projectDir, { type: 'transcript_digest', ts, seq: next, run_id: state.runStart.run_id, ...validated });
    return next;
  });
  return { recorded: true, seq };
}

async function wait(projectDir, action, { now, haltSubtype, skill, what, resolveIdentity } = {}) {
  await adoptIdentity(projectDir, { now, resolveIdentity });
  const ts = nowIso(now);
  const first = withJournalLock(projectDir, () => {
    const { state } = requireActiveRun(projectDir);
    if (action === 'start') {
      if (state.openWait) {
        throw new ValidationError(['a wait interval is already open']);
      }
      const attributedStage = state.openStage || state.lastStage || 'config';
      if (haltSubtype !== undefined && haltSubtype !== null) {
        // The paired halt_needs_user event is emitted mechanically so a stalled
        // run never loses the "why" to a missed record call.
        const validated = validateEvent({
          event_type: 'halt_needs_user',
          subtype: haltSubtype,
          stage: attributedStage,
          skill: skill || 'wix-replatform',
          severity: 'blocking',
          what_happened: what || `Run halted at ${attributedStage} waiting on the user (${haltSubtype}).`,
        }, state);
        const eventRecord = {
          type: 'event', ts, seq: state.seq + 1, class_key: validated.classKey, scrub_hits: validated.scrubHits, event: validated.event,
        };
        appendJournal(projectDir, eventRecord);
        return { kind: 'halt', state, attributedStage, eventRecord };
      }
      if (skill !== undefined || what !== undefined) {
        throw new ValidationError(['--skill/--what on wait start are only valid together with --halt <subtype>']);
      }
      appendJournal(projectDir, { type: 'wait_start', ts, seq: state.seq + 1, stage: attributedStage });
      return { kind: 'started', attributedStage };
    }
    if (action === 'end') {
      if (!state.openWait) {
        throw new ValidationError(['no wait interval is open']);
      }
      appendJournal(projectDir, { type: 'wait_end', ts, seq: state.seq + 1 });
      return { kind: 'ended', waitedMs: Math.max(0, tsMs(ts) - tsMs(state.openWait.start)) };
    }
    throw new ValidationError([`wait action must be start or end (got: ${action})`]);
  });
  if (first.kind === 'ended') return { waiting: false, waited_ms: first.waitedMs };
  if (first.kind === 'started') return { waiting: true, stage: first.attributedStage, halt_recorded: false };

  const { state, attributedStage, eventRecord } = first;
  {
    const { subject, operatorId, view } = resolveSendSubject(projectDir, state, ts);
    if (subject) {
      const rows = biBacklog(view)
        ? healRows(state, operatorId, [eventRecord])
        : [foldedRowForClass(state, eventRecord)];
      await pushToBiForSubject(operatorId, projectDir, ts, eventRecord.seq, rows, subject, 'run_event',
        { throughSeq: eventRecord.seq });
    }
  }
  // The wait opens after the push's markers, with a seq from a fresh read.
  withJournalLock(projectDir, () => {
    const { state: fresh } = requireActiveRun(projectDir);
    appendJournal(projectDir, { type: 'wait_start', ts, seq: fresh.seq + 1, stage: attributedStage });
  });
  return { waiting: true, stage: attributedStage, halt_recorded: true };
}

async function finalize(projectDir, rollupInput, { now, resolveIdentity } = {}) {
  // Last chance to find the operator's id before the run closes under its placeholder.
  await adoptIdentity(projectDir, { now, resolveIdentity });
  const ts = nowIso(now);
  const { state, payload, provisionalJournal } = withJournalLock(projectDir, () => {
    const { state: fresh } = requireActiveRun(projectDir);
    const checked = validateFinalizeInput(rollupInput, fresh);
    // A journal that predates placeholders gets one before the closing record is written.
    ensureSubjectLocked(projectDir, fresh, ts);
    const records = [];
    let seq = fresh.seq;
    if (fresh.openWait) {
      seq += 1;
      records.push({ type: 'wait_end', ts, seq });
    }
    if (fresh.openStage) {
      seq += 1;
      records.push({
        type: 'stage_end',
        ts,
        seq,
        stage: fresh.openStage,
        outcome: OUTCOME_FOR_TERMINAL[checked.terminal_state],
      });
    }
    seq += 1;
    records.push({ type: 'finalize', ts, seq, payload: checked });
    appendJournalRecords(projectDir, records);
    const provisionalJournal = readJournal(projectDir);
    return { state: fresh, payload: checked, provisionalJournal };
  });
  const finalizeSeq = replay(provisionalJournal.records).seq;

  // BI push before archival, so a failed push is journaled into the run and
  // lands in the archived telemetry_health. The full re-push (started row +
  // every folded event class + the finalized rollup) heals any mid-run failure
  // or identity-not-yet-known hold-back: the stream is append-only and the
  // reviewer dedupes on the natural keys, latest row wins. The pushed rollup's
  // own health can't include this push's outcome (it is assembled first) — if
  // the push fails there is no BI row at all, and the local document, which is
  // authoritative, records the failure.
  let biPush = { sent: 0, failed: 0, skipped: 0 };
  {
    const provisional = assembleDocument(projectDir, provisionalJournal.records, provisionalJournal.malformed);
    const rollup = provisional.document.rollup;
    const provisionalState = replay(provisionalJournal.records);
    const operatorId = biSink.isGuid(rollup.wix_user_id) ? rollup.wix_user_id : null;
    // Without the operator's id the closing record goes under the run's placeholder, with no
    // operator on it: the run is counted as unattributed, not lost.
    const subject = operatorId || provisionalState.provisionalSubject;
    if (subject) {
      const rows = [
        biSink.startedRunRow(provisionalState.runStart, operatorId),
        ...provisional.document.events.map(biSink.eventRow),
        biSink.finalizedRunRow(rollup),
      ];
      // A disabled sink lands here too: pushRows reports the rows as skipped
      // and pushToBi journals them, so the kill-switch is never silent.
      const pushed = await pushToBiForSubject(operatorId, projectDir, ts, finalizeSeq, rows, subject, 'finalize',
        { throughSeq: finalizeSeq, deadlineMs: FINALIZE_DEADLINE_MS, force: true });
      biPush = { sent: pushed.sent, failed: pushed.failed, skipped: pushed.skipped };
    }
  }

  const { finalState, assembled } = withJournalLock(projectDir, () => {
    let finalJournal = readJournal(projectDir);
    let current = finalJournal ? replay(finalJournal.records) : null;
    if (!current || !current.runStart || current.runStart.run_id !== state.runStart.run_id || !current.finalize) {
      // Another writer already completed this finalize's archival (for example a `start`
      // that found the finalize record); read the archived journal instead.
      finalJournal = readJournalFile(path.join(runsDir(projectDir), `${archiveName(state)}.jsonl`));
      current = replay(finalJournal.records);
      return {
        finalState: current,
        assembled: assembleDocument(projectDir, finalJournal.records, finalJournal.malformed),
      };
    }
    return { finalState: current, assembled: archiveFinalizedJournal(projectDir, finalJournal, current) };
  });

  const health = assembled.document.rollup.telemetry_health;
  return {
    finalized: true,
    run_id: finalState.runStart.run_id,
    attempt: finalState.runStart.attempt,
    terminal_state: payload.terminal_state,
    bi_push: biPush,
    signal_file: 'run-telemetry.json',
    event_classes: assembled.document.events.length,
    telemetry_health: health,
    evidence_bytes: assembled.evidenceBytes,
    ...(assembled.evidenceBytes > EVIDENCE_SIZE_FLAG_BYTES
      ? {
          evidence_breakdown: assembled.evidenceFiles
            .sort((a, b) => b.bytes - a.bytes)
            .slice(0, 10),
        }
      : {}),
  };
}

async function rebuild(projectDir, { attempt, push } = {}) {
  // Re-assemble a finalized run's signal document from its archived journal.
  // The document is a pure function of the journal, so recorder fixes (fold
  // semantics, manifest rules, health flags) can be applied to past runs
  // without touching the captured record itself. With `push`, the rebuilt run
  // is re-emitted to BI (outage backfill / post-fix re-push) — idempotent by
  // the reviewer's query-time dedup on the natural keys.
  const dir = runsDir(projectDir);
  const archived = fs.existsSync(dir)
    ? fs.readdirSync(dir)
      .map((name) => {
        const match = /^run-(\d+)-.+\.jsonl$/.exec(name);
        return match ? { attempt: Number(match[1]), name } : null;
      })
      .filter(Boolean)
    : [];
  if (archived.length === 0) {
    throw new ValidationError(
      ['no archived run journals to rebuild from'],
      'rebuild re-assembles a finalized signal document from telemetry/runs/run-<attempt>-<run_id>.jsonl',
    );
  }
  const latestAttempt = archived.reduce((max, a) => Math.max(max, a.attempt), 0);
  const target = attempt === undefined || attempt === null
    ? archived.find((a) => a.attempt === latestAttempt)
    : archived.find((a) => a.attempt === attempt);
  if (!target) {
    throw new ValidationError([`no archived journal for attempt ${attempt} (have: ${archived.map((a) => a.attempt).join(', ')})`]);
  }
  const journal = readJournalFile(path.join(dir, target.name));
  const state = replay(journal.records);
  if (!state.runStart || !state.finalize) {
    throw new ValidationError([`archived journal ${target.name} is not a finalized run`]);
  }
  const assembled = assembleDocument(projectDir, journal.records, journal.malformed);
  writeJsonAtomic(path.join(dir, `${archiveName(state)}.json`), assembled.document);
  const signalUpdated = target.attempt === latestAttempt;
  if (signalUpdated) {
    writeJsonAtomic(signalPath(projectDir), assembled.document);
  }
  // The archived journal is immutable, so a backfill push's outcome cannot be
  // journaled into the run — it is reported here, to the operator running the
  // explicit backfill, instead.
  let biPush;
  if (push) {
    const rollup = assembled.document.rollup;
    const rows = [
      biSink.startedRunRow(state.runStart, rollup.wix_user_id),
      ...assembled.document.events.map(biSink.eventRow),
      biSink.finalizedRunRow(rollup),
    ];
    const rebuildSubject = biSink.isGuid(rollup.wix_user_id) ? rollup.wix_user_id : state.provisionalSubject;
    const result = await biSink.pushRows(rows, rebuildSubject);
    biPush = {
      sent: result.sent,
      failed: result.failures.reduce((sum, f) => sum + f.rows, 0),
      ...(result.failures.length > 0 ? { failure_details: [...new Set(result.failures.map((f) => f.detail))] } : {}),
    };
  }
  return {
    rebuilt: true,
    run_id: state.runStart.run_id,
    attempt: target.attempt,
    signal_file_updated: signalUpdated,
    event_classes: assembled.document.events.length,
    telemetry_health: assembled.document.rollup.telemetry_health,
    ...(biPush ? { bi_push: biPush } : {}),
  };
}

function status(projectDir) {
  const journal = readJournal(projectDir);
  if (!journal) {
    const lastAttempt = lastFinalizedAttempt(projectDir);
    return { active: false, finalized_attempts: lastAttempt };
  }
  const state = replay(journal.records);
  return {
    active: !state.finalize,
    run_id: state.runStart ? state.runStart.run_id : null,
    attempt: state.runStart ? state.runStart.attempt : null,
    session_count: state.sessionCount,
    open_stage: state.openStage,
    open_wait: Boolean(state.openWait),
    events_recorded: state.events.length,
    distinct_classes: new Set(state.events.map((r) => r.class_key)).size,
    rejected_calls: state.rejected.length,
    last_seq: state.seq,
    // The confirmed watermark for the run's current subject (the operator's id once known, else its
    // placeholder), matching `bi_backlog`; a mark written before subjects were recorded counts for the
    // operator's id only.
    bi_push_ok_seq: currentSubjectOkSeq(state),
    bi_backlog: state.runStart ? biBacklog(state) : true,
  };
}

module.exports = {
  TELEMETRY_SCHEMA_VERSION,
  STAGES,
  STAGE_OUTCOMES,
  TERMINAL_STATES,
  SEVERITIES,
  EVENT_SUBTYPES,
  FIELD_TIERS,
  FREE_TEXT_MAX,
  SHAPES_KEPT_PER_CLASS,
  ValidationError,
  scrubText,
  validateDims,
  validateEvent,
  validateFinalizeInput,
  validateTranscriptDigest,
  journalPath,
  signalPath,
  runsDir,
  readJournal,
  replay,
  start,
  resumeIfStale,
  withJournalLock,
  ROUTE_RESUME_MIN_GAP_MS,
  SESSION_ADOPT_WINDOW_MS,
  HEAL_DEADLINE_MS,
  LOCK_WAIT_MS,
  MARKER_LOCK_WAIT_MS,
  IMPLICIT_WAIT_MIN_MS,
  biBacklog,
  adoptIdentity,
  flush,
  attention,
  closingAttention,
  dims,
  record,
  stage,
  meter,
  sourceRead,
  summarizeSourceReads,
  resolveSkillsVersion,
  validateSourceRead,
  SOURCE_READ_COUNT_KEYS,
  transcriptDigest,
  wait,
  finalize,
  rebuild,
  status,
  validateMeter,
  computeCost,
  METER_DURATION_KEYS,
  METER_COUNT_KEYS,
};