// nano-workforce — the transcript READ projection (ADR 0056, H3 / #146, read path #222).
//
// The write path (relay.family.ts) flushes an ephemeral agent's PTY stream to a durable transcript on
// job completion; this module is the READ counterpart the advisory `GET /agentic/transcripts*`
// endpoints share. It projects a {@link TranscriptStore} row (+ its retained chunks) onto the wire
// shape and enriches it with the H6 correlation (`app/agentic/correlation.ts`) so a captured session
// lines up with "that process instance / this plan" — even after the ephemeral agent has exited.
//
// Correlation is BEST-EFFORT and advisory: the correlation registry is in-memory and only holds
// currently-linked jobs, so a completed session's process-instance / plan context is present only
// while the job is still live. The jobKey itself is always recoverable — it is encoded in the stream
// id (`composeStreamId(instance, jobKey)`, decoded with `parseStreamId`), so a past session is never
// anonymous even once its correlation has been released.
//
// Pure and side-effect-free apart from reading the store: no I/O beyond the injected store, so it is
// unit-testable on the injected env (Node, no browser), and never touches the engine or a BPMN flow.

import { parseStreamId } from "@nanobpm/agentic/emit";
import type { TranscriptChunk, TranscriptRing, TranscriptStore, TranscriptStream } from "@nanobpm/agentic/transcript";
import type { AgenticTranscript, AgenticTranscriptData, ErrorBody } from "../../nano-generated/api-io.d.ts";
import type { CorrelationRegistry } from "./correlation.ts";
import { jobKeyOfJobStream } from "./correlation.ts";
import type { AgenticCorrelationStore } from "./correlation-store.ts";
import type { RelayTranscriptService } from "./families/relay.family.ts";
import { utf8ByteLength } from "./transcript-events.ts";

/** Total captured bytes across a set of retained chunks (UTF-8, the on-the-wire terminal encoding). */
export function byteLengthOf(chunks: readonly TranscriptChunk[]): number {
  let total = 0;
  for (const c of chunks) total += utf8ByteLength(c.chunk);
  return total;
}

/** The correlation fields (jobKey + engine context + worker attribution) a stream id resolves to, best-effort. */
interface CorrelationFields {
  jobKey?: string;
  processInstanceKey?: string;
  bpmnProcessId?: string;
  elementId?: string;
  /** The engine element-instance key the job's token occupied (#544) — per-occupancy, unlike elementId. */
  elementInstanceKey?: string;
  planKey?: string;
  /** The worker instance that ran the job (durable — survives release / restart). */
  instance?: string;
  /** The worker's durable identity, when recorded. */
  identity?: string;
  /** The worker's host, when recorded. */
  host?: string;
}

/**
 * Resolve a stream id to its correlation fields. The jobKey is decoded from the instance-scoped
 * stream id (`composeStreamId(instance, jobKey)`) via {@link parseStreamId}, falling back to the bare
 * `job:<jobKey>` Stage-0 alias ({@link jobKeyOfJobStream}) so a session addressed by either scheme stays
 * attributable. Engine context (process instance / plan) + worker attribution (instance / identity /
 * host) come from the LIVE registry while the job is still linked, and fall back to the DURABLE store
 * (`AgenticCorrelationStore`) once the job has completed and its live correlation was released — so a
 * PAST session stays attributable to its worker after the worker exits or the process restarts.
 * Non-job streams yield an empty object.
 */
