import { randomUUID } from 'node:crypto';
import { spawnSync } from 'node:child_process';
import {
  closeSync,
  openSync,
  readFileSync,
  renameSync,
  unlinkSync,
  writeFileSync,
} from 'node:fs';
import { join } from 'node:path';
import type { AuthProfile } from '../../lib/auth.js';
import { GameClient } from '../../lib/game-client.js';
import { getProfileRuntimeStateDir } from '../../lib/init-command.js';
import { readWorkspacePerceptionConfig } from '../../lib/perception-config.js';
import {
  isRoomMode,
  isTrainingRoomMode,
  queueMode,
  queueRoomCode,
  queueStatus,
} from '../../lib/room.js';
import type { RuntimeEpochPlugin } from '../framework/plugins/index.js';
import { isMailboxRepository } from '../framework/mailbox/index.js';
import type { RuntimeReadyView } from '../framework/runtime.js';
import {
  MailboxRelayBroker,
  runStdoutMailboxRelay,
} from '../mailbox-host-relay.js';
import {
  gameStartRuntimePath,
  readGameStartRuntime,
  sendProfileOwnerControlRequest,
  startOwnerControlServer,
  type OwnerControlServer,
} from '../owner-control.js';
import {
  createOwnerTelemetryRecorder,
  observabilityRetention,
  resolveOwnerObservabilityLevel,
} from '../observability/index.js';
import {
  watchParentProcess,
  type ParentProcessWatchHandle,
} from '../parent-process-watch.js';
import { BufferedRawWsLog, ownerRawWsLogPath } from '../raw-ws-log.js';
import { presentCanonicalEventForCclEvents } from '../../pipeline/canonical-event-presentation.js';
import { compactStateForEvents } from '../../pipeline/event-format.js';
import { createClawclawRuntime } from './runtime.js';
import { assembleClawclawPlugins } from './plugins/assembly.js';
import {
  ClawclawMailboxPlugin,
  ClawclawMailboxService,
  CompositeClawclawMailboxDriver,
  DesktopEventSocketSink,
  AgentMailboxCoordinator,
  MailboxEventBatchPump,
  type DesktopEventSocketConfig,
} from './plugins/mailbox/index.js';
import {
  resolveDefaultClawclawStrategies,
  resolveClawclawStrategySource,
} from './plugins/strategy/index.js';
import { syncOfficialStrategies } from '../../lib/strategy-export.js';
import {
  clawclawQueueGameId,
  ClawclawSourcePlugin,
  normalizeClawclawQueueState,
  type ClawclawSourceCompletion,
  type ClawclawSourceStartIntent,
} from './plugins/source/index.js';
import {
  type ClawclawOwnerNotificationHost,
} from './notifications.js';

export interface ClawclawOwnerOptions {
  profile: AuthProfile;
  intent: ClawclawSourceStartIntent;
  forceExistingRuntime?: boolean;
  epochPlugins?: RuntimeEpochPlugin<Record<string, any>, Record<string, unknown>>[];
  notificationHost?: ClawclawOwnerNotificationHost;
  /** Desktop Host adapter for this run; Channels can be wired as a sibling adapter later. */
  desktopEvents?: DesktopEventSocketConfig;
}

export interface ClawclawOwnerRuntimeStart {
  initialQueueState: Record<string, any>;
  initialSegment: { kind: string; identity?: string };
  mode?: string;
  resume: boolean;
  resumeKey?: string;
}

/**
 * Derives backend participation continuity from one authoritative Lobby
 * observation. `resume` permits recovery of a matching unsealed scope after an
 * unclean process loss; a normally stopped Owner has already sealed its local
 * scope and the next Owner starts a new one. Physical WebSocket connections are
 * never restored here.
 */
export function deriveClawclawOwnerRuntimeStart(
  queueValue: unknown,
  intent: ClawclawSourceStartIntent,
): ClawclawOwnerRuntimeStart {
  const initialQueueState = normalizeClawclawQueueState(queueValue);
  const status = queueStatus(initialQueueState);
  const mode = queueMode(initialQueueState);
  const roomCode = queueRoomCode(initialQueueState);
  const gameId = clawclawQueueGameId(initialQueueState);

  if (status === 'in_room') {
    const switchingRooms = intent.kind === 'join_room' && intent.roomCode !== roomCode;
    if (!switchingRooms && roomCode) {
      return {
        initialQueueState,
        initialSegment: { kind: 'room', identity: roomCode },
        mode: mode ?? 'room',
        resume: true,
        resumeKey: roomCode,
      };
    }
  }

  if (status === 'allocated' && gameId) {
    return {
      initialQueueState,
      initialSegment: { kind: 'game', identity: gameId },
      ...(mode ? { mode } : {}),
      // Match sessions use game_id as their session key. Room sessions keep
      // room_code as the session key, but can still match the current game
      // segment by game_id when /queue/status omits room_code during play.
      resume: true,
      ...(isRoomMode(mode)
        ? roomCode ? { resumeKey: roomCode } : {}
        : { resumeKey: gameId }),
    };
  }

  if (
    (status === 'queued' || status === 'already_in_queue' || status === 'allocating')
    && mode
  ) {
    return {
      initialQueueState,
      initialSegment: { kind: 'matching', identity: mode },
      mode,
      resume: true,
    };
  }

  return {
    initialQueueState,
    initialSegment: { kind: 'bootstrap' },
    resume: false,
  };
}

