/**
 * Hook Executor
 *
 * Executes hook scripts in a controlled environment with:
 * - Timeout management (total + idle)
 * - Input validation
 * - Output validation
 * - Structured logging
 *
 * **Execution model:** a hook script is a program celilo RUNS, not a library
 * celilo loads. `executeHookScript` spawns `bun hook-runner.ts <script>` and
 * becomes a broker: it keeps the database, the master key and the live
 * capability objects, and answers the child over a versioned NDJSON protocol
 * on a Unix socket (`hook-protocol.ts`, `broker.ts`). See
 * `openspec/changes/hook-process-boundary/design.md`.
 *
 * Three things follow, and they are the point:
 *
 *   - The child's environment is an ALLOW-LIST, not `...process.env`. celilo's
 *     own paths and the operator's tokens stop being visible (design D5).
 *   - A timeout is a real `proc.kill()`. The old implementation raced the
 *     hook's promise against a timer and cancelled nothing, so a hook that
 *     "timed out" kept its capability objects and went on registering DNS
 *     records after celilo reported the deploy failed (celilo#1003).
 *   - A hook's memory is its own process's, not celilo's heap.
 *
 * **Stage 2 has landed and the child is JAILED where a backend exists.** The
 * filesystem view is derived per run (`mount-set.ts`, design D9) and built with
 * bubblewrap (`jail.ts`, design D8). A jailed hook reaching for celilo's master
 * key gets `ENOENT` — the path is not denied, it is absent. Where no backend
 * exists (a Mac today, task 4.8) the hook runs unjailed and the mode is
 * RECORDED, so a host that quietly stops jailing is visible rather than silent.
 *
 * Hooks still do NOT execute on the target machine — a remote operation is
 * initiated from the celilo host. **Stage 3 has landed and scopes that
 * reachability by the credential (design D12):** `~/.ssh` is not bound into
 * the jail, so a hand-built `ssh` cannot authenticate, and the remote
 * primitives cross to the remote-ops broker (`remote-broker.ts`), which holds
 * the key and checks every target against the policy in
 * `services/remote-access.ts` — the module's own systems, or the module's own
 * credential.
 *
 * Execution function (Rule 10.1) - performs side effects (script execution)
 */

import { existsSync, mkdirSync, readdirSync, rmdirSync, statSync } from 'node:fs';
import { dirname, join, resolve } from 'node:path';
import {
  type DeployedSystem,
  isMissingProviderInputError,
  moduleArtifactDir,
  moduleStateDir,
} from '@celilo/capabilities';
import { getDbPath } from '../config/paths';
import { getDb } from '../db/client';
import {
  type ContractHookSignature,
  contractHookSignature,
  resolveContract,
  supportedContractVersions,
} from '../manifest/contracts';
import { isPrivilegedCapability } from '../manifest/validate';
import { pruneModuleArtifacts } from './artifact-retention';
import { startBroker } from './broker';
import type { HookStoresProvider } from './broker';
import {
  HOOK_MOUNT_SET_ENV,
  HOOK_PROTOCOL_VERSION,
  HOOK_PROTOCOL_VERSION_ENV,
  HOOK_REMOTE_SOCKET_ENV,
  HOOK_SOCKET_ENV,
  createLineReader,
  deserializeError,
} from './hook-protocol';
import {
  type JailPlan,
  detectJailBackend,
  planJailedSpawn,
  realpathRequest,
  recordJailMode,
  resolveJailPolicy,
  runtimeModulePathsFor,
} from './jail';
import {
  type DeclaredPathInput,
  type MountSet,
  type MountSetRequest,
  deriveMountSet,
} from './mount-set';
import { type RemoteAccessPolicy, startRemoteBroker } from './remote-broker';
import type { HookContext, HookDefinition, HookLogger, HookResult } from './types';
import { mountSetEnvValue } from './unjailed-lint';

/** Default total timeout: 60 seconds */
const DEFAULT_TIMEOUT_MS = 60_000;

/**
 * Default idle timeout: 30 seconds since last log message.
 *
 * This is a *heuristic* for hooks that declared no bound of their own — it
 * catches a wedged hook faster than the total timeout would. It is NOT a
 * contract, and it must not silently override one: when a hook or a
 * subscription declares an explicit timeout, that declaration replaces this
 * heuristic entirely (see `invokeHook`).
 */
const IDLE_TIMEOUT_MS = 30_000;

/** How often the idle tracker looks. Unchanged from the promise-based timer. */
const IDLE_POLL_MS = 5_000;

/**
 * SIGTERM is a request. A hook that traps it and keeps working would leave the
 * boundary buying nothing over the promise race it replaced, so SIGKILL
 * follows. Short, because by this point celilo has already told the operator
 * the hook failed.
 */
const SIGKILL_GRACE_MS = 2_000;

