/**
 * `celilo events <subcommand>` — operator surface for the SQLite event
 * bus. Thin wrappers over `@celilo/event-bus`'s programmatic API. The
 * standalone `event-bus` CLI binary is still available; these commands
 * exist so operators don't have to set EVENT_BUS_DB or remember a
 * second tool.
 *
 * Subcommands:
 *   status               bus.health() as JSON
 *   tail                 recent events
 *   list-subscribers     persistent subscribers
 *   list-pending         pending deliveries (subscriber fan-out — NOT questions)
 *   list-failed          failed/abandoned deliveries + a TRUE total
 *   list-unanswered      interview questions nobody has answered yet
 *   drain                process pending deliveries once
 *   run                  long-running dispatcher (foreground; SIGINT to stop)
 *   emit <type> [json]   emit an event (operator/test path; bypasses schema)
 *   reply <id> <value>   answer one pending interview query by event id
 *   ack <event_id>       mark a running delivery succeeded
 *   fail <event_id>      mark a running delivery failed
 *   repair               crash-recovery sweep without starting the dispatcher
 *   resume               alias for repair (acknowledges halt-on-recovery)
 */

import { spawnSync } from 'node:child_process';
import {
  BUS_VERSION,
  type FailedBySubscriber,
  defineEvents,
  describeError,
  drainOnce,
  openBus,
  recoverFromCrash,
  runDispatcher,
} from '@celilo/event-bus';
import { eq } from 'drizzle-orm';
import { sessionParkedOn } from '../../api/sessions';
import { getEventBusPath, shortenPath } from '../../config/paths';
import { getDb } from '../../db/client';
import { modules } from '../../db/schema';
import { createConsoleLogger } from '../../hooks/logger';
import { runNamedHook } from '../../hooks/run-named-hook';
import type { HookName } from '../../hooks/types';
import type { ModuleManifest } from '../../manifest/schema';
import type { EnsureRequiredPayload } from '../../services/bus-interview';
import {
  detectPlatform,
  installDaemon,
  planDaemonInstall,
  readInstalledUnit,
  resolveRestartScope,
  restartDaemon,
  supervisorCommands,
  uninstallDaemon,
  unitInstalledInAnyScope,
} from '../../services/events-daemon';
import { getArg, hasFlag } from '../parser';
import type { CommandResult, CommandSuccess } from '../types';

const NO_SCHEMAS = defineEvents({});

function openCliBus() {
  return openBus({ dbPath: getEventBusPath(), events: NO_SCHEMAS });
}

/**
 * Every `celilo events` JSON verb returns through here, so `rawOutput` is set
 * once: without it the payload goes through the CLI's decorating renderer and
 * reaches stdout prefixed and re-wrapped, and no longer parses.
 */
function jsonResult(data: unknown): CommandResult {
  return {
    success: true,
    message: JSON.stringify(data, null, 2),
    rawOutput: true,
    data,
  };
}

export async function handleEventsStatus(): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    return jsonResult(bus.health());
  } finally {
    bus.close();
  }
}

/** camelCase → snake_case for every key (payloads are camelCase; hook inputs snake_case). */
function snakeCaseKeys(obj: Record<string, unknown>): Record<string, unknown> {
  const out: Record<string, unknown> = {};
  for (const [k, v] of Object.entries(obj)) {
    out[k.replace(/[A-Z]/g, (m) => `_${m.toLowerCase()}`)] = v;
  }
  return out;
}

/**
 * `celilo events run-hook <module> <sub-name> [<event_id>]` — the generic
 * runner a `hook:` subscription resolves to (openspec/specs/event-driven-hook-subscriptions/spec.md).
 *
 * The dispatcher spawns this as a fault-isolated subprocess. It re-reads the
 * module's manifest to find the named subscription's `hook` + `hook_inputs`,
 * assembles the hook inputs as `{ ...hook_inputs, ...snakeCase(eventPayload) }`,
 * and runs the hook through the normal backend executor (config + decrypted
 * secrets + the module's own capabilities injected). It is dumb pass-through —
 * it knows nothing about DNS or any specific event type.
 */
