pikasTech-HWLAB/internal/cloud/workbench-kafka-debug-replay-contract.ts

// Responsibility: classify isolated Kafka debug replay progress without turning debug state into a product authority.

export const WORKBENCH_KAFKA_DEBUG_REPLAY_CONTRACT_VERSION = "workbench-kafka-debug-replay-v2";

export function createWorkbenchKafkaDebugReplayTracker(input = {}) {
  return {
    replayId: textValue(input.replayId),
    traceId: textValue(input.traceId),
    topic: textValue(input.topic),
    producer: {
      invoked: booleanOrNull(input.producer?.invoked),
      sourceMatched: booleanOrNull(input.producer?.sourceMatched),
      publishedCount: integerOrNull(input.producer?.publishedCount)
    },
    consumer: { ready: false, groupId: null },
    requestedOffsetRange: normalizeOffsetRange(input.offsetRange),
    counts: {
      scanned: 0,
      barrierScanned: 0,
      parsed: 0,
      traceMatched: 0,
      replayMatched: 0,
      matched: 0,
      delivered: 0,
      clientReceived: null,
      decoded: null,
      applied: null
    },
    rejectedByReason: {},
    offsets: {
      scanned: new Map(),
      matched: new Map(),
      delivered: new Map()
    },
    sourceLineage: null
  };
}

export function markWorkbenchKafkaDebugReplayConsumerReady(tracker, input = {}) {
  tracker.consumer.ready = true;
  tracker.consumer.groupId = textValue(input.groupId);
  return tracker;
}

export function observeWorkbenchKafkaDebugReplayRecord(tracker, observation = {}) {
  tracker.counts.scanned += 1;
  recordOffset(tracker.offsets.scanned, observation.partition, observation.offset);
  if (observation.inOffsetRange === false) {
    reject(tracker, "offset_range_mismatch");
    return tracker;
  }
  tracker.counts.barrierScanned += 1;
  if (observation.parsed !== true) {
    reject(tracker, "record_parse_rejected");
    return tracker;
  }
  tracker.counts.parsed += 1;
  if (observation.filterMatches?.traceId === true) tracker.counts.traceMatched += 1;
  if (observation.filterMatches?.replayId === true) tracker.counts.replayMatched += 1;
  if (observation.matched !== true) {
    reject(tracker, "filter_mismatch");
    return tracker;
  }
  tracker.counts.matched += 1;
  recordOffset(tracker.offsets.matched, observation.partition, observation.offset);
  return tracker;
}

export function recordWorkbenchKafkaDebugReplayDelivery(tracker, record = {}, delivered = true) {
  if (!delivered) {
    reject(tracker, "sse_write_failed");
    return tracker;
  }
  tracker.counts.delivered += 1;
  recordOffset(tracker.offsets.delivered, record.partition, record.offset);
  if (!tracker.sourceLineage) tracker.sourceLineage = safeSourceLineage(record.value);
  return tracker;
}

export function completeWorkbenchKafkaDebugReplay(tracker, input = {}) {
  const counts = {
    ...tracker.counts,
    clientReceived: nullableCount(input.clientReceived, tracker.counts.clientReceived),
    decoded: nullableCount(input.decoded, tracker.counts.decoded),
    applied: nullableCount(input.applied, tracker.counts.applied)
  };
  const terminalObserved = input.terminalObserved === true;
  const result = classifyWorkbenchKafkaDebugReplayResult({
    replayId: tracker.replayId,
    traceId: tracker.traceId,
    producer: tracker.producer,
    consumer: tracker.consumer,
    requestedOffsetRange: tracker.requestedOffsetRange,
    counts,
    completionReason: textValue(input.reason),
    terminalObserved
  });
  return {
    contractVersion: WORKBENCH_KAFKA_DEBUG_REPLAY_CONTRACT_VERSION,
    replayId: tracker.replayId,
    traceId: tracker.traceId,
    topic: tracker.topic,
    producer: tracker.producer,
    consumer: tracker.consumer,
    phase: result.phase,
    code: result.code,
    ok: result.ok,
    message: result.message,
    completionReason: textValue(input.reason),
    terminalObserved,
    counts,
    rejectedByReason: { ...tracker.rejectedByReason },
    offsetRange: {
      requested: tracker.requestedOffsetRange,
      scanned: summarizeOffsets(tracker.offsets.scanned),
      matched: summarizeOffsets(tracker.offsets.matched),
      delivered: summarizeOffsets(tracker.offsets.delivered)
    },
    sourceLineage: tracker.sourceLineage,
    valuesPrinted: false
  };
}

