pikasTech-HWLAB/internal/cloud/workbench-kafka-refresh-handoff.ts

// SPEC: PJ2026-0104010803 Workbench Kafka retention refresh handoff.
// Responsibility: prove one bounded retained hwlab.event.v1 scan before handing the same SSE writer to shared live fanout.

export function createWorkbenchKafkaRefreshHandoff(options = {}) {
  const sessionId = textValue(options.sessionId);
  const traceId = textValue(options.traceId);
  const allowUnscoped = options.allowUnscoped === true;
  const liveBufferLimit = positiveInteger(options.liveBufferLimit);
  const identityWindowLimit = positiveInteger(options.identityWindowLimit) ?? liveBufferLimit;
  const resumeAfterId = textValue(options.resumeAfterId);
  if (!sessionId && !traceId && !allowUnscoped) throw handoffError("workbench_kafka_refresh_scope_missing", "Kafka refresh handoff requires a session or trace scope.", "idle");
  if (!liveBufferLimit) throw handoffError("workbench_kafka_refresh_buffer_limit_invalid", "Kafka refresh handoff requires an explicit positive live buffer limit.", "idle");
  if (typeof options.subscribeLive !== "function" || typeof options.queryRetention !== "function" || typeof options.deliverEvent !== "function" || typeof options.deliverConnected !== "function") {
    throw handoffError("workbench_kafka_refresh_runtime_invalid", "Kafka refresh handoff runtime callbacks are incomplete.", "idle");
  }

  const bufferedLive = [];
  const barrierByPartition = new Map();
  const replayProofByTransport = new Map();
  const stableIdentityByTransport = new Map();
  const seenEventIds = new Map();
  const seenSourceEventIds = new Map();
  const sourceEventIdByEventId = new Map();
  const eventIdBySourceEventId = new Map();
  const identityWindow = [];
  const abortController = new AbortController();
  let phase = "idle";
  let unsubscribe = null;
  let failure = null;
  let bootstrapSettled = false;
  let scopedTopic = null;
  let scopedPartition = null;
  let queryDiagnostics = null;
  let deliveryChain = Promise.resolve(true);
  let pendingLiveDeliveries = 0;
  const counts = {
    matched: 0,
    filteredNonHwlab: 0,
    filteredMissingIdentity: 0,
    replayed: 0,
    buffered: 0,
    bufferedDelivered: 0,
    liveDelivered: 0,
    deduplicated: 0,
    resumeSkipped: 0
  };
  let resumeFound = resumeAfterId === null;
  let retentionWindowAdvanced = false;

  function stop(reason = "connection-closed") {
    if (phase === "closed") return;
    phase = "closed";
    abortController.abort(reason);
    unsubscribeLive();
    bufferedLive.length = 0;
  }

  async function start() {
    if (phase !== "idle") throw handoffError("workbench_kafka_refresh_already_started", "Kafka refresh handoff can only be started once.", phase);
    phase = "buffering";
    unsubscribe = options.subscribeLive(onLiveEnvelope);
    try {
      const queryResult = await options.queryRetention({ signal: abortController.signal });
      queryDiagnostics = boundedQueryDiagnostics(queryResult);
      if (failure) throw failure;
      if (phase === "closed") throw handoffError("workbench_kafka_refresh_stopped", "Kafka refresh handoff stopped before the retention barrier completed.", phase);
      validateQueryCompletion(queryResult);
      installBarrier(queryResult);
      const replayRecords = validateReplayRecords(queryResult);

      phase = "replaying";
      for (const record of replayRecords) {
        if (failure) throw failure;
        if (!record.deliver) continue;
        await deliverSerial(record.envelope, record.transport, "replay");
        counts.replayed += 1;
      }

      phase = "handoff";
      await deliverHandoffSerial(handoffSummary("handoff"));
      phase = "flushing";
      while (bufferedLive.length > 0) {
        if (failure) throw failure;
        const item = bufferedLive.shift();
        const decision = classifyLiveRecord(item.envelope, item.transport);
        if (!decision.deliver) continue;
        await deliverSerial(decision.envelope, decision.transport, "buffered-live");
        counts.bufferedDelivered += 1;
      }

      const summary = handoffSummary("live");
      const connectedDelivery = deliverConnectedSerial(summary);
      phase = "live";
      await connectedDelivery;
      if (failure) throw failure;
      bootstrapSettled = true;
      return summary;
    } catch (error) {
      const typed = asHandoffError(error, phase);
      fail(typed);
      throw typed;
    }
  }

  function onLiveEnvelope(envelope, transport) {
    if (phase === "closed" || phase === "failed") return;
    if (allowUnscoped) {
      const filtered = unscopedEnvelopeFilter(envelope);
      if (filtered) {
        counts[filtered] += 1;
        return;
      }
    }
    if (!envelopeMatchesScope(envelope, sessionId, traceId)) return;
    if (phase !== "live") {
      if (bufferedLive.length >= liveBufferLimit) {
        fail(handoffError("workbench_kafka_refresh_live_buffer_overflow", "Kafka refresh live buffer exceeded its YAML-owned capacity before handoff.", phase, { liveBufferLimit }));
        return;
      }
      bufferedLive.push({ envelope, transport });
      counts.buffered += 1;
      return;
    }
    let decision;
    try {
      decision = classifyLiveRecord(envelope, transport);
    } catch (error) {
      fail(asHandoffError(error, phase));
      return;
    }
    if (!decision.deliver) return;
    if (pendingLiveDeliveries >= liveBufferLimit) {
      fail(handoffError("workbench_kafka_refresh_live_buffer_overflow", "Kafka refresh live delivery queue exceeded its YAML-owned capacity after handoff.", phase, { liveBufferLimit }));
      return;
    }
    pendingLiveDeliveries += 1;
    void deliverSerial(decision.envelope, decision.transport, "live")
      .then(() => { counts.liveDelivered += 1; })
      .catch((error) => fail(asHandoffError(error, phase)))
      .finally(() => { pendingLiveDeliveries = Math.max(0, pendingLiveDeliveries - 1); });
  }

  function validateQueryCompletion(result) {
    const reason = textValue(result?.completion?.reason ?? result?.completionReason) || "unknown";
    const complete = result?.completion?.complete === true
      && result?.completion?.barrierReached === true
      && result?.completion?.retentionStartVerified === true
      && result?.reachedEndOffsets === true;
    if (!complete) {
      const codeByReason = {
        timeout: "workbench_kafka_refresh_timeout",
        "scan-limit": "workbench_kafka_refresh_scan_limit",
        limit: "workbench_kafka_refresh_matched_event_limit",
        aborted: "workbench_kafka_refresh_aborted",
        "retention-start-moved": "workbench_kafka_refresh_retention_start_moved",
        "retention-start-unavailable": "workbench_kafka_refresh_retention_start_unavailable",
        "retention-start-unverified": "workbench_kafka_refresh_retention_start_unverified"
      };
      throw handoffError(codeByReason[reason] ?? "workbench_kafka_refresh_barrier_incomplete", `Kafka retention scan did not prove its captured barrier (${reason}).`, phase, {
        completionReason: reason,
        matchedEventLimit: result?.limit,
        scanLimit: result?.scanLimit,
        timeoutMs: result?.timeoutMs,
        scannedCount: result?.scannedCount,
        matchedCount: result?.matchedCount,
        partitions: Array.isArray(result?.endOffsets) ? result.endOffsets.map((entry) => entry?.partition) : []
      });
    }
    if (result?.endOffsetsAvailable !== true || !Array.isArray(result?.endOffsets) || result.endOffsets.length === 0) {
      throw handoffError("workbench_kafka_refresh_barrier_missing", "Kafka retention scan completed without a usable end-offset barrier.", phase);
    }
  }

  function installBarrier(result) {
    scopedTopic = textValue(result?.topic);
    if (!scopedTopic) {
      throw handoffError("workbench_kafka_refresh_transport_identity_missing", "Kafka retention barrier is missing the current HWLAB topic identity.", phase);
    }
    barrierByPartition.clear();
    for (const entry of result.endOffsets) {
      const partition = partitionValue(entry?.partition);
      const endOffset = offsetValue(entry?.endOffset);
      if (partition === null || endOffset === null || barrierByPartition.has(partition)) {
        throw handoffError("workbench_kafka_refresh_barrier_invalid", "Kafka retention barrier contains an invalid or duplicate partition entry.", phase);
      }
      barrierByPartition.set(partition, endOffset);
    }
  }

  function validateReplayRecords(result) {
    const records = Array.isArray(result?.events) ? result.events : [];
    counts.matched = records.length;
    if (!allowUnscoped && traceId && records.length === 0) {
      throw handoffError("workbench_kafka_refresh_trace_empty", "Trace-scoped Kafka refresh found no retained events after a complete scan.", phase);
    }
    const replayPartitions = new Set();
    let previousOffset = null;
    const normalized = [];
    for (const record of records) {
      const envelope = record?.value;
      if (allowUnscoped) {
        const filtered = unscopedEnvelopeFilter(envelope);
        if (filtered) {
          counts[filtered] += 1;
          continue;
        }
      }
      if (!envelopeMatchesScope(envelope, sessionId, traceId) || envelope?.schema !== "hwlab.event.v1") {
        throw handoffError("workbench_kafka_refresh_scope_mismatch", "Kafka retention query returned an envelope outside the authorized scope.", phase);
      }
      const transport = requireTransportIdentity(record, scopedTopic, phase);
      const stableIdentity = requireStableIdentity(envelope, phase);
      const stableKey = stableIdentityKey(stableIdentity);
      const barrier = barrierByPartition.get(transport.partition);
      if (barrier === undefined || compareOffsets(transport.offset, barrier) >= 0) {
        throw handoffError("workbench_kafka_refresh_post_barrier_record", "Kafka retention query returned a record at or beyond its exclusive barrier.", phase);
      }
      replayPartitions.add(transport.partition);
      if (replayPartitions.size > 1) throw handoffError("workbench_kafka_refresh_multi_partition", "One authorized Kafka refresh scope spans multiple HWLAB topic partitions.", phase);
      if (previousOffset !== null && compareOffsets(transport.offset, previousOffset) <= 0) {
        throw handoffError("workbench_kafka_refresh_order_invalid", "Kafka retention records are not strictly ordered within the authorized partition.", phase);
      }
      previousOffset = transport.offset;
      scopedPartition = transport.partition;
      const transportKey = transportIdentityKey(transport);
      const priorProof = replayProofByTransport.get(transportKey);
      if (priorProof && priorProof.stableKey !== stableKey) throw handoffError("workbench_kafka_refresh_transport_conflict", "One Kafka transport identity maps to conflicting stable events.", phase);
      const deliver = rememberIdentity(transportKey, stableIdentity, phase);
      replayProofByTransport.set(transportKey, stableIdentityByTransport.get(transportKey));
      normalized.push({ envelope, transport, stableIdentity, deliver });
    }
    if (resumeAfterId) {
      const resumeIndex = normalized.findIndex((record) => record.stableIdentity.eventId === resumeAfterId || record.stableIdentity.sourceEventId === resumeAfterId);
      resumeFound = resumeIndex >= 0;
      retentionWindowAdvanced = resumeIndex < 0;
      if (resumeIndex >= 0) {
        counts.resumeSkipped = resumeIndex + 1;
        for (let index = 0; index <= resumeIndex; index += 1) normalized[index].deliver = false;
      }
    }
    return normalized;
  }

  function classifyLiveRecord(envelope, rawTransport) {
    if (envelope?.schema !== "hwlab.event.v1" || !envelopeMatchesScope(envelope, sessionId, traceId)) {
      throw handoffError("workbench_kafka_refresh_scope_mismatch", "Shared live fanout delivered an envelope outside the authorized scope.", phase);
    }
    const transport = requireTransportIdentity(rawTransport, scopedTopic, phase);
    const stableIdentity = requireStableIdentity(envelope, phase);
    const stableKey = stableIdentityKey(stableIdentity);
    if (scopedPartition !== null && transport.partition !== scopedPartition) {
      throw handoffError("workbench_kafka_refresh_multi_partition", "One authorized Kafka refresh scope spans multiple HWLAB topic partitions.", phase, {
        expectedPartition: scopedPartition,
        actualPartition: transport.partition
      });
    }
    const barrier = barrierByPartition.get(transport.partition);
    if (barrier === undefined) throw handoffError("workbench_kafka_refresh_barrier_partition_missing", "Shared live fanout used a partition absent from the captured barrier.", phase, { partition: transport.partition });
    if (scopedPartition === null) scopedPartition = transport.partition;
    const transportKey = transportIdentityKey(transport);
    if (compareOffsets(transport.offset, barrier) < 0) {
      const replayProof = replayProofByTransport.get(transportKey);
      if (replayProof?.stableKey !== stableKey) {
        throw handoffError("workbench_kafka_refresh_barrier_gap", "A pre-barrier shared live record was not proven by the retained replay scan.", phase, { partition: transport.partition });
      }
    }
    return { envelope, transport, deliver: rememberIdentity(transportKey, stableIdentity, phase) };
  }

  function rememberIdentity(transportKey, stableIdentity, currentPhase) {
    const stableKey = stableIdentityKey(stableIdentity);
    const priorStable = stableIdentityByTransport.get(transportKey);
    if (priorStable && priorStable.stableKey !== stableKey) {
      throw handoffError("workbench_kafka_refresh_transport_conflict", "One Kafka transport identity maps to conflicting stable events.", currentPhase);
    }
    const priorSourceEventId = stableIdentity.eventId ? sourceEventIdByEventId.get(stableIdentity.eventId)?.sourceEventId : null;
    if (priorSourceEventId && stableIdentity.sourceEventId && priorSourceEventId !== stableIdentity.sourceEventId) {
      throw handoffError("workbench_kafka_refresh_stable_identity_conflict", "One outer eventId maps to conflicting sourceEventId identities.", currentPhase);
    }
    const priorEventId = stableIdentity.sourceEventId ? eventIdBySourceEventId.get(stableIdentity.sourceEventId)?.eventId : null;
    if (priorEventId && stableIdentity.eventId && priorEventId !== stableIdentity.eventId) {
      throw handoffError("workbench_kafka_refresh_stable_identity_conflict", "One outer sourceEventId maps to conflicting eventId identities.", currentPhase);
    }
    const duplicate = Boolean(
      (stableIdentity.eventId && seenEventIds.has(stableIdentity.eventId))
      || (stableIdentity.sourceEventId && seenSourceEventIds.has(stableIdentity.sourceEventId))
    );
    const entry = { transportKey, stableKey, eventId: stableIdentity.eventId, sourceEventId: stableIdentity.sourceEventId };
    stableIdentityByTransport.set(transportKey, entry);
    if (stableIdentity.eventId) {
      seenEventIds.set(stableIdentity.eventId, entry);
      if (stableIdentity.sourceEventId) sourceEventIdByEventId.set(stableIdentity.eventId, entry);
    }
    if (stableIdentity.sourceEventId) {
      seenSourceEventIds.set(stableIdentity.sourceEventId, entry);
      if (stableIdentity.eventId) eventIdBySourceEventId.set(stableIdentity.sourceEventId, entry);
    }
    identityWindow.push(entry);
    trimIdentityWindow();
    if (duplicate) {
      counts.deduplicated += 1;
      return false;
    }
    return true;
  }

  function trimIdentityWindow() {
    while (identityWindow.length > identityWindowLimit) {
      const entry = identityWindow.shift();
      if (!entry) return;
      if (stableIdentityByTransport.get(entry.transportKey) === entry) stableIdentityByTransport.delete(entry.transportKey);
      if (replayProofByTransport.get(entry.transportKey) === entry) replayProofByTransport.delete(entry.transportKey);
      if (entry.eventId && seenEventIds.get(entry.eventId) === entry) seenEventIds.delete(entry.eventId);
      if (entry.sourceEventId && seenSourceEventIds.get(entry.sourceEventId) === entry) seenSourceEventIds.delete(entry.sourceEventId);
      if (entry.eventId && sourceEventIdByEventId.get(entry.eventId) === entry) sourceEventIdByEventId.delete(entry.eventId);
      if (entry.sourceEventId && eventIdBySourceEventId.get(entry.sourceEventId) === entry) eventIdBySourceEventId.delete(entry.sourceEventId);
    }
  }

  function deliverSerial(envelope, transport, delivery) {
    const run = async () => {
      if (failure || phase === "failed" || phase === "closed") throw failure ?? handoffError("workbench_kafka_refresh_stopped", "Kafka refresh delivery stopped before the queued business event could be written.", phase);
      const written = await options.deliverEvent(envelope, transport, delivery);
      if (written === false) throw handoffError("workbench_kafka_refresh_sse_write_failed", "Kafka refresh business event could not be written to SSE.", phase);
      return true;
    };
    deliveryChain = deliveryChain.then(run);
    return deliveryChain;
  }

  function deliverConnectedSerial(summary) {
    const run = async () => {
      if (failure || phase === "failed" || phase === "closed") throw failure ?? handoffError("workbench_kafka_refresh_stopped", "Kafka refresh delivery stopped before the connected contract could be written.", phase);
      const written = await options.deliverConnected(summary);
      if (written === false) throw handoffError("workbench_kafka_refresh_sse_write_failed", "Kafka refresh connected contract could not be written to SSE.", phase);
      return true;
    };
    deliveryChain = deliveryChain.then(run);
    return deliveryChain;
  }

  function deliverHandoffSerial(summary) {
    if (typeof options.deliverHandoff !== "function") return Promise.resolve(true);
    const run = async () => {
      if (failure || phase === "failed" || phase === "closed") throw failure ?? handoffError("workbench_kafka_refresh_stopped", "Kafka refresh delivery stopped before the handoff contract could be written.", phase);
      const written = await options.deliverHandoff(summary);
      if (written === false) throw handoffError("workbench_kafka_refresh_sse_write_failed", "Kafka refresh handoff contract could not be written to SSE.", phase);
      return true;
    };
    deliveryChain = deliveryChain.then(run);
    return deliveryChain;
  }

  function handoffSummary(summaryPhase = phase) {
    return {
      phase: summaryPhase,
      topic: scopedTopic,
      topicPartitions: scopedPartition === null ? [] : [scopedPartition],
      barrier: [...barrierByPartition.entries()].map(([partition, endOffset]) => ({ partition, endOffset })).sort((left, right) => left.partition - right.partition),
      counts: { ...counts },
      resume: {
        requested: resumeAfterId !== null,
        found: resumeFound,
        windowAdvanced: retentionWindowAdvanced
      },
      identityWindowLimit,
      query: queryDiagnostics,
      valuesRedacted: true
    };
  }

  function fail(error) {
    if (failure || phase === "closed") return;
    failure = asHandoffError(error, phase);
    phase = "failed";
    abortController.abort(failure.code);
    unsubscribeLive();
    bufferedLive.length = 0;
    if (bootstrapSettled && typeof options.onFailure === "function") {
      void Promise.resolve(options.onFailure(failure)).catch(() => undefined);
    }
  }

  function unsubscribeLive() {
    if (typeof unsubscribe !== "function") return;
    const stop = unsubscribe;
    unsubscribe = null;
    try { stop(); } catch {}
  }

  return {
    start,
    stop,
    phase: () => phase,
    summary: () => handoffSummary(phase),
    valuesPrinted: false
  };
}