export async function handleEventsRunHook(args: string[]): Promise<CommandResult> {
  const moduleId = args[0];
  const subName = args[1];
  // Event id: an explicit 3rd arg wins, else the dispatcher's $EVENT_ID.
  const idStr = args[2] ?? process.env.EVENT_ID;
  if (!moduleId || !subName) {
    return {
      success: false,
      error: 'usage: celilo events run-hook <module> <sub-name> [<event_id>]',
    };
  }
  const eventId = Number(idStr);
  if (!idStr || !Number.isInteger(eventId)) {
    return {
      success: false,
      error: 'events run-hook requires an event id (3rd arg or $EVENT_ID)',
    };
  }

  const db = getDb();
  const module = db.select().from(modules).where(eq(modules.id, moduleId)).get();
  if (!module) return { success: false, error: `Module not found: ${moduleId}` };

  // Quiescence (openspec/changes/module-pause-lifecycle, task 2.1/2.2). Pausing
  // drops the module's bus subscriptions, so ordinarily nothing reaches here at
  // all; this is the second line, and it is load-bearing rather than belt-and-
  // braces. The subscribers table lives in a DIFFERENT database (events.db) to
  // the module state, so the two can disagree — `events resync-subscriptions`
  // rebuilds subscribers from celilo.db, a restore starts events.db empty, and
  // a hand-written row is always possible. Every one of those paths ends here,
  // where the module's actual state is readable.
  //
  // Success, not failure: the event was delivered correctly and the module is
  // deliberately not listening. Reporting a failure would retry it up to
  // max_attempts and then surface as an alert about the pause the operator
  // themselves took.
  if (module.state === 'PAUSED') {
    return {
      success: true,
      message: `Skipped ${moduleId}.${subName}: module is paused`,
    };
  }

  const manifest = module.manifestData as ModuleManifest;
  const sub = (manifest.subscriptions ?? []).find((s) => s.name === subName);
  if (!sub) {
    return { success: false, error: `Module '${moduleId}' has no subscription named '${subName}'` };
  }
  if (!sub.hook) {
    return {
      success: false,
      error: `Subscription '${subName}' on '${moduleId}' is not a hook subscription`,
    };
  }

  const bus = openCliBus();
  let payload: Record<string, unknown> = {};
  let eventType: string;
  try {
    const event = bus.getEvent(eventId);
    if (!event) return { success: false, error: `Event ${eventId} not found on the bus` };
    eventType = event.type;
    if (event.payload && typeof event.payload === 'object') {
      payload = event.payload as Record<string, unknown>;
    }
  } finally {
    bus.close();
  }

  const inputs: Record<string, unknown> = {
    ...(sub.hook_inputs ?? {}),
    ...snakeCaseKeys(payload),
  };

  const logger = createConsoleLogger(moduleId, sub.hook);
  const result = await runNamedHook(moduleId, sub.hook as HookName, db, logger, {
    inputs,
    timeoutMs: sub.timeout_ms,
  });

  if (result.notDefined) {
    return { success: false, error: `Module '${moduleId}' declares no '${sub.hook}' hook to run` };
  }
  if (!result.success) {
    return { success: false, error: result.error ?? `hook '${sub.hook}' failed` };
  }
  return {
    success: true,
    message: `Ran ${moduleId}.${sub.hook} for event ${eventId} (${eventType})`,
  };
}

export async function handleEventsTail(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const limit = flags.limit ? Number(flags.limit) : 50;
    const type = typeof flags.type === 'string' ? flags.type : undefined;
    return jsonResult(bus.recentEvents({ limit, type }));
  } finally {
    bus.close();
  }
}

export async function handleEventsListSubscribers(): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const rows = bus.db
      .query<
        {
          name: string;
          pattern: string;
          handler: string;
          max_attempts: number;
          timeout_ms: number;
        },
        []
      >('SELECT name, pattern, handler, max_attempts, timeout_ms FROM subscribers ORDER BY name')
      .all();
    return jsonResult(rows);
  } finally {
    bus.close();
  }
}

/**
 * `celilo events resync-subscriptions` — rebuild the bus `subscribers` table
 * from every deployed module's manifest. The reactive layer is registered at
 * DEPLOY time and lives in the bus, which a restore/migration starts EMPTY — so
 * after a cutover this re-establishes who-reacts-to-what without redeploying
 * every module (ISS-0088). Idempotent.
 */
export async function handleEventsResyncSubscriptions(): Promise<CommandResult> {
  try {
    const { resyncAllSubscriptions } = await import('../../services/module-subscriptions');
    const result = resyncAllSubscriptions();
    const lines = [
      `Re-registered ${result.registered} subscription(s) from ${result.modules} deployed module(s).`,
    ];
    if (result.failures.length > 0) {
      const detail = result.failures.map((f) => `  ${f.moduleId}: ${f.error}`).join('\n');
      return {
        success: false,
        error: `${result.failures.length} module(s) failed to re-register subscriptions:\n${detail}`,
      };
    }
    lines.push('', 'A running dispatcher will now deliver events to these handlers.');
    return { success: true, message: lines.join('\n'), data: result };
  } catch (err) {
    return { success: false, error: err instanceof Error ? err.message : String(err) };
  }
}

export async function handleEventsListPending(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const subscriber = typeof flags.subscriber === 'string' ? flags.subscriber : undefined;
    const limit = flags.limit ? Number(flags.limit) : 100;
    return jsonResult(bus.pendingDeliveries({ subscriber, limit }));
  } finally {
    bus.close();
  }
}

/** One failed/abandoned delivery, as `events list-failed` reports it. */
export interface FailedDelivery {
  eventId: number;
  eventType: string | null;
  subscriber: string | null;
  status: string;
  attempts: number;
  finishedAt: number | null;
  error: string | null;
}

/** What `events list-failed` returns: the true total, the shape, and a sample. */
export interface FailedDeliveryReport {
  total: number;
  bySubscriber: FailedBySubscriber[];
  shown: number;
  deliveries: FailedDelivery[];
}

/**
 * `celilo events list-failed` — what has failed or been abandoned, who it was
 * for, and why.
 *
 * The instrument celilo#623 lacked: `failedDeliveries` had exactly one caller
 * (the doctor) and no operator surface at all, so the only signal was a count
 * that was really a LIMIT. `total` here is a real COUNT — `deliveries` is the
 * sample, capped by `--limit`, and `shown` says so.
 *
 * This is NOT `list-pending`, which reads deliveries still queued.
 */
