/**
 * The single seam between omp-conductor and the omp harness.
 *
 * The harness is a `peerDependency`, not a `devDependency`: this package has to
 * type-check and publish without it on disk. So every untyped access lives in
 * this file, behind one cast at the dynamic-import boundary, and the rest of the
 * package only ever sees {@link AgentSessionLike}. If the harness renames
 * `createAgentSession`, `subscribe` or `abort`, exactly one file breaks.
 */

import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
import { createServer, type Server, type Socket } from "node:net";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";

import { OMP_HARNESS_PACKAGE, packageNodeModulesRoot } from "./host.ts";
import { harnessVersion, resolveHarnessEntry } from "./harness-loader.ts";
import { readOnlySession, worktreeConfinement } from "./confinement.ts";
import { observeGraphTools } from "./graph.ts";
import { releasePolicyTripwire, type GateShape, type ReleaseBlockContext } from "./release-policy.ts";
import type {
  HostToParent,
  ParentToHost,
  SessionHostSpec,
} from "./session-host.ts";
import { decodeFrames, encodeFrame } from "./session-host.ts";
import type { GraphToolsObservation, ResolvedGrants, SessionRole } from "./types.ts";
import { SESSION_ROLE_ENV } from "./types.ts";
import { conductorVerbs } from "./verbs/client.ts";

/**
 * Harness package name. Held in a variable and cast to `string` at the call
 * site on purpose: a non-literal specifier stops `tsc` from trying to resolve
 * the module, which is the whole reason this shim exists.
 */
const OMP_PACKAGE = OMP_HARNESS_PACKAGE;

/**
 * Why the explicitly anchored harness entry is not the one installed beside
 * omp-conductor, or `undefined` when it is — or when there is nothing to
 * compare, because this package is running from a source checkout with no
 * install root of its own.
 *
 * The failure it names is silent otherwise (#828): Bun's ambient bare-package
 * resolution can choose its install cache even when the worker binding is
 * mounted and inode-current. The production resolver is therefore
 * {@link resolveHarnessEntry}, rooted at this module's directory; this check
 * remains the fail-closed attestation that the resolved absolute path stayed
 * inside the same install.
 */
export function harnessOutsideInstall(
  moduleDir: string = import.meta.dir,
  resolveSpecifier: (specifier: string) => string = () => resolveHarnessEntry(moduleDir),
): string | undefined {
  const root = packageNodeModulesRoot(moduleDir);
  if (root === undefined) return undefined;
  let resolved: string;
  try {
    const result = resolveSpecifier(OMP_PACKAGE);
    resolved = result.startsWith("file:") ? fileURLToPath(result) : result;
  } catch {
    // Nothing resolved at all. The import below reports that far better than a
    // guess here could, and reporting it twice in two spellings is worse.
    return undefined;
  }
  if (resolved.startsWith(`${root}/`)) return undefined;
  return (
    `omp-conductor could not load its peer dependency ${OMP_PACKAGE} from its own installation: it resolves to ` +
    `${resolved}, outside ${root}. Refusing Bun's ambient auto-install cache prevents a different harness build ` +
    "and native addon from standing in for the operator's install. Run `omp-conductor setup host` to verify the " +
    "worker binding, or install the harness alongside omp-conductor."
  );
}

/**
 * The only session surface the dispatcher is allowed to know about: send one
 * prompt, watch the event stream, and kill it. Caps are enforced by the caller
 * against this interface, never by the model inside the session.
 */
export interface AgentSessionLike {
  prompt(text: string, opts?: Record<string, unknown>): Promise<unknown>;
  /**
   * Subscribe to one harness event type (`"message_end"`, `"agent_end"`, …), or
   * `"*"` for every event. The payload is the harness's own event union, which
   * this package cannot name without the peer dependency, so it arrives as
   * `unknown` and each caller narrows the two or three fields it reads.
   *
   * One event is not the harness's at all: `"session_exit"` fires when the
   * underlying session *process* has terminated, carrying
   * `{ type: "session_exit", code?: number | null }`. It is the single real
   * terminal signal — a crashed or killed session is never confused with one
   * that merely stopped streaming (`"agent_end"`). A supervisor may receive it
   * after its own `dispose()`, and must decide what an exit means from its own
   * state rather than assuming every exit is a crash.
   */
  on(event: string, cb: (e: unknown) => void): void;
  abort(): void;
  /**
   * Cooperative operator park: abort the active turn and resolve when the
   * harness is idle (#238). Distinct from `abort()`, which is the
   * fire-and-forget kill path and must stay cheap for cap kills.
   */
  park(): Promise<void>;
  /**
   * Absolute transcript path the harness opened, so a human — and the
   * arm/monitor tooling — can read what the worker actually did. `undefined`
   * when the session has no file backing; never a path we merely asked for.
   */
  sessionFile?: string;
  /**
   * Set when the harness could not honour the requested model and quietly used
   * another one. Carried out through this seam so the daemon can log the
   * downgrade, instead of it vanishing and a run just reading dumber.
   */
  modelFallbackMessage?: string;
  /**
   * The code-graph session observation (#726): whether the graph MCP tools
   * were in this session's registry, as a one-shot async fact. The harness
   * tool registry settles late in a session's life (MCP wiring finalises
   * after creation), so the in-process session under {@link createLocalSession}
   * exposes this promise; the session-host child relays the resolved value
   * over the wire, and the proxy session {@link createSession} exposes the
   * settled observation as {@link graphTools}. Resolving `undefined` means no
   * observation could be recorded (surface absent or never readable) — never
   * "graph tools absent", which is the `present: false` truth value.
   */
  graphToolsReady?: Promise<GraphToolsObservation | undefined>;
  /**
   * The code-graph session observation (#726): whether the graph MCP tools
   * were in this session's registry. Set on the proxy session once the child's
   * relayed observation lands. Absent means no observation was recorded — never
   * "graph tools absent", which is the `present: false` truth value.
   */
  graphTools?: GraphToolsObservation;
  /**
   * The version of the omp-conductor module this session process actually
   * LOADED, attested by the live process at session start (#832): the module
   * tree this session's tools execute out of, read from the package this
   * process resolved — never from the package files on disk after the fact.
   * Absent only when the version cannot be read (a hand-built fake, a
   * packaging oddity) — absence is not "the newest", and an upgrade that
   * cannot see the loaded version must not claim the session reloaded.
   */
  extensionVersion?: string;
  /** Absolute entry path of the harness module this session process loaded. */
  harnessPath?: string;
  /** Version read from the package owning {@link harnessPath}. */
  harnessVersion?: string;
}