/**
 * Environment variables a hook inherits. Everything else is dropped (D5).
 *
 * Two kinds, and the distinction is the whole rule.
 *
 * **How to run** — `PATH`, `HOME`, `LANG`, `TZ`, `TMPDIR`. A hook that spawns
 * anything needs these, and withholding them hides no path.
 *
 * **How to reach the network, and whom to trust** — the CA and proxy families.
 * These configure the RUNTIME, not the module: they say which certificate
 * authorities this host trusts and how it reaches the internet. They name no
 * secret, they are not module-specific, and withholding them reduces a hook's
 * authority by nothing at all. It only makes its TLS connections wrong.
 *
 * `NODE_EXTRA_CA_CERTS` is why this second group exists rather than being an
 * afterthought. `packages/e2e/docker/Dockerfile.management:107` sets it because
 * `bun fetch()` otherwise ignores `/etc/ssl/certs/ca-certificates.crt`
 * entirely and rejects every certificate the simulated CA issues. Twelve
 * hook-reachable module files make HTTPS requests. Dropping it would have
 * broken all of them, in the rig only, as a TLS error naming no cause.
 *
 * The proxy family is not set anywhere today. It is here because it is the
 * same kind of thing, it costs nothing when absent, and the alternative is
 * discovering it the way `NODE_EXTRA_CA_CERTS` was discovered.
 *
 * **`TF_CLI_CONFIG_FILE` is deliberately NOT here.** It is toolchain
 * configuration too, but no hook invokes the terraform binary — every mention
 * of terraform in a module script is a `generated/terraform/` PATH. celilo runs
 * terraform from its own process, which keeps its own environment. Recorded so
 * the next reader does not have to re-derive it.
 *
 * What is NOT in either group: anything module-specific. A module that needs a
 * value declares it (config) or is handed it (a hook input). See the note on
 * `DDNS_ENDPOINT` in openspec/changes/hook-process-boundary/baseline.md for the
 * one case that got this wrong and what was done about it.
 */
const FORWARDED_ENV = [
  // How to run.
  'PATH',
  'HOME',
  'LANG',
  'TZ',
  'TMPDIR',
  // Whom to trust.
  'NODE_EXTRA_CA_CERTS',
  'SSL_CERT_FILE',
  'SSL_CERT_DIR',
  // How to reach the network.
  'HTTP_PROXY',
  'HTTPS_PROXY',
  'NO_PROXY',
  'http_proxy',
  'https_proxy',
  'no_proxy',
] as const;

/**
 * The directory holding the `celilo` bin of the running CLI package, resolved
 * from this file's own location: src/hooks -> the package's bin/. Every
 * distribution layout lands there. The .deb wrapper execs
 * `$CELILO_HOME/node_modules/@celilo/cli/bin/celilo` directly (packaging/celilo/
 * scripts/wrapper.sh), and a bun global install symlinks ~/.bun/bin/celilo into
 * that same bin directory, whose script self-locates its package dir.
 */
export const CLI_BIN_DIR = join(import.meta.dir, '..', '..', 'bin');

/**
 * The PATH a hook child gets: the CLI package's own bin directory, then the
 * runtime's, ahead of the inherited one — so the child can exec the SAME
 * `celilo` the parent runs, and that celilo can still find its interpreter.
 *
 * `dirname(process.execPath)` alone is NOT the CLI's directory under the fleet
 * layout: the .deb runs the CLI through a wrapper at /usr/local/bin/celilo that
 * execs bun on `node_modules/@celilo/cli/src/cli/index.ts`, so the directory
 * beside the bun binary holds only bun (measured on the fleet, 2026-09-07,
 * ce-y0fd). Hooks then found `celilo` only because the systemd unit happened to
 * inherit /usr/local/bin on its PATH — the same accident that broke every hook
 * spawning `celilo` by name in the e2e management image (celilo#1300).
 * CLI_BIN_DIR resolves where the bin actually lives, whatever layout launched
 * the parent, and comes FIRST so `celilo` is the package's own bin rather than
 * an operator's wrapper. The runtime directory stays second because bin/celilo
 * needs `command -v bun` (it is bash, not a shebanged TS file).
 *
 * Already-present wins: an operator who has a directory on PATH sees an
 * unchanged value. An absent inherited PATH still yields both directories, so
 * a hook can always reach `celilo` even from a stripped parent; bash resolves
 * through the loader's default search.
 *
 * Exported for the allow-list tests (hook-trespass.test.ts), which pin this
 * contract the same way they pin the rest of the child environment.
 */
export function childPath(inherited: string | undefined): string {
  const base = inherited === undefined ? [] : inherited.split(':');
  const prepended: string[] = [];
  for (const dir of [CLI_BIN_DIR, dirname(process.execPath)]) {
    if (!base.includes(dir) && !prepended.includes(dir)) prepended.push(dir);
  }
  return [...prepended, ...base].join(':');
}

/** The shim celilo spawns. Resolved from here so an npm install finds it too. */
const HOOK_RUNNER_PATH = join(import.meta.dir, 'hook-runner.ts');
/**
 * What celilo actually SPAWNS: the entry that installs the advisory lint
 * before the runner shim's own import graph can load `node:fs` (task 4.7).
 * The runner path above is still what the mount set binds — the entry and the
 * shim live in the same directory and resolve through the same
 * `node_modules`.
 */
const HOOK_RUNNER_SPAWN_PATH = join(import.meta.dir, 'hook-runner-entry.ts');

/**
 * Validate hook inputs against a contract signature.
 *
 * Policy function (Rule 10.1) - validates inputs.
 *
 * The set of required inputs comes from the contract version registered for
 * the manifest's `celilo_contract`, NOT from the manifest itself. See
 * `apps/celilo/src/manifest/contracts/v1.ts`.
 *
 * @param signature - Contract hook signature for this hook name
 * @param inputs - Provided input values
 * @returns Error message if validation fails, null if valid
 */
export function validateHookInputs(
  signature: ContractHookSignature,
  inputs: Record<string, unknown>,
): string | null {
  const required = Object.entries(signature.inputs)
    .filter(([, def]) => def.required)
    .map(([name]) => name);

  const missing = required.filter((name) => !(name in inputs));
  if (missing.length > 0) {
    return `Missing required hook inputs: ${missing.join(', ')}`;
  }

  return null;
}