export async function handleEventsListFailed(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const subscriber = typeof flags.subscriber === 'string' ? flags.subscriber : undefined;
    const limit = flags.limit ? Number(flags.limit) : 50;
    const { total, bySubscriber } = bus.failedDeliveryTotals();
    // Subscriber/event names live in other tables; the delivery row has ids.
    const named = new Map(
      bus.db
        .query<{ id: number; name: string }, []>('SELECT id, name FROM subscribers')
        .all()
        .map((r) => [r.id, r.name]),
    );
    const deliveries: FailedDelivery[] = bus.failedDeliveries({ limit, subscriber }).map((d) => ({
      eventId: d.eventId,
      eventType: bus.getEvent(d.eventId)?.type ?? null,
      subscriber: named.get(d.subscriberId) ?? null,
      status: d.status,
      attempts: d.attempts,
      finishedAt: d.finishedAt,
      error: describeError(d.lastError),
    }));
    const report: FailedDeliveryReport = {
      total: subscriber
        ? (bySubscriber.find((s) => s.subscriber === subscriber)?.count ?? 0)
        : total,
      bySubscriber,
      shown: deliveries.length,
      deliveries,
    };
    return jsonResult(report);
  } finally {
    bus.close();
  }
}

/** One unanswered interview question, as `events list-unanswered` reports it. */
export interface UnansweredInterview {
  eventId: number;
  type: string;
  family: InterviewFamily;
  /** `<scope>.<key>` — the identity a responder pre-stages an answer under. */
  key: string;
  question: string;
  ageMs: number;
  /** The parked api-serve session waiting on this answer, if any. */
  sessionId: string | null;
}

/**
 * `celilo events list-unanswered` — interview queries with no correlated reply:
 * what is waiting on a decision right now.
 *
 * The instrument celilo#609 lacked. `events list-pending` was reached for and
 * silently answered a different question (it reads subscriber *deliveries*), so
 * a parked command looked like no command at all. Non-empty here for as long as
 * something is parked is the recurrence gate for that whole class of bug.
 */
export async function handleEventsListUnanswered(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const limit = flags.limit ? Number(flags.limit) : 50;
    const now = Date.now();
    const rows: UnansweredInterview[] = [];
    for (const event of bus.unansweredQueries({ limit })) {
      const family = interviewFamily(event.type);
      if (!family) continue; // e.g. responder.probe — not a question for an operator.
      const payload = (event.payload ?? {}) as { message?: string; description?: string };
      rows.push({
        eventId: event.id,
        type: event.type,
        family,
        key: event.type.slice(`${family}.required.`.length),
        question: payload.message ?? payload.description ?? event.type,
        ageMs: now - event.emittedAt,
        sessionId: sessionParkedOn(String(event.id))?.sessionId ?? null,
      });
    }
    return jsonResult(rows);
  } finally {
    bus.close();
  }
}

export async function handleEventsDrain(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const concurrency = flags.concurrency ? Number(flags.concurrency) : undefined;
    const result = await drainOnce(bus, { concurrency });
    return jsonResult({ ...result, dbPath: shortenPath(getEventBusPath()) });
  } finally {
    bus.close();
  }
}

/**
 * Run the long-running dispatcher in the foreground. Blocks until
 * SIGINT/SIGTERM. Pair with `celilo events status` from another shell
 * to confirm it's healthy.
 */
export async function handleEventsRun(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  const handle = runDispatcher(bus, {
    pollIntervalMs: flags['poll-ms'] ? Number(flags['poll-ms']) : undefined,
    concurrency: flags.concurrency ? Number(flags.concurrency) : undefined,
    haltOnRecovery: hasFlag(flags, 'halt-on-recovery'),
  });
  console.error(
    `[celilo events] dispatcher running (pid ${process.pid}, db ${shortenPath(getEventBusPath())}). Ctrl-C to stop.`,
  );

  let stopping = false;
  const stop = async (signal: string) => {
    if (stopping) return;
    stopping = true;
    console.error(`[celilo events] received ${signal}, stopping...`);
    await handle.stop();
    bus.close();
    process.exit(0);
  };
  process.on('SIGINT', () => stop('SIGINT'));
  process.on('SIGTERM', () => stop('SIGTERM'));
  await new Promise(() => {}); // hold open
  return { success: true, message: '' }; // unreachable
}

export async function handleEventsEmit(
  args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const type = getArg(args, 0);
  if (!type) {
    return { success: false, error: 'Usage: celilo events emit <type> [<payload-json>]' };
  }
  const payloadRaw = getArg(args, 1);
  let payload: unknown = undefined;
  if (payloadRaw !== undefined) {
    try {
      payload = JSON.parse(payloadRaw);
    } catch (err) {
      return {
        success: false,
        error: `Invalid JSON payload: ${err instanceof Error ? err.message : String(err)}`,
      };
    }
  }
  const bus = openCliBus();
  try {
    const event = bus.emitRaw(type, payload, {
      dedupKey: typeof flags['dedup-key'] === 'string' ? flags['dedup-key'] : undefined,
      emittedBy: typeof flags['emitted-by'] === 'string' ? flags['emitted-by'] : 'celilo-cli',
    });
    return jsonResult(event);
  } finally {
    bus.close();
  }
}

