/**
 * # sdk/remote — the HTTP transport: the 109 client over the 5rb wire (kestrel-djm.5)
 *
 * The LIGHT transport. `remoteTransport({ baseUrl, fetch })` binds the kestrel-109 {@link KestrelClient}
 * (mint / validate / sim / grade over the 5rb wire dialect) and re-exposes it under the ONE typed SDK
 * contract. Catalog discovery / validation / grade delegate to the client verbatim; an incremental Session
 * is a CLIENT-SIDE REPLAY of the server-computed, catalog-pinned transcript the `POST /sim` stream carries
 * (the wire deliberately omits an `advance` — a face is a pure encoder over the controller, never a forked
 * progression). Because the server ran the SAME engine the LOCAL transport runs, the replayed pentad chain /
 * Blotter / grade / root are BYTE-IDENTICAL to the LOCAL run (the djm.5 headline).
 *
 * FAIL-CLOSED, PROTOCOL-UNIFORM: a 402/Offer is DATA on {@link Gated} (surfaced, never thrown / redirected);
 * an unknown subject is a typed {@link KestrelClientError} (the 109 client's non-2xx throw); an off-contract
 * SSE event is the 109 client's `SSE_PROTOCOL_VIOLATION` (fail-closed, never laundered into a face state);
 * a replay turn that diverges from the recorded script is a typed djm.4 {@link SessionResponse} refusal.
 *
 * DEPENDENCY BOUNDARY (djm.3 / djm.5): the ONLY value import is the 109 client (+ its re-exported wire
 * types) and this SDK's light core (`facade.ts` / `types.ts`). Every djm.2/djm.4 shape is a TYPE-ONLY import
 * (erased). NO local runtime — no djm.4 controller, no simulate/sim driver, no engine tree, no djm.8 loader
 * — reaches this graph. Pinned by tests/package.boundary.test.ts.
 *
 * ── DESIGN FORK LOGGED (hard rules forbid .beads writes) — the HTTP Session is a server-run REPLAY ──
 * A catalog Session's SUBJECT is a FIXED, server-pinned script (arm-once/pass-forever). The HTTP transport
 * runs it server-side ONCE at `openSession` (`POST /sim` resolves the catalog dataset to its pre-computed
 * record) and REPLAYS the record verb-by-verb: `start`/`advance` deliver the recorded Frames and return the
 * recorded pentad-chained {@link import("../protocol/session.ts").TurnEntry}s verbatim (never re-sealed — the
 * server's entries ARE the truth), verifying each caller turn against the recorded script (a divergent turn
 * is a typed `turn-conflict` refusal — fail-closed). This mirrors "HTTP is canonical; the SDK is a lossless
 * projection" (djm.5 design). The minimal shape consistent with djm.4's Delivery/SessionResponse.
 */

import { KestrelClient, KestrelClientError } from "../client/index.ts";
import type { FetchLike, TrialCapability } from "../client/index.ts";

import type { Blotter } from "../protocol/index.ts";
import { decodeCatalogPage } from "../catalog/catalog-listing.ts";
import type { SessionDiagnostic, SessionId, TurnBody, TurnEntry } from "../protocol/session.ts";
import type {
  AuthoredResponse,
  BoundResponse,
  Delivery,
  SessionFrame,
  SessionResponse,
  SessionTranscript,
  TurnDisposition,
} from "../session/controller-types.ts";

import { SDK_VERSION, createSdk } from "./facade.ts";
import type {
  ArtifactResult,
  Gated,
  GradeOutcome,
  GradeRequest,
  KestrelSession,
  OperationRef,
  OperationResumption,
  SdkFinalized,
  SessionRef,
  Transport,
  ValidateOutcome,
} from "./types.ts";

// Re-export the light SDK core so a remote-only consumer imports EVERYTHING it needs from this ONE light
// entry: `import { createSdk, remoteTransport } from "kestrel.markets/sdk/remote"`.
export { createSdk, SDK_VERSION } from "./facade.ts";
export type {
  KestrelSdk,
  KestrelSession,
  SessionRef,
  SdkFinalized,
  GradeRequest,
  ArtifactResult,
  OperationRef,
  OperationResumption,
  Transport,
  Gated,
  GradeOutcome,
  ValidateOutcome,
} from "./types.ts";

/** Options for {@link remoteTransport}: the managed API base + an optional `fetch` override (tests / a
 *  custom agent) + an optional pre-seeded trial capability. Mirrors the 109 client's `ClientOptions`. */
export interface RemoteTransportOptions {
  readonly baseUrl: string;
  readonly fetch?: FetchLike;
  readonly capability?: TrialCapability;
}