export interface OwnerLock {
  release(): void;
}

interface OwnerRuntimePointer {
  release(): void;
}

const MAILBOX_DRAIN_TIMEOUT_MS = 2_000;
const DESKTOP_RESOURCE_LOCALE = 'zh-CN';
const EVENTS_QUERY_BARRIER_TIMEOUT_MS = 10_000;
const LOCAL_STOP_FORCE_EXIT_MS = 10_000;
const FORCE_STOP_GRACE_MS = 1_500;
const FORCE_STOP_TERM_MS = 1_500;
const FORCE_STOP_KILL_MS = 2_000;
const FORCE_STOP_POLL_MS = 50;

interface EventsQueryBoundaryResult {
  ready: RuntimeReadyView;
  querySeq: number;
  sourceBoundaryMs: number;
  pluginReadyMs: number;
  totalMs: number;
}

interface PreparedEventsQueryTrace {
  querySeq: number;
  ownerStartedAtMs: number;
  ownerPreparedAtMs: number;
}

function eventsQueryId(value: unknown): string {
  if (typeof value === 'string' && /^[A-Za-z0-9_.:-]{1,120}$/.test(value)) return value;
  return randomUUID();
}

function boundedTiming(value: unknown): number | undefined {
  return typeof value === 'number'
    && Number.isFinite(value)
    && value >= 0
    && value <= 3_600_000
    ? Math.round(value)
    : undefined;
}

function eventsOutputCount(output: Record<string, unknown>): number {
  if (Array.isArray(output.events)) return output.events.length;
  return output.found === true && output.event && typeof output.event === 'object' ? 1 : 0;
}

async function waitWithin<T>(promise: Promise<T>, timeoutMs: number, error: string): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    return await Promise.race([
      promise,
      new Promise<T>((_resolve, reject) => {
        timer = setTimeout(() => reject(new Error(error)), timeoutMs);
        timer.unref?.();
      }),
    ]);
  } finally {
    if (timer) clearTimeout(timer);
  }
}