/**
 * Module members this shim calls. Checked at the import boundary so a harness
 * bump that drops one fails by name, instead of as `undefined is not a
 * constructor` in the middle of session setup.
 */
const REQUIRED_EXPORTS = ["createAgentSession", "SessionManager", "AgentRegistry"] as const;

/** The three module members this shim calls, all verified before the cast. */
interface OmpModule {
  createAgentSession(opts: unknown): Promise<unknown>;
  /** Only the file-backed factories: `inMemory()` would defeat the transcript. */
  SessionManager: {
    create(cwd: string, sessionDir?: string): unknown;
    /**
     * Resume path. Continues the newest transcript for `cwd`, and the harness
     * itself starts a fresh one when there is nothing to continue. Declared
     * optional — and absent from {@link REQUIRED_EXPORTS} — so a harness build
     * without it degrades to a fresh session instead of failing to start.
     */
    continueRecent?(cwd: string, sessionDir?: string): Promise<unknown>;
  };
  /** Constructed once per session — see the note at the call site. */
  AgentRegistry: new () => unknown;
}

/**
 * The members of the harness `AgentSession` this package touches. The real
 * class has hundreds; declaring four keeps the blast radius of an SDK bump
 * proportional to what we actually depend on.
 */
interface RawSession {
  prompt(text: string, opts?: unknown): Promise<unknown>;
  subscribe(listener: (event: unknown) => void): unknown;
  abort(opts?: {
    goalReason?: "interrupted" | "internal";
    reason?: string;
    preserveCompaction?: boolean;
  }): Promise<void> | void;
  dispose?(opts?: unknown): Promise<unknown>;
  readonly sessionFile?: string;
}

/**
 * Teardown handles, keyed by the adapter we handed out. Kept off
 * {@link AgentSessionLike} so the interface stays exactly the shared contract
 * every other slice codes against, while callers still have a way to release
 * the harness's background work (MCP clients, watchers) instead of leaking it
 * for the lifetime of the daemon.
 */
const disposers = new WeakMap<AgentSessionLike, () => Promise<void>>();

/**
 * Start one omp coding session **in this process**, rooted at `cwd`, with a
 * private agent registry and a file-backed transcript of its own.
 *
 * Production callers want {@link createSession}, which runs this same function
 * inside a **child process**; the only caller of this one is `session-host.ts`,
 * plus the tests that pin what it builds. Calling it directly from the daemon
 * puts the harness back inside the dispatcher, where a crash, a hang or an
 * exhausted heap takes the whole fleet down with it and no `MemoryMax` bounds
 * it — that is what the child process buys. It is not a security boundary: the
 * child runs as the daemon's own user.
 *
 * `sessionDir` chooses the directory the harness writes that transcript into;
 * omitted, the harness picks its default location for `cwd`. Either way the
 * resolved path comes back on `session.sessionFile` — this function never
 * invents one.
 *
 * `resume` continues the most recent transcript for `cwd` instead of opening a
 * blank one. That is what a long-lived session (the orchestrator) wants: a
 * daemon restart should not erase its memory of what it has already escalated.
 * A worker wants the opposite, so it stays off by default.
 *
 * @throws if the peer dependency is absent, naming it — a missing harness is a
 * deployment mistake, and a stack trace about a failed dynamic import sends the
 * reader looking in the wrong place.
 */
