/*
 * Durable AgentRun Kafka projector storage operations.
 * Kafka I/O is intentionally outside every transaction in this module.
 */
import { createHash } from "node:crypto";

import {
  indexWorkbenchAggregateEvents,
  inputFactWithAggregateSeq,
  nonNegativeInteger,
  normalizeWorkbenchAggregateEventsForFacts,
  normalizeWorkbenchFacts,
  parseJsonColumn,
  stableJson,
  textOr
} from "./runtime-store-core.ts";

export async function commitAgentRunKafkaProjection(store, params = {}) {
  const transport = normalizeTransport(params.transport);
  const canonicalEvent = objectValue(params.canonicalEvent);
  const projectedEvent = objectValue(params.projectedEvent);
  const sourceEventId = requiredText(canonicalEvent.eventId, "canonical AgentRun eventId");
  const traceId = requiredText(projectedEvent.traceId ?? canonicalEvent.traceId, "canonical AgentRun traceId");
  const sessionId = text(projectedEvent.sessionId ?? canonicalEvent.hwlabSessionId);
  const sourceSeq = nonNegativeInteger(canonicalEvent.sourceSeq ?? canonicalEvent.event?.seq);
  const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
  const now = store.now();
  const result = await store.withDurableTransaction("workbench.kafka-projection.commit", async (client) => {
    await lockKafkaInboxIdentity(client, transport, sourceEventId);
    await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${traceId}`]);
    const existing = await client.query(
      "SELECT source_topic, source_partition, source_offset, source_event_id, input_sha256, status, projected_seq FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR (source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1",
      [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]
    );
    const prior = existing.rows?.[0] ?? null;
    if (prior) {
      if (prior.input_sha256 !== inputSha256) {
        const diagnostic = { code: "workbench_kafka_inbox_hash_conflict", message: "Kafka transport or source event identity was reused with a different payload hash.", valuesRedacted: true };
        if (prior.status !== "projected") {
          await client.query(
            "UPDATE workbench_kafka_inbox SET status = 'failed', error_json = $1, updated_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5",
            [stableJson(diagnostic), now, prior.source_topic, prior.source_partition, prior.source_offset]
          );
        }
        await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(canonicalEvent), createdAt: now });
        return { conflict: true, diagnostic, duplicate: false, projectedSeq: nonNegativeInteger(prior.projected_seq) };
      }
      return { duplicate: true, conflict: false, status: prior.status, projectedSeq: nonNegativeInteger(prior.projected_seq) };
    }
    await client.query(
      "INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,'processing',NULL,$12,'{}',$13,$13,NULL)",
      [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, traceId, sessionId, text(canonicalEvent.runId), text(canonicalEvent.commandId), sourceSeq, stableJson(redactedEnvelope(canonicalEvent)), now]
    );
    const checkpointResult = await client.query("SELECT checkpoint_json FROM workbench_projection_checkpoints WHERE trace_id = $1 FOR UPDATE", [traceId]);
    const previousCheckpoint = parseJsonColumn(checkpointResult.rows?.[0]?.checkpoint_json, null);
    const sessionResult = sessionId
      ? await client.query("SELECT owner_user_id, project_id, conversation_id, thread_id, session_json FROM workbench_sessions WHERE session_id = $1 FOR UPDATE", [sessionId])
      : { rows: [] };
    const existingSession = sessionResult.rows?.[0] ?? null;
    const maxResult = await client.query(
      "SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq",
      [traceId]
    );
    const projectedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1;
    if (typeof params.factsFactory !== "function") throw codedError("workbench_kafka_facts_factory_required", "Kafka projection requires a pure facts factory.");
    const built = await params.factsFactory({ projectedSeq, previousCheckpoint, projectedAt: now });
    let facts = null;
    let persistedEvents = [];
    if (built?.written !== false) {
      facts = normalizeWorkbenchFacts(mergeKafkaSessionIdentity(built?.facts ?? {}, existingSession), params.requestMeta ?? {}, now);
      const persisted = await persistFacts(store, client, facts, params.requestMeta ?? {});
      persistedEvents = persisted.events;
    }
    const committedProjectedSeq = built?.written === false
      ? nonNegativeInteger(built?.projectedSeq) || nonNegativeInteger(previousCheckpoint?.projectedSeq) || projectedSeq
      : projectedSeq;
    const projectionWritten = built?.written !== false;
    if (projectionWritten) {
      const hwlabEvent = objectValue(params.hwlabEvent);
      const hwlabEventId = requiredText(hwlabEvent.eventId, "HWLAB Kafka eventId");
      await client.query(
        "INSERT INTO hwlab_kafka_outbox (event_id, topic, partition_key, payload_json, headers_json, attempt_count, next_attempt_at, lease_owner, lease_expires_at, published_at, last_error, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,0,$6,NULL,NULL,NULL,NULL,$6,$6) ON CONFLICT (event_id) DO NOTHING",
        [hwlabEventId, requiredText(params.hwlabTopic, "HWLAB Kafka topic"), requiredText(params.partitionKey, "HWLAB Kafka partition key"), stableJson(hwlabEvent), stableJson(objectValue(params.headers)), now]
      );
    }
    await client.query(
      "UPDATE workbench_kafka_inbox SET status = 'projected', projected_seq = $1, updated_at = $2, projected_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5",
      [committedProjectedSeq, now, transport.sourceTopic, transport.sourcePartition, transport.sourceOffset]
    );
    return { duplicate: false, conflict: false, projectedSeq: committedProjectedSeq, facts, events: persistedEvents, projectionWritten, suppressedAfterSeal: built?.suppressedAfterSeal === true };
  });
  if (result.facts) store.memory.writeWorkbenchFacts({ facts: result.facts }, params.requestMeta ?? {});
  return { ...result, transport, sourceEventId, traceId, sessionId, sourceSeq, valuesPrinted: false };
}

export async function recordFailedAgentRunKafkaMessage(store, params = {}) {
  const transport = normalizeTransport(params.transport);
  const now = store.now();
  const error = redactedError(params.error);
  const sourceEventId = text(params.sourceEventId);
  const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
  const result = await store.withDurableTransaction("workbench.kafka-projection.dlq", async (client) => {
    await lockKafkaInboxIdentity(client, transport, sourceEventId);
    const existing = await client.query("SELECT source_topic, source_partition, source_offset, input_sha256, status, projected_seq FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR ($4::text IS NOT NULL AND source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1 FOR UPDATE", [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]);
    const prior = existing.rows?.[0] ?? null;
    if (prior) {
      const sameTransport = prior.source_partition === transport.sourcePartition && String(prior.source_offset) === transport.sourceOffset;
      const diagnostic = prior.input_sha256 !== inputSha256
        ? { ...error, code: "workbench_kafka_inbox_hash_conflict" }
        : sameTransport ? error : { ...error, code: "workbench_kafka_inbox_source_event_duplicate" };
      if (prior.status !== "projected") {
        await client.query("UPDATE workbench_kafka_inbox SET status = 'failed', error_json = $1, updated_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5", [stableJson(diagnostic), now, prior.source_topic, prior.source_partition, prior.source_offset]);
      }
      await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(params.envelope), createdAt: now });
      return { duplicate: sameTransport, conflict: !sameTransport || prior.input_sha256 !== inputSha256, priorStatus: prior.status, projectedSeq: nonNegativeInteger(prior.projected_seq), diagnostic };
    }
    await client.query(
      "INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,NULL,NULL,NULL,NULL,0,'failed',NULL,$7,$8,$9,$9,NULL) ON CONFLICT (source_topic, source_partition, source_offset) DO UPDATE SET status = 'failed', error_json = EXCLUDED.error_json, updated_at = EXCLUDED.updated_at",
      [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, stableJson(redactedEnvelope(params.envelope)), stableJson(error), now]
    );
    await insertDlq(client, { ...transport, sourceEventId, inputSha256, error, envelope: redactedEnvelope(params.envelope), createdAt: now });
    return { duplicate: false, conflict: false, priorStatus: null, diagnostic: error };
  });
  return { recorded: true, transport, sourceEventId, error: result.diagnostic, ...result, valuesPrinted: false };
}

export async function recordIgnoredAgentRunKafkaMessage(store, params = {}) {
  const transport = normalizeTransport(params.transport);
  const envelope = objectValue(params.envelope);
  const sourceEventId = requiredText(envelope.eventId, "canonical AgentRun eventId");
  const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
  const now = store.now();
  return store.withDurableTransaction("workbench.kafka-projection.ignore", async (client) => {
    await lockKafkaInboxIdentity(client, transport, sourceEventId);
    const existing = await client.query("SELECT input_sha256, status FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR (source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1", [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]);
    const prior = existing.rows?.[0] ?? null;
    if (prior) {
      if (prior.input_sha256 !== inputSha256) {
        const diagnostic = { code: "workbench_kafka_inbox_hash_conflict", message: "Ignored Kafka identity was reused with a different payload hash.", valuesRedacted: true };
        await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(envelope), createdAt: now });
        return { ignored: true, duplicate: false, conflict: true, status: prior.status, diagnostic, valuesPrinted: false };
      }
      return { ignored: true, duplicate: true, status: prior.status, valuesPrinted: false };
    }
    await client.query(
      "INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,NULL,NULL,$7,$8,$9,'ignored',NULL,$10,'{}',$11,$11,NULL)",
      [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, text(envelope.runId), text(envelope.commandId), nonNegativeInteger(envelope.sourceSeq), stableJson(redactedEnvelope(envelope)), now]
    );
    return { ignored: true, duplicate: false, status: "ignored", valuesPrinted: false };
  });
}

export async function claimHwlabKafkaOutbox(store, { owner, leaseMs, limit } = {}) {
  const leaseOwner = requiredText(owner, "HWLAB Kafka relay owner");
  const boundedLimit = boundedInteger(limit, 100, 1, 500);
  const now = store.now();
  const leaseExpiresAt = new Date(Date.parse(now) + boundedInteger(leaseMs, 60_000, 1_000, 3_600_000)).toISOString();
  const result = await store.withDurableTransaction("hwlab.kafka-outbox.claim", (client) => client.query(
    `WITH candidates AS (
       SELECT candidate.outbox_seq FROM hwlab_kafka_outbox candidate
       WHERE candidate.published_at IS NULL
         AND candidate.next_attempt_at <= $1
         AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at <= $1)
         AND NOT EXISTS (
           SELECT 1 FROM hwlab_kafka_outbox earlier
           WHERE earlier.published_at IS NULL
             AND earlier.partition_key = candidate.partition_key
             AND earlier.outbox_seq < candidate.outbox_seq
         )
       ORDER BY candidate.outbox_seq ASC
       FOR UPDATE SKIP LOCKED
       LIMIT $2
     )
     UPDATE hwlab_kafka_outbox o SET attempt_count = o.attempt_count + 1, lease_owner = $3, lease_expires_at = $4, updated_at = $1
     FROM candidates WHERE o.outbox_seq = candidates.outbox_seq
     RETURNING o.*`,
    [now, boundedLimit, leaseOwner, leaseExpiresAt]
  ));
  return (result.rows ?? []).map(outboxRow);
}

export async function completeHwlabKafkaOutbox(store, item = {}) {
  const result = await store.query(
    "UPDATE hwlab_kafka_outbox SET published_at = $1, lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, updated_at = $1 WHERE outbox_seq = $2 AND lease_owner = $3 AND attempt_count = $4 AND lease_expires_at = $5 AND lease_expires_at > $1 AND published_at IS NULL RETURNING outbox_seq",
    [store.now(), item.outboxSeq, item.leaseOwner, item.attemptCount, item.leaseExpiresAt]
  );
  if (!result.rows?.[0]) throw codedError("hwlab_kafka_outbox_lease_conflict", "HWLAB Kafka outbox claim is stale.");
  return { completed: true, outboxSeq: Number(result.rows[0].outbox_seq), valuesPrinted: false };
}

export async function retryHwlabKafkaOutbox(store, item = {}, error = null, retryAt = null) {
  const now = store.now();
  const nextAttemptAt = text(retryAt) ?? now;
  const result = await store.query(
    "UPDATE hwlab_kafka_outbox SET next_attempt_at = $1, lease_owner = NULL, lease_expires_at = NULL, last_error = $2, updated_at = $3 WHERE outbox_seq = $4 AND lease_owner = $5 AND attempt_count = $6 AND lease_expires_at = $7 AND lease_expires_at > $3 AND published_at IS NULL RETURNING outbox_seq",
    [nextAttemptAt, stableJson(redactedError(error)), now, item.outboxSeq, item.leaseOwner, item.attemptCount, item.leaseExpiresAt]
  );
  if (!result.rows?.[0]) throw codedError("hwlab_kafka_outbox_lease_conflict", "HWLAB Kafka outbox claim is stale.");
  return { retried: true, outboxSeq: Number(result.rows[0].outbox_seq), nextAttemptAt, valuesPrinted: false };
}

export async function hwlabKafkaProjectorStatus(store) {
  const result = await store.query(
    `SELECT
       COUNT(*) FILTER (WHERE published_at IS NULL)::int AS backlog_count,
       COUNT(*) FILTER (WHERE published_at IS NULL AND attempt_count > 0)::int AS retry_count,
       MIN(created_at) FILTER (WHERE published_at IS NULL) AS oldest_pending_at,
       (SELECT COUNT(*)::int FROM workbench_kafka_inbox WHERE status = 'failed') AS failed_inbox_count,
       (SELECT COUNT(*)::int FROM workbench_kafka_dlq) AS dlq_count
     FROM hwlab_kafka_outbox`, []
  );
  const row = result.rows?.[0] ?? {};
  return { backlogCount: Number(row.backlog_count ?? 0), retryCount: Number(row.retry_count ?? 0), oldestPendingAt: row.oldest_pending_at ?? null, failedInboxCount: Number(row.failed_inbox_count ?? 0), dlqCount: Number(row.dlq_count ?? 0), valuesPrinted: false };
}

export async function readAtomicWorkbenchProjectionSync(store, params = {}) {
  const since = nonNegativeInteger(params.afterOutboxSeq ?? params.afterSeq ?? params.since);
  const limit = boundedInteger(params.limit, 100, 1, 500);
  const traceId = text(params.traceId);
  const sessionId = text(params.sessionId);
  const snapshotOnly = params.snapshotOnly === true;
  const deltaOnly = params.deltaOnly === true;
  if (!traceId && !sessionId) throw codedError("workbench_sync_scope_required", "Atomic Workbench sync requires sessionId or traceId.");
  if (snapshotOnly && deltaOnly) throw codedError("workbench_sync_mode_invalid", "snapshotOnly and deltaOnly are mutually exclusive.");
  return store.withDurableTransaction("workbench.projection-sync.read", async (client) => {
    await client.query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
    const filter = traceId ? { column: "trace_id", value: traceId } : { column: "session_id", value: sessionId };
    const cutoffResult = await client.query(`SELECT COALESCE(MAX(outbox_seq), 0)::bigint AS cutoff FROM workbench_projection_outbox WHERE ${filter.column} = $1`, [filter.value]);
    const cutoff = Number(cutoffResult.rows?.[0]?.cutoff ?? 0);
    const outboxResult = snapshotOnly ? { rows: [] } : await client.query(
      `SELECT outbox_seq, outbox_event_id, entity_family, entity_id, event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE outbox_seq > $1 AND outbox_seq <= $2 AND ${filter.column} = $3 ORDER BY outbox_seq ASC LIMIT $4`,
      [since, cutoff, filter.value, limit + 1]
    );
    const allRows = (outboxResult.rows ?? []).map(projectionOutboxRow);
    const events = allRows.slice(0, limit);
    const facts = deltaOnly ? {} : await readSyncFacts(client, { traceId, sessionId, limit: 500, families: snapshotOnly ? ["sessions", "messages", "turns"] : null });
    const hasMore = allRows.length > limit;
    return { facts, events, cutoffOutboxSeq: cutoff, cursorOutboxSeq: snapshotOnly ? cutoff : hasMore ? (events.at(-1)?.outboxSeq ?? since) : cutoff, hasMore, valuesPrinted: false };
  });
}

async function persistFacts(store, client, facts, requestMeta) {
  const aggregateEvents = normalizeWorkbenchAggregateEventsForFacts(facts, requestMeta, store.now());
  const events = [];
  for (const event of aggregateEvents) events.push(await store.persistWorkbenchAggregateEvent(event, client));
  const eventIndex = indexWorkbenchAggregateEvents(events);
  const inputFacts = facts.inputs.map((fact) => inputFactWithAggregateSeq(fact, eventIndex));
  for (const fact of inputFacts) await store.persistWorkbenchSessionInputFact(fact, client);
  for (const fact of facts.sessions) await store.persistWorkbenchSessionFact(fact, client);
  for (const fact of facts.messages) await store.persistWorkbenchMessageFact(fact, client);
  for (const fact of facts.parts) await store.persistWorkbenchPartFact(fact, client);
  for (const fact of facts.turns) await store.persistWorkbenchTurnFact(fact, client);
  for (const fact of facts.traceEvents) await store.persistWorkbenchTraceEventFact(fact, client);
  for (const fact of facts.checkpoints) await store.persistWorkbenchProjectionCheckpoint(fact, client);
  for (const fact of facts.messages) {
    const event = eventIndex.get(`message:${fact.messageId}`) ?? eventIndex.get(`messages:${fact.messageId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
    await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "messages", "message", { role: fact.role, status: fact.status }), client);
  }
  for (const fact of facts.traceEvents) {
    const event = eventIndex.get(`traceEvent:${fact.id}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
    await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "traceEvents", fact.terminal ? "terminal" : "event", { type: fact.eventType }), client);
  }
  for (const fact of facts.turns.filter((item) => item.terminal || item.sealed)) {
    const event = eventIndex.get(`turn:${fact.turnId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
    await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "turns", "terminal", { status: fact.status, failureKind: fact.failureKind }), client);
  }
  return { events, inputFacts };
}