function boundedQueryDiagnostics(result) {
  const timing = result?.timing && typeof result.timing === "object" ? result.timing : {};
  const index = result?.index && typeof result.index === "object" ? result.index : {};
  return {
    completionReason: textValue(result?.completion?.reason ?? result?.completionReason),
    complete: result?.completion?.complete === true,
    scannedCount: numericValue(result?.scannedCount),
    parsedCount: numericValue(result?.parsedCount),
    matchedCount: numericValue(result?.matchedCount),
    totalMatchedCount: numericValue(result?.totalMatchedCount),
    droppedMatchedCount: numericValue(result?.droppedMatchedCount),
    filterRejectedCount: numericValue(result?.filterRejectedCount),
    keyRejectedCount: numericValue(result?.keyRejectedCount),
    partitionKeyScoped: result?.partitionKeyScoped === true,
    targetPartition: partitionValue(result?.targetPartition),
    index: {
      hit: index.hit === true,
      source: textValue(index.source),
      fallbackReason: textValue(index.fallbackReason),
      indexedEventCount: numericValue(index.indexedEventCount ?? index.status?.indexedEventCount),
      indexedSessionCount: numericValue(index.indexedSessionCount ?? index.status?.indexedSessionCount),
      incompleteSessionCount: numericValue(index.incompleteSessionCount ?? index.status?.incompleteSessionCount),
      blocking: false,
      valuesRedacted: true
    },
    timing: {
      endOffsetSnapshotMs: numericValue(timing.endOffsetSnapshotMs),
      groupOffsetsMs: numericValue(timing.groupOffsetsMs),
      consumerConnectMs: numericValue(timing.consumerConnectMs),
      consumerSubscribeMs: numericValue(timing.consumerSubscribeMs),
      scanMs: numericValue(timing.scanMs),
      cleanupMs: numericValue(timing.cleanupMs),
      indexLookupMs: numericValue(timing.indexLookupMs),
      totalMs: numericValue(timing.totalMs),
      valuesPrinted: false
    },
    valuesPrinted: false
  };
}