export async function handleEventsAck(
  args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const eventId = getArg(args, 0);
  if (!eventId) return { success: false, error: 'Usage: celilo events ack <event_id>' };
  const eId = Number(eventId);
  if (!Number.isFinite(eId)) return { success: false, error: 'event_id must be a number' };
  const bus = openCliBus();
  try {
    const subscriberId = resolveSubscriberId(bus, eId, flags);
    if (subscriberId === null) {
      return {
        success: false,
        error: `No running delivery for event ${eId}; pass --subscriber <id>`,
      };
    }
    bus.markSucceeded({ eventId: eId, subscriberId });
    return jsonResult({ ok: true, eventId: eId, subscriberId });
  } finally {
    bus.close();
  }
}

export async function handleEventsFail(
  args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const eventId = getArg(args, 0);
  if (!eventId)
    return { success: false, error: 'Usage: celilo events fail <event_id> --error <msg>' };
  const eId = Number(eventId);
  if (!Number.isFinite(eId)) return { success: false, error: 'event_id must be a number' };
  const errorMsg = typeof flags.error === 'string' ? flags.error : 'handler failed';
  const noRetry = hasFlag(flags, 'no-retry');
  const bus = openCliBus();
  try {
    const subscriberId = resolveSubscriberId(bus, eId, flags);
    if (subscriberId === null) {
      return {
        success: false,
        error: `No running delivery for event ${eId}; pass --subscriber <id>`,
      };
    }
    if (noRetry) {
      bus.markFailed({ eventId: eId, subscriberId }, new Error(errorMsg), { abandoned: true });
    } else {
      bus.markFailed({ eventId: eId, subscriberId }, new Error(errorMsg));
    }
    return jsonResult({ ok: true, eventId: eId, subscriberId, abandoned: noRetry });
  } finally {
    bus.close();
  }
}

/**
 * `celilo events repair` / `celilo events resume` — run the
 * crash-recovery sweep. Resets stuck `running` deliveries to `pending`
 * so the next dispatcher tick picks them up. `resume` is the
 * operator-facing alias for the halt-on-recovery acknowledgment flow.
 */
export async function handleEventsRepair(): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const result = recoverFromCrash(bus);
    return jsonResult({
      recovered: result.recovered,
      stuckCount: result.stuckCount,
      lastHeartbeatAgeMs: result.lastHeartbeatAgeMs,
    });
  } finally {
    bus.close();
  }
}

const INTERVIEW_FAMILIES = ['config', 'secret', 'ensure', 'aspect', 'interview'] as const;
type InterviewFamily = (typeof INTERVIEW_FAMILIES)[number];

/** Classify a query event type into its interview family, or null if it isn't one. */
function interviewFamily(type: string): InterviewFamily | null {
  if (type.startsWith('config.required.')) return 'config';
  if (type.startsWith('secret.required.')) return 'secret';
  if (type.startsWith('ensure.required.')) return 'ensure';
  if (type.startsWith('aspect.required.')) return 'aspect';
  if (type.startsWith('interview.required.')) return 'interview';
  return null;
}

/**
 * `celilo events reply <query-event-id> <value-json>` — answer ONE pending
 * interview query by its event id. The one-shot reply primitive a
 * `claude-config-responder` uses: find the question with
 * `celilo events list-unanswered`, ask the operator, emit the answer here.
 * (It used to say "read the log with `events tail --type '…'`" — hand-scraping
 * the log because no command listed unanswered questions. `list-unanswered` is
 * that command; it also names the parked session, which `tail` cannot.)
 * Unlike `events respond` (which must be subscribed BEFORE the query is
 * emitted — bus watches don't replay history) this looks the query up by id
 * and emits a correlated reply carrying `replyFor`, which plain `events emit`
 * can't set.
 *
 * The query's event type selects how `<value-json>` is interpreted:
 *   - config.required.<m>.<k>  → value is the answer itself ('"foo"', '8080',
 *                                '["a","b"]'); replies { value }.
 *   - secret.required.<m>.<k>  → value is the secret (string, or JSON object for
 *                                structured secrets). Written out-of-band to the
 *                                encrypted store; replies { acknowledged: true }
 *                                so the value never lands on the bus.
 *   - ensure.required.<p>.<id> → value is an object keyed by each input `target`
 *                                (e.g. '{"config.x":"v","secret.y":"v"}'). Config
 *                                targets go in the reply's `values`; secret targets
 *                                are written out-of-band (merged into the JSON
 *                                object keyed by the input's objectKey); replies
 *                                { values, acknowledged? }.
 *   - aspect.required.<m>.<r>  → value is a boolean — 'true' approves the
 *                                base-module aspect, 'false' refuses it (ISS-0027).
 *                                Replies { consented }; the deploy records the
 *                                approval/denial.
 *   - interview.required.<scope>.<key> → value is the answer itself, shaped per
 *                                the question's kind ('"node3"', 'true',
 *                                '["a","b"]'); replies { value }. The generic
 *                                operator-command interview family (ISS-0127).
 */