function outboxRecord(fact, event, family, commitType, metadata) {
  const entityId = text(family === "messages" ? fact.messageId : family === "turns" ? fact.turnId : fact.id) ?? text(fact.traceId) ?? "projection";
  return {
    outboxEventId: projectionOutboxEventId({ fact, event, family, entityId, commitType }),
    entityFamily: family,
    entityId,
    eventSeq: event?.eventSeq ?? null,
    aggregateId: event?.aggregateId ?? null,
    aggregateSeq: event?.aggregateSeq ?? 0,
    projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
    traceId: fact.traceId,
    sessionId: fact.sessionId,
    turnId: fact.turnId,
    messageId: fact.messageId,
    projectedSeq: fact.projectedSeq,
    sourceSeq: fact.sourceSeq,
    sourceEventId: fact.sourceEventId,
    commitType,
    terminal: Boolean(fact.terminal),
    sealed: Boolean(fact.sealed),
    payload: {
      family,
      fact,
      metadata: { ...metadata, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true },
      valuesRedacted: true
    },
    createdAt: fact.occurredAt ?? fact.updatedAt ?? fact.createdAt
  };
}

export function projectionOutboxEventId({ fact = {}, event = null, family, entityId, commitType }) {
  const sourceEventId = text(fact.sourceEventId);
  if (sourceEventId) return `${sourceEventId}:${family}:${entityId}:${commitType}`;
  const identity = stableJson({
    family,
    entityId,
    commitType,
    eventSeq: event?.eventSeq ?? null,
    aggregateId: event?.aggregateId ?? null,
    aggregateSeq: event?.aggregateSeq ?? 0,
    projectionRevision: event?.projectionRevision ?? fact.projectedSeq ?? 0,
    projectedSeq: fact.projectedSeq ?? 0,
    sourceSeq: fact.sourceSeq ?? 0
  });
  return `projection:${createHash("sha256").update(identity).digest("hex")}`;
}