export async function createLocalSession(opts: {
  cwd: string;
  sessionDir?: string;
  model?: string;
  resume?: boolean;
  /**
   * What kind of session this is. Explicit rather than derived from the caller:
   * it selects the worktree confinement gate *and* is the role the release gate
   * compares grants against, and those two must never be able to disagree about
   * whether a session is a worker (#122).
   */
  role: SessionRole;
  /** Install the release/deploy tool-call gate with these per-shape grants. */
  releaseGrants?: ResolvedGrants;
  /** Durable audit callback invoked only when that gate rejects a call. */
  onReleaseBlocked?: (shape: GateShape, context: ReleaseBlockContext) => void;
  /**
   * The fleet-owned omp settings overlay staged into this session's omp
   * settings (#537): an absolute path to a YAML overlay the daemon materialised
   * from the project's `ompSettings` map (plus the retry keys derived from
   * `modelFallbacks`), passed to `Settings.init({ configFiles: [<path>] })` so
   * it layers on top of the ordinary global/project discovery — never an
   * isolated instance, which would drop providers and approvals out from under
   * the worker. Absent, no settings are passed and the harness discovers
   * exactly as it does today.
   */
  ompSettingsFile?: string;
  /**
   * The conductor verb socket this session's mutation tools call (#126).
   *
   * The tools are registered either way. Omitted, every one of them fails
   * closed with an explanation, which is the correct behaviour for a session
   * whose daemon never gave it a channel — a tool that quietly vanished would
   * leave the model to improvise `git push` instead.
   */
  verbSocketPath?: string;
  /**
   * Deny every tool but reading and searching (#307).
   *
   * For sessions asked to *read* a repository and answer a question — the setup
   * probes. Distinct from {@link worktreeConfinement}, which scopes where writes
   * may land: this removes the shell, the editors and the verbs entirely, so the
   * session has no route to a mutation to scope in the first place.
   */
  readOnly?: boolean;
  /**
   * The structured-settlement contract (#540): a JSON Schema the session's
   * `yield` tool validates its `data` payload against. The worker yields its
   * settlement as structured output instead of writing prose for the caller to
   * regex. A subagent-oriented harness option, forwarded as-is.
   */
  outputSchema?: unknown;
  /**
   * Enforcement policy for {@link outputSchema}. The worker contract is
   * `"permissive"`: a schema violation must still settle with whatever the
   * worker produced, never discard the report.
   */
  outputSchemaMode?: "permissive" | "strict";
  /**
   * Force the hidden `yield` tool into this session's toolset (#540). Without
   * it the model has no sanctioned way to submit structured output.
   */
  requireYieldTool?: boolean;
}): Promise<AgentSessionLike> {
  let harnessPath: string;
  try {
    // An explicit Bun-supported root is the resolver contract. A bare import
    // from this same bound module was observed selecting the worker's ambient
    // auto-install cache even while the peer beside it was readable (#828).
    harnessPath = resolveHarnessEntry(import.meta.dir);
  } catch (cause) {
    throw new Error(
      `omp-conductor could not resolve its peer dependency ${OMP_PACKAGE} from ${import.meta.dir}. Install it alongside omp-conductor (it is deliberately not bundled, so the dispatcher runs the same harness build as the operator).`,
      { cause },
    );
  }
  const stray = harnessOutsideInstall(import.meta.dir, () => harnessPath);
  if (stray !== undefined) throw new Error(stray);
  let loaded: unknown;
  try {
    // Dynamic import is load-bearing: the runtime-selected absolute path pins
    // the operator's peer without making it a build-time dependency.
    loaded = await import(harnessPath);
  } catch (cause) {
    throw new Error(
      `omp-conductor could not load its peer dependency ${OMP_PACKAGE} from ${harnessPath}. Install it alongside omp-conductor (it is deliberately not bundled, so the dispatcher runs the same harness build as the operator).`,
      { cause },
    );
  }
  // Narrowed onto a `const` so the checks survive into the closure below.
  const namespace: unknown = loaded;
  if (namespace === null || typeof namespace !== "object") {
    throw new Error(
      `${OMP_PACKAGE} loaded but is not a module namespace; omp-conductor needs a harness build that still exposes the SDK entrypoint.`,
    );
  }
  const missing = REQUIRED_EXPORTS.filter(
    (name) => typeof Reflect.get(namespace, name) !== "function",
  );
  if (missing.length > 0) {
    throw new Error(
      `${OMP_PACKAGE} loaded but exports no ${missing.join(", ")}; omp-conductor needs a harness build exposing createAgentSession (the session), SessionManager (a file-backed transcript) and AgentRegistry (so concurrent workers do not collide on the "Main" identity).`,
    );
  }
  // The one unchecked cast in this package: the shape is verified immediately
  // above, but only the harness itself can name its own types.
  const mod = namespace as unknown as OmpModule;

  // File-backed on purpose. `SessionManager.inMemory()` leaves
  // `session.sessionFile` undefined, and once the worktree is gone the
  // transcript is the only record of what the worker actually did.
  const sessionManager = await openSessionManager(mod, opts);
  const extensions = [
    // Read-only sessions get the deny-by-default gate *first*, so an unknown tool
    // is refused before any later extension can form an opinion about it.
    ...(opts.readOnly === true ? [readOnlySession()] : []),
    // Workers only. The orchestrator runs unconfined by operator ruling
    // (#143): the gate could only ever be installed in sessions this daemon
    // spawns, so an external orchestrator — the supported shape — never had it,
    // and a boundary that exists in one deployment out of two is not a
    // boundary. What holds the orchestrator is its brief and the verb ledger.
    ...(opts.role === "worker" ? [worktreeConfinement(opts.cwd)] : []),
    ...(opts.releaseGrants === undefined
      ? []
      : [releasePolicyTripwire(opts.releaseGrants, opts.role, opts.onReleaseBlocked)]),
    // The sanctioned way back to GitHub (#126). A thin client: it forwards
    // arguments over the run's own socket and renders what the daemon decided.
    // Registered unconditionally, including on a session with no socket, so a
    // missing channel is a named refusal rather than an absent tool the model
    // routes around with bash.
    conductorVerbs(opts.verbSocketPath),
  ];

  // The fleet-owned omp settings overlay (#537): when the daemon staged one,
  // layer it into a Settings instance over the ordinary global/project
  // discovery. `Settings.init({ cwd, configFiles })` keeps the daemon account's
  // global config (providers, approvals, modelRoles) and any project
  // `.omp/config.yml` intact — the overlay is deep-merged *after* them — and
  // never `Settings.isolated()`, which would build from the overlay alone and
  // drop everything out from under the worker. Absent a staged overlay, no
  // settings are passed and the harness discovers exactly as it does today.
  let overlaySettings: unknown = undefined;
  if (opts.ompSettingsFile !== undefined) {
    const Settings = Reflect.get(namespace, "Settings");
    if (typeof Settings === "function" && typeof Settings.init === "function") {
      overlaySettings = await Settings.init({
        cwd: opts.cwd,
        configFiles: [opts.ompSettingsFile],
      });
    }
  }

  const created = await mod.createAgentSession({
    cwd: opts.cwd,
    // A raw pattern rather than a resolved Model: the harness resolves it
    // after extensions load, so we never have to import its model registry.
    ...(opts.model === undefined ? {} : { modelPattern: opts.model }),
    ...(overlaySettings === undefined ? {} : { settings: overlaySettings }),
    sessionManager,
    // A private registry per session, never the process-global default: that
    // one admits only one "Main" identity per generation, so a second session
    // sharing it fails to start. The daemon runs `maxConcurrentWorkers`
    // (2 by default) workers at once, which makes this the normal path.
    agentRegistry: new mod.AgentRegistry(),
    // Default true in the SDK, but pass it explicitly so a harness change
    // cannot silently strip MCP from workers. Discovery walks cwd + user
    // agentDir (~/.omp/agent/mcp.json) — without this, workers grep-only and
    // burn the turns cap on discovery (#29).
    enableMCP: true,
    // The structured-settlement contract (#540): subagent-oriented options
    // threaded to the harness so the worker can yield its report as a schema-
    // validated object instead of prose. Absent for every other session kind —
    // probes and orchestrator sessions never yield.
    ...(opts.outputSchema === undefined ? {} : { outputSchema: opts.outputSchema }),
    ...(opts.outputSchemaMode === undefined ? {} : { outputSchemaMode: opts.outputSchemaMode }),
    ...(opts.requireYieldTool === undefined ? {} : { requireYieldTool: opts.requireYieldTool }),
    // Mechanical tool gates: confinement prevents structured worktree I/O from
    // escaping cwd; release policy blocks release/deploy calls when configured.
    ...(extensions.length === 0 ? {} : { extensions }),
  });
  const raw = asRawSession(created);
  // Surfaced rather than swallowed: this is how a quiet downgrade to a weaker
  // model reaches the daemon's log instead of only the operator's surprise.
  const fallback =
    created !== null && typeof created === "object"
      ? Reflect.get(created, "modelFallbackMessage")
      : undefined;
  const modelFallbackMessage =
    typeof fallback === "string" && fallback !== "" ? fallback : undefined;

  // The code-graph session observation (#726): whether the graph MCP tools are
  // in this session's registry. The harness tool registry is structural and
  // settles late — MCP wiring finalises through a deferred discovery pass after
  // `createAgentSession` resolves, so the read can throw in the creation
  // window, and the graph tools surface under the *enabled* names, never the
  // active ones. `observeGraphTools` polls that surface until it stops
  // throwing, bounded; an absent surface (older harness build) or an
  // unreadable one records no observation — a registry answer is never
  // guessed. The observation resolves asynchronously so the session handshake
  // never waits for it.
  const sessionTools = raw as { getEnabledToolNames?: () => string[] };
  const graphToolsReady = observeGraphTools(sessionTools.getEnabledToolNames?.bind(sessionTools));

  // One real subscription fanned out per event type, so N `on()` calls cost one
  // listener on the harness stream and unknown event types cost nothing.
  const handlers = new Map<string, ((e: unknown) => void)[]>();
  raw.subscribe((event) => {
    const type = (event as { type?: unknown } | null | undefined)?.type;
    if (typeof type !== "string") return;
    for (const cb of handlers.get(type) ?? []) cb(event);
    for (const cb of handlers.get("*") ?? []) cb(event);
  });

  const loadedHarnessVersion = harnessVersion(harnessPath);
  const session: AgentSessionLike = {
    prompt: (text, promptOpts) => raw.prompt(text, promptOpts),
    on(event, cb) {
      const list = handlers.get(event);
      if (list) list.push(cb);
      else handlers.set(event, [cb]);
    },
    abort() {
      raw.abort();
    },
    park() {
      return Promise.resolve(
        raw.abort({ goalReason: "interrupted", reason: "operator pause" }),
      ).then(() => undefined);
    },
    // The loaded-module attestation (#832): this session process executes the
    // conductor module it resolved at start; the version is a property of this
    // process's loaded tree, deliberately not of what is on disk now.
    ...(conductorModuleVersion() === undefined ? {} : { extensionVersion: conductorModuleVersion() }),
    harnessPath,
    ...(loadedHarnessVersion === undefined ? {} : { harnessVersion: loadedHarnessVersion }),
    // The path the session actually opened, never one we asked for: the
    // arm/monitor tooling reads this file as proof of activity, so a path
    // nothing ever writes to is worse than no path at all.
    get sessionFile() {
      return raw.sessionFile;
    },
    ...(modelFallbackMessage === undefined ? {} : { modelFallbackMessage }),
    graphToolsReady,
  };

  disposers.set(session, async () => {
    await raw.dispose?.();
  });
  return session;
}