export async function handleEventsReply(
  args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const idArg = getArg(args, 0);
  const valueArg = getArg(args, 1);
  if (!idArg || valueArg === undefined) {
    return {
      success: false,
      error: `Usage: celilo events reply <query-event-id> <value-json>
  e.g. celilo events reply 42 '"example.net"'`,
    };
  }
  const queryId = Number(idArg);
  if (!Number.isInteger(queryId)) {
    return { success: false, error: 'query-event-id must be an integer (from `events tail`)' };
  }

  let value: unknown;
  try {
    value = JSON.parse(valueArg);
  } catch (err) {
    return {
      success: false,
      error: `Invalid JSON value: ${err instanceof Error ? err.message : String(err)}
  Encode the answer as JSON, e.g. '"example.net"', '8080', '["a","b"]'.`,
    };
  }

  const emittedBy =
    typeof flags['emitted-by'] === 'string' ? flags['emitted-by'] : 'claude-config-responder';

  const bus = openCliBus();
  try {
    const query = bus.getEvent(queryId);
    if (!query) {
      return {
        success: false,
        error: `No event with id ${queryId} on the bus (check \`celilo events tail\`).`,
      };
    }

    const family = interviewFamily(query.type);
    if (!family) {
      return {
        success: false,
        error: `Event ${queryId} is type '${query.type}', not an interview query.
  Expected one of: config.required.* / secret.required.* / ensure.required.* / aspect.required.* / interview.required.*`,
      };
    }

    // First-reply-wins: if a reply already landed for this query, don't
    // double-answer (and don't re-write a secret). Report who answered so the
    // caller's loop can move on rather than treat it as an error.
    const existingReply = bus
      .recentEvents({ limit: 500, type: `${query.type}.reply` })
      .find((e) => e.replyFor === queryId);
    if (existingReply) {
      return jsonResult({
        ok: true,
        status: 'already-answered',
        queryId,
        type: query.type,
        repliedBy: existingReply.emittedBy ?? null,
      });
    }

    if (family === 'config' || family === 'interview') {
      bus.emitRaw(`${query.type}.reply`, { value }, { replyFor: queryId, emittedBy });
      return jsonResult({ ok: true, status: 'replied', queryId, type: query.type, family, value });
    }

    if (family === 'secret') {
      const payload = query.payload as { module?: unknown; key?: unknown };
      if (typeof payload?.module !== 'string' || typeof payload?.key !== 'string') {
        return {
          success: false,
          error: `Secret query ${queryId} has a malformed payload (missing module/key).`,
        };
      }
      const { getOrCreateMasterKey } = await import('../../secrets/master-key');
      const { writeModuleSecretKey } = await import('../../services/config-interview');
      const plaintext = typeof value === 'string' ? value : JSON.stringify(value);
      const masterKey = await getOrCreateMasterKey();
      await writeModuleSecretKey(payload.module, payload.key, plaintext, getDb(), masterKey);
      bus.emitRaw(`${query.type}.reply`, { acknowledged: true }, { replyFor: queryId, emittedBy });
      return jsonResult({
        ok: true,
        status: 'replied',
        queryId,
        type: query.type,
        family,
        secretWritten: `${payload.module}.${payload.key}`,
      });
    }

    if (family === 'aspect') {
      // Aspect-consent (ISS-0027): the reply carries only the decision; the
      // deploy process records the approval/denial (it holds module/version/
      // scope). Config-like — `'true'` approves, `'false'` refuses.
      if (typeof value !== 'boolean') {
        return {
          success: false,
          error: `Aspect-consent reply must be a JSON boolean — 'true' to approve the aspect, 'false' to refuse. Got: ${valueArg}`,
        };
      }
      bus.emitRaw(`${query.type}.reply`, { consented: value }, { replyFor: queryId, emittedBy });
      return jsonResult({
        ok: true,
        status: 'replied',
        queryId,
        type: query.type,
        family,
        consented: value,
      });
    }

    // family === 'ensure'
    const payload = query.payload as EnsureRequiredPayload;
    if (typeof payload?.provider !== 'string' || !Array.isArray(payload?.inputs)) {
      return {
        success: false,
        error: `Ensure query ${queryId} has a malformed payload (missing provider/inputs).`,
      };
    }
    if (typeof value !== 'object' || value === null || Array.isArray(value)) {
      const targets = payload.inputs.map((i) => i.target).join(', ');
      return {
        success: false,
        error: `Ensure reply must be a JSON object keyed by input target, e.g.
  '{"config.foo":"x","secret.bar":"y"}'
  Required targets: ${targets}`,
      };
    }
    const provided = value as Record<string, unknown>;

    const { getOrCreateMasterKey } = await import('../../secrets/master-key');
    const { readModuleSecretKey, writeModuleSecretKey } = await import(
      '../../services/config-interview'
    );
    const db = getDb();
    const replyValues: Record<string, unknown> = {};
    let acknowledged = false;
    let masterKey: Buffer | null = null;

    for (const input of payload.inputs) {
      const providedValue = provided[input.target];
      if (providedValue === undefined) {
        const targets = payload.inputs.map((i) => i.target).join(', ');
        return {
          success: false,
          error: `Ensure reply is missing target '${input.target}'. Provide all of: ${targets}`,
        };
      }

      if (input.target.startsWith('config.')) {
        replyValues[input.target] = providedValue;
        continue;
      }

      // Secret target: read-merge-write the JSON-encoded secret object so we
      // never clobber sibling keys, then ack (value stays off the bus).
      const name = input.target.slice('secret.'.length);
      if (!masterKey) masterKey = await getOrCreateMasterKey();
      let obj: Record<string, unknown> = {};
      const currentRaw = await readModuleSecretKey(payload.provider, name, db, masterKey);
      if (currentRaw) {
        try {
          const parsed = JSON.parse(currentRaw);
          if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
            obj = parsed as Record<string, unknown>;
          }
        } catch {
          /* malformed existing secret — overwrite */
        }
      }
      obj[input.objectKey] =
        typeof providedValue === 'string' ? providedValue : JSON.stringify(providedValue);
      await writeModuleSecretKey(payload.provider, name, JSON.stringify(obj), db, masterKey);
      acknowledged = true;
    }

    bus.emitRaw(
      `${query.type}.reply`,
      acknowledged ? { values: replyValues, acknowledged: true } : { values: replyValues },
      { replyFor: queryId, emittedBy },
    );
    return jsonResult({
      ok: true,
      status: 'replied',
      queryId,
      type: query.type,
      family,
      values: replyValues,
      acknowledged,
    });
  } finally {
    bus.close();
  }
}