function numericValue(value) {
  const number = Number(value);
  return Number.isFinite(number) && number >= 0 ? number : null;
}

export function workbenchKafkaRefreshErrorPayload(error, scope = {}) {
  const typed = asHandoffError(error, textValue(error?.phase) || "failed");
  const details = safeHandoffDiagnosticDetails(typed.details);
  return {
    type: "error",
    status: "blocked",
    phase: typed.phase,
    traceId: textValue(scope.traceId),
    sessionId: textValue(scope.sessionId),
    error: {
      code: typed.code,
      message: typed.message,
      phase: typed.phase,
      retryable: true,
      details,
      valuesRedacted: true
    },
    fallback: false,
    valuesRedacted: true
  };
}

function safeHandoffDiagnosticDetails(value) {
  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
  const details = {};
  const safeTextKeys = ["completionReason", "expectedTopic", "actualTopic"];
  const safeIntegerKeys = [
    "matchedEventLimit",
    "scanLimit",
    "timeoutMs",
    "scannedCount",
    "matchedCount",
    "liveBufferLimit",
    "partition",
    "expectedPartition",
    "actualPartition"
  ];
  for (const key of safeTextKeys) {
    const text = textValue(value[key]);
    if (text) details[key] = text.slice(0, 160);
  }
  for (const key of safeIntegerKeys) {
    const number = Number(value[key]);
    if (Number.isSafeInteger(number) && number >= 0) details[key] = number;
  }
  if (Array.isArray(value.partitions)) {
    details.partitions = [...new Set(value.partitions
      .map((partition) => partitionValue(partition))
      .filter((partition) => partition !== null))]
      .slice(0, 32);
  }
  return Object.keys(details).length > 0 ? details : null;
}