/**
 * Validate hook outputs against a contract signature.
 *
 * Policy function (Rule 10.1) - validates outputs.
 *
 * @param signature - Contract hook signature for this hook name
 * @param outputs - Returned output values
 * @returns Error message if validation fails, null if valid
 */
export function validateHookOutputs(
  signature: ContractHookSignature,
  outputs: Record<string, unknown>,
): string | null {
  const required = Object.entries(signature.outputs)
    .filter(([, def]) => def.required)
    .map(([name]) => name);

  const missing = required.filter((name) => !(name in outputs));
  if (missing.length > 0) {
    return `Hook did not return required outputs: ${missing.join(', ')}`;
  }

  return null;
}

/**
 * Resolve the absolute path to a hook script
 *
 * Policy function (Rule 10.1) - validates path
 *
 * @param modulePath - Absolute path to module directory
 * @param scriptPath - Relative script path from manifest
 * @returns Resolved absolute path
 */
export function resolveHookScript(modulePath: string, scriptPath: string): string {
  const resolved = resolve(modulePath, scriptPath);

  // Security: ensure resolved path is within module directory
  const normalizedModule = resolve(modulePath);
  if (!resolved.startsWith(normalizedModule)) {
    throw new Error(`Hook script path escapes module directory: ${scriptPath}`);
  }

  return resolved;
}

/**
 * What the jail needs that only the caller knows.
 *
 * Everything else in a `MountSetRequest` the executor already holds or is: the
 * writable directories are on the context, the socket belongs to the broker it
 * just started, and the runtime and the shim are its own. These two are not.
 * `modulePath` is the module's identity and `pathInputs` needs the contract
 * signature, and neither reaches `executeHookScript` any other way.
 *
 * Absent means "do not jail this invocation" rather than "jail it with
 * nothing" — an empty mount set is a hook that cannot read its own script.
 */
export interface HookJailInputs {
  /** The module's own tree, `mod.sourcePath`. Realpath'd here, not by the caller. */
  readonly modulePath: string;
  /** Contract-declared path inputs, paired with the values this run resolved. */
  readonly pathInputs: readonly DeclaredPathInput[];
}

export interface ExecuteHookOptions {
  /** Total timeout in milliseconds. */
  timeoutMs?: number;
  /**
   * Kill after this much silence (no output of any kind from the child).
   * Pass `timeoutMs` to disable the idle heuristic — see `IDLE_TIMEOUT_MS`.
   */
  idleTimeoutMs?: number;
  /** What to jail this run with. Absent runs the hook unjailed. */
  jail?: HookJailInputs;
  /**
   * D12's target policy for this run's remote operations, built by the caller
   * (`services/remote-access.ts`) so the executor stays decoupled from the
   * database the same way it is for `systems`. Absent means every
   * fleet-credential operation is REFUSED, naming the gap (Rule 6.4).
   */
  remoteAccess?: RemoteAccessPolicy;
  /**
   * The module's hook-owned-state stores (hook-owned-state task 3.5), built
   * lazily so a hook that never touches `context.secrets` / `context.config`
   * pays nothing. Absent means the broker refuses store writes with an error
   * naming the gap — never a silent drop.
   */
  hookStores?: HookStoresProvider;
}

/**
 * The stored `hooks.jail_policy` value, or undefined when there is no row or
 * no celilo DB to ask (a dev box). Same row-read pattern as
 * `ipam/auto-allocator.ts` and the executor's fellow consumer
 * `cli/commands/system-doctor.ts`; the drizzle table is `systemConfig` in
 * `db/schema.ts`.
 *
 * Unlike `remoteAccess`, this read happens here rather than through a
 * caller-built input because D2 names the executor itself as one of the two
 * consumers that fetch the stored row and hand it to the pure resolver.
 */
function readStoredJailPolicy(): string | undefined {
  if (!existsSync(getDbPath())) return undefined;
  const row = getDb()
    .$client.prepare('SELECT value FROM system_config WHERE key = ?')
    .get('hooks.jail_policy') as { value: string } | undefined;
  return row?.value;
}

/**
 * The module's own recorded jail policy, or undefined when there is no row
 * (per-module-jail-policy task 1.4). The system-key read stays its own
 * function because the two answers come from different tables.
 *
 * The executor is handed the module's tree, `options.jail.modulePath`, which
 * is `mod.sourcePath` — not its id, which the caller holds. The row is found
 * the way the jail inputs identify the module: by joining `modules` on
 * `source_path`, the exact string the caller passed, so no second identity
 * is invented and none of invokeHook's callers need plumbing. Absent jail
 * inputs (an invocation that cannot be jailed) also means no module identity
 * and no module row to read.
 *
 * A bad stored value is not coerced here — `resolveJailPolicy` throws on it,
 * the same rule the system key follows (peba's ruling on ce-8832).
 */
function readStoredModuleJailPolicy(modulePath: string | undefined): string | undefined {
  if (!modulePath || !existsSync(getDbPath())) return undefined;
  const row = getDb()
    .$client.prepare(
      'SELECT p.policy AS policy FROM module_jail_policies p JOIN modules m ON m.id = p.module_id WHERE m.source_path = ?',
    )
    .get(modulePath) as { policy: string } | undefined;
  return row?.policy;
}

/**
 * Execute a hook script
 *
 * Execution function (Rule 10.1) - performs side effects
 *
 * @param scriptPath - Absolute path to the hook script
 * @param context - Hook context with config, secrets, logger, and inputs
 * @param options - Timeouts and the jail inputs
 * @returns Hook result with outputs
 */