/**
 * `celilo events respond` — start a responder against the bus and
 * block. Two modes:
 *
 *   - **Interactive (default).** Same code path as the in-deploy
 *     terminal-responder, but spawned from a separate shell so an
 *     operator can answer config / secret / ensure prompts for a
 *     deploy running elsewhere. Blocks on SIGINT/SIGTERM. Use case:
 *     deploy in shell A; this shell B types the answers.
 *
 *   - **Non-interactive (`--values <file>`).** Reads a JSON values
 *     map from the given file and replies programmatically. Used by
 *     the `celilo-config-responder` AI subagent and any other
 *     scripted operator that wants to drive a deploy without typing.
 *     Exits when the bus has been quiet for `--idle-timeout` (default
 *     30s) or after `--max-duration` (default 10m), whichever comes
 *     first. Outputs a final JSON summary on stdout.
 *
 * The first reply still wins, so multiple responders racing on the
 * same query is fine — whoever answers first wins; the rest log a
 * "stale-reply" event that's audit-only.
 *
 * Values-file shape:
 * ```json
 * {
 *   "config":  { "module.key": "value", ... },
 *   "secrets": { "module.key": "value", ... },
 *   "ensures": {
 *     "provider.ensureId": {
 *       "configValues": { "config.target": "value" },
 *       "secretValues": { "secret.target": "value" }
 *     }
 *   }
 * }
 * ```
 */
export async function handleEventsRespond(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const valuesPath = typeof flags.values === 'string' ? flags.values : undefined;
  if (valuesPath) {
    return runProgrammaticResponder(valuesPath, flags);
  }

  const { startTerminalResponder } = await import('../../services/terminal-responder');
  const handle = startTerminalResponder();
  console.error(
    `[celilo events respond] terminal responder running (pid ${process.pid}). Ctrl-C to stop.`,
  );

  let stopping = false;
  const stop = (signal: string) => {
    if (stopping) return;
    stopping = true;
    console.error(`[celilo events respond] received ${signal}, stopping...`);
    handle.close();
    process.exit(0);
  };
  process.on('SIGINT', () => stop('SIGINT'));
  process.on('SIGTERM', () => stop('SIGTERM'));
  await new Promise(() => {}); // hold open
  return { success: true, message: '' }; // unreachable
}

/**
 * Read values JSON, start a programmatic responder, and exit when
 * the bus is quiet for `--idle-timeout` or after `--max-duration`.
 * Returns a CommandResult with the summary so the JSON-result CLI
 * envelope wraps it.
 */