function envelopeMatchesScope(envelope, sessionId, traceId) {
  if (!envelope || typeof envelope !== "object" || Array.isArray(envelope)) return false;
  const envelopeSessionId = textValue(envelope.hwlabSessionId ?? envelope.sessionId);
  const envelopeTraceId = textValue(envelope.traceId);
  if (sessionId && envelopeSessionId !== sessionId) return false;
  if (traceId && envelopeTraceId !== traceId) return false;
  return true;
}

function unscopedEnvelopeFilter(envelope) {
  if (!envelope || typeof envelope !== "object" || Array.isArray(envelope) || envelope.schema !== "hwlab.event.v1") return "filteredNonHwlab";
  if (!textValue(envelope.eventId) && !textValue(envelope.sourceEventId)) return "filteredMissingIdentity";
  return null;
}

function requireTransportIdentity(value, expectedTopic, phase) {
  const topic = textValue(value?.topic);
  const partition = partitionValue(value?.partition);
  const offset = offsetValue(value?.offset);
  if (!topic || partition === null || offset === null) {
    throw handoffError("workbench_kafka_refresh_transport_identity_missing", "Kafka refresh requires the current HWLAB topic/partition/offset identity.", phase);
  }
  if (expectedTopic && topic !== expectedTopic) {
    throw handoffError("workbench_kafka_refresh_transport_topic_mismatch", "Kafka refresh received a live record from a topic other than the captured HWLAB retention topic.", phase, { expectedTopic, actualTopic: topic });
  }
  return { topic, partition, offset };
}