function mergeKafkaSessionIdentity(facts, row) {
  if (!row) return facts;
  const existing = parseJsonColumn(row.session_json, {});
  const sessions = Array.isArray(facts?.sessions) ? facts.sessions.map((fact) => ({
    ...existing,
    ...fact,
    ownerUserId: fact.ownerUserId ?? row.owner_user_id ?? existing.ownerUserId ?? null,
    projectId: fact.projectId ?? row.project_id ?? existing.projectId ?? null,
    conversationId: fact.conversationId ?? row.conversation_id ?? existing.conversationId ?? null,
    threadId: fact.threadId ?? row.thread_id ?? existing.threadId ?? null,
    ownerRole: fact.ownerRole ?? existing.ownerRole ?? null
  })) : [];
  return { ...facts, sessions };
}

async function readSyncFacts(client, { traceId, sessionId, limit, families = null }) {
  const familySet = Array.isArray(families) ? new Set(families) : null;
  const specs = [
    ["sessions", "workbench_sessions", "session_json", traceId ? "last_trace_id" : "session_id"],
    ["messages", "workbench_messages", "message_json", traceId ? "trace_id" : "session_id"],
    ["parts", "workbench_parts", "part_json", traceId ? "trace_id" : "session_id"],
    ["turns", "workbench_turns", "turn_json", traceId ? "trace_id" : "session_id"],
    ["traceEvents", "workbench_trace_events", "event_json", traceId ? "trace_id" : "session_id"],
    ["checkpoints", "workbench_projection_checkpoints", "checkpoint_json", traceId ? "trace_id" : "session_id"]
  ].filter(([family]) => !familySet || familySet.has(family));
  const value = traceId ?? sessionId;
  const facts = {};
  for (const [family, table, jsonColumn, column] of specs) {
    const result = await client.query(`SELECT ${jsonColumn} FROM ${table} WHERE ${column} = $1 ORDER BY updated_at ASC LIMIT $2`, [value, limit]);
    facts[family] = (result.rows ?? []).map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
  }
  return facts;
}