/** The server-computed Session record the `POST /sim` stream carries for a catalog subject — the truth the
 * HTTP transport replays. Structurally the wire form of a baked catalog-records.ts record (the wire is the
 * contract between server and client; the client does not import the server fixture). */
interface WireSessionRecord {
  readonly sessionId: string;
  readonly blotter: Blotter;
  readonly tipHash: string;
  readonly conformanceRoot: string;
  readonly artifacts: readonly string[];
  readonly entries: readonly WireTurnEntry[];
  readonly frames: readonly unknown[];
}
interface WireTurnEntry {
  readonly kind: "turn";
  readonly sessionId: string;
  readonly ordinal: number;
  readonly parentHash: string;
  readonly frameRoot: string;
  readonly authoredSha256: string;
  readonly body: TurnBody;
  readonly entryHash: string;
}

const isRecord = (v: unknown): v is Record<string, unknown> => typeof v === "object" && v !== null;

/** The disposition a replayed response earns (the djm.4 receipt label; not asserted by the parity test but
 * kept honest). */
function dispositionOf(r: AuthoredResponse): TurnDisposition {
  switch (r.kind) {
    case "authored":
      return "armed";
    case "pass":
      return "pass";
    case "stand-down":
      return "stood-down";
  }
}

/**
 * The RECORDED body a caller response must match for a faithful replay. Returns `null` on a match, else the
 * typed djm.4 diagnostic for the divergence (fail-closed — the HTTP replay honors only the server-pinned
 * script; a divergent turn cannot be re-run against a fixed server transcript).
 */
function matchOrDiagnose(body: TurnBody, r: AuthoredResponse): SessionDiagnostic | null {
  switch (body.kind) {
    case "authored":
      // An empty authored body is a `pass` (a no-op continuation); a non-empty one is an authored document.
      if (body.bytes === "") return r.kind === "pass" ? null : "turn-conflict";
      return r.kind === "authored" && r.document === body.bytes ? null : "turn-conflict";
    case "stand-down":
      return r.kind === "stand-down" && r.reason === body.reason ? null : "turn-conflict";
    case "failure":
      // A recorded host failure has no re-authorable content; only a `pass` continues past it.
      return r.kind === "pass" ? null : "turn-conflict";
  }
}

/**
 * A HTTP Session — a client-side REPLAY of the server-computed, catalog-pinned record. Every returned object
 * is the server's verbatim djm.2/djm.4 protocol shape (byte-identical to the LOCAL run).
 */
class RemoteSession implements KestrelSession {
  readonly sessionId: SessionId;
  private cursor = 0; // index of the next uncommitted recorded turn
  private started = false;

  constructor(private readonly record: WireSessionRecord) {
    this.sessionId = record.sessionId as SessionId;
  }

  async start(): Promise<Delivery> {
    if (this.started) throw new KestrelClientError({ code: "ALREADY_STARTED", message: "start() may only be called once, first" });
    this.started = true;
    const first = this.deliveryAt(0);
    if (first === null) throw new KestrelClientError({ code: "EMPTY_TRANSCRIPT", message: "the server returned an empty Session transcript" });
    return first;
  }

  advance(response: AuthoredResponse): Promise<SessionResponse> {
    return Promise.resolve(this.commit(response));
  }

  revise(response: AuthoredResponse): Promise<SessionResponse> {
    return Promise.resolve(this.commit(response));
  }

  async submit(response: BoundResponse): Promise<SessionResponse> {
    const entry = this.record.entries.find((e) => e.ordinal === response.ordinal);
    if (entry === undefined) return { ok: false, diagnostic: "broken-chain" };
    if (response.sessionId !== this.sessionId) return { ok: false, diagnostic: "wrong-session" };
    if (response.parentHash !== entry.parentHash) return { ok: false, diagnostic: "broken-chain" };
    if (response.frameRoot !== entry.frameRoot) return { ok: false, diagnostic: "stale-frame" };
    if (!sameBody(entry.body, response.body)) return { ok: false, diagnostic: "turn-conflict" };
    const receipt = { entry: entry as unknown as TurnEntry, disposition: bodyDisposition(entry.body) };
    const idx = this.record.entries.indexOf(entry);
    const next = this.deliveryAt(idx + 1);
    return { ok: true, receipt, next };
  }

  async resume(): Promise<SessionTranscript> {
    return { sessionId: this.sessionId, entries: this.record.entries as unknown as readonly TurnEntry[] };
  }

  async finalize(): Promise<SdkFinalized> {
    return {
      blotter: this.record.blotter,
      sessionId: this.sessionId,
      tipHash: this.record.tipHash,
      conformanceRoot: this.record.conformanceRoot,
      artifacts: this.record.artifacts,
    };
  }