function requireStableIdentity(envelope, phase) {
  const eventId = textValue(envelope?.eventId);
  const sourceEventId = textValue(envelope?.sourceEventId);
  if (!eventId && !sourceEventId) throw handoffError("workbench_kafka_refresh_stable_identity_missing", "Kafka refresh requires outer eventId or sourceEventId identity.", phase);
  return { eventId, sourceEventId };
}

function stableIdentityKey(identity) {
  return `event:${identity.eventId ?? ""}|source:${identity.sourceEventId ?? ""}`;
}

function transportIdentityKey(transport) {
  return `${transport.topic}:${transport.partition}:${transport.offset}`;
}

function partitionValue(value) {
  const number = Number(value);
  return Number.isInteger(number) && number >= 0 ? number : null;
}

function offsetValue(value) {
  const text = textValue(value);
  if (!text || !/^\d+$/u.test(text)) return null;
  try {
    return BigInt(text) >= 0n ? text : null;
  } catch {
    return null;
  }
}

function compareOffsets(left, right) {
  const leftValue = BigInt(left);
  const rightValue = BigInt(right);
  return leftValue < rightValue ? -1 : leftValue > rightValue ? 1 : 0;
}

function positiveInteger(value) {
  const number = Number(value);
  return Number.isInteger(number) && number > 0 ? number : null;
}

function textValue(value) {
  return typeof value === "string" && value.trim() ? value.trim() : null;
}

function asHandoffError(error, phase) {
  if (error?.code && error?.phase) return error;
  return handoffError(error?.code ?? "workbench_kafka_refresh_failed", error instanceof Error ? error.message : String(error ?? "Kafka refresh handoff failed."), phase);
}

function handoffError(code, message, phase, details = null) {
  return Object.assign(new Error(message), {
    code,
    phase,
    details,
    statusCode: 503,
    valuesRedacted: true
  });
}