#!/usr/bin/env bun
/**
 * The child process one omp session runs in.
 *
 * Sessions run out of process so that one session's memory, crash or runaway
 * turn is the child's and not the dispatcher's: a harness that exits, hangs or
 * exhausts its heap takes down a process the daemon supervises rather than the
 * daemon itself. It is also what lets `MemoryMax` on the unit govern the whole
 * fleet, and what makes a kill a real kill.
 *
 * This file is the far side: it loads the harness, runs the real session, and
 * speaks a small protocol back over a unix socket to the
 * {@link AgentSessionLike} proxy in `omp.ts`. It holds no conductor state, opens
 * no database, and reads no config — everything it needs arrives in
 * {@link SessionHostSpec}, so the child's inputs are data a caller can see
 * rather than ambient state it inherits.
 *
 * The protocol deliberately mirrors the narrow session surface and carries
 * only data.
 */

import { connect } from "node:net";

import { createLocalSession, disposeSession, type AgentSessionLike } from "./omp.ts";
import type { GraphToolsObservation } from "./types.ts";
import type { GateShape, ReleaseBlockContext } from "./release-policy.ts";
import type { ResolvedGrants, SessionRole } from "./types.ts";

/**
 * Everything the child needs to build the session. Plain JSON on purpose:
 * callbacks (`onReleaseBlocked`) cannot cross a process boundary, so they
 * become messages instead — see {@link HostToParent}.
 */
export interface SessionHostSpec {
  socket: string;
  cwd: string;
  role: SessionRole;
  sessionDir?: string;
  model?: string;
  resume?: boolean;
  releaseGrants?: ResolvedGrants;
  /**
   * The conductor verb socket this session may call mutations on (#126).
   *
   * Carried in the spec rather than read from the environment by the child,
   * for the same reason the grants are: the child deciding *which* socket it
   * owns is the child deciding which run it is, and that is the question the
   * transport exists to stop it answering.
   * Absent, the verbs are registered and every one of them fails closed.
   */
  verbSocketPath?: string;
  /** Deny every tool but reading and searching (#307) — the setup probes. */
  readOnly?: boolean;
  /** The fleet-owned omp settings overlay (#537): absolute path to the YAML
   *  overlay the daemon materialised from the project's `ompSettings` map.
   *  Carried as a plain scalar like the other fields; absent, nothing is staged
   *  on the far side either. */
  ompSettingsFile?: string;
  /**
   * The structured-settlement contract (#540): a JSON Schema the session's
   * `yield` tool validates its `data` payload against. Carried as plain JSON
   * through the spec, like everything else; absent, the far side passes nothing
   * to the harness.
   */
  outputSchema?: unknown;
  /** Enforcement policy for {@link outputSchema} — the worker contract is
   *  `"permissive"`: a schema violation settles with whatever the worker
   *  produced, never a lost report. */
  outputSchemaMode?: "permissive" | "strict";
  /** Force the hidden `yield` tool into this session's toolset (#540). */
  requireYieldTool?: boolean;
}

/** Parent → child. */
export type ParentToHost =
  | { t: "prompt"; id: number; text: string; opts?: Record<string, unknown> }
  | { t: "park"; id: number }
  | { t: "abort" }
  | { t: "dispose" };

/** Child → parent. */
export type HostToParent =
  | {
      t: "ready";
      sessionFile?: string;
      modelFallbackMessage?: string;
      /**
       * The omp-conductor version this child process loaded, attested at the
       * ready handshake (#832): the session host's `ready` is the exact-session
       * handoff the upgrade verifies against, so the live process's own loaded
       * module version travels with it instead of being read off disk later.
       * Absent only when the version cannot be read — absence is "not
       * attested", never "the newest".
       */
      extensionVersion?: string;
      /** Exact peer entry and package version loaded by this child (#828). */
      harnessPath?: string;
      harnessVersion?: string;
    }
  | {
      t: "graph-tools";
      /**
       * The code-graph session observation (#726): whether the graph MCP tools
       * were in this session's registry. Sent once the session's tool registry
       * settles (it finalises after creation, so the ready handshake never
       * waits for it). Absent value never travels — no message means no
       * observation was recorded, never "graph tools absent", which is the
       * `present: false` truth value.
       */
      graphTools: GraphToolsObservation;
    }
  | { t: "start-error"; message: string }
  | { t: "event"; event: unknown }
  | { t: "session-file"; path: string }
  | { t: "prompt-result"; id: number; ok: boolean; error?: string }
  | { t: "park-result"; id: number; ok: boolean; error?: string }
  | ({ t: "release-blocked"; shape: GateShape } & ReleaseBlockContext);

/**
 * Depth at which a harness event stops being copied for the wire.
 *
 * Every field this package actually reads — `type`, `message.role`,
 * `message.content[].text`, `message.usage.cost.total`,
 * `telemetry.cost.estimatedUsd`, `isTerminal` — is within five levels. The cap
 * exists for the other direction: a harness event holding a client, a socket or
 * a parent pointer would otherwise be walked until something threw, and losing
 * the event stream is losing turn counting, spend accounting and the run's
 * report.
 */