async function waitForMailboxDrain(
  broker: MailboxRelayBroker,
  timeoutMs = MAILBOX_DRAIN_TIMEOUT_MS,
): Promise<boolean> {
  const deadline = Date.now() + timeoutMs;
  while (broker.hasPendingCurrent()) {
    if (Date.now() >= deadline) return false;
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  return true;
}

function processAlive(pid: number): boolean {
  if (!Number.isSafeInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

function terminateRuntimeProcessTree(pid: number, signal: NodeJS.Signals): boolean {
  if (!Number.isSafeInteger(pid) || pid <= 0 || pid === process.pid) return false;
  if (process.platform === 'win32') {
    const systemRoot = process.env.SystemRoot ?? process.env.SYSTEMROOT ?? 'C:\\Windows';
    const result = spawnSync(join(systemRoot, 'System32', 'taskkill.exe'), [
      '/PID',
      String(pid),
      '/T',
      '/F',
    ], {
      stdio: 'ignore',
      windowsHide: true,
      timeout: FORCE_STOP_TERM_MS,
    });
    return result.status === 0 || !processAlive(pid);
  }

  const childSignal = signal === 'SIGKILL' ? '-KILL' : '-TERM';
  try {
    spawnSync('pkill', [childSignal, '-P', String(pid)], {
      stdio: 'ignore',
      timeout: FORCE_STOP_TERM_MS,
    });
  } catch {}
  try {
    process.kill(pid, signal);
    return true;
  } catch {
    return !processAlive(pid);
  }
}

async function waitForProcessExit(
  pid: number,
  timeoutMs: number,
  isAlive = processAlive,
): Promise<boolean> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (!isAlive(pid)) return true;
    await new Promise((resolve) => setTimeout(resolve, FORCE_STOP_POLL_MS));
  }
  return !isAlive(pid);
}

export interface OwnerLockOperations {
  processAlive(pid: number): boolean;
  requestStop(
    profile: Pick<AuthProfile, 'apiKey' | 'runtimeAccountId'>,
    expectedPid: number,
  ): Promise<void>;
  terminateProcessTree(pid: number, signal: NodeJS.Signals): boolean;
  waitForExit(pid: number, timeoutMs: number): Promise<boolean>;
}

const defaultOwnerLockOperations: OwnerLockOperations = {
  processAlive,
  requestStop: async (profile, expectedPid) => {
    const stateDir = getProfileRuntimeStateDir(profile);
    const pointer = readGameStartRuntime(stateDir);
    const pointerPid = Number(pointer?.owner_pid ?? pointer?.pid);
    if (pointerPid !== expectedPid) return;
    try {
      await sendProfileOwnerControlRequest(profile, 'stop');
    } catch {}
  },
  terminateProcessTree: terminateRuntimeProcessTree,
  waitForExit: (pid, timeoutMs) => waitForProcessExit(pid, timeoutMs),
};

function readOwnerLockPid(path: string): number | null {
  try {
    const pid = Number((JSON.parse(readFileSync(path, 'utf8')) as { pid?: number }).pid);
    return Number.isSafeInteger(pid) && pid > 0 ? pid : null;
  } catch {
    return null;
  }
}

function removeOwnerLockForPid(path: string, pid: number): boolean {
  try {
    if (readOwnerLockPid(path) !== pid) return false;
    unlinkSync(path);
    return true;
  } catch {
    return false;
  }
}

function removeRuntimePointerForPid(stateDir: string, pid: number): void {
  const pointer = readGameStartRuntime(stateDir);
  const pointerPid = Number(pointer?.owner_pid ?? pointer?.pid);
  if (pointerPid !== pid) return;
  const controlPath = typeof pointer?.control?.path === 'string' ? pointer.control.path : '';
  try { unlinkSync(gameStartRuntimePath(stateDir)); } catch {}
  if (process.platform !== 'win32' && controlPath) {
    try { unlinkSync(controlPath); } catch {}
  }
}

async function stopExistingRuntime(
  profile: Pick<AuthProfile, 'apiKey' | 'runtimeAccountId'>,
  pid: number,
  operations: OwnerLockOperations,
): Promise<void> {
  try { await operations.requestStop(profile, pid); } catch {}
  if (await operations.waitForExit(pid, FORCE_STOP_GRACE_MS)) return;

  operations.terminateProcessTree(pid, 'SIGTERM');
  if (await operations.waitForExit(pid, FORCE_STOP_TERM_MS)) return;

  operations.terminateProcessTree(pid, 'SIGKILL');
  if (await operations.waitForExit(pid, FORCE_STOP_KILL_MS)) return;

  throw new Error(`Unable to replace the running ClawClaw runtime (pid ${pid}).`);
}

export async function acquireOwnerLock(
  profile: Pick<AuthProfile, 'apiKey' | 'runtimeAccountId'>,
  options: {
    force?: boolean;
    operations?: OwnerLockOperations;
  } = {},
): Promise<OwnerLock> {
  const stateDir = getProfileRuntimeStateDir(profile);
  const path = join(stateDir, 'owner.lock');
  const operations = options.operations ?? defaultOwnerLockOperations;
  let replacementPid: number | undefined;
  for (let attempt = 0; attempt < 4; attempt += 1) {
    try {
      const fd = openSync(path, 'wx');
      writeFileSync(fd, JSON.stringify({ pid: process.pid, started_at: new Date().toISOString() }));
      closeSync(fd);
      let released = false;
      return {
        release: () => {
          if (released) return;
          released = true;
          try {
            const current = JSON.parse(readFileSync(path, 'utf8')) as { pid?: number };
            if (current.pid === process.pid) unlinkSync(path);
          } catch {}
        },
      };
    } catch (error: any) {
      if (error?.code !== 'EEXIST') throw error;
      const ownerPid = readOwnerLockPid(path);
      if (ownerPid === null) {
        try { unlinkSync(path); } catch {}
        continue;
      }
      if (!operations.processAlive(ownerPid)) {
        removeRuntimePointerForPid(stateDir, ownerPid);
        removeOwnerLockForPid(path, ownerPid);
        continue;
      }
      if (!options.force) {
        throw new Error(
          `A ClawClaw runtime is already running (pid ${ownerPid}). Re-run with --force to replace it.`,
        );
      }
      if (ownerPid === process.pid) {
        throw new Error('Refusing to replace the current ClawClaw runtime process.');
      }
      if (replacementPid !== undefined && replacementPid !== ownerPid) {
        throw new Error(
          `The active ClawClaw runtime changed from pid ${replacementPid} to pid ${ownerPid}; refusing to replace the new process.`,
        );
      }
      replacementPid = ownerPid;
      await stopExistingRuntime(profile, ownerPid, operations);
      if (operations.processAlive(ownerPid)) {
        throw new Error(`Unable to replace the running ClawClaw runtime (pid ${ownerPid}).`);
      }
      removeRuntimePointerForPid(stateDir, ownerPid);
      removeOwnerLockForPid(path, ownerPid);
    }
  }
  throw new Error('Unable to acquire the ClawClaw runtime owner lock.');
}

function publishOwnerRuntimePointer(
  stateDir: string,
  control: OwnerControlServer,
): OwnerRuntimePointer {
  const path = gameStartRuntimePath(stateDir);
  const tempPath = `${path}.${process.pid}.tmp`;
  const value = {
    schema: 'clawclaw.owner.v1',
    owner_pid: process.pid,
    started_at: new Date().toISOString(),
    control: control.control,
    control_token: control.token,
    relay_token: control.relayToken,
  };
  writeFileSync(tempPath, JSON.stringify(value));
  try {
    renameSync(tempPath, path);
  } catch (error) {
    try {
      unlinkSync(path);
      renameSync(tempPath, path);
    } catch {
      try { unlinkSync(tempPath); } catch {}
      throw error;
    }
  }
  let released = false;
  return {
    release: () => {
      if (released) return;
      released = true;
      try {
        const current = JSON.parse(readFileSync(path, 'utf8')) as Record<string, any>;
        if (current.owner_pid === process.pid && current.control?.path === control.control.path) {
          unlinkSync(path);
        }
      } catch {}
      try { unlinkSync(tempPath); } catch {}
    },
  };
}

/** Runs one foreground owner with the explicitly configured Runtime Plugins. */
export async function runClawclawOwner(
  options: ClawclawOwnerOptions,
): Promise<ClawclawSourceCompletion> {
  let lock: OwnerLock | undefined;
  let telemetry: ReturnType<typeof createOwnerTelemetryRecorder> | undefined;
  let rawWsLog: BufferedRawWsLog | undefined;
  let runtime: ReturnType<typeof createClawclawRuntime> | undefined;
  let source: ClawclawSourcePlugin | undefined;
  let mailbox: ClawclawMailboxPlugin | undefined;
  let mailboxService: ClawclawMailboxService | undefined;
  let agentMailbox: AgentMailboxCoordinator | undefined;
  let desktopEventPump: MailboxEventBatchPump | undefined;
  let mailboxBroker: MailboxRelayBroker | undefined;
  let ownerControl: OwnerControlServer | undefined;
  let runtimePointer: OwnerRuntimePointer | undefined;
  let relayAlive = false;
  let relayAbort: AbortController | undefined;
  let relayPromise: Promise<void> | undefined;
  let parentWatch: ParentProcessWatchHandle | undefined;
  let localStopRequested = false;
  let forceExitTimer: ReturnType<typeof setTimeout> | undefined;
  const requestLocalStop = (): void => {
    if (localStopRequested) return;
    localStopRequested = true;
    forceExitTimer = setTimeout(() => process.exit(1), LOCAL_STOP_FORCE_EXIT_MS);
    forceExitTimer.unref?.();
    void source?.submitIntent({ kind: 'stop' }).catch(() => {});
  };
  const onSignal = (): void => requestLocalStop();

  try {
    lock = await acquireOwnerLock(options.profile, {
      force: options.forceExistingRuntime === true,
    });
    telemetry = createOwnerTelemetryRecorder(options.profile);
    if (resolveOwnerObservabilityLevel() === 'full') {
      const retention = observabilityRetention();
      rawWsLog = new BufferedRawWsLog(
        ownerRawWsLogPath(getProfileRuntimeStateDir(options.profile)),
        retention,
      );
      telemetry.log('info', 'runtime.raw_ws.capture_started', {
        max_file_bytes: retention.maxFileBytes,
        retained_files: retention.retainedFiles,
      }, { component: 'runtime.transport' });
    }
    const client = new GameClient({
      lobbyUrl: options.profile.serverUrl,
      apiKey: options.profile.apiKey,
      agentName: options.profile.agentName,
      enableWs: true,
      rawWsMessageSink: rawWsLog,
    });
    const perceptionConfig = readWorkspacePerceptionConfig();
    await syncOfficialStrategies();
    const defaultStrategies = resolveDefaultClawclawStrategies();
    const assembly = assembleClawclawPlugins(client, {
      epochPlugins: options.epochPlugins,
      defaultStrategies,
      perception: perceptionConfig.perception,
    });
    const epochPlugins = assembly.plugins;
    mailbox = assembly.mailbox;
    const runtimeStart = deriveClawclawOwnerRuntimeStart(
      await client.getQueueStatus('clawclaw'),
      options.intent,
    );
    runtime = createClawclawRuntime(options.profile, {
      initialSegment: runtimeStart.initialSegment,
      mode: runtimeStart.mode,
      resume: runtimeStart.resume,
      resumeKey: runtimeStart.resumeKey,
      observability: telemetry,
      runtimeEpochPlugins: epochPlugins,
      visualEventDelivery: perceptionConfig.events.delivery,
      visualEventTypes: assembly.perception.produces.events,
    });
    if (!isMailboxRepository(runtime.repository)) {
      throw new Error('The configured Runtime repository does not support durable Mailbox delivery.');
    }
    const mailboxRepository = runtime.repository;
    agentMailbox = new AgentMailboxCoordinator({
      repository: mailboxRepository,
      readyView: () => runtime!.readReadyView(0, 1),
      presentEvent: presentCanonicalEventForCclEvents,
      compactState: compactStateForEvents,
      diagnostic: (name, attributes) => runtime?.diagnostic(name, attributes),
    });
    mailboxService = new ClawclawMailboxService({
      repository: mailboxRepository,
      readyView: () => runtime!.readReadyView(0, 1),
      agentMailbox,
      monitorEnabled: !options.desktopEvents,
      diagnostic: (name, attributes) => runtime?.diagnostic(name, attributes),
    });
    if (options.desktopEvents) {
      const desktopSink = new DesktopEventSocketSink({
        ...options.desktopEvents,
        activeResources: () => ({
          locale: DESKTOP_RESOURCE_LOCALE,
          strategy: assembly.strategy.activeResourceDescriptor(DESKTOP_RESOURCE_LOCALE),
          perceptions: assembly.perception.localizedModuleDescriptors(DESKTOP_RESOURCE_LOCALE),
        }),
        eventDescriptors: () => assembly.perception
          .localizedEventDescriptors(DESKTOP_RESOURCE_LOCALE)
          .map((descriptor) => ({
            type: descriptor.type,
            perception_id: descriptor.perceptionId,
            name: descriptor.name,
            description: descriptor.description,
          })),
        onConnected: () => desktopEventPump?.retryNow(),
        diagnostic: (name, attributes) => runtime?.diagnostic(name, attributes),
      });
      desktopEventPump = new MailboxEventBatchPump({
        source: {
          prepare: () => agentMailbox!.prepareEvents({
            deliveryVia: 'mailbox_full',
            ownerIdPrefix: 'desktop_ipc',
          }),
          renew: (deliveryToken) => agentMailbox!.renew(deliveryToken),
          ack: (deliveryToken) => agentMailbox!.ack(deliveryToken),
          release: (deliveryToken) => agentMailbox!.release(deliveryToken),
        },
        sink: desktopSink,
        diagnostic: (name, attributes) => runtime?.diagnostic(name, attributes),
      });
      mailbox.attachDriver(new CompositeClawclawMailboxDriver([
        mailboxService,
        agentMailbox,
        desktopEventPump,
      ]));
      desktopSink.start();
    } else {
      mailbox.attachDriver(new CompositeClawclawMailboxDriver([
        mailboxService,
        agentMailbox,
      ]));
    }
    const ownerSource = new ClawclawSourcePlugin({
      client,
      profile: options.profile,
      initialIntent: options.intent,
      initialQueueState: runtimeStart.initialQueueState,
    });
    source = ownerSource;
    let eventsQueryBarrier: Promise<EventsQueryBoundaryResult> | undefined;
    let eventsQuerySequence = 0;
    const preparedEventsQueryTraces = new Map<string, PreparedEventsQueryTrace>();
    const readyViewAfterEventsQueryBoundary = (queryId: string, querySeq: number) => {
      if (eventsQueryBarrier) return eventsQueryBarrier;
      const requestId = queryId;
      const startedAt = Date.now();
      const operation = (async () => {
        let epochSeq: number | undefined;
        try {
          const acceptance = await ownerSource.submitEventsQueryBoundary(requestId);
          const acceptedAt = Date.now();
          epochSeq = acceptance.epoch.epochSeq;
          runtime!.diagnostic('runtime.events_query.boundary_accepted', {
            query_seq: querySeq,
            epoch_seq: epochSeq,
            source_boundary_ms: acceptedAt - startedAt,
          });
          const ready = await waitWithin(
            runtime!.waitForEpochReadyView(epochSeq),
            EVENTS_QUERY_BARRIER_TIMEOUT_MS,
            'events_query_barrier_timeout',
          );
          runtime!.diagnostic('runtime.events_query.boundary_ready', {
            query_seq: querySeq,
            epoch_seq: ready.epochSeq,
            event_watermark: ready.eventWatermark,
            source_boundary_ms: acceptedAt - startedAt,
            plugin_ready_ms: Date.now() - acceptedAt,
            duration_ms: Date.now() - startedAt,
          });
          return {
            ready,
            querySeq,
            sourceBoundaryMs: acceptedAt - startedAt,
            pluginReadyMs: Date.now() - acceptedAt,
            totalMs: Date.now() - startedAt,
          };
        } catch (error) {
          runtime!.diagnostic('runtime.events_query.boundary_failed', {
            query_seq: querySeq,
            ...(epochSeq === undefined ? {} : { epoch_seq: epochSeq }),
            duration_ms: Date.now() - startedAt,
            error: error instanceof Error ? error.message : String(error),
          }, 'warn');
          throw error;
        }
      })();
      eventsQueryBarrier = operation.finally(() => {
        eventsQueryBarrier = undefined;
      });
      return eventsQueryBarrier;
    };
    if (options.notificationHost && !options.desktopEvents) {
      mailboxBroker = new MailboxRelayBroker(
        mailboxRepository,
        'short',
      );
      relayAbort = new AbortController();
      relayAlive = true;
      relayPromise = runStdoutMailboxRelay({
        request: async (type, payload) => {
          if (type === 'mailbox_claim') mailboxService?.prepareMonitor();
          return mailboxBroker?.handle({ type, ...payload }) ?? null;
        },
        emit: (payload) => options.notificationHost!.emit(payload),
        compose: (payloads) => mailboxService!.composeMonitor(payloads),
        isOwnerAlive: () => relayAlive,
        mode: 'short',
        signal: relayAbort.signal,
      }).catch((error) => {
        runtime?.diagnostic('runtime.mailbox.host_relay_failed', {
          error: error instanceof Error ? error.message : String(error),
        }, 'error');
      });
    }
    ownerControl = await startOwnerControlServer(
      getProfileRuntimeStateDir(options.profile),
      async (request, context) => {
        if (context.principal === 'host_relay') {
          if (request.type === 'mailbox_claim') mailboxService?.prepareMonitor();
          return mailboxBroker?.handle(request) ?? { ok: false, error: 'mailbox_monitor_unavailable' };
        }
        if (context.principal !== 'master') return { ok: false, error: 'owner_request_not_supported' };
        if (request.type === 'mailbox_claim' || request.type === 'mailbox_ack' || request.type === 'mailbox_nack') {
          if (request.type === 'mailbox_claim') mailboxService?.prepareMonitor();
          return mailboxBroker?.handle(request) ?? { ok: false, error: 'mailbox_monitor_unavailable' };
        }
        if (request.type === 'mailbox_events_prepare') {
          const ownerStartedAtMs = Date.now();
          const queryId = eventsQueryId(request.query_id);
          const querySeq = ++eventsQuerySequence;
          const eventType = typeof request.event_type === 'string' ? request.event_type : undefined;
          const mode = request.mode === 'backlog' ? 'backlog' : 'current';
          const queryKind = eventType
            ? 'latest_by_type'
            : mode === 'backlog' ? 'unread_backlog' : 'unread_current';
          let stage = 'query_boundary';
          try {
            const boundary = !eventType && mode === 'current'
              ? await readyViewAfterEventsQueryBoundary(queryId, querySeq)
              : undefined;
            const mailboxStartedAtMs = Date.now();
            stage = 'mailbox_prepare';
            const prepared = await mailboxService!.prepareEventsWhenAvailable({
              ...(eventType ? { eventType } : {}),
              mode,
            }, boundary?.ready);
            const ownerPreparedAtMs = Date.now();
            for (const [token, trace] of preparedEventsQueryTraces) {
              if (ownerPreparedAtMs - trace.ownerPreparedAtMs > 60_000) {
                preparedEventsQueryTraces.delete(token);
              }
            }
            if (prepared.deliveryToken) {
              preparedEventsQueryTraces.set(prepared.deliveryToken, {
                querySeq,
                ownerStartedAtMs,
                ownerPreparedAtMs,
              });
            }
            const state = prepared.output.state;
            const stateRecord = state && typeof state === 'object'
              ? state as Record<string, unknown>
              : undefined;
            runtime!.diagnostic('runtime.events_query.prepared', {
              query_seq: querySeq,
              query_kind: queryKind,
              ...(eventType ? { event_type: eventType } : {}),
              boundary_used: !!boundary,
              ...(boundary ? {
                boundary_query_seq: boundary.querySeq,
                boundary_shared: boundary.querySeq !== querySeq,
                source_boundary_ms: boundary.sourceBoundaryMs,
                plugin_ready_ms: boundary.pluginReadyMs,
                boundary_total_ms: boundary.totalMs,
                ready_epoch_seq: boundary.ready.epochSeq,
                event_watermark: boundary.ready.eventWatermark,
              } : {}),
              mailbox_prepare_ms: ownerPreparedAtMs - mailboxStartedAtMs,
              owner_total_ms: ownerPreparedAtMs - ownerStartedAtMs,
              event_count: eventsOutputCount(prepared.output),
              delivery_token_present: !!prepared.deliveryToken,
              ...(typeof stateRecord?.phase === 'string' ? { phase: stateRecord.phase } : {}),
              ...(typeof stateRecord?.tick === 'number' ? { tick: stateRecord.tick } : {}),
            }, 'info');
            return {
              ok: true,
              type: 'mailbox_events_prepare',
              query_id: queryId,
              output: prepared.output,
              ...(prepared.deliveryToken ? { delivery_token: prepared.deliveryToken } : {}),
            };
          } catch (error) {
            runtime!.diagnostic('runtime.events_query.prepare_failed', {
              query_seq: querySeq,
              query_kind: queryKind,
              ...(eventType ? { event_type: eventType } : {}),
              stage,
              duration_ms: Date.now() - ownerStartedAtMs,
              error: error instanceof Error ? error.message : String(error),
            }, 'warn');
            throw error;
          }
        }
        if (request.type === 'mailbox_events_ack') {
          const ackStartedAtMs = Date.now();
          const token = typeof request.delivery_token === 'string' ? request.delivery_token : '';
          const trace = preparedEventsQueryTraces.get(token);
          preparedEventsQueryTraces.delete(token);
          const accepted = mailboxService!.ackEvents(token);
          runtime!.diagnostic('runtime.events_query.client_completed', {
            ...(trace ? { query_seq: trace.querySeq } : {}),
            accepted,
            ...(trace ? {
              prepared_to_ack_ms: ackStartedAtMs - trace.ownerPreparedAtMs,
              owner_observed_total_ms: Date.now() - trace.ownerStartedAtMs,
            } : {}),
            ...(boundedTiming(request.client_elapsed_ms) === undefined ? {} : {
              client_elapsed_before_ack_ms: boundedTiming(request.client_elapsed_ms),
            }),
            ...(boundedTiming(request.prepare_rpc_ms) === undefined ? {} : {
              prepare_rpc_ms: boundedTiming(request.prepare_rpc_ms),
            }),
            ...(boundedTiming(request.stdout_write_ms) === undefined ? {} : {
              stdout_write_ms: boundedTiming(request.stdout_write_ms),
            }),
            ...(boundedTiming(request.output_characters) === undefined ? {} : {
              output_characters: boundedTiming(request.output_characters),
            }),
            ...(boundedTiming(request.output_bytes) === undefined ? {} : {
              output_bytes: boundedTiming(request.output_bytes),
            }),
            ack_processing_ms: Date.now() - ackStartedAtMs,
          }, accepted ? 'info' : 'warn');
          return accepted
            ? { ok: true, type: 'mailbox_events_ack' }
            : { ok: false, error: 'mailbox_delivery_not_found' };
        }
        if (request.type === 'mailbox_events_release') {
          const releaseStartedAtMs = Date.now();
          const token = typeof request.delivery_token === 'string' ? request.delivery_token : '';
          const trace = preparedEventsQueryTraces.get(token);
          preparedEventsQueryTraces.delete(token);
          const released = mailboxService!.releaseEvents(token);
          runtime!.diagnostic('runtime.events_query.client_aborted', {
            ...(trace ? { query_seq: trace.querySeq } : {}),
            released,
            outcome: typeof request.outcome === 'string' ? request.outcome : 'released',
            ...(trace ? {
              prepared_to_release_ms: releaseStartedAtMs - trace.ownerPreparedAtMs,
              owner_observed_total_ms: Date.now() - trace.ownerStartedAtMs,
            } : {}),
            ...(boundedTiming(request.client_elapsed_ms) === undefined ? {} : {
              client_elapsed_before_release_ms: boundedTiming(request.client_elapsed_ms),
            }),
            ...(boundedTiming(request.prepare_rpc_ms) === undefined ? {} : {
              prepare_rpc_ms: boundedTiming(request.prepare_rpc_ms),
            }),
            release_processing_ms: Date.now() - releaseStartedAtMs,
          }, 'warn');
          return released
            ? { ok: true, type: 'mailbox_events_release' }
            : { ok: false, error: 'mailbox_delivery_not_found' };
        }
        if (request.type === 'snapshot') {
          const ready = runtime!.readReadyView(0, 1);
          return {
            ok: true,
            type: 'snapshot',
            scope: ready.scope,
            state: ready.state.state,
            summary: ready.state.state,
            state_status: ready.state.stateStatus,
            ready: {
              epoch_seq: ready.epochSeq,
              frame_seq: ready.frameSeq,
              event_commit_seq: ready.eventWatermark,
            },
          };
        }
        if (request.type === 'action') {
          const capability = typeof request.capability === 'string' ? request.capability : '';
          const expected = request.expected_scope;
          if (
            !capability
            || !expected
            || typeof expected.sessionId !== 'string'
            || typeof expected.segmentId !== 'string'
          ) return { ok: false, error: 'invalid_action_request' };
          const result = await runtime!.performAction(
            capability,
            request.action,
            { sessionId: expected.sessionId, segmentId: expected.segmentId },
            { kind: 'agent' },
          );
          return { ok: true, type: 'action', result };
        }
        if (request.type === 'switch_strategy') {
          const requested = typeof request.strategy === 'string' ? request.strategy.trim() : '';
          if (!requested) return { ok: false, error: 'strategy_id_required' };
          try {
            const resolved = resolveClawclawStrategySource(requested);
            assembly.strategy.select({
              ...resolved,
            });
            return {
              ok: true,
              type: 'switch_strategy',
              strategy: resolved.id,
              effective: 'next_epoch',
            };
          } catch (error) {
            return {
              ok: false,
              error: error instanceof Error ? error.message : String(error),
            };
          }
        }
        if (request.type === 'update_strategy_control') {
          const control = request.control;
          if (!control || typeof control !== 'object' || Array.isArray(control)) {
            return { ok: false, error: 'invalid_strategy_control' };
          }
          try {
            const ack = await assembly.strategy.control(control);
            return {
              ...ack,
              ok: true,
              type: 'update_strategy_control',
            };
          } catch (error) {
            return {
              ok: false,
              type: 'update_strategy_control',
              error: error instanceof Error ? error.message : String(error),
            };
          }
        }
        if (request.type === 'stop') {
          await ownerSource.submitIntent({ kind: 'stop' });
          return { ok: true, type: request.type };
        }
        if (request.type === 'quit' || request.type === 'leave') {
          await ownerSource.submitIntent({ kind: 'leave' });
          return { ok: true, type: request.type };
        }
        if (request.type === 'training_mode') {
          if (!isTrainingRoomMode(request.room_type)) {
            return { ok: false, type: request.type, error: 'invalid_training_room_type' };
          }
          const result = await ownerSource.submitIntent({
            kind: 'training_mode',
            roomType: request.room_type,
          });
          return { ok: true, type: request.type, result };
        }
        if (request.type === 'training_role_lock') {
          if (!Number.isInteger(request.character_id) || request.character_id <= 0) {
            return { ok: false, type: request.type, error: 'invalid_character_id' };
          }
          const result = await ownerSource.submitIntent({
            kind: 'training_role_lock',
            characterId: request.character_id,
          });
          return { ok: true, type: request.type, result };
        }
        if (request.type === 'training_role_clear') {
          const result = await ownerSource.submitIntent({ kind: 'training_role_clear' });
          return { ok: true, type: request.type, result };
        }
        return { ok: false, error: 'owner_request_not_supported' };
      },
      {
        observability: telemetry,
      },
    );
    runtimePointer = publishOwnerRuntimePointer(
      getProfileRuntimeStateDir(options.profile),
      ownerControl,
    );
    parentWatch = watchParentProcess(() => {
      runtime?.diagnostic('runtime.owner.parent_exited', {
        parent_pid: parentWatch?.parentPid ?? process.ppid,
      }, 'warn');
      requestLocalStop();
    });
    process.on('SIGINT', onSignal);
    process.on('SIGTERM', onSignal);
    await runtime.startSourcePlugin(ownerSource);
    await desktopEventPump?.flush();
    if (localStopRequested) await ownerSource.submitIntent({ kind: 'stop' });
    if (!options.desktopEvents) options.notificationHost?.markMailboxReady();
    const completion = await source.waitForCompletion();
    await runtime.flush();
    await mailboxService.flush();
    await agentMailbox.flush();
    await desktopEventPump?.drain();
    if (mailboxBroker && !(await waitForMailboxDrain(mailboxBroker))) {
      runtime.diagnostic('runtime.shutdown.mailbox_drain_incomplete', {
        timeout_ms: MAILBOX_DRAIN_TIMEOUT_MS,
      }, 'warn');
    }
    return completion;
  } finally {
    process.off('SIGINT', onSignal);
    process.off('SIGTERM', onSignal);
    if (forceExitTimer) clearTimeout(forceExitTimer);
    parentWatch?.stop();
    relayAlive = false;
    relayAbort?.abort(new Error('owner_stopping'));
    await relayPromise;
    try {
      try {
        await ownerControl?.close();
      } finally {
        runtimePointer?.release();
        await runtime?.close();
      }
    } finally {
      try {
        if (rawWsLog) {
          await rawWsLog.close();
          const stats = rawWsLog.stats();
          telemetry?.log(
            stats.dropped > 0 || stats.failedBatches > 0 ? 'warn' : 'info',
            'runtime.raw_ws.capture_stopped',
            { ...stats },
            { component: 'runtime.transport' },
          );
        }
        await telemetry?.closeWithin(2_000).catch(() => false);
      } finally {
        lock?.release();
      }
    }
  }
}