export function correlationFieldsFor(
  stream: string,
  correlation: CorrelationRegistry | undefined,
  durable?: AgenticCorrelationStore | undefined,
): CorrelationFields {
  const jobKey = parseStreamId(stream)?.stream ?? jobKeyOfJobStream(stream);
  if (jobKey === undefined) return {};
  const fields: CorrelationFields = { jobKey };
  const context = correlation?.resolve(jobKey);
  if (context) {
    if (context.processInstanceKey !== undefined) fields.processInstanceKey = context.processInstanceKey;
    if (context.bpmnProcessId !== undefined) fields.bpmnProcessId = context.bpmnProcessId;
    if (context.elementId !== undefined) fields.elementId = context.elementId;
    if (context.elementInstanceKey !== undefined) fields.elementInstanceKey = context.elementInstanceKey;
    if (context.planKey !== undefined) fields.planKey = context.planKey;
  }
  // Durable fallback: fill any field the live registry did not supply (a released past session, or a
  // worker-attribution field the in-memory registry never carried). Live values take precedence.
  const row = durable?.get(jobKey);
  if (row) {
    if (fields.processInstanceKey === undefined && row.processInstanceKey !== undefined) {
      fields.processInstanceKey = row.processInstanceKey;
    }
    if (fields.bpmnProcessId === undefined && row.bpmnProcessId !== undefined) fields.bpmnProcessId = row.bpmnProcessId;
    if (fields.elementId === undefined && row.elementId !== undefined) fields.elementId = row.elementId;
    if (fields.elementInstanceKey === undefined && row.elementInstanceKey !== undefined) {
      fields.elementInstanceKey = row.elementInstanceKey;
    }
    if (fields.planKey === undefined && row.planKey !== undefined) fields.planKey = row.planKey;
    if (row.instance !== undefined) fields.instance = row.instance;
    if (row.identity !== undefined) fields.identity = row.identity;
    if (row.host !== undefined) fields.host = row.host;
  }
  return fields;
}

/** Project a stored transcript's metadata (+ its retained chunks) onto the list wire shape. */
export function toTranscript(
  meta: TranscriptStream,
  store: TranscriptStore,
  correlation: CorrelationRegistry | undefined,
  durable?: AgenticCorrelationStore | undefined,
): AgenticTranscript {
  const chunks = store.read(meta.stream);
  const out: AgenticTranscript = {
    stream: meta.stream,
    lifecycle: meta.lifecycle,
    status: meta.status,
    createdAt: meta.createdAt,
    nextOffset: meta.nextOffset,
    byteLength: byteLengthOf(chunks),
    chunkCount: chunks.length,
  };
  if (meta.completedAt !== undefined) out.completedAt = meta.completedAt;
  if (meta.firstOffset !== undefined) out.firstOffset = meta.firstOffset;
  const fields = correlationFieldsFor(meta.stream, correlation, durable);
  if (fields.jobKey !== undefined) out.jobKey = fields.jobKey;
  if (fields.processInstanceKey !== undefined) out.processInstanceKey = fields.processInstanceKey;
  if (fields.bpmnProcessId !== undefined) out.bpmnProcessId = fields.bpmnProcessId;
  if (fields.elementId !== undefined) out.elementId = fields.elementId;
  if (fields.elementInstanceKey !== undefined) out.elementInstanceKey = fields.elementInstanceKey;
  if (fields.planKey !== undefined) out.planKey = fields.planKey;
  if (fields.instance !== undefined) out.instance = fields.instance;
  if (fields.identity !== undefined) out.identity = fields.identity;
  if (fields.host !== undefined) out.host = fields.host;
  return out;
}

/** The filters {@link listTranscripts} understands (all optional; an empty filter returns everything). */
export interface TranscriptFilter {
  readonly jobKey?: string;
  readonly processInstanceKey?: string;
  /** The engine element-instance key (#544) — resolves a session to one occupancy of a looping activity. */
  readonly elementInstanceKey?: string;
  readonly planKey?: string;
  /** The worker instance that ran the session (durable attribution) — powers the worker-history view. */
  readonly instance?: string;
  /** ISO-8601 lower bound (inclusive) on the session's createdAt. */
  readonly since?: string;
  /** ISO-8601 upper bound (inclusive) on the session's createdAt. */
  readonly until?: string;
}

/**
 * List every captured session projected to the wire shape, sorted newest-first by createdAt (then by
 * stream for a stable tie-break), after applying the (advisory) filters. jobKey / process-instance /
 * plan / instance filters match the correlation-enriched fields; since/until bound createdAt.
 */