const EVENT_DEPTH = 8;

/**
 * A JSON-safe copy of one harness event.
 *
 * Cycle- and depth-guarded rather than a bare `JSON.stringify`, because the
 * event union belongs to the peer dependency and this package does not get to
 * assume it is a tree. Functions, symbols and `undefined` drop out the way they
 * would through `JSON.stringify`; a `Date` keeps its ISO form; anything that
 * refuses to be read is replaced by `null` rather than taking the event with it.
 */
export function serializableEvent(value: unknown, depth = EVENT_DEPTH, seen = new WeakSet<object>()): unknown {
  if (value === null || typeof value !== "object") {
    return typeof value === "function" || typeof value === "symbol" || typeof value === "bigint"
      ? undefined
      : value;
  }
  if (value instanceof Date) return value.toISOString();
  if (seen.has(value)) return null;
  if (depth <= 0) return null;
  seen.add(value);
  try {
    if (Array.isArray(value)) return value.map((item) => serializableEvent(item, depth - 1, seen));
    const out: Record<string, unknown> = {};
    for (const [key, raw] of Object.entries(value)) {
      const copied = serializableEvent(raw, depth - 1, seen);
      if (copied !== undefined) out[key] = copied;
    }
    return out;
  } catch {
    return null;
  } finally {
    seen.delete(value);
  }
}

/** Newline-delimited JSON both ways. Values are escaped, so a newline is a frame. */
export function encodeFrame(message: HostToParent | ParentToHost): string {
  return `${JSON.stringify(message)}\n`;
}

/**
 * Split a growing buffer into whole frames. Returns the frames and the tail that
 * is not a frame yet — a socket read boundary lands mid-message routinely, and
 * dropping that tail loses whichever message straddled it.
 */
export function decodeFrames(buffer: string): { frames: unknown[]; rest: string } {
  const parts = buffer.split("\n");
  const rest = parts.pop() ?? "";
  const frames: unknown[] = [];
  for (const part of parts) {
    if (part === "") continue;
    try {
      frames.push(JSON.parse(part));
    } catch {
      // A torn frame must not stop the frames written after it: losing one
      // event costs a turn count, losing the stream costs the run.
    }
  }
  return { frames, rest };
}

export interface SessionHostDeps {
  createSession: typeof createLocalSession;
}

/**
 * Run one session until the parent disposes it or the socket drops.
 *
 * Resolves when the session is over. Never throws for a session that failed to
 * start: the failure is reported as `start-error` and the parent turns it back
 * into the same `Error` the in-process path would have raised, so a missing
 * peer dependency still reads as a deployment mistake rather than as an
 * unexplained child exit.
 *
 * Always resolves `true` now that the worker-identity refusal is gone (#894):
 * the boolean return survives so the entry point's exit-code contract keeps
 * its shape for any wrapper that learned it.
 */