export function classifyWorkbenchKafkaDebugReplayResult(input = {}) {
  const counts = normalizeCounts(input.counts);
  const producer = input.producer ?? {};
  const consumer = input.consumer ?? {};
  const hasTraceFilter = Boolean(textValue(input.traceId));
  const hasReplayFilter = Boolean(textValue(input.replayId));
  const hasOffsetBarrier = Boolean(normalizeOffsetRange(input.requestedOffsetRange));

  if (consumer.ready === false) return failure("consumer", "consumer_not_ready", "The isolated Kafka consumer did not reach its ready barrier.");
  if (producer.sourceMatched === false) return failure("producer", "source_trace_missing", "The source AgentRun topic did not contain the requested trace lineage.");
  if (producer.invoked === false && counts.matched === 0) return failure("producer", "producer_not_invoked", "No isolated debug producer was invoked for this replay.");
  if (hasOffsetBarrier && counts.barrierScanned === 0) return failure("kafka", "topic_no_append_for_replay", "The requested replay offset range was not appended or observed.");
  if (counts.scanned === 0) return failure("producer", "producer_not_invoked", "The isolated debug topic contained no records for this replay.");
  if (counts.barrierScanned > 0 && counts.parsed === 0) return failure("server", "record_parse_rejected", "Kafka records were scanned but none decoded as JSON objects.");
  if (hasTraceFilter && counts.traceMatched === 0) {
    if (producer.sourceMatched === true && producer.invoked === true) return failure("kafka", "topic_no_append_for_replay", "The producer reported success but the debug topic did not expose the requested trace append.");
    if (producer.sourceMatched === true) return failure("producer", "producer_not_invoked", "The source trace exists but no isolated debug producer append was observed.");
    return failure("producer", "source_trace_missing", "Kafka records were scanned but none carried the requested trace lineage.");
  }
  if (hasReplayFilter && counts.replayMatched === 0) return failure("server", "records_scanned_but_filter_mismatch", "Kafka records were scanned but none matched the requested replayId.");
  if (counts.matched === 0) return failure("server", "records_scanned_but_filter_mismatch", "Kafka records were scanned but none matched the replay filters.");
  if (counts.delivered < counts.matched && input.completionReason !== "limit") return failure("server", "sse_write_failed", "Matched Kafka records were not all delivered over SSE.");
  if (counts.clientReceived !== null && counts.clientReceived < counts.delivered) return failure("client", "transport_failed", "The browser received fewer records than the server delivered.");
  if (counts.decoded !== null && counts.clientReceived !== null && counts.decoded < counts.clientReceived) return failure("client", "decoder_rejected", "The browser decoder rejected one or more delivered records.");
  if (counts.applied !== null && counts.decoded !== null && counts.applied < counts.decoded) return failure("client", "reducer_rejected", "The production reducer rejected one or more decoded records.");
  if (input.terminalObserved !== true) return failure("terminal", "terminal_missing", "Matching events were delivered but no terminal event was observed.");
  return { ok: true, phase: "terminal", code: "terminal_complete", message: "The isolated replay reached a terminal event." };
}

export function normalizeWorkbenchKafkaDebugOffsetRange(value) {
  return normalizeOffsetRange(value);
}

function failure(phase, code, message) {
  return { ok: false, phase, code, message };
}

function reject(tracker, reason) {
  tracker.rejectedByReason[reason] = (tracker.rejectedByReason[reason] ?? 0) + 1;
}

function safeSourceLineage(value) {
  const lineage = objectValue(value?.debugLineage);
  const sourceEvent = objectValue(value?.sourceEvent);
  if (Object.keys(lineage).length === 0) return null;
  return {
    inputSchema: textValue(lineage.inputSchema),
    inputEventType: textValue(lineage.inputEventType),
    inputTraceId: textValue(lineage.inputTraceId),
    inputSessionId: textValue(lineage.inputSessionId),
    inputHwlabSessionId: textValue(lineage.inputHwlabSessionId),
    inputRunId: textValue(lineage.inputRunId),
    inputCommandId: textValue(lineage.inputCommandId),
    sourceSeq: integerOrNull(lineage.sourceSeq),
    sourceTopic: textValue(sourceEvent.topic),
    sourcePartition: integerOrNull(sourceEvent.partition),
    sourceOffset: kafkaOffset(sourceEvent.offset),
    valuesPrinted: false
  };
}

function recordOffset(target, partitionValue, offsetValue) {
  const partition = integerOrNull(partitionValue);
  const offset = kafkaOffset(offsetValue);
  if (partition === null || offset === null) return;
  const current = target.get(partition);
  if (!current) {
    target.set(partition, { partition, firstOffset: offset, lastOffset: offset, count: 1 });
    return;
  }
  if (BigInt(offset) < BigInt(current.firstOffset)) current.firstOffset = offset;
  if (BigInt(offset) > BigInt(current.lastOffset)) current.lastOffset = offset;
  current.count += 1;
}

function summarizeOffsets(target) {
  return [...target.values()].sort((left, right) => left.partition - right.partition).map((entry) => ({ ...entry }));
}

function normalizeOffsetRange(value) {
  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
  const partition = integerOrNull(value.partition);
  const firstOffset = kafkaOffset(value.firstOffset ?? value.startOffset);
  const lastOffset = kafkaOffset(value.lastOffset ?? value.endOffset);
  if (partition === null || firstOffset === null || lastOffset === null || BigInt(firstOffset) > BigInt(lastOffset)) return null;
  return { partition, firstOffset, lastOffset };
}

function normalizeCounts(value = {}) {
  const nullable = (name) => nullableCount(value[name], null);
  return {
    scanned: count(value.scanned),
    barrierScanned: count(value.barrierScanned),
    parsed: count(value.parsed),
    traceMatched: count(value.traceMatched),
    replayMatched: count(value.replayMatched),
    matched: count(value.matched),
    delivered: count(value.delivered),
    clientReceived: nullable("clientReceived"),
    decoded: nullable("decoded"),
    applied: nullable("applied")
  };
}

function nullableCount(value, fallback) {
  const parsed = integerOrNull(value);
  return parsed === null || parsed < 0 ? fallback : parsed;
}

function count(value) {
  return nullableCount(value, 0);
}

function kafkaOffset(value) {
  const text = textValue(value);
  if (!text || !/^\d+$/u.test(text)) return null;
  try { return BigInt(text).toString(); } catch { return null; }
}

function integerOrNull(value) {
  if (value === null || value === undefined || (typeof value === "string" && value.trim() === "")) return null;
  const parsed = Number(value);
  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : null;
}

function booleanOrNull(value) {
  return value === true ? true : value === false ? false : null;
}

function objectValue(value) {
  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}

function textValue(value) {
  const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
  return text.length > 0 ? text : null;
}