export function listTranscripts(
  store: TranscriptStore,
  correlation: CorrelationRegistry | undefined,
  filter: TranscriptFilter = {},
  durable?: AgenticCorrelationStore | undefined,
): AgenticTranscript[] {
  const sinceMs = filter.since !== undefined ? Date.parse(filter.since) : undefined;
  const untilMs = filter.until !== undefined ? Date.parse(filter.until) : undefined;
  const rows = store
    .list()
    .map((meta) => toTranscript(meta, store, correlation, durable))
    .filter((t) => {
      if (filter.jobKey !== undefined && t.jobKey !== filter.jobKey) return false;
      if (filter.processInstanceKey !== undefined && t.processInstanceKey !== filter.processInstanceKey) return false;
      if (filter.elementInstanceKey !== undefined && t.elementInstanceKey !== filter.elementInstanceKey) return false;
      if (filter.planKey !== undefined && t.planKey !== filter.planKey) return false;
      if (filter.instance !== undefined && t.instance !== filter.instance) return false;
      const createdMs = Date.parse(t.createdAt);
      if (sinceMs !== undefined && Number.isFinite(createdMs) && createdMs < sinceMs) return false;
      if (untilMs !== undefined && Number.isFinite(createdMs) && createdMs > untilMs) return false;
      return true;
    });
  // Newest session first (a "past sessions" feed reads best most-recent-first); stable on stream id.
  rows.sort((a, b) => {
    const byTime = b.createdAt.localeCompare(a.createdAt);
    return byTime !== 0 ? byTime : a.stream.localeCompare(b.stream);
  });
  return rows;
}

/**
 * Fetch a stored transcript's bytes from offset `from` (inclusive), projected onto the range/offset
 * wire shape — the SAME resume-from-offset contract the live terminal renders, so the cockpit replays
 * a closed stream through its existing renderer.
 *
 * Reads the durable {@link TranscriptStore} first; when the store has no row for the stream (it was
 * never persisted, or — the #486 caveat — the job completed on a still-live multiplexing worker whose
 * ring has not been flushed yet) it falls back to the injected live ring so a freshly-emitted
 * `transcriptUrl` is readable the moment it is emitted. Returns undefined only when neither the store
 * nor a live ring has the stream.
 */
export function readTranscriptFrom(
  stream: string,
  from: number,
  store: TranscriptStore | undefined,
  correlation: CorrelationRegistry | undefined,
  durable?: AgenticCorrelationStore | undefined,
  live?: { ring: TranscriptRing; createdAt: string } | undefined,
): AgenticTranscriptData | undefined {
  const meta = store?.get(stream);
  let out: AgenticTranscriptData;
  if (store !== undefined && meta !== undefined) {
    const slice = store.since(stream, from);
    const entries = slice.entries.map((c) => ({ offset: c.offset, chunk: c.chunk }));
    out = {
      stream: meta.stream,
      lifecycle: meta.lifecycle,
      status: meta.status,
      createdAt: meta.createdAt,
      nextOffset: slice.nextOffset,
      byteLength: byteLengthOf(slice.entries),
      chunkCount: entries.length,
      from,
      gap: slice.gap,
      entries,
    };
    if (meta.completedAt !== undefined) out.completedAt = meta.completedAt;
  } else if (live !== undefined) {
    // #486: no durable row yet — the job completed on a still-live multiplexing worker whose ring has
    // not been flushed. Serve the captured bytes straight from the live ring so a freshly-emitted
    // `transcriptUrl` is readable immediately (an open ephemeral stream), rather than 404-ing until the
    // worker disconnects. `gap` is derived structurally: the first returned entry sitting past `from`
    // means retention already evicted the requested prefix.
    const slice = live.ring.since(from);
    const entries = slice.entries.map((c) => ({ offset: c.offset, chunk: c.chunk }));
    out = {
      stream,
      lifecycle: "ephemeral",
      status: "open",
      createdAt: live.createdAt,
      nextOffset: live.ring.nextOffset,
      byteLength: byteLengthOf(slice.entries),
      chunkCount: entries.length,
      from,
      gap: entries.length > 0 && entries[0].offset > from,
      entries,
    };
  } else {
    return undefined;
  }
  const fields = correlationFieldsFor(out.stream, correlation, durable);
  if (fields.jobKey !== undefined) out.jobKey = fields.jobKey;
  if (fields.processInstanceKey !== undefined) out.processInstanceKey = fields.processInstanceKey;
  if (fields.bpmnProcessId !== undefined) out.bpmnProcessId = fields.bpmnProcessId;
  if (fields.elementId !== undefined) out.elementId = fields.elementId;
  if (fields.elementInstanceKey !== undefined) out.elementInstanceKey = fields.elementInstanceKey;
  if (fields.planKey !== undefined) out.planKey = fields.planKey;
  if (fields.instance !== undefined) out.instance = fields.instance;
  if (fields.identity !== undefined) out.identity = fields.identity;
  if (fields.host !== undefined) out.host = fields.host;
  return out;
}