/**
 * Release a session's harness-side resources. Safe to call on any
 * {@link AgentSessionLike} — a hand-rolled fake, or a harness build with no
 * `dispose()`, is a no-op rather than a crash during teardown.
 */
export async function disposeSession(session: AgentSessionLike): Promise<void> {
  await disposers.get(session)?.();
  disposers.delete(session);
}

/**
 * Pick the session manager for one `createSession` call.
 *
 * On the resume path the harness's own `continueRecent` decides whether there
 * is anything to continue, so this never has to list or stat transcripts
 * itself. `create` is the fallback for two cases: a harness build predating
 * `continueRecent`, and a resume that failed outright.
 */
async function openSessionManager(
  mod: OmpModule,
  opts: { cwd: string; sessionDir?: string; resume?: boolean },
): Promise<unknown> {
  const fresh = (): unknown =>
    opts.sessionDir === undefined
      ? mod.SessionManager.create(opts.cwd)
      : mod.SessionManager.create(opts.cwd, opts.sessionDir);

  if (opts.resume !== true || typeof mod.SessionManager.continueRecent !== "function") return fresh();
  try {
    const resumed = await mod.SessionManager.continueRecent(opts.cwd, opts.sessionDir);
    if (resumed !== null && typeof resumed === "object") return resumed;
  } catch {
    // ponytail: an unreadable newest transcript costs this session its memory
    // rather than blocking startup — a conductor that will not boot because an
    // old .jsonl is corrupt is worse than one that starts forgetful. Upgrade
    // path: walk `SessionManager.list(cwd)` for the newest readable transcript.
  }
  return fresh();
}

/**
 * Unwrap whatever `createAgentSession` handed back. It returns
 * `{ session, … }` today; a bare session is accepted too so a minor SDK change
 * in either direction fails loudly here instead of as `undefined is not a
 * function` three events later.
 */
function asRawSession(created: unknown): RawSession {
  const wrapper = created as { session?: unknown } | null | undefined;
  const candidate = (
    wrapper && typeof wrapper === "object" && "session" in wrapper ? wrapper.session : created
  ) as RawSession | null | undefined;
  if (
    !candidate ||
    typeof candidate.prompt !== "function" ||
    typeof candidate.subscribe !== "function" ||
    typeof candidate.abort !== "function"
  ) {
    throw new Error(
      `${OMP_PACKAGE} returned an unrecognised session from createAgentSession(); omp-conductor needs { prompt, subscribe, abort }.`,
    );
  }
  return candidate;
}


// --------------------------------------------------------- the session proxy

/** Absolute path of the child entrypoint, resolved against this package. */
export const SESSION_HOST = join(import.meta.dir, "session-host.ts");

/**
 * The version of the omp-conductor package this *process* loaded — the module
 * tree this process's session machinery actually executes, read from the
 * package.json next to `import.meta.dir` (#832). Read once per process and
 * cached: the upgrade's session-reload proof compares this against the target
 * release to tell a genuinely reloaded OMP process from a package install the
 * live session never picked up. `undefined` when the file cannot be read —
 * absence is a missing attestation, never "the newest".
 */