async function insertDlq(client, record) {
  await client.query(
    "INSERT INTO workbench_kafka_dlq (source_topic, source_partition, source_offset, source_event_id, input_sha256, error_code, error_message, envelope_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT (source_topic, source_partition, source_offset) DO UPDATE SET error_code = EXCLUDED.error_code, error_message = EXCLUDED.error_message, envelope_json = EXCLUDED.envelope_json",
    [record.sourceTopic, record.sourcePartition, record.sourceOffset, record.sourceEventId, record.inputSha256, record.error.code, record.error.message, stableJson(record.envelope), record.createdAt]
  );
}

async function lockKafkaInboxIdentity(client, transport, sourceEventId) {
  const transportIdentity = `${transport.sourceTopic}:transport:${transport.sourcePartition}:${transport.sourceOffset}`;
  await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${transportIdentity}`]);
  if (sourceEventId) {
    const eventIdentity = `${transport.sourceTopic}:event:${sourceEventId}`;
    await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${eventIdentity}`]);
  }
}

function normalizeTransport(value = {}) {
  const input = objectValue(value);
  const sourceTopic = requiredText(input.sourceTopic ?? input.topic, "Kafka source topic");
  const sourcePartition = Number(input.sourcePartition ?? input.partition);
  const sourceOffset = text(input.sourceOffset ?? input.offset);
  if (!Number.isInteger(sourcePartition) || sourcePartition < 0) throw codedError("workbench_kafka_partition_invalid", "Kafka source partition must be a non-negative integer.");
  if (!/^\d+$/u.test(sourceOffset ?? "")) throw codedError("workbench_kafka_offset_invalid", "Kafka source offset must be a non-negative integer string.");
  return { sourceTopic, sourcePartition, sourceOffset, inputSha256: text(input.inputSha256), sourceKey: text(input.sourceKey), valuesPrinted: false };
}