  private deliveryAt(i: number): Delivery | null {
    const entry = this.record.entries[i];
    if (entry === undefined) return null;
    return {
      sessionId: this.sessionId,
      ordinal: entry.ordinal,
      parentHash: entry.parentHash,
      frameRoot: entry.frameRoot,
      frame: this.record.frames[i] as SessionFrame,
    };
  }

  private commit(response: AuthoredResponse): SessionResponse {
    const entry = this.record.entries[this.cursor];
    if (entry === undefined) return { ok: false, diagnostic: "not-pending" };
    const mismatch = matchOrDiagnose(entry.body, response);
    if (mismatch !== null) return { ok: false, diagnostic: mismatch };
    this.cursor += 1;
    const receipt = { entry: entry as unknown as TurnEntry, disposition: dispositionOf(response) };
    return { ok: true, receipt, next: this.deliveryAt(this.cursor) };
  }
}

/** Structural body equality for the explicit-binding submit path (no re-seal). */
function sameBody(a: TurnBody, b: TurnBody): boolean {
  if (a.kind !== b.kind) return false;
  if (a.kind === "authored" && b.kind === "authored") return a.bytes === b.bytes;
  if (a.kind === "stand-down" && b.kind === "stand-down") return a.reason === b.reason;
  if (a.kind === "failure" && b.kind === "failure") return a.failureClass === b.failureClass;
  return false;
}

/** The disposition of a recorded body (the submit path's receipt label). */
function bodyDisposition(body: TurnBody): TurnDisposition {
  switch (body.kind) {
    case "authored":
      return body.bytes === "" ? "pass" : "armed";
    case "stand-down":
      return "stood-down";
    case "failure":
      return "failure";
  }
}

/** Resolve a {@link SessionRef} (+ an optional caller-supplied strategy `document`) to the `POST /sim` inputs.
 *  A catalog subject rides its `id` DIRECTLY as `dataset.artifact_id` — the BARE platform artifact handle the
 *  server resolves against its catalog store; a raw dataset carries its own `source` + `dataset`.
 *
 *  kestrel-p6hz: the `catalog:` PREFIX this used to prepend (`dataset: "catalog:" + subject.id`) is a
 *  CLIENT-INVENTED namespace the deployed server never speaks — `POST /sim` with `dataset.artifact_id =
 *  "catalog:art_…"` fail-closes `HTTP_404 'dataset catalog:art_… is not present in the catalog store'`, while
 *  the BARE `dataset.artifact_id = "art_…"` opens cleanly (both verified against live prod). So a discovered
 *  free catalog row 404'd through the agent face even though the same artifact opens over raw HTTP — the
 *  equal-projections violation. Sending the bare handle removes the prefix defect; the SimRequest's
 *  `artifact_id` is a bare Id (the vendored OpenAPI contract), and `catalog()` now surfaces exactly that bare
 *  handle as each row's `id`, so `openSession({ kind:"catalog", id: row.id })` is self-consistent end-to-end.
 *
 *  The optional `document` is the customer strategy the managed author-no-strategy fence requires (ADR-0012,
 *  kestrel-88u2) — matching the sim verb's `--plans` semantics: when present it is the sim `source` (for a
 *  catalog subject, whose `source` is structurally `""` otherwise; for a dataset subject it overrides the
 *  ref's own source). A bare openSession (no document) keeps `source:""`: the platform's fence answers, now
 *  legibly (Bug A). */
function simInputs(subject: SessionRef, document?: string): { source: string; dataset: string } {
  return subject.kind === "catalog"
    ? { source: document ?? "", dataset: subject.id }
    : { source: document ?? subject.source, dataset: subject.dataset };
}

/**
 * Build the HTTP transport over the managed API. Wrap it with `createSdk(remoteTransport({ baseUrl, fetch }))`.
 * Delegates catalog / validate / grade to the 109 client; runs + replays a catalog Session's server-computed
 * transcript for the incremental surface. Fail-closed and protocol-uniform throughout.
 */