export async function executeHookScript(
  scriptPath: string,
  context: HookContext,
  options: ExecuteHookOptions = {},
): Promise<Record<string, unknown>> {
  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
  const idleTimeoutMs = options.idleTimeoutMs ?? IDLE_TIMEOUT_MS;

  if (!existsSync(scriptPath)) {
    throw new Error(`Hook script not found: ${scriptPath}`);
  }

  // Activity is now ANY frame or ANY byte the child writes, rather than only a
  // `ctx.logger` call. That is strictly more generous than the old tracker and
  // it removes a class of false idle-kill: a hook that prints progress to
  // stdout used to look silent (celilo#622 is the same family).
  let lastActivity = Date.now();
  const markActive = () => {
    lastActivity = Date.now();
  };

  const { logger, ...serializableContext } = context;
  delete (serializableContext as Record<string, unknown>).capabilities;

  const broker = await startBroker({
    capabilities: context.capabilities,
    context: serializableContext as Record<string, unknown>,
    scriptPath,
    logger,
    onActivity: markActive,
    stores: options.hookStores,
  });
  let remoteBroker: Awaited<ReturnType<typeof startRemoteBroker>> | undefined;

  try {
    // The remote-ops socket lives beside the capability socket, so the jail's
    // one socket-directory mount covers both channels (design D12).
    const roots = remoteRoots(context, options.jail);
    remoteBroker = await startRemoteBroker({
      socketDir: dirname(broker.socketPath),
      policy: options.remoteAccess,
      readableRoots: roots.readable,
      writableRoots: roots.writable,
      logger,
      onActivity: markActive,
    });

    const mountSet = options.jail
      ? deriveMountSet(realpathRequest(jailRequest(options.jail, context, broker.socketPath)))
      : undefined;
    const jail = planJailedSpawn(
      [process.execPath, HOOK_RUNNER_SPAWN_PATH],
      mountSet,
      detectJailBackend(),
      resolveJailPolicy(
        process.env.CELILO_HOOK_JAIL,
        readStoredModuleJailPolicy(options.jail?.modulePath),
        readStoredJailPolicy(),
      ).policy,
    );
    reportJail(jail, logger);
    // Only a real module hook says anything about whether this HOST jails. An
    // invocation with no module tree is unjailable whatever the host can do,
    // and recording it would overwrite a `jailed` record with an `unjailed`
    // one — which is precisely the transition task 4.4 raises an alert on.
    if (options.jail) recordJailMode(jail);

    const child = Bun.spawn({
      cmd: [...jail.cmd],
      // `sandbox-exec` has no `--chdir`, so the plan names one and the spawn
      // sets it. Without it the child inherits the operator's shell directory,
      // which the profile does not name — and `bun` reads its cwd before it
      // runs anything, so the hook dies with a message about nothing in
      // particular. bubblewrap plans leave this absent and use `--chdir`.
      ...(jail.cwd ? { cwd: jail.cwd } : {}),
      env: hookChildEnv(
        broker.socketPath,
        remoteBroker.socketPath,
        // The lint rides only on an unjailed run — its presence in the child's
        // environment is the shim's signal to install it (task 4.7).
        jail.mode === 'unjailed' ? mountSet : undefined,
      ),
      stdout: 'pipe',
      stderr: 'pipe',
    });

    // The child's stdout and stderr stay what they are — human output. They
    // now reach `ctx.logger` instead of celilo's own stdout, which is the
    // attribution: a `console.log` from an in-process hook landed unlabelled
    // in the CLI's output with nothing saying which hook wrote it.
    const forwarding = Promise.all([
      forwardStream(child.stdout, logger.info.bind(logger), markActive),
      forwardStream(child.stderr, logger.warn.bind(logger), markActive),
    ]);

    let killedFor: string | null = null;
    const kill = (reason: string) => {
      if (killedFor) return;
      killedFor = reason;
      // Refuse capability calls FIRST. A hook between `kill` and its own death
      // can still have a call in flight, and answering it is exactly the
      // failure this replaces. The remote channel refuses for the same reason:
      // a killed hook must not go on reaching systems (celilo#1003, D12).
      broker.stop();
      remoteBroker?.stop();
      child.kill('SIGTERM');
      setTimeout(() => child.kill('SIGKILL'), SIGKILL_GRACE_MS).unref();
    };

    const totalTimer = setTimeout(() => kill('total'), timeoutMs);
    // Debug runs get no idle timer, same as before: an operator stepping
    // through a browser hook is silent for minutes on purpose.
    const idleTimer = context.debug
      ? undefined
      : setInterval(() => {
          if (Date.now() - lastActivity >= idleTimeoutMs) kill('idle');
        }, IDLE_POLL_MS);

    let exitCode: number;
    try {
      exitCode = await child.exited;
      // Both channels, not just the process: the terminal frame can still be
      // in the socket buffer when the exit is observed.
      await broker.drained();
      await forwarding;
    } finally {
      clearTimeout(totalTimer);
      if (idleTimer) clearInterval(idleTimer);
    }

    // Process exit is the ONE terminal event, so there is no race to lose and
    // nothing keeps running past this line. That is the difference from the
    // `Promise.race` this replaces (celilo#1003).
    if (killedFor === 'total') {
      throw new Error(`Hook total timeout exceeded (${Math.round(timeoutMs / 1000)}s)`);
    }
    if (killedFor === 'idle') {
      throw new Error(
        `Hook idle timeout exceeded (no log output for ${Math.round(idleTimeoutMs / 1000)}s)`,
      );
    }

    const outcome = broker.outcome();
    if (!outcome) {
      const faults = broker.faults();
      throw new Error(
        `Hook process exited with code ${exitCode} without returning a result.${
          faults.length > 0 ? ` ${faults.join('; ')}` : ''
        }`,
      );
    }

    if (!outcome.ok) throw deserializeError(outcome.error);
    return outcome.outputs;
  } finally {
    // Remote first: its socket file lives inside the directory the capability
    // broker's close removes.
    remoteBroker?.close();
    broker.close();
  }
}