async function runProgrammaticResponder(
  valuesPath: string,
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const { readFileSync } = await import('node:fs');
  const { startProgrammaticResponder } = await import('../../services/programmatic-responder');
  const { getDb } = await import('../../db/client');
  const { getEventBusPath } = await import('../../config/paths');

  let values: import('../../services/programmatic-responder').ResponderValues;
  try {
    values = JSON.parse(readFileSync(valuesPath, 'utf-8'));
  } catch (err) {
    return {
      success: false,
      error: `Failed to read --values file ${valuesPath}: ${err instanceof Error ? err.message : String(err)}`,
    };
  }

  const idleTimeoutMs = parseDurationMs(flags['idle-timeout'], 30_000);
  const maxDurationMs = parseDurationMs(flags['max-duration'], 600_000);

  const db = getDb();
  const startedAt = Date.now();
  const handle = startProgrammaticResponder({
    busDbPath: getEventBusPath(),
    db,
    values,
    onMissing: 'skip',
    emittedBy: typeof flags.emittedBy === 'string' ? flags.emittedBy : 'cli:respond',
  });

  console.error(
    `[celilo events respond] programmatic responder running (pid ${process.pid}, idle ${idleTimeoutMs}ms, max ${maxDurationMs}ms).`,
  );

  let stopping = false;
  let stopReason: 'idle' | 'max-duration' | 'signal' = 'idle';
  const buildSummary = (): CommandSuccess => {
    if (!stopping) {
      stopping = true;
      handle.close();
    }
    const summary = {
      exitReason: stopReason,
      durationMs: Date.now() - startedAt,
      answered: handle.answered(),
      missed: handle.missed(),
    };
    return {
      success: true,
      message: JSON.stringify(summary, null, 2),
      rawOutput: true,
      data: summary,
    };
  };

  process.on('SIGINT', () => {
    stopReason = 'signal';
    const r = buildSummary();
    console.error(r.message);
    process.exit(0);
  });
  process.on('SIGTERM', () => {
    stopReason = 'signal';
    const r = buildSummary();
    console.error(r.message);
    process.exit(0);
  });

  // Poll for idle/max-duration. Each tick: if (now - lastActivity) > idle and
  // we've seen at least one event, exit. If now - start > max-duration,
  // exit regardless.
  const pollMs = 500;
  while (true) {
    await new Promise((r) => setTimeout(r, pollMs));
    const now = Date.now();
    if (now - startedAt > maxDurationMs) {
      stopReason = 'max-duration';
      break;
    }
    if (handle.eventCount() > 0 && now - handle.lastActivityAt() > idleTimeoutMs) {
      stopReason = 'idle';
      break;
    }
  }

  return buildSummary();
}

/**
 * Parse a duration flag value. Accepts plain milliseconds (e.g.
 * `30000`), or suffixed values (`30s`, `5m`, `1h`). Falls back to
 * the default if the flag isn't a string.
 */
function parseDurationMs(flag: string | boolean | undefined, defaultMs: number): number {
  if (typeof flag !== 'string') return defaultMs;
  const m = flag.match(/^(\d+)(ms|s|m|h)?$/);
  if (!m) return defaultMs;
  const n = Number(m[1]);
  const unit = m[2] ?? 'ms';
  switch (unit) {
    case 'ms':
      return n;
    case 's':
      return n * 1000;
    case 'm':
      return n * 60_000;
    case 'h':
      return n * 3_600_000;
    default:
      return defaultMs;
  }
}

function daemonScopeFrom(flags: Record<string, string | boolean>): 'user' | 'system' {
  return flags.system ? 'system' : 'user';
}

/**
 * `celilo events install-daemon` — write a systemd unit (Linux) or
 * launchd plist (macOS) that runs the dispatcher under supervision.
 * Default is a per-user unit; `--system` writes the system-scope unit
 * (/etc/systemd/system, or a /Library/LaunchDaemons LaunchDaemon on
 * macOS) that survives logout/reboot and runs as the celilo state-dir
 * owner — the management-plane shape. Writes the file but doesn't
 * touch supervisor state — the operator (or the celilo-mgmt Ansible
 * role) runs the enable/bootstrap steps themselves so any change is
 * visible.
 */
export async function handleEventsInstallDaemon(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  try {
    const options = {
      celiloPath: typeof flags['celilo-path'] === 'string' ? flags['celilo-path'] : undefined,
      pollMs: flags['poll-ms'] ? Number(flags['poll-ms']) : undefined,
      concurrency: flags.concurrency ? Number(flags.concurrency) : undefined,
      scope: daemonScopeFrom(flags),
    };
    // --print: render-only, raw unit content on stdout. The celilo-mgmt
    // Ansible role captures this and does the root-owned write itself
    // (the deb wrapper runs celilo as the unprivileged celilo user, so
    // a direct --system write would EACCES on /etc/systemd/system).
    if (flags.print) {
      const plan = planDaemonInstall(options);
      return { success: true, message: plan.unitContent, rawOutput: true, data: plan };
    }
    const result = installDaemon(options);
    const lines = [
      `Wrote ${result.platform} ${result.scope} unit: ${shortenPath(result.unitPath)}`,
      `  celilo:   ${shortenPath(result.celiloPath)}`,
      `  bus DB:   ${shortenPath(result.busDbPath)}`,
      ...(result.runAsUser ? [`  runs as:  ${result.runAsUser}`] : []),
      '',
      'Next steps (run these yourself — install does not touch supervisor state):',
      ...result.nextSteps.map((s) => `  ${s}`),
    ];
    return { success: true, message: lines.join('\n'), data: result };
  } catch (err) {
    return {
      success: false,
      error: err instanceof Error ? err.message : String(err),
    };
  }
}

/**
 * `celilo events uninstall-daemon` — remove the supervisor unit file
 * (`--system` for the system-scope unit). Symmetrical: prints the
 * operator-facing disable steps, doesn't run them.
 */
export async function handleEventsUninstallDaemon(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  try {
    const result = uninstallDaemon({ scope: daemonScopeFrom(flags) });
    const lines = [
      result.removed
        ? `Removed unit: ${shortenPath(result.unitPath)}`
        : `No unit file at ${shortenPath(result.unitPath)} — nothing to remove.`,
      '',
      'Next steps:',
      ...result.nextSteps.map((s) => `  ${s}`),
    ];
    return { success: true, message: lines.join('\n'), data: result };
  } catch (err) {
    return {
      success: false,
      error: err instanceof Error ? err.message : String(err),
    };
  }
}