let loadedConductorVersion: string | undefined;
export function conductorModuleVersion(): string | undefined {
  if (loadedConductorVersion !== undefined) return loadedConductorVersion;
  try {
    const raw = readFileSync(join(import.meta.dir, "..", "package.json"), "utf8");
    const parsed = JSON.parse(raw) as { version?: unknown };
    loadedConductorVersion = typeof parsed.version === "string" ? parsed.version : undefined;
  } catch {
    loadedConductorVersion = undefined;
  }
  return loadedConductorVersion;
}

/**
 * How long the parent waits for a child to connect and report `ready`.
 *
 * Generous on purpose: the child loads the harness, discovers MCP servers and
 * opens a transcript before it can answer, and on a cold cache that is seconds
 * rather than milliseconds. The timeout is not a performance guard, it is the
 * difference between "the launcher was rejected" surfacing as a named error and
 * surfacing as a dispatch that never returns.
 */
const START_TIMEOUT_MS = 180_000;

/** Bounded tail of the child's stderr, so a crash is legible without unbounded buffering. */
const STDERR_TAIL = 8_000;

/**
 * How long a dying child's pipes get to finish before its failure is reported
 * without them. Generous because it only ever delays an already-failed startup,
 * and the streams normally reach EOF the instant the process does.
 */
const DRAIN_GRACE_MS = 2_000;

/** How long a disposed child gets to exit before it is signalled. */
const DISPOSE_GRACE_MS = 5_000;

/**
 * Thrown by {@link createSession} when the pre-spawn admission gate closes: a
 * daemon stop landed while the session socket was binding, so the child was
 * never spawned. The caller maps this to a stopped run, not a failed one —
 * a shutdown must not charge an attempt against a worker it refused to start
 * (#374).
 */
export class SessionAdmissionClosedError extends Error {}

export interface CreateSessionOptions {
  cwd: string;
  sessionDir?: string;
  model?: string;
  resume?: boolean;
  role: SessionRole;
  releaseGrants?: ResolvedGrants;
  onReleaseBlocked?: (shape: GateShape, context: ReleaseBlockContext) => void;
  /**
   * Absolute path to the fleet-owned omp settings overlay (#537), forwarded to
   * the child through the host spec so the far side's `createLocalSession`
   * loads it via `Settings.init({ configFiles: [<path>] })`. Absent, nothing
   * is staged and the session discovers settings as it does today.
   */
  ompSettingsFile?: string;

  /**
   * Where the control socket is bound. The daemon puts it beside the run's own
   * session directory; omitted, a private temp directory is used, which is what
   * the tests want.
   */
  socketPath?: string;
  /**
   * The conductor verb socket this session's mutation tools call (#126).
   *
   * A different socket from `socketPath` above, and deliberately so: that one
   * is the parent's own control channel to the child, this one is the child's
   * only route back to a mutation, and collapsing them would make the daemon's
   * session plumbing an attack surface for verbs.
   */
  verbSocketPath?: string;
  /**
   * The child's pid, the instant it exists.
   *
   * Called synchronously after `Bun.spawn` and before the child could have
   * connected to anything, because that ordering is the point: the daemon binds
   * its verb channel to this pid, and a channel accepts nothing until it is bound.
   */
  onSpawn?: (pid: number) => void;
  /**
   * The session's Herdr representation (#840), opened from the exact pid the
   * spawn below reports and released when the child is gone.
   *
   * Injected rather than imported so this module keeps knowing nothing about
   * Herdr: `createSession` owns the process, and whoever launches a *worker*
   * owns whether that process should be visible in an operator's workspace. A
   * caller that passes nothing gets today's behaviour exactly.
   *
   * `open` is called with the same pid, at the same instant, as
   * {@link CreateSessionOptions.onSpawn} — the representation is of the
   * authoritative child or of nothing. It must never throw and never block the
   * launch: visibility is not a precondition for work.
   */
  pane?: {
    open(pid: number): void;
    release(): void;
  };
  /**
   * Pre-spawn admission gate (#374). Consulted once, immediately after the
   * socket bind await and immediately before `Bun.spawn` — the last window a
   * daemon stop can land in before a child exists. A gate that returns false
   * closes the listener, removes the socket, and throws
   * {@link SessionAdmissionClosedError} instead of spawning; the run then
   * settles as stopped rather than adding a worker the shutdown would have to
   * wait for. Absent, the spawn always proceeds.
   */
  maySpawn?: () => boolean;
  /** Child stderr, line by line. Defaults to the process's own stderr. */
  onChildLog?: (line: string) => void;
  startupTimeoutMs?: number;
  /**
   * Test seam, and only that — the same shape as
   * `OrchestratorOpts.createSessionImpl`. Production always wants
   * `session-host.ts`; the parity suite points this at a child that calls
   * {@link runSessionHost} with a hand-written session, so the proxy, the
   * socket, the framing and the process lifecycle are all exercised for real
   * without a live harness or a model bill.
   */
  hostModule?: string;
  /**
   * Deny every tool but reading and searching (#307).
   *
   * Forwarded to the child, because that is where the session — and therefore the
   * gate — actually lives.
   */
  readOnly?: boolean;
  /**
   * The structured-settlement contract (#540): a JSON Schema the session's
   * `yield` tool validates its `data` payload against. Forwarded to the child
   * through the host spec, exactly like {@link ompSettingsFile}.
   */
  outputSchema?: unknown;
  /**
   * Enforcement policy for {@link outputSchema} — the worker contract is
   * `"permissive"`, so a schema violation settles with whatever the worker
   * produced instead of losing the report.
   */
  outputSchemaMode?: "permissive" | "strict";
  /** Force the hidden `yield` tool into this session's toolset (#540). */
  requireYieldTool?: boolean;
}