/**
 * The local directories this run's stream primitives may touch (design D12).
 *
 * Derived from the same inputs as the jail's mount set, which is what keeps
 * the two views from drifting: everything writable here is writable there,
 * and a path outside these roots is a path the jailed hook could not see
 * anyway — the broker writing to it would produce a file the hook cannot
 * read, reported as success.
 *
 * Planning function (Rule 10.4) — pure.
 */
function remoteRoots(
  context: HookContext,
  jail: HookJailInputs | undefined,
): { readable: string[]; writable: string[] } {
  const writable: string[] = [context.stateDir, context.screenshotDir];
  const readable: string[] = [];
  if (jail) {
    readable.push(jail.modulePath);
    writable.push(join(jail.modulePath, 'generated'));
    for (const input of jail.pathInputs) {
      (input.access === 'write' ? writable : readable).push(input.value);
    }
  }
  // Whatever a hook may write, it may also read back.
  return { readable: [...readable, ...writable], writable };
}

/**
 * The child's environment, built from an allow-list (design D5).
 *
 * NOT `...process.env`. `packages/event-bus/src/dispatcher.ts` spawns handlers
 * that way and it grants a subprocess MORE than the in-process call did — the
 * whole environment plus an explicit database path. Copying that pattern would
 * buy the crash isolation and none of the authority reduction.
 *
 * Measured on the operator's own shell: what a hook inherited before this was
 * 24 variables naming celilo state or carrying a credential, twenty of them
 * third-party API tokens with nothing to do with celilo
 * (`openspec/changes/hook-process-boundary/baseline.md`).
 *
 * This is not the boundary — a hook can still COMPUTE the default master-key
 * path from `$HOME` and read it. It is the cheap half, and it is what stops the
 * dispatcher's mistake repeating here. Stage 2's mount set is what makes the
 * path unreachable.
 *
 * Planning function (Rule 10.4) — pure.
 */
export function hookChildEnv(
  socketPath: string,
  remoteSocketPath: string,
  mountSet: MountSet | undefined,
): Record<string, string> {
  const env: Record<string, string> = {};

  for (const name of FORWARDED_ENV) {
    if (name === 'PATH') {
      env.PATH = childPath(process.env.PATH);
      continue;
    }
    const value = process.env[name];
    if (value !== undefined) env[name] = value;
  }

  env[HOOK_SOCKET_ENV] = socketPath;
  env[HOOK_REMOTE_SOCKET_ENV] = remoteSocketPath;
  // The unjailed advisory lint's input (task 4.7). Absent when jailed — the
  // shim installs the lint when and only when this variable is set.
  if (mountSet) env[HOOK_MOUNT_SET_ENV] = mountSetEnvValue(mountSet);
  env[HOOK_PROTOCOL_VERSION_ENV] = String(HOOK_PROTOCOL_VERSION);
  // Forwarded when the operator set it, not synthesised from `debug` — a hook
  // reads the flag off `ctx.debug`, which crosses in the context frame.
  if (process.env.CELILO_DEBUG !== undefined) env.CELILO_DEBUG = process.env.CELILO_DEBUG;

  return env;
}

/**
 * Assemble this run's mount-set request (design D9).
 *
 * Planning function (Rule 10.4) — pure. It names every path the jail will
 * bind, and reading it is how you answer "why can the hook not see X".
 *
 * The two writable directories are carved out of the module's own read-only
 * tree, which is the one exception to D9's sentence that the tree is bound
 * read-only. `state/` and `screenshots/<run>` both sit INSIDE `<store>/<id>`,
 * and bubblewrap resolves that by order: a later `--bind` wins over an earlier
 * `--ro-bind`. The derivation emits them in that order deliberately.
 */
function jailRequest(
  inputs: HookJailInputs,
  context: HookContext,
  socketPath: string,
): MountSetRequest {
  return {
    modulePath: inputs.modulePath,
    stateDir: context.stateDir,
    screenshotDir: context.screenshotDir,
    socketDir: dirname(socketPath),
    runtimePath: process.execPath,
    runnerPath: HOOK_RUNNER_PATH,
    runtimeModulePaths: runtimeModulePathsFor(HOOK_RUNNER_PATH),
    pathInputs: inputs.pathInputs,
  };
}

/**
 * Say what the jail did, once per run, at a level that matches how surprising
 * it is. A dropped row is now either FATAL or SILENT, never a warning.
 *
 * This used to warn about every absent row, because it could not tell a routine
 * absence from a damaging one: "both look identical here, so the line names the
 * paths and lets a reader tell them apart." That reasoning was sound and the
 * result was not. The line fired 96 times with a byte-identical payload in a
 * single `cele2e run --all`, so the case it existed to catch was buried in
 * ninety-six copies of the case that does not matter. A signal that repeats
 * unchanged is one every reader learns to skip.
 *
 * Each row now states what its own absence means (`MountAbsence`), so there is
 * no classification left for a human to do from a path string:
 *
 *   required       throw. The hook cannot do what it was asked.
 *   declared-only  silent until a module can declare the facility (ce-qani).
 *   runtime        silent. A genuinely needed one fails the runtime, louder.
 *   conditional    silent. Expected at this point in the lifecycle.
 */