function projectionOutboxRow(row) {
  return { outboxSeq: Number(row.outbox_seq), outboxEventId: row.outbox_event_id, entityFamily: row.entity_family, entityId: row.entity_id, eventSeq: row.event_seq == null ? null : Number(row.event_seq), aggregateId: row.aggregate_id, aggregateSeq: Number(row.aggregate_seq ?? 0), projectionRevision: Number(row.projection_revision ?? 0), traceId: row.trace_id, sessionId: row.session_id, turnId: row.turn_id, messageId: row.message_id, projectedSeq: Number(row.projected_seq ?? 0), sourceSeq: Number(row.source_seq ?? 0), sourceEventId: row.source_event_id, commitType: row.commit_type, terminal: Boolean(row.terminal), sealed: Boolean(row.sealed), payload: parseJsonColumn(row.payload_json, {}), createdAt: row.created_at, valuesPrinted: false };
}

function outboxRow(row) {
  return { outboxSeq: Number(row.outbox_seq), eventId: row.event_id, topic: row.topic, partitionKey: row.partition_key, payload: parseJsonColumn(row.payload_json, {}), headers: parseJsonColumn(row.headers_json, {}), attemptCount: Number(row.attempt_count ?? 0), nextAttemptAt: row.next_attempt_at, leaseOwner: row.lease_owner, leaseExpiresAt: row.lease_expires_at, createdAt: row.created_at, valuesPrinted: false };
}

function redactedEnvelope(value) {
  const input = objectValue(value);
  return { schema: text(input.schema), eventType: text(input.eventType), eventId: text(input.eventId), outboxSeq: nonNegativeInteger(input.outboxSeq), sourceSeq: nonNegativeInteger(input.sourceSeq), traceId: text(input.traceId), sessionId: text(input.sessionId), hwlabSessionId: text(input.hwlabSessionId), runId: text(input.runId), commandId: text(input.commandId), valuesRedacted: true };
}

function redactedError(error) {
  return { code: text(error?.code) ?? "workbench_kafka_message_invalid", message: String(error?.message ?? error ?? "Kafka message is invalid.").slice(0, 500), valuesRedacted: true };
}

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

function text(value) {
  const normalized = textOr(value, "");
  return normalized || null;
}

function requiredText(value, label) {
  const normalized = text(value);
  if (!normalized) throw codedError("workbench_kafka_contract_invalid", `${label} is required.`);
  return normalized;
}

function boundedInteger(value, fallback, min, max) {
  const parsed = Number(value);
  return Number.isInteger(parsed) && parsed >= min && parsed <= max ? parsed : fallback;
}

function codedError(code, message) {
  const error = new Error(message);
  error.code = code;
  return error;
}