/**
 * Start one omp coding session in a **child process** and return a proxy for it.
 *
 * The proxy is the whole of `omp-conductor`'s view of a session, and it is
 * deliberately thin: {@link AgentSessionLike} is the entire forwarded
 * surface. `prompt`, `park`, and `abort` go out over a unix socket, harness
 * events come back and are re-emitted to the same `on()` subscribers the
 * in-process version served, and `sessionFile` is
 * whatever path the child reports the session actually opened — never one this
 * side invented.
 *
 * Two behaviours that were free in-process and are bought explicitly here:
 *
 * - **A child outliving its parent.** The session used to die with the daemon
 *   because it *was* the daemon. Now the child watches its socket and exits
 *   when the parent goes away, so a daemon crash still leaves no orphaned
 *   session holding a checkout and a model budget.
 * - **A wedged harness.** An in-process `abort()` that the harness ignored hung
 *   the daemon. Disposal here escalates: dispose message, then `SIGTERM`, then
 *   `SIGKILL`. The cap-kill path therefore always terminates.
 *
 * @throws if the child cannot be launched or the session cannot be built,
 * carrying the child's own message — a missing peer dependency still reads as
 * the deployment mistake it is, not as an unexplained exit code.
 */
export async function createSession(opts: CreateSessionOptions): Promise<AgentSessionLike> {
  const owned = opts.socketPath === undefined;
  // 0711, not 0700: searchable but not listable, which keeps *other local
  // accounts* from enumerating this directory. It is not a restraint on the
  // session — that runs as this process's own uid, matches the owner class and
  // therefore has `rwx` here regardless. Same reasoning, and the same limit, as
  // the run workspace parent.
  const socketDir = owned ? mkdtempSync(join(tmpdir(), "omp-session-")) : dirname(opts.socketPath ?? "");
  const socketPath = opts.socketPath ?? join(socketDir, "s");
  mkdirSync(socketDir, { recursive: true, mode: 0o711 });
  chmodSync(socketDir, 0o711);
  rmSync(socketPath, { force: true });

  const server = createServer();
  const { promise: attached, resolve: onAttach, reject: onAttachFail } =
    Promise.withResolvers<Socket>();
  server.once("connection", (socket: Socket) => {
    socket.setNoDelay(true);
    onAttach(socket);
  });
  server.once("error", (err: Error) => {
    onAttachFail(err);
  });
  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(socketPath, () => {
      resolve();
    });
  });
  // Pre-spawn admission recheck (#374): SIGTERM/SIGINT can land during the
  // listen await above — after the daemon's own pre-launch checks and before
  // this child exists. A closed gate means no child: close the listener,
  // remove the socket (and, when this side created the directory, the
  // directory itself — the same cleanup the failure paths below run), and
  // signal the caller to settle the run as stopped instead of spawning a
  // worker the shutdown would then have to wait for.
  if (opts.maySpawn?.() === false) {
    server.close();
    rmSync(socketPath, { force: true });
    if (owned) rmSync(socketDir, { recursive: true, force: true });
    throw new SessionAdmissionClosedError(
      "daemon shutdown began while the session socket was binding",
    );
  }
  // The socket is the run's own channel, and this process's own uid — the
  // fleet account every session runs as (#894) — is the only one that speaks
  // on it.
  chmodSync(socketPath, 0o600);

  const spec: SessionHostSpec = {
    socket: socketPath,
    cwd: opts.cwd,
    role: opts.role,
    ...(opts.sessionDir === undefined ? {} : { sessionDir: opts.sessionDir }),
    ...(opts.model === undefined ? {} : { model: opts.model }),
    ...(opts.resume === undefined ? {} : { resume: opts.resume }),
    ...(opts.releaseGrants === undefined ? {} : { releaseGrants: opts.releaseGrants }),
    ...(opts.verbSocketPath === undefined ? {} : { verbSocketPath: opts.verbSocketPath }),
    ...(opts.readOnly === undefined ? {} : { readOnly: opts.readOnly }),
    ...(opts.ompSettingsFile === undefined ? {} : { ompSettingsFile: opts.ompSettingsFile }),
    ...(opts.outputSchema === undefined ? {} : { outputSchema: opts.outputSchema }),
    ...(opts.outputSchemaMode === undefined ? {} : { outputSchemaMode: opts.outputSchemaMode }),
    ...(opts.requireYieldTool === undefined ? {} : { requireYieldTool: opts.requireYieldTool }),
  };

  const log = opts.onChildLog ?? ((line: string) => process.stderr.write(`${line}\n`));
  const hostModule = opts.hostModule ?? SESSION_HOST;
  // The child is the same installed tree this process runs from, launched
  // under the fleet account (#894). `--no-install` keeps Bun off its ambient
  // auto-install cache, so the peer and its native addon resolve from the
  // operator's exact install or fail loudly.
  const argv = [process.execPath, "--no-install", hostModule, JSON.stringify(spec)];
  // The session's own role is stamped on the child, so a process the agent
  // runs — `omp-conductor report` from its sandbox — can tell a worker session
  // from the operator's shell. Direct CLI runs outside a spawned session
  // inherit nothing and stay the orchestrator surface.
  let env: Record<string, string | undefined> | undefined;
  if (opts.role !== undefined) {
    env = { ...process.env, [SESSION_ROLE_ENV]: opts.role };
  }
  const child = Bun.spawn(argv, {
    cwd: opts.cwd,
    env,
    stdin: "ignore",
    // The harness writes progress to stdout; both streams are the daemon's log,
    // never the protocol. The protocol has its own socket precisely so a chatty
    // harness build cannot corrupt it.
    stdout: "pipe",
    stderr: "pipe",
  });

  // Synchronously, before a single await: the child cannot have connected yet, so
  // the caller's channel is bound before it can be reached.
  if (child.pid !== undefined) opts.onSpawn?.(child.pid);
  // The workspace representation is of THIS pid or of nothing (#840). After the
  // verb-channel bind above, because a pane is a convenience and the bind is a
  // correctness boundary — and never before the child exists, which is what
  // makes "bound to the exact session-host pid" true rather than nominal.
  if (child.pid !== undefined) opts.pane?.open(child.pid);

  let stderrTail = "";
  const drain = async (stream: ReadableStream<Uint8Array> | undefined, prefix: string): Promise<void> => {
    if (stream === undefined) return;
    const decoder = new TextDecoder();
    let rest = "";
    for await (const chunk of stream) {
      rest += decoder.decode(chunk, { stream: true });
      const lines = rest.split("\n");
      rest = lines.pop() ?? "";
      for (const line of lines) {
        if (line === "") continue;
        stderrTail = `${stderrTail}${line}\n`.slice(-STDERR_TAIL);
        log(`${prefix}${line}`);
      }
    }
  };
  // Kept, not discarded: the child's own stderr is the whole value of the startup
  // failure path, and it is a *race* against `child.exited` rather than a
  // guarantee. A child that dies before connecting resolves `exited` as soon as
  // the process is gone, which can be before these loops have read what it
  // printed on the way out — and then the operator gets "exited 1" instead of the
  // reason, which is precisely the diagnosis this path exists to give. Observed
  // failing on ubuntu and passing on darwin, which is what a race looks like.
  const drained = Promise.all([drain(child.stdout, "session: "), drain(child.stderr, "session: ")]);
  drained.catch(() => undefined);

  /**
   * The child's output, once the pipes have actually finished — the whole point of
   * every failure path below.
   *
   * Bounded rather than awaited outright: a grandchild that inherited stderr keeps
   * the pipe open after the child is gone, and a failure nobody reports is worse
   * than one reported without its tail. Both exit routes go through here so they
   * cannot drift apart again; the pre-connect one did, which is how a release run
   * caught this.
   */
  const settledTail = async (): Promise<string> => {
    await Promise.race([
      drained,
      new Promise<void>((resolve) => {
        setTimeout(resolve, DRAIN_GRACE_MS).unref?.();
      }),
    ]);
    return stderrTail.trim();
  };

  /**
   * The child's own start failure, if it got one out over the socket before it
   * died — the socket counterpart of the pipe drain.
   *
   * A child that cannot start can connect and write `start-error` in the same
   * event-loop turn its exit is dispatched in: the connect already completed
   * in the kernel, the accept is still queued, and closing the listener in
   * that window drops the frame with the reason. So every exit route waits
   * this out BEFORE {@link cleanup} closes the server, then prefers the frame
   * over a bare exit code. Bounded like the pipes: a dead child that never
   * connected has nothing on the wire, and 2s only ever delays an
   * already-failed startup.
   *
   * Memoized: the pre-connect catch and the exit handler both arrive here, and
   * both must read the SAME socket — two independent readers would race on the
   * first `destroy()` and one of them would lose the frame.
   */
  let startErrorReading: Promise<string | undefined> | undefined;
  const startErrorFromSocket = (): Promise<string | undefined> => {
    startErrorReading ??= (async (): Promise<string | undefined> => {
      const socket = await Promise.race([
        attached.catch(() => undefined),
        new Promise<undefined>((resolve) => {
          setTimeout(() => resolve(undefined), DRAIN_GRACE_MS).unref?.();
        }),
      ]);
      if (socket === undefined) return undefined;
      try {
        return await new Promise<string | undefined>((resolve) => {
          let buffer = "";
          let settled = false;
          const finish = (message: string | undefined): void => {
            if (settled) return;
            settled = true;
            clearTimeout(timer);
            resolve(message);
          };
          const timer = setTimeout(() => finish(undefined), DRAIN_GRACE_MS);
          timer.unref?.();
          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 { t?: unknown; message?: unknown };
              if (message.t === "start-error" && typeof message.message === "string") {
                finish(message.message);
                return;
              }
            }
          });
          socket.once("error", () => finish(undefined));
          socket.once("close", () => finish(undefined));
        });
      } finally {
        socket.destroy();
      }
    })();
    return startErrorReading;
  };

  const cleanup = (): void => {
    server.close();
    rmSync(socketPath, { force: true });
    if (owned) rmSync(socketDir, { recursive: true, force: true });
  };

  const handlers = new Map<string, ((e: unknown) => void)[]>();
  const pending = new Map<number, { resolve: () => void; reject: (err: Error) => void }>();
  let sessionFile: string | undefined;
  let modelFallbackMessage: string | undefined;
  let graphTools: GraphToolsObservation | undefined;
  let extensionVersion: string | undefined;
  let loadedHarnessPath: string | undefined;
  let loadedHarnessVersion: string | undefined;
  let promptSeq = 0;

  const { promise: ready, resolve: onReady, reject: onReadyFail } = Promise.withResolvers<void>();
  // Parked immediately. `ready` is rejected from the child's exit handler,
  // which can fire before anything awaits it — and an unhandled rejection in a
  // daemon that supervises sessions is a process-level crash, not a failed run.
  // The real awaiters below still see the rejection.
  ready.catch(() => undefined);
  const { promise: exited, resolve: onExit } = Promise.withResolvers<void>();
  /** Set once teardown has begun, so a clean exit is not reported as a crash. */
  let disposing = false;

  const fail = (message: string): void => {
    const tail = stderrTail.trim();
    const err = new Error(tail === "" ? message : `${message}\nchild output:\n${tail}`);
    onReadyFail(err);
    for (const [, waiter] of pending) waiter.reject(err);
    pending.clear();
  };

  // The one terminal event this proxy owns outright: the session process is
  // gone, whether by crash, kill, or a clean dispose of its own. Delivered to
  // subscribers *before* any failure is surfaced, so a supervisor learning of
  // the death can also read its exit code off the same event. `disposing` is
  // the proxy's own intent (its disposer asked the child to stop), not the
  // subscriber's, so the event is emitted in both branches and each observer
  // decides what an exit means from its own state.
  const emitSessionExit = (code: number | null): void => {
    const type = "session_exit";
    const event = { type, code };
    for (const cb of handlers.get(type) ?? []) cb(event);
    for (const cb of handlers.get("*") ?? []) cb(event);
  };

  void child.exited.then(async (code) => {
    onExit();
    const exitCode = typeof code === "number" ? code : null;
    emitSessionExit(exitCode);
    // The pid this pane spoke for is gone, so conductor stops speaking for it
    // (#840) — on every exit path, crash and clean dispose alike, because a
    // representation that outlives its process is a worker the workspace still
    // shows as live. What happens to the pane itself is #841's policy; this only
    // gives back the authority.
    opts.pane?.release();
    // A child that cannot start writes `start-error` over the socket, not the
    // pipes — and its exit can be dispatched before the accept of a connection
    // that already completed in the kernel. Let the socket settle before the
    // listener closes, so the child's own words survive the ordering.
    if (disposing) {
      cleanup();
      return;
    }
    const startError = await startErrorFromSocket();
    cleanup();
    // The tail first, so the message carries the child's own words.
    await settledTail();
    fail(
      startError ?? `omp-conductor session child exited ${String(code)} before the session ended`,
    );
  });

  let socket: Socket;
  try {
    socket = await Promise.race([
      attached,
      new Promise<never>((_resolve, reject) => {
        setTimeout(() => {
          reject(
            new Error(
              `omp-conductor session child did not connect within ${String(
                (opts.startupTimeoutMs ?? START_TIMEOUT_MS) / 1000,
              )}s`,
            ),
          );
        }, opts.startupTimeoutMs ?? START_TIMEOUT_MS).unref?.();
      }),
      child.exited.then((code): never => {
        throw new Error(`omp-conductor session child exited ${String(code)} before connecting`);
      }),
    ]);
  } catch (err) {
    child.kill("SIGKILL");
    // The child's exit can win the race above against the accept of a
    // connection that completed in the kernel — then its `start-error` frame
    // is still queued, and closing the listener now would throw it away.
    // Read the socket first (bounded, like the pipes), and prefer its words
    // over a bare exit code.
    const startError = await startErrorFromSocket();
    cleanup();
    // Killed first, so the pipes are already closing, and only then read. This is
    // the route a child that dies before connecting takes — a missing peer
    // dependency, say — and reading `stderrTail` synchronously here raced the
    // drain loops and reported a bare exit code instead of the reason.
    const tail = await settledTail();
    const reason = startError ?? (err instanceof Error ? err.message : String(err));
    throw new Error(`${reason}${tail === "" ? "" : `\nchild output:\n${tail}`}`);
  }

  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 HostToParent;
      switch (message.t) {
        case "ready":
          sessionFile = message.sessionFile;
          modelFallbackMessage = message.modelFallbackMessage;
          extensionVersion = message.extensionVersion;
          loadedHarnessPath = message.harnessPath;
          loadedHarnessVersion = message.harnessVersion;
          onReady();
          break;
        case "graph-tools":
          graphTools = message.graphTools;
          break;
        case "start-error":
          fail(message.message);
          break;
        case "session-file":
          sessionFile = message.path;
          break;
        case "event": {
          const type = (message.event as { type?: unknown } | null | undefined)?.type;
          if (typeof type !== "string") break;
          for (const cb of handlers.get(type) ?? []) cb(message.event);
          for (const cb of handlers.get("*") ?? []) cb(message.event);
          break;
        }
        case "prompt-result": {
          const waiter = pending.get(message.id);
          if (waiter === undefined) break;
          pending.delete(message.id);
          if (message.ok) waiter.resolve();
          else waiter.reject(new Error(message.error ?? "session prompt failed"));
          break;
        }
        case "park-result": {
          const waiter = pending.get(message.id);
          if (waiter === undefined) break;
          pending.delete(message.id);
          if (message.ok) waiter.resolve();
          else waiter.reject(new Error(message.error ?? "session park failed"));
          break;
        }
        case "release-blocked":
          opts.onReleaseBlocked?.(message.shape, {
            tool: message.tool,
            reason: message.reason,
            args: message.args,
            ...(message.matched === undefined ? {} : { matched: message.matched }),
            ...(message.invocation === undefined ? {} : { invocation: message.invocation }),
          });
          break;
      }
    }
  });

  const write = (message: ParentToHost): void => {
    if (!socket.writableEnded) socket.write(encodeFrame(message));
  };

  await Promise.race([
    ready,
    new Promise<never>((_resolve, reject) => {
      setTimeout(() => {
        reject(new Error("omp-conductor session child connected but never reported a session"));
      }, opts.startupTimeoutMs ?? START_TIMEOUT_MS).unref?.();
    }),
  ]).catch((err: unknown) => {
    child.kill("SIGKILL");
    cleanup();
    throw err;
  });

  const session: AgentSessionLike = {
    prompt(text, promptOpts) {
      promptSeq += 1;
      const id = promptSeq;
      const { promise, resolve, reject } = Promise.withResolvers<void>();
      pending.set(id, { resolve, reject });
      write({ t: "prompt", id, text, ...(promptOpts === undefined ? {} : { opts: promptOpts }) });
      // Resolves to `undefined`: no caller in this package reads a prompt's
      // return value, and shipping the harness's own object across the wire
      // would make this seam depend on a shape it deliberately does not name.
      return promise.then(() => undefined);
    },
    park() {
      promptSeq += 1;
      const id = promptSeq;
      const { promise, resolve, reject } = Promise.withResolvers<void>();
      pending.set(id, { resolve, reject });
      write({ t: "park", id });
      return promise.then(() => undefined);
    },
    on(event, cb) {
      const list = handlers.get(event);
      if (list) list.push(cb);
      else handlers.set(event, [cb]);
    },
    abort() {
      write({ t: "abort" });
    },
    get sessionFile() {
      return sessionFile;
    },
    get modelFallbackMessage() {
      return modelFallbackMessage;
    },
    get graphTools() {
      return graphTools;
    },
    get extensionVersion() {
      return extensionVersion;
    },
    get harnessPath() {
      return loadedHarnessPath;
    },
    get harnessVersion() {
      return loadedHarnessVersion;
    },
  };

  disposers.set(session, async () => {
    disposing = true;
    write({ t: "dispose" });
    socket.end();
    const settled = await Promise.race([
      exited.then(() => true),
      new Promise<false>((resolve) => {
        setTimeout(() => {
          resolve(false);
        }, DISPOSE_GRACE_MS).unref?.();
      }),
    ]);
    if (!settled) {
      // The escalation that makes a cap-kill final. In-process, a harness that
      // ignored `abort()` hung the daemon; here it costs one signal.
      child.kill("SIGTERM");
      const stopped = await Promise.race([
        exited.then(() => true),
        new Promise<false>((resolve) => {
          setTimeout(() => {
            resolve(false);
          }, DISPOSE_GRACE_MS).unref?.();
        }),
      ]);
      if (!stopped) child.kill("SIGKILL");
    }
    cleanup();
  });

  return session;
}