function reportJail(plan: JailPlan, logger: HookLogger): void {
  if (plan.mode === 'jailed') {
    const fatal = plan.skipped.filter((m) => m.absence === 'required');
    if (fatal.length > 0) {
      // Throw rather than warn. A dropped contract input means the hook writes
      // into the run's PRIVATE TMPFS and reports success over a directory that
      // is discarded when it exits (task 4.2j), and a dropped interpreter or
      // module tree means it could never have run at all. Both were previously
      // survivable-looking warnings.
      const detail = fatal.map((m) => `${m.path} (${m.reason})`).join(', ');
      throw new Error(
        `Hook jail cannot be built: ${fatal.length} required mount(s) absent on this host: ${detail}`,
      );
    }
    return;
  }
  // Not a warning. D8 is explicit that a steady-state unjailed host (a Mac) is
  // not an event, and a per-invocation warning on a fleet that deploys often
  // is noise that gets filtered. The recorded mode is what raises the alarm.
  if (plan.reason) logger.info(`Hook jail unavailable: ${plan.reason}`);
}

/** Read a piped stream line by line into the logger. */
async function forwardStream(
  stream: ReadableStream<Uint8Array>,
  emit: (line: string) => void,
  markActive: () => void,
): Promise<void> {
  const decoder = new TextDecoder();
  const feed = createLineReader((line) => {
    markActive();
    emit(line);
  });

  for await (const chunk of stream) {
    feed(decoder.decode(chunk, { stream: true }));
  }
  // A final line with no trailing newline would otherwise be dropped.
  feed('\n');
}

export interface InvokeHookOptions {
  debug?: boolean;
  /** Pre-loaded capability function interfaces to inject into context */
  capabilities?: Record<string, unknown>;
  /**
   * The module's 0..N deployed systems, injected as `ctx.systems`
   * (openspec/specs/module-systems-addressing/spec.md). Callers load these from the DB via
   * `getModuleSystems` — the executor stays decoupled from the database, the
   * same way `capabilities` is loaded by the caller. Defaults to `[]`.
   */
  systems?: DeployedSystem[];
  /**
   * Names of capabilities the hook's module declares in
   * `requires.capabilities`. The executor checks each one is present in
   * `capabilities` before invoking the hook handler and fails fast with a
   * clear error if any are missing. Added in HOOK_API_V2 Phase 3 (D3).
   *
   * Optional for backwards compatibility — callers that don't pass this
   * skip the pre-flight check, matching the pre-Phase-3 behavior.
   */
  requiredCapabilities?: string[];
  /**
   * Total timeout for THIS invocation, in milliseconds. Set by the caller
   * that knows the per-invocation bound — for a bus delivery that's the
   * subscription's `timeout_ms` (celilo#622). Takes precedence over the
   * manifest hook's own `timeout`, being the more specific declaration.
   */
  timeoutMs?: number;
  /**
   * D12's target policy for the hook's remote operations, built by the caller
   * via `remoteAccessPolicy(moduleId, db)` — the same caller-loads-it pattern
   * as `systems`, and for the same reason: the executor stays decoupled from
   * the database. Absent means every fleet-credential remote operation is
   * refused (Rule 6.4).
   */
  remoteAccess?: RemoteAccessPolicy;
  /**
   * The module's hook-owned-state stores (hook-owned-state task 3.5), as a
   * lazy provider: `() => createHookStores(db, moduleId)`. Built on first use
   * inside the run, so a hook that never persists state never reads the
   * manifest or touches the master key.
   */
  hookStores?: HookStoresProvider;
}

/**
 * Decide the total and idle bounds for one hook invocation.
 *
 * Planning function (Rule 10.4) — pure, so the crux below is testable
 * without waiting on a real clock.
 *
 * **Idle-vs-total (celilo#622): an explicit declaration REPLACES the idle
 * heuristic rather than being capped by it** — the idle bound becomes the
 * total, which is the same as switching it off. The idle timer measures
 * silence since the last `ctx.logger` call, which is a proxy for "this hook
 * told us nothing about how long it needs". Once a hook or subscription
 * states a bound, that proxy is not merely redundant but wrong: honoring a
 * declared 120s total while still killing at 30s of silence honors the
 * declaration in name only. That is exactly how namecheap's ddns
 * `refresh_registrations` died — it declares `timeout_ms: 120000`, logs once
 * per name, and a single slow upstream response is 30s of silence.
 *
 * The cost is deliberate: a hook that declares a bound and then truly wedges
 * is now bounded only by that declared total, not killed early. That is the
 * declaring subscriber's call to make, and the dispatcher kills the handler
 * subprocess at the same `timeout_ms` regardless.
 *
 * @param declaredTimeoutMs - Explicitly declared total, or undefined
 * @param debug - Debug mode uses a long total and no idle timer at all
 */
export function resolveHookTimeouts(
  declaredTimeoutMs: number | undefined,
  debug: boolean,
): { timeoutMs: number; idleTimeoutMs: number } {
  const timeoutMs = debug
    ? 600_000 // 10 minutes for interactive debugging
    : (declaredTimeoutMs ?? DEFAULT_TIMEOUT_MS);

  return {
    timeoutMs,
    idleTimeoutMs: declaredTimeoutMs === undefined ? IDLE_TIMEOUT_MS : timeoutMs,
  };
}

/**
 * Verify every required capability is present in the loaded capability map.
 *
 * Policy function (Rule 10.1) — pure check, no side effects.
 *
 * @returns Error message listing missing capabilities, or null if all present
 */