/** The single-stream read result both transcript READ routes share (#744): the bytes on 200, the
 * same 400 (malformed `from`) / 404 (no such stream, or no service mounted) outcomes on failure. */
export type SingleTranscriptResult =
  | { status: 200; body: AgenticTranscriptData }
  | { status: 400; body: ErrorBody }
  | { status: 404; body: ErrorBody };

/**
 * Resolve a requested stream id to the instance-scoped id the transcript is actually STORED under.
 *
 * The data plane keys every job transcript by `composeStreamId(instance, jobKey)` (issue #738), but the
 * Explorer Stage-0 `transcriptUrl` (`app/agentic/transcript-url.ts`, #543) still addresses a job by the
 * bare `job:<jobKey>` alias — the completing worker cannot know its own instance at output-mapping time.
 * So a Stage-0 read arrives keyed by `job:<jobKey>` and would MISS the store (and 404) unless it is first
 * mapped back to the instance-scoped id. The jobKey → instance-scoped stream is recovered from the LIVE
 * correlation registry while the job is still linked, else from the DURABLE store (which persists the
 * instance-scoped `stream` keyed by jobKey at completion). An already-instance-scoped id, a non-job
 * stream, or an unknown jobKey is returned unchanged (the caller then reads it as-is, 404-ing if absent).
 */
export function canonicalStreamFor(
  stream: string,
  correlation: CorrelationRegistry | undefined,
  durable: AgenticCorrelationStore | undefined,
): string {
  // Already instance-scoped (or a non-alias id) — nothing to resolve.
  if (parseStreamId(stream) !== undefined) return stream;
  const jobKey = jobKeyOfJobStream(stream);
  if (jobKey === undefined) return stream;
  return correlation?.resolve(jobKey)?.stream ?? durable?.get(jobKey)?.stream ?? stream;
}

/**
 * The ONE canonical single-stream transcript read, shared by BOTH routes that serve it (#744):
 * `GET /agentic/transcripts?stream=<id>&from=<n>` (the proxy-safe QUERY form the cockpit clients
 * build — a gateway that peels one percent-encoding layer before routing splits an encoded slash
 * in a PATH segment into an extra segment and 404s, while a query value survives intact) and
 * `GET /agentic/transcripts/{stream}` (the legacy path form the worker-emitted `transcriptUrl`
 * resolves — safe there because `job:<jobKey>` ids structurally never contain a slash). One
 * implementation so the two addressings can never answer differently for the same stream/from.
 *
 * A Stage-0 `job:<jobKey>` alias is first resolved to the instance-scoped id the bytes are stored under
 * ({@link canonicalStreamFor}), so both the durable store read AND the live-ring fallback address the
 * stream the producer actually wrote (issue #738) rather than 404-ing the alias.
 */
export function readSingleTranscript(
  stream: string,
  from: number | undefined,
  service: RelayTranscriptService | undefined,
  correlation: CorrelationRegistry | undefined,
): SingleTranscriptResult {
  const offset = from ?? 0;
  if (!Number.isSafeInteger(offset) || offset < 0) {
    return { status: 400, body: { error: "invalid from: expected a non-negative integer offset" } };
  }
  if (!service) {
    // No relay/transcript service mounted at all - nothing to replay.
    return { status: 404, body: { error: "no transcript for stream" } };
  }
  const readStream = canonicalStreamFor(stream, correlation, service.correlationStore);
  const data = readTranscriptFrom(
    readStream,
    offset,
    service.store,
    correlation,
    service.correlationStore,
    service.liveFallback(readStream),
  );
  if (data === undefined) {
    return { status: 404, body: { error: "no transcript for stream" } };
  }
  return { status: 200, body: data };
}