export async function runSessionHost(
  spec: SessionHostSpec,
  deps: SessionHostDeps = { createSession: createLocalSession },
): Promise<boolean> {
  const socket = connect(spec.socket);
  socket.setNoDelay(true);
  const send = (message: HostToParent): void => {
    if (!socket.writableEnded) socket.write(encodeFrame(message));
  };

  const { promise: connected, resolve: onConnect, reject: onConnectFail } = Promise.withResolvers<void>();
  socket.once("connect", () => {
    onConnect();
  });
  socket.once("error", (err: Error) => {
    onConnectFail(err);
  });
  await connected;

  let session: AgentSessionLike | undefined;
  try {
    session = await deps.createSession({
      cwd: spec.cwd,
      role: spec.role,
      ...(spec.sessionDir === undefined ? {} : { sessionDir: spec.sessionDir }),
      ...(spec.model === undefined ? {} : { model: spec.model }),
      ...(spec.resume === undefined ? {} : { resume: spec.resume }),
      ...(spec.releaseGrants === undefined ? {} : { releaseGrants: spec.releaseGrants }),
      ...(spec.verbSocketPath === undefined ? {} : { verbSocketPath: spec.verbSocketPath }),
      ...(spec.readOnly === undefined ? {} : { readOnly: spec.readOnly }),
      ...(spec.ompSettingsFile === undefined ? {} : { ompSettingsFile: spec.ompSettingsFile }),
      ...(spec.outputSchema === undefined ? {} : { outputSchema: spec.outputSchema }),
      ...(spec.outputSchemaMode === undefined ? {} : { outputSchemaMode: spec.outputSchemaMode }),
      ...(spec.requireYieldTool === undefined ? {} : { requireYieldTool: spec.requireYieldTool }),
      // The release audit lives in the daemon's state directory, which this
      // process may not be able to write and must not be trusted to. It
      // becomes a message; the parent performs the durable write.
      onReleaseBlocked: (shape, context) => {
        send({ t: "release-blocked", shape, ...context });
      },
    });
  } catch (err) {
    send({ t: "start-error", message: err instanceof Error ? err.message : String(err) });
    // The frame must reach the parent before this process exits: `process.exit`
    // does not flush the socket, and a parent that only observes the exit used
    // to report a bare exit code instead of the child's own words.
    // This is a failed start, not an identity refusal: the process still walks
    // away with a clean exit code, because the parent's verdict comes from the
    // frame, not the status — 91 is reserved for the boundary refusal above.
    await new Promise<void>((resolve) => {
      socket.once("finish", () => resolve());
      socket.once("error", () => resolve());
      socket.end();
    });
    return true;
  }

  const live = session;
  // `sessionFile` is a getter that can materialise after the session opens, and
  // the daemon records it as the only readable evidence a run leaves behind. So
  // it is re-read on every event rather than captured once.
  let lastSessionFile = live.sessionFile;
  live.on("*", (event) => {
    send({ t: "event", event: serializableEvent(event) });
    const current = live.sessionFile;
    if (current !== undefined && current !== lastSessionFile) {
      lastSessionFile = current;
      send({ t: "session-file", path: current });
    }
  });

  send({
    t: "ready",
    ...(live.sessionFile === undefined ? {} : { sessionFile: live.sessionFile }),
    ...(live.modelFallbackMessage === undefined
      ? {}
      : { modelFallbackMessage: live.modelFallbackMessage }),
    // The loaded-module attestation (#832): what THIS process's session
    // machinery actually executes, carried in the exact-session handoff the
    // upgrade verifies against.
    ...(live.extensionVersion === undefined ? {} : { extensionVersion: live.extensionVersion }),
    ...(live.harnessPath === undefined ? {} : { harnessPath: live.harnessPath }),
    ...(live.harnessVersion === undefined ? {} : { harnessVersion: live.harnessVersion }),
  });

  // The code-graph session observation (#726) rides its own frame: the factory
  // reads the session's tool registry asynchronously (the harness finalises
  // MCP wiring after creation), so the ready handshake never waits for it. The
  // child is the session's own process — this is the runtime observation the
  // config check in `resolvePrereqs` deliberately is not, and an observation
  // that never resolves simply never travels: no message is "not recorded",
  // never "graph tools absent".
  const graphToolsReady = live.graphToolsReady;
  if (graphToolsReady !== undefined) {
    void graphToolsReady.then((graphTools) => {
      if (graphTools !== undefined) send({ t: "graph-tools", graphTools });
    });
  }

  const { promise: finished, resolve: finish } = Promise.withResolvers<void>();
  let buffer = "";
  socket.on("data", (chunk: Buffer) => {
    buffer += chunk.toString("utf8");
    const { frames, rest } = decodeFrames(buffer);
    buffer = rest;
    for (const frame of frames) {
      const message = frame as ParentToHost;
      if (message.t === "prompt") {
        void live.prompt(message.text, message.opts).then(
          () => {
            send({ t: "prompt-result", id: message.id, ok: true });
          },
          (err: unknown) => {
            send({
              t: "prompt-result",
              id: message.id,
              ok: false,
              error: err instanceof Error ? err.message : String(err),
            });
          },
        );
        continue;
      }
      if (message.t === "park") {
        const id = message.id;
        void live.park().then(
          () => send({ t: "park-result", id, ok: true }),
          (err: unknown) =>
            send({
              t: "park-result",
              id,
              ok: false,
              error: err instanceof Error ? err.message : String(err),
            }),
        );
        continue;
      }
      if (message.t === "abort") {
        live.abort();
        continue;
      }
      if (message.t === "dispose") finish();
    }
  });

  // A dead parent must not leave a live session holding a checkout and a model
  // budget. The daemon dying used to take its in-process sessions with it; the
  // socket closing is how that stays true now it does not.
  socket.on("close", () => {
    finish();
  });
  await finished;

  try {
    live.abort();
  } catch {
    // Already finished, or a harness that dislikes a second abort. Either way
    // teardown continues: this path runs on the way out.
  }
  try {
    await disposeSession(live);
  } catch {
    // Teardown noise must not become the run's outcome.
  }
  socket.end();
  return true;
}

/** Reads the spec the parent handed over. Argv, not stdin: stdin is the harness's. */
function specFromArgv(argv: readonly string[]): SessionHostSpec {
  const raw = argv[2];
  if (raw === undefined) {
    throw new Error("omp-conductor session host: expected a JSON session spec as argv[2]");
  }
  return JSON.parse(raw) as SessionHostSpec;
}

if (import.meta.main) {
  // Exit code is the parent's only signal when the socket never opened, so a
  // spec that will not parse fails loudly here rather than as a silent hang.
  const spec = specFromArgv(process.argv);
  const started = await runSessionHost(spec);
  // 91 was the worker-identity boundary refusal; the code stays reserved (and
  // unreachable — runSessionHost always resolves true since #894) so nothing
  // that learned to read it can mistake a future refusal for a clean exit.
  process.exit(started ? 0 : 91);
}