export function checkRequiredCapabilities(
  hookName: string,
  requiredCapabilities: string[],
  loadedCapabilities: Record<string, unknown>,
): string | null {
  // Framework-granted privileges (e.g. cross_module_read) aren't loaded as
  // provider capability-functions — they're granted by the framework to the
  // specific hooks that use them (backup/restore), via a separate mechanism.
  // Don't treat them as missing providers for every hook (e.g. on_install).
  const missing = requiredCapabilities.filter(
    (name) => !isPrivilegedCapability(name) && !(name in loadedCapabilities),
  );
  if (missing.length === 0) return null;

  const lines = [
    `Hook '${hookName}' requires capability ${missing.length === 1 ? '' : 'capabilities '}${missing.map((n) => `'${n}'`).join(', ')} but no provider${missing.length === 1 ? ' is' : 's are'} loaded.`,
    `Install a module that provides ${missing.length === 1 ? missing[0] : 'each missing capability'} and retry.`,
  ];
  return lines.join(' ');
}

/**
 * The contract's declared path inputs, paired with the values this run got.
 *
 * Policy function (Rule 10.1) — pure.
 *
 * Walks the DECLARATION, never the values. A heuristic over the input names
 * ("does it end in `_dir`?") silently changes what a hook can reach the day
 * somebody adds an input called `workspace`, and the symptom is an `ENOENT` on
 * a path that visibly exists on the box. `ContractField.path` says which
 * fields are paths and at what access, and this reads only that.
 *
 * The corollary is that an undeclared path input contributes nothing to the
 * jail, so the hook cannot reach it. That is the correct outcome and it is why
 * `db_path` had to be declared (celilo#1118, task 4.2h): the contract lying
 * about what a backup hook receives now costs the hook the file.
 */
export function declaredPathInputs(
  signature: ContractHookSignature,
  inputs: Record<string, unknown>,
): DeclaredPathInput[] {
  const declared: DeclaredPathInput[] = [];
  for (const [name, field] of Object.entries(signature.inputs)) {
    if (!field.path) continue;
    const value = inputs[name];
    if (typeof value !== 'string' || value === '') continue;
    declared.push({ name, value, access: field.path.access });
  }
  return declared;
}

/**
 * Find the most recently created screenshot in a directory
 *
 * @param dir - Directory to scan
 * @param since - Only consider files created after this timestamp (ms)
 * @returns Path to screenshot, or undefined
 */
/**
 * Every file the hook wrote to this run's artifact directory.
 *
 * No mtime filter is needed and none is wanted: the directory is created
 * fresh for this run, so everything in it was written by this run. The
 * previous version returned only the newest `.png`, which discarded the
 * page content and the observed-request log — the two things that make a
 * screenshot diagnosable.
 */
function collectArtifacts(dir: string): string[] {
  try {
    return readdirSync(dir)
      .map((file) => join(dir, file))
      .filter((path) => {
        try {
          return statSync(path).isFile();
        } catch {
          return false;
        }
      })
      .sort();
  } catch {
    // Best effort — never mask the original error with a readdir failure.
    return [];
  }
}

/** `undefined` rather than `[]`, so an empty result carries no field at all. */
function nonEmpty(paths: string[]): string[] | undefined {
  return paths.length > 0 ? paths : undefined;
}

/** Remove a run's artifact directory when the hook wrote nothing into it. */
function discardIfEmpty(dir: string): void {
  try {
    if (readdirSync(dir).length === 0) rmdirSync(dir);
  } catch {
    // Best effort — a leftover directory is not worth failing a hook over.
  }
}

/**
 * Execute a named hook for a module
 *
 * Orchestration function - coordinates validation, resolution, and execution.
 *
 * Resolves the hook's inputs/outputs signature from the contract registry
 * keyed by `contractVersion`, validates the provided inputs against it,
 * runs the script, and validates the returned outputs.
 *
 * @param modulePath - Absolute path to module directory
 * @param hookName - Name of the hook to execute
 * @param contractVersion - Celilo contract version from `manifest.celilo_contract`
 * @param definition - Hook definition from manifest (script + timeout only)
 * @param inputs - Input values for the hook
 * @param config - Module configuration values
 * @param secrets - Module secret values (decrypted)
 * @param logger - Logger instance
 * @param options - Debug and other options
 * @returns Hook result
 */