export function remoteTransport(options: RemoteTransportOptions): Transport {
  const base = options.baseUrl.replace(/\/+$/, "");
  const fetchImpl: FetchLike = options.fetch ?? (globalThis.fetch as FetchLike);
  const client = new KestrelClient({
    baseUrl: options.baseUrl,
    ...(options.fetch !== undefined ? { fetch: options.fetch } : {}),
    ...(options.capability !== undefined ? { capability: options.capability } : {}),
  });

  return {
    kind: "http",
    version: `${SDK_VERSION}+http`,

    async catalog() {
      const res = await fetchImpl(`${base}/catalog`, { headers: { accept: "application/json" } });
      if (!res.ok) {
        throw new KestrelClientError({ code: `HTTP_${res.status}`, message: `GET /catalog failed (HTTP ${res.status})`, status: res.status });
      }
      // FAITHFUL wire decode into the canonical CatalogPage (kestrel-adge) — the byte-identical
      // CatalogListing[] (kestrel-dnxq; `artifact_id → id`, `frame_count → frameCount`, drop the
      // `content_hash` replay pin, fail-closed on an unaddressable row) PLUS the optional ex-ante `pricing`
      // block surfaced VERBATIM when the served `{ entries, pricing }` body carries one (absent ⇒ a bare page).
      return decodeCatalogPage(await res.json());
    },

    validate(document): Promise<ValidateOutcome> {
      return client.validate(document);
    },

    async openSession(subject, document): Promise<Gated<KestrelSession>> {
      const { source, dataset } = simInputs(subject, document);
      let record: WireSessionRecord | undefined;
      const opened = await client.sim({
        source,
        dataset,
        onEvent: (e) => {
          const rec = e.data?.["session"];
          if (isRecord(rec)) record = rec as unknown as WireSessionRecord;
        },
      });
      if (opened.gated) return { gated: true, payment: opened.payment };
      if (record === undefined) {
        // A 2xx sim that carried no Session record cannot back an incremental replay — fail closed, never
        // fabricate a Session (kestrel-markets-vuiy: the managed transport ran the subject as a BATCH
        // simulation but emits no incremental Session transcript to replay; managed incremental openSession
        // over a catalog subject is not yet a served surface — tracked cross-repo). Rather than the opaque
        // "not an incremental subject" (which contradicted describe's own catalog contract), the refusal now
        // POINTS AT THE BATCH PATH the same subject DOES run on, so a wire-first integrator is redirected
        // instead of stranded. The self-hosted LOCAL transport still opens the SAME catalog subject
        // incrementally today (`createSdk(localTransport())`).
        const subjectClause =
          subject.kind === "catalog"
            ? `the managed transport does not (yet) serve an incremental Session over catalog subject ${JSON.stringify(subject.id)} — the /sim stream carried no Session transcript to replay. `
            : "the /sim stream carried no Session transcript to replay for this subject. ";
        throw new KestrelClientError({
          code: "NO_SESSION_RECORD",
          message:
            subjectClause +
            "Run it as a BATCH simulation instead: the `sim` verb, or the SDK `client.sim({ source, dataset })` — the same drive to a Certified grade, minus the incremental start/advance/finalize surface. (The self-hosted LOCAL transport opens catalog subjects incrementally today.)",
        });
      }
      return { gated: false, value: new RemoteSession(record) };
    },

    grade(request: GradeRequest): Promise<Gated<GradeOutcome>> {
      return client.grade({ blotters: request.blotters });
    },

    async artifact(ref: string): Promise<ArtifactResult> {
      const res = await fetchImpl(`${base}/artifacts/${encodeURIComponent(ref)}`, { headers: { accept: "application/json" } });
      if (!res.ok) {
        throw new KestrelClientError({ code: `HTTP_${res.status}`, message: `artifact ${JSON.stringify(ref)} not resolvable (HTTP ${res.status})`, status: res.status });
      }
      // The server returns the SAME { ref, kind, value } the LOCAL projection yields (src/sdk/local.ts
      // artifact()); the transport re-exposes it verbatim so both faces return a byte-identical ArtifactResult.
      const body = (await res.json()) as { kind?: unknown; value?: unknown };
      return { ref, kind: typeof body.kind === "string" ? body.kind : "artifact", value: body.value };
    },

    async resumeOperation(ref: OperationRef): Promise<OperationResumption> {
      // A COMPLETED operation's OUTCOME (blotter / report / receipt + artifact refs) is carried on its
      // CANONICAL RESUMABLE EVENT STREAM (`GET /operations/{id}/events`) — the SAME stream `sim`/`grade`
      // consume — NOT on the plain `GET /operations/{id}` status envelope, which carries NO domain payload for
      // a completed op in production (it returns empty, so the old plain-endpoint read handed a wire-first
      // integrator an EMPTY payload — kestrel-o3bi: the contract mock had masked this by fabricating a payload
      // the real endpoint never serves). Replay the event stream (resumable via the opaque `after` cursor) and
      // accumulate the terminal payload, so a resumed Operation returns the SAME { blotter, grade, artifacts }
      // the LOCAL projection yields (src/sdk/local.ts resumeOperation) — byte-identical across transports (the
      // djm.5 parity headline), now over the stream production actually serves. The stream is read to its
      // TERMINAL event, so the resumption cursor is `null` (nothing remains to resume) — matching LOCAL.
      const { payload } = await client.resumeOperation({
        operationId: ref.operationId,
        ...(ref.after?.token !== undefined ? { after: ref.after.token } : {}),
      });
      return { operationId: ref.operationId, cursor: null, payload };
    },
  };
}