/**
 * `celilo events restart-daemon` — cycle the dispatcher through its supervisor
 * and PROVE the new process is live on the installed code (celilo#604).
 *
 * The interesting case is the orphan: a dispatcher the supervisor doesn't own
 * can't be killed by `systemctl restart`, and the one-dispatcher-per-bus guard
 * (#584) then crash-loops the unit while the old code keeps serving. So this
 * stops unmanaged dispatchers first, and verifies on the bus afterwards rather
 * than trusting systemctl's exit code.
 */
export async function handleEventsRestartDaemon(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const bus = openCliBus();
  try {
    const platform = detectPlatform();
    // No unit installed at all. apt-upgrade runs this on every box, including
    // ones that never installed the daemon — failing those would break an
    // upgrade that has nothing stale to fix. Distinguish the two cases:
    // nothing running is a genuine no-op; something running is unsupervised
    // and may be stale, and there is no supervisor to cycle it through.
    if (!flags.system && !unitInstalledInAnyScope(platform)) {
      const live = bus.liveDispatchers();
      if (live.length === 0) {
        return {
          success: true,
          message: 'No supervisor unit installed and no dispatcher running — nothing to restart.',
        };
      }
      return {
        success: false,
        error: `A dispatcher is running unsupervised (pid ${live.map((d) => d.pid).join(', ')}, code v${live[0]?.version ?? '?'}) and no supervisor unit is installed, so it cannot be cycled — it may be serving stale code. Run \`celilo events install-daemon\`, enable the unit, then retry.`,
      };
    }
    const scope = resolveRestartScope({ platform, scope: flags.system ? 'system' : undefined });
    const cmds = supervisorCommands(platform, scope);
    const result = await restartDaemon(
      { platform, scope, expectedVersion: BUS_VERSION },
      {
        liveDispatchers: () =>
          bus.liveDispatchers().map((d) => ({ pid: d.pid, version: d.version })),
        supervisorPid: () => {
          if (!cmds.mainPid) return null;
          const out = spawnSync(cmds.mainPid[0], cmds.mainPid.slice(1), { encoding: 'utf-8' });
          const pid = Number((out.stdout ?? '').trim());
          return Number.isInteger(pid) && pid > 0 ? pid : null;
        },
        kill: (pid, signal) => {
          try {
            process.kill(pid, signal);
          } catch {
            // Already gone, or not ours to signal — the bus poll is the arbiter.
          }
        },
        restartUnit: () => {
          const out = spawnSync(cmds.restart[0], cmds.restart.slice(1), { encoding: 'utf-8' });
          if (out.status !== 0) {
            throw new Error(
              `${cmds.restart.join(' ')} failed (exit ${out.status ?? 'signal'}): ${(out.stderr ?? '').trim()}`,
            );
          }
        },
        sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
      },
    );
    const lines = [
      `Dispatcher restarted under the ${result.scope}-scope unit: ${shortenPath(result.unitPath)}`,
      `  now running: pid ${result.dispatcher.pid}, code v${result.dispatcher.version}`,
      ...(result.orphansKilled.length > 0
        ? [
            `  stopped ${result.orphansKilled.length} unsupervised dispatcher(s): pid ${result.orphansKilled.join(', ')}`,
          ]
        : []),
    ];
    return { success: true, message: lines.join('\n'), data: result };
  } catch (err) {
    return { success: false, error: err instanceof Error ? err.message : String(err) };
  } finally {
    bus.close();
  }
}

/**
 * `celilo events show-daemon` — print whatever unit file is currently
 * installed (`--system` for the system-scope unit) so the operator can
 * see exactly what the supervisor will run, without grepping into
 * platform-specific paths.
 */
export async function handleEventsShowDaemon(
  _args: string[],
  flags: Record<string, string | boolean>,
): Promise<CommandResult> {
  const scope = daemonScopeFrom(flags);
  const result = readInstalledUnit({ scope });
  if (!result.exists) {
    return {
      success: true,
      message: `No supervisor unit installed at ${shortenPath(result.path)}\n\nRun \`celilo events install-daemon${scope === 'system' ? ' --system' : ''}\` to create one.`,
    };
  }
  return {
    success: true,
    message: `${shortenPath(result.path)}:\n\n${result.content}`,
    data: { path: result.path, content: result.content },
  };
}

function resolveSubscriberId(
  bus: ReturnType<typeof openCliBus>,
  eventId: number,
  flags: Record<string, string | boolean>,
): number | null {
  if (typeof flags.subscriber === 'string' || typeof flags.subscriber === 'number') {
    return Number(flags.subscriber);
  }
  const row = bus.db
    .query<{ subscriber_id: number }, [number]>(
      `SELECT subscriber_id FROM deliveries
        WHERE event_id = ? AND status = 'running'
        ORDER BY started_at DESC LIMIT 1`,
    )
    .get(eventId);
  return row?.subscriber_id ?? null;
}