export async function invokeHook(
  modulePath: string,
  hookName: string,
  contractVersion: string,
  definition: HookDefinition,
  inputs: Record<string, unknown>,
  config: Record<string, unknown>,
  secrets: Record<string, string>,
  logger: HookLogger,
  options: InvokeHookOptions = {},
): Promise<HookResult> {
  const startTime = Date.now();
  const debug = options.debug ?? false;

  // Resolve the hook's contract signature
  const contract = resolveContract(contractVersion);
  if (!contract) {
    return {
      success: false,
      outputs: {},
      error: `Unsupported celilo_contract version '${contractVersion}'. Supported: ${supportedContractVersions().join(', ')}`,
      duration: Date.now() - startTime,
    };
  }

  const signature = contractHookSignature(contract.hooks, hookName);
  if (!signature) {
    return {
      success: false,
      outputs: {},
      error: `Hook '${hookName}' is not part of celilo_contract ${contractVersion}`,
      duration: Date.now() - startTime,
    };
  }

  // Validate inputs against the contract signature
  const inputError = validateHookInputs(signature, inputs);
  if (inputError) {
    return {
      success: false,
      outputs: {},
      error: inputError,
      duration: Date.now() - startTime,
    };
  }

  // Resolve script path
  let scriptPath: string;
  try {
    scriptPath = resolveHookScript(modulePath, definition.script);
  } catch (error) {
    return {
      success: false,
      outputs: {},
      error: error instanceof Error ? error.message : String(error),
      duration: Date.now() - startTime,
    };
  }

  // Prepare this run's artifact directory. Per-run, because a hook writing
  // fixed filenames — a reasonable thing to do — would otherwise overwrite
  // its own previous artifacts and leave retention nothing to retain.
  // `moduleArtifactDir` is the one definition of that layout; a bus
  // subscriber, which gets no HookContext, calls it directly.
  // Path now, directory later. Creating it here would leak an empty
  // directory on every early return between this point and the try/finally
  // below — the capability pre-flight is one — and nothing ever cleans
  // those up, so a module accrues one per affected run forever.
  const screenshotDir = moduleArtifactDir(modulePath, `${hookName}-${startTime}`);
  // Unlike the artifact directory, this one is NOT per-run and is created
  // unconditionally: a hook must be able to write state on its first ever run,
  // and should never have to mkdir its own sanctioned location (celilo#1000).
  const stateDir = moduleStateDir(modulePath);

  // Build context
  const loadedCapabilities = options.capabilities ?? {};
  // The store accessor methods attach on the CHILD side (hook-runner.ts,
  // buildStoreView) where the socket is, so the object serialized here
  // carries the map half only. Same cast rationale as the runner's own
  // context build. (hook-owned-state task 3.5)
  const context = {
    ...inputs,
    config,
    secrets,
    systems: options.systems ?? [],
    logger,
    debug,
    screenshotDir,
    stateDir,
    capabilities: loadedCapabilities,
  } as unknown as HookContext;

  // Pre-flight: every required capability must be loaded (HOOK_API_V2 D3).
  // Fail before invoking the handler so the script never sees a missing
  // capability. Skipped if the caller didn't pass requiredCapabilities,
  // for backwards compatibility with pre-Phase-3 invocation sites.
  if (options.requiredCapabilities && options.requiredCapabilities.length > 0) {
    const capError = checkRequiredCapabilities(
      hookName,
      options.requiredCapabilities,
      loadedCapabilities,
    );
    if (capError) {
      return {
        success: false,
        outputs: {},
        error: capError,
        duration: Date.now() - startTime,
      };
    }
  }

  // An explicitly declared bound — the subscription's `timeout_ms` for a bus
  // delivery, else the manifest hook's own `timeout`. The subscription wins:
  // it is the more specific declaration for this invocation.
  const { timeoutMs, idleTimeoutMs } = resolveHookTimeouts(
    options.timeoutMs ?? definition.timeout,
    debug,
  );

  // Create the artifact directory only once every early return is behind
  // us, so the `finally` below is guaranteed to run and reclaim it.
  mkdirSync(screenshotDir, { recursive: true });
  // Created every run and reclaimed on none. An empty `state/` is not litter
  // the way an empty per-run artifact directory is: there is one of them, it is
  // where the module is told to write, and deleting it between runs would make
  // its existence depend on whether the last hook happened to use it.
  mkdirSync(stateDir, { recursive: true });
  // Prune on write, so retention needs no scheduler of its own and cannot
  // fall behind a module that runs often.
  pruneModuleArtifacts(dirname(screenshotDir));

  // Execute
  try {
    logger.info(`Executing hook: ${hookName}`);
    const outputs = await executeHookScript(scriptPath, context, {
      timeoutMs,
      idleTimeoutMs,
      jail: { modulePath, pathInputs: declaredPathInputs(signature, inputs) },
      remoteAccess: options.remoteAccess,
      hookStores: options.hookStores,
    });

    // Validate outputs against the contract signature
    const outputError = validateHookOutputs(signature, outputs);
    if (outputError) {
      return {
        success: false,
        outputs,
        error: outputError,
        duration: Date.now() - startTime,
      };
    }

    logger.success(`Hook ${hookName} completed successfully`);
    return {
      success: true,
      outputs,
      artifactPaths: nonEmpty(collectArtifacts(screenshotDir)),
      duration: Date.now() - startTime,
    };
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error);
    logger.error(`Hook ${hookName} failed: ${message}`);

    // Artifacts the hook wrote before it failed — the post-mortem.
    const artifactPaths = nonEmpty(collectArtifacts(screenshotDir));
    if (artifactPaths) {
      logger.info(`Artifacts saved:\n  ${artifactPaths.join('\n  ')}`);
    }

    // Recognise the cross-module structured error so the orchestrator can
    // run the ensure interview and retry the hook. The check goes through
    // a duck-typed guard because @celilo/capabilities can be loaded from
    // either the workspace package or a module's bundled copy, and
    // `instanceof` fails across those identity boundaries.
    if (isMissingProviderInputError(error)) {
      return {
        success: false,
        outputs: {},
        error: message,
        artifactPaths,
        duration: Date.now() - startTime,
        missingProviderInput: {
          providerModuleId: error.providerModuleId,
          ensureId: error.ensureId,
          value: error.value,
          humanContext: error.humanContext,
        },
      };
    }

    return {
      success: false,
      outputs: {},
      error: message,
      artifactPaths,
      duration: Date.now() - startTime,
    };
  } finally {
    // Most hooks write nothing, and every hook invocation would otherwise
    // leave an empty directory behind. Discard it; a run that produced
    // artifacts keeps its directory.
    discardIfEmpty(screenshotDir);
  }
}