import http from 'node:http';
import { randomBytes } from 'node:crypto';
import { EventEmitter } from 'node:events';
import fs from 'fs';
import path from 'path';
import type { ChildProcess } from 'child_process';

import {
  checkUsageLimit,
  deleteSession,
  getAllTasks,
  getTasksForTenant,
  getGroupByHooksToken,
  logUsage,
  setSession,
} from './db.js';
import {
  runContainerAgent,
  writeGroupsSnapshot,
  writeTasksSnapshot,
  type ContainerInput,
} from './container-runner.js';
import type { GroupQueue } from './group-queue.js';
import { resolveGroupFolderPath, resolveGroupIpcPath } from './group-folder.js';
import { logger } from './logger.js';
import { storeTurnPair } from './lcm.js';
import { ASSISTANT_NAME, HOOKS_PORT } from './config.js';
import {
  getDefaultModelForProvider,
  getTenantProvider,
} from './runtime-provider.js';
import type { RegisteredGroup, SessionRecord } from './types.js';
import type { HookIpcTask } from './ipc.js';
import type { TenantManager } from './tenant-manager.js';



/** Tools never callable via /hooks/invoke */
const INVOKE_DENY_LIST = new Set(['bash', 'computer', 'web_search']);

/** Session key suffix so hook runs use an isolated session */
const HOOK_SESSION_SUFFIX = '--hook';

/**
 * Per-group hook concurrency control. Hooks are serialized per group but do NOT
 * wait behind scheduled task containers in the main group queue. Each new hook
 * is chained after any currently-running hook for the same group.
 */
const hookChains = new Map<string, Promise<void>>();

function runHookConcurrent(groupJid: string, fn: () => Promise<void>): void {
  const prev = hookChains.get(groupJid) ?? Promise.resolve();
  const next = prev.then(() => fn()).catch((err) => {
    logger.error({ groupJid, err }, 'hooks: unhandled error in hook chain');
  });
  hookChains.set(groupJid, next);
  // Clean up resolved chains to avoid memory leaks
  next.then(() => {
    if (hookChains.get(groupJid) === next) hookChains.delete(groupJid);
  }).catch(() => {});
}

/** Timeout for synchronous /hooks/invoke (ms) */
const INVOKE_TIMEOUT_MS = 120_000;

/** Close container stdin after first result with this delay (ms) */
const HOOK_CLOSE_DELAY_MS = 5_000;

/** UUID v4 pattern for caller-supplied session IDs */
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

// ---------------------------------------------------------------------------
// Warm session tracking
// ---------------------------------------------------------------------------
interface WarmSession {
  /** Emitter for container output events */
  emitter: EventEmitter;
  /** The group folder this warm container uses (e.g. koala/hook-{uuid}) */
  groupFolder: string;
  /** Resolves when the container exits */
  done: Promise<void>;
  /** Serializes pipe calls so only one query is active at a time */
  pipeChain: Promise<void>;
}

/**
 * Map of session UUID → WarmSession.
 * Only populated by /hooks/warm, consumed by /hooks/invoke and /hooks/stream.
 */
const warmSessions = new Map<string, WarmSession>();

// ---------------------------------------------------------------------------
// On-demand hook group folder creation
// ---------------------------------------------------------------------------

/** Build the hook group folder key for a tenant + session UUID */
function hookGroupFolder(tenantId: string, sessionUuid: string): string {
  return `${tenantId}/hook-${sessionUuid}`;
}

/**
 * Ensure the on-demand hook group dirs exist with correct permissions.
 * Creates groups/, data/ipc/, data/sessions/ subdirs under the tenant home.
 */
function ensureHookGroupDirs(groupFolder: string): void {
  const groupDir = resolveGroupFolderPath(groupFolder);
  const ipcDir = resolveGroupIpcPath(groupFolder);
  const ipcInputDir = path.join(ipcDir, 'input');
  fs.mkdirSync(groupDir, { recursive: true });
  fs.mkdirSync(ipcInputDir, { recursive: true });
  // Sessions dir created by container-runner as needed
}

/**
 * Write a prompt as an IPC input message for a warm container to pick up.
 * Returns the filename written.
 */
function writeIpcPrompt(
  groupFolder: string,
  prompt: string,
  attachments?: ContainerInput['attachments'],
): string {
  const ipcDir = resolveGroupIpcPath(groupFolder);
  const inputDir = path.join(ipcDir, 'input');
  fs.mkdirSync(inputDir, { recursive: true });
  const filename = `msg-${Date.now()}-${randomBytes(3).toString('hex')}.json`;
  const filePath = path.join(inputDir, filename);
  const tmpPath = filePath + '.tmp';
  fs.writeFileSync(tmpPath, JSON.stringify({ type: 'message', text: prompt, attachments }));
  fs.renameSync(tmpPath, filePath);
  return filename;
}

export interface HooksServerDeps {
  registeredGroups: () => Record<string, RegisteredGroup>;
  sessions: Record<string, SessionRecord>;
  queue: GroupQueue;
  tenantManager: TenantManager;
  onProcess: (jid: string, proc: ChildProcess, name: string, folder: string) => void;
}

function sendJson(res: http.ServerResponse, status: number, body: unknown): void {
  const payload = JSON.stringify(body);
  res.writeHead(status, {
    'Content-Type': 'application/json',
    'Content-Length': Buffer.byteLength(payload),
  });
  res.end(payload);
}

function getBearerToken(req: http.IncomingMessage): string | null {
  const auth = req.headers['authorization'];
  if (!auth || !auth.startsWith('Bearer ')) return null;
  const token = auth.slice(7).trim();
  return token || null;
}

async function readJsonBody(req: http.IncomingMessage, maxBytes = 64 * 1024): Promise<unknown> {
  return new Promise((resolve, reject) => {
    let body = '';
    let size = 0;
    req.on('data', (chunk: Buffer) => {
      size += chunk.length;
      if (size > maxBytes) {
        reject(new Error('Request body too large'));
        return;
      }
      body += chunk.toString();
    });
    req.on('end', () => {
      try {
        resolve(body ? JSON.parse(body) : {});
      } catch {
        resolve({});
      }
    });
    req.on('error', reject);
  });
}

/**
 * Spin up a container in warm-standby mode — no initial query, just wait for IPC.
 * Registers in warmSessions so invoke/stream can pipe prompts into it.
 */
function launchWarmContainer(
  sessionUuid: string,
  group: RegisteredGroup & { jid: string },
  warmGroupFolder: string,
  deps: HooksServerDeps,
): WarmSession {
  const emitter = new EventEmitter();
  const { tenantManager, sessions, onProcess, registeredGroups } = deps;

  const tenant = group.tenantId ? tenantManager.getTenant(group.tenantId) : undefined;
  const provider = getTenantProvider(tenant);
  const sessionRecord = sessions[warmGroupFolder];
  const sessionId = sessionRecord?.runtime === provider ? sessionRecord.id : undefined;

  const secretsResult = group.tenantId
    ? tenantManager.getSecretsForTenant(group.tenantId)
    : undefined;
  if (group.tenantId && !secretsResult) {
    throw new Error(`No credentials available for tenant ${group.tenantId}`);
  }
  const secrets = secretsResult?.secrets;
  const effectiveName = tenant?.name || ASSISTANT_NAME;

  const allRegisteredGroups = registeredGroups();
  const mainGroup = group.tenantId
    ? Object.values(allRegisteredGroups).find(
        (g) => g.tenantId === group.tenantId && g.isMain,
      )
    : undefined;

  // Write context snapshots into the warm group folder
  const allTasks = group.tenantId ? getTasksForTenant(group.tenantId) : getAllTasks();
  writeTasksSnapshot(
    warmGroupFolder,
    false,
    allTasks.map((t) => ({
      id: t.id,
      groupFolder: t.group_folder,
      prompt: t.prompt,
      scheduleType: t.schedule_type,
      scheduleValue: t.schedule_value,
      status: t.status,
      nextRun: t.next_run,
    })),
  );
  writeGroupsSnapshot(warmGroupFolder, false, [], new Set(Object.keys(allRegisteredGroups)));

  const done = runContainerAgent(
    { ...group, folder: warmGroupFolder },
    {
      prompt: '',
      warmOnly: true,
      sessionId,
      groupFolder: warmGroupFolder,
      chatJid: group.jid,
      isMain: false,
      mainGroupFolder: mainGroup?.folder,
      assistantName: effectiveName,
      tenantId: group.tenantId,
      provider,
      model: getDefaultModelForProvider(provider, tenant),
    },
    (proc, containerName) => onProcess(group.jid, proc, containerName, warmGroupFolder),
    async (streamedOutput) => {
      if (streamedOutput.newSessionId && streamedOutput.status !== 'error') {
        sessions[warmGroupFolder] = { id: streamedOutput.newSessionId, runtime: provider };
        setSession(warmGroupFolder, streamedOutput.newSessionId, provider, group.tenantId);
      }
      // Forward all output events to subscribers
      emitter.emit('output', streamedOutput);
    },
    tenant,
    secrets,
  ).then((output) => {
    if (output.newSessionId && output.status !== 'error') {
      sessions[warmGroupFolder] = { id: output.newSessionId, runtime: provider };
      setSession(warmGroupFolder, output.newSessionId, provider, group.tenantId);
    }
    emitter.emit('exit', output);
  }).catch((err) => {
    emitter.emit('error', err);
  }).finally(() => {
    warmSessions.delete(sessionUuid);
    emitter.removeAllListeners();
  });

  const warm: WarmSession = { emitter, groupFolder: warmGroupFolder, done, pipeChain: Promise.resolve() };
  warmSessions.set(sessionUuid, warm);
  return warm;
}

/**
 * Pipe a prompt into a warm container via IPC and collect the result.
 * Subscribes to the warm session's emitter for output events.
 */
function pipeToWarmContainer(
  warm: WarmSession,
  prompt: string,
  opts?: {
    onResult?: (result: string) => void;
    onChunk?: (chunk: { type: 'text' | 'progress'; text: string }) => void;
    attachments?: ContainerInput['attachments'];
    timeoutMs?: number;
  },
): Promise<{ result: string | null; error?: string }> {
  // Serialize: wait for any prior pipe call to finish before starting ours.
  // This ensures the emitter only has one active listener at a time.
  const resultPromise = new Promise<{ result: string | null; error?: string }>((resolve) => {
    warm.pipeChain = warm.pipeChain.then(() => new Promise<void>((chainResolve) => {
    let resolved = false;
    let result: string | null = null;

    const cleanup = () => {
      warm.emitter.removeListener('output', onOutput);
      warm.emitter.removeListener('exit', onExit);
      warm.emitter.removeListener('error', onError);
    };

    const finish = (r: { result: string | null; error?: string }) => {
      if (resolved) return;
      resolved = true;
      cleanup();
      resolve(r);
      chainResolve(); // release the pipe chain for the next caller
    };

    const onOutput = (output: { status: string; result?: string; error?: string; assistantText?: string; progress?: string }) => {
      if (output.assistantText) {
        opts?.onChunk?.({ type: 'text', text: output.assistantText });
      }
      if (output.progress) {
        opts?.onChunk?.({ type: 'progress', text: output.progress });
      }
      if (output.result) {
        const raw = typeof output.result === 'string' ? output.result : JSON.stringify(output.result);
        result = raw.replace(/<internal>[\s\S]*?<\/internal>/g, '').trim();
        opts?.onResult?.(result);
        finish({ result });
      }
      if (output.status === 'error') {
        finish({ result: null, error: output.error || 'unknown error' });
      }
    };

    const onExit = () => {
      finish(result ? { result } : { result: null, error: 'container exited before producing result' });
    };

    const onError = (err: Error) => {
      finish({ result: null, error: err.message });
    };

    warm.emitter.on('output', onOutput);
    warm.emitter.on('exit', onExit);
    warm.emitter.on('error', onError);

    // Write the prompt to IPC — the warm container will pick it up
    writeIpcPrompt(warm.groupFolder, prompt, opts?.attachments);

    // Optional timeout
    if (opts?.timeoutMs) {
      setTimeout(() => {
        finish({ result: null, error: 'invoke timeout exceeded' });
      }, opts.timeoutMs);
    }
    }));
  });
  return resultPromise;
}

/** Run an agent container for a hook request using an isolated session */
async function runHookAgent(
  group: RegisteredGroup & { jid: string },
  prompt: string,
  deps: HooksServerDeps,
  hookSessionFolder: string,
  onResult?: (result: string) => void,
  attachments?: ContainerInput['attachments'],
  onChunk?: (chunk: { type: 'text' | 'progress'; text: string }) => void,
  keepAlive?: boolean,
): Promise<{ result: string | null; error?: string }> {
  const { tenantManager, sessions, queue, onProcess, registeredGroups } = deps;

  const startTime = Date.now();
  const isMain = group.isMain === true;
  const tenant = group.tenantId ? tenantManager.getTenant(group.tenantId) : undefined;
  const provider = getTenantProvider(tenant);

  // Use hook-isolated session (separate from the group's Discord session)
  const sessionRecord = sessions[hookSessionFolder];
  const sessionId = sessionRecord?.runtime === provider ? sessionRecord.id : undefined;

  const secretsResult = group.tenantId
    ? tenantManager.getSecretsForTenant(group.tenantId)
    : undefined;
  if (group.tenantId && !secretsResult) {
    throw new Error(`No credentials available for tenant ${group.tenantId}`);
  }
  const secrets = secretsResult?.secrets;
  const effectiveName = tenant?.name || ASSISTANT_NAME;

  // Write context snapshots so the container has up-to-date state
  const allRegisteredGroups = registeredGroups();
  const mainGroup = !isMain && group.tenantId
    ? Object.values(allRegisteredGroups).find(
        (g) => g.tenantId === group.tenantId && g.isMain,
      )
    : undefined;

  // Determine the effective folder for snapshots and IPC
  const snapshotFolder = hookSessionFolder.includes('/hook-') ? hookSessionFolder : group.folder;

  const allTasks = group.tenantId ? getTasksForTenant(group.tenantId) : getAllTasks();
  writeTasksSnapshot(
    snapshotFolder,
    false,
    allTasks.map((t) => ({
      id: t.id,
      groupFolder: t.group_folder,
      prompt: t.prompt,
      scheduleType: t.schedule_type,
      scheduleValue: t.schedule_value,
      status: t.status,
      nextRun: t.next_run,
    })),
  );
  writeGroupsSnapshot(snapshotFolder, false, [], new Set(Object.keys(allRegisteredGroups)));

  // Hooks use their own IPC directory (hookSessionFolder) to avoid interfering
  // with the main group's IPC close sentinel. This isolates hook containers
  // from scheduled task containers and regular chat containers.
  const hookIpcInputDir = path.join(resolveGroupIpcPath(hookSessionFolder), 'input');
  fs.mkdirSync(hookIpcInputDir, { recursive: true });

  let closeTimer: ReturnType<typeof setTimeout> | null = null;
  const scheduleClose = () => {
    if (closeTimer) return;
    if (keepAlive) return; // caller wants the container to stay warm (idle timeout handles cleanup)
    closeTimer = setTimeout(() => {
      // Write _close directly to the hook-specific IPC dir (not the main group dir)
      try { fs.writeFileSync(path.join(hookIpcInputDir, '_close'), ''); } catch { /* ignore */ }
    }, HOOK_CLOSE_DELAY_MS);
  };

  let result: string | null = null;
  let error: string | null = null;

  // If the hook session folder is a dedicated hook group (koala/hook-xxx),
  // use it as the container's group folder so paths are consistent.
  const isSessionScoped = hookSessionFolder !== `${group.folder}${HOOK_SESSION_SUFFIX}`;
  const containerGroupFolder = isSessionScoped ? hookSessionFolder : group.folder;
  const containerGroup = isSessionScoped ? { ...group, folder: hookSessionFolder } : group;

  try {
    const output = await runContainerAgent(
      containerGroup,
      {
        prompt,
        sessionId,
        groupFolder: containerGroupFolder,
        ipcFolder: isSessionScoped ? undefined : hookSessionFolder,
        chatJid: group.jid,
        isMain: false,
        mainGroupFolder: mainGroup?.folder,
        assistantName: effectiveName,
        tenantId: group.tenantId,
        provider,
        model: getDefaultModelForProvider(provider, tenant),
        attachments,
      },
      (proc, containerName) => onProcess(group.jid, proc, containerName, group.folder),
      async (streamedOutput) => {
        // Only save session ID from successful outputs — saving from errors
        // causes death loops when the session is corrupted
        if (streamedOutput.newSessionId && streamedOutput.status !== 'error') {
          sessions[hookSessionFolder] = { id: streamedOutput.newSessionId, runtime: provider };
          setSession(hookSessionFolder, streamedOutput.newSessionId, provider, group.tenantId);
        }
        if (streamedOutput.assistantText) {
          onChunk?.({ type: 'text', text: streamedOutput.assistantText });
        }
        if (streamedOutput.progress) {
          onChunk?.({ type: 'progress', text: streamedOutput.progress });
        }
        if (streamedOutput.result) {
          const raw =
            typeof streamedOutput.result === 'string'
              ? streamedOutput.result
              : JSON.stringify(streamedOutput.result);
          result = raw.replace(/<internal>[\s\S]*?<\/internal>/g, '').trim();
          onResult?.(result);
          scheduleClose();
        }
        if (streamedOutput.status === 'success') {
          // Hook containers use isolated IPC dirs — don't signal the main group queue
        }
        if (streamedOutput.status === 'error') {
          error = streamedOutput.error || 'unknown error';
        }
      },
      tenant,
      secrets,
    );

    if (closeTimer) clearTimeout(closeTimer);

    if (output.newSessionId && output.status !== 'error') {
      sessions[hookSessionFolder] = { id: output.newSessionId, runtime: provider };
      setSession(hookSessionFolder, output.newSessionId, provider, group.tenantId);
    }

    if (output.status === 'error') {
      error = output.error || 'unknown error';
      // Auto-recover: if the agent crashed quickly, the session is likely
      // corrupted. Clear it so the next run starts fresh.
      const elapsed = Date.now() - startTime;
      if (elapsed < 10_000 && sessions[hookSessionFolder]) {
        logger.warn(
          { groupFolder: group.folder, elapsed, sessionId: sessions[hookSessionFolder].id },
          'hooks: quick crash detected, clearing potentially corrupted session',
        );
        delete sessions[hookSessionFolder];
        deleteSession(hookSessionFolder);
      }
    } else if (output.result && !result) {
      result = output.result;
    }
  } catch (err) {
    if (closeTimer) clearTimeout(closeTimer);
    error = err instanceof Error ? err.message : String(err);
  }

  if (group.tenantId) {
    logUsage(group.tenantId, group.folder, Date.now() - startTime);
  }

  // LCM: persist hook turn
  if (result && prompt) {
    storeTurnPair({
      groupFolder: group.folder,
      sessionId: sessions[hookSessionFolder]?.id ?? group.folder,
      userContent: prompt,
      agentContent: result,
    });
  }

  return error ? { result: null, error } : { result };
}

async function handleHooksRequest(
  req: http.IncomingMessage,
  res: http.ServerResponse,
  deps: HooksServerDeps,
): Promise<void> {
  const url = new URL(req.url ?? '/', 'http://localhost');
  const pathname = url.pathname;

  if (!pathname.startsWith('/hooks/')) {
    sendJson(res, 404, { ok: false, error: 'not found' });
    return;
  }

  if (req.method !== 'POST') {
    res.writeHead(405, { Allow: 'POST' });
    res.end();
    return;
  }

  const token = getBearerToken(req);
  if (!token) {
    sendJson(res, 401, { ok: false, error: 'missing bearer token' });
    return;
  }

  const group = getGroupByHooksToken(token);
  if (!group) {
    sendJson(res, 401, { ok: false, error: 'invalid or disabled token' });
    return;
  }

  if (group.tenantId) {
    const limitCheck = checkUsageLimit(group.tenantId);
    if (!limitCheck.allowed) {
      sendJson(res, 429, { ok: false, error: `Usage limit reached: ${limitCheck.reason}` });
      return;
    }
  }

  const hookSessionFolder = `${group.folder}${HOOK_SESSION_SUFFIX}`;

  switch (pathname) {
    case '/hooks/agent': {
      let body: unknown;
      try {
        body = await readJsonBody(req, 20 * 1024 * 1024); // 20MB — supports base64-encoded image attachments
      } catch {
        sendJson(res, 400, { ok: false, error: 'invalid request body' });
        return;
      }
      const agentBody = body as { prompt?: unknown; session?: unknown };
      const prompt =
        typeof agentBody.prompt === 'string'
          ? agentBody.prompt.trim()
          : '';
      if (!prompt) {
        sendJson(res, 400, { ok: false, error: 'body.prompt is required' });
        return;
      }

      const agentSessionKey = typeof agentBody.session === 'string' ? agentBody.session.trim() : undefined;
      if (agentSessionKey && !UUID_RE.test(agentSessionKey)) {
        sendJson(res, 400, { ok: false, error: 'session must be a valid UUID' });
        return;
      }

      // If session provided + warm container exists, pipe to it
      const agentWarm = agentSessionKey ? warmSessions.get(agentSessionKey) : undefined;
      if (agentWarm) {
        pipeToWarmContainer(agentWarm, prompt).catch((err) => {
          logger.error({ err, session: agentSessionKey }, 'hooks: agent pipe to warm failed');
        });
        sendJson(res, 202, { ok: true, queued: true });
        return;
      }

      const effectiveAgentSession = agentSessionKey && group.tenantId
        ? hookGroupFolder(group.tenantId, agentSessionKey)
        : agentSessionKey
          ? `${group.folder}${HOOK_SESSION_SUFFIX}-${agentSessionKey}`
          : hookSessionFolder;
      if (agentSessionKey && group.tenantId) {
        ensureHookGroupDirs(effectiveAgentSession);
      }

      const taskId = `hook:agent:${Date.now()}:${randomBytes(3).toString('hex')}`;
      const agentConcurrencyKey = agentSessionKey ? `${group.jid}:${agentSessionKey}` : group.jid;
      runHookConcurrent(agentConcurrencyKey, async () => {
        logger.info({ groupFolder: effectiveAgentSession, taskId, session: agentSessionKey }, 'hooks: running /hooks/agent');
        await runHookAgent(group, prompt, deps, effectiveAgentSession);
      });

      sendJson(res, 202, { ok: true, queued: true });
      return;
    }

    case '/hooks/wake': {
      const groupDir = resolveGroupFolderPath(group.folder);
      const heartbeatPath = path.join(groupDir, 'HEARTBEAT.md');

      let heartbeatContent: string;
      try {
        heartbeatContent = fs.readFileSync(heartbeatPath, 'utf-8').trim();
      } catch {
        sendJson(res, 404, { ok: false, error: 'HEARTBEAT.md not found in group workspace' });
        return;
      }

      if (!heartbeatContent) {
        sendJson(res, 400, { ok: false, error: 'HEARTBEAT.md is empty' });
        return;
      }

      const taskId = `hook:wake:${Date.now()}:${randomBytes(3).toString('hex')}`;
      runHookConcurrent(group.jid, async () => {
        logger.info({ groupFolder: group.folder, taskId }, 'hooks: running /hooks/wake');
        await runHookAgent(group, heartbeatContent, deps, hookSessionFolder);
      });

      sendJson(res, 202, { ok: true, queued: true });
      return;
    }

    case '/hooks/warm': {
      let body: unknown;
      try {
        body = await readJsonBody(req, 1024);
      } catch {
        sendJson(res, 400, { ok: false, error: 'invalid request body' });
        return;
      }

      const wb = body as { session?: unknown };
      const warmSessionKey = typeof wb.session === 'string' ? (wb.session as string).trim() : undefined;

      if (!warmSessionKey || !UUID_RE.test(warmSessionKey)) {
        sendJson(res, 400, { ok: false, error: 'session (UUID) is required' });
        return;
      }

      if (!group.tenantId) {
        sendJson(res, 400, { ok: false, error: 'warm requires a tenant-scoped group' });
        return;
      }

      // Already warm?
      if (warmSessions.has(warmSessionKey)) {
        sendJson(res, 200, { ok: true, session: warmSessionKey, warm: true, existing: true });
        return;
      }

      const warmFolder = hookGroupFolder(group.tenantId, warmSessionKey);
      ensureHookGroupDirs(warmFolder);

      const taskId = `hook:warm:${Date.now()}:${randomBytes(3).toString('hex')}`;
      logger.info({ groupFolder: warmFolder, taskId, session: warmSessionKey }, 'hooks: launching warm container');

      launchWarmContainer(warmSessionKey, group, warmFolder, deps);

      sendJson(res, 202, { ok: true, session: warmSessionKey, warm: true });
      return;
    }

    case '/hooks/invoke': {
      let body: unknown;
      try {
        body = await readJsonBody(req, 20 * 1024 * 1024); // 20MB — supports base64-encoded image attachments
      } catch {
        sendJson(res, 400, { ok: false, error: 'invalid request body' });
        return;
      }

      const b = body as { prompt?: unknown; tool?: unknown; args?: unknown; attachments?: unknown; keepAlive?: unknown; session?: unknown };
      const keepAlive = b.keepAlive === true;
      const sessionKey = typeof b.session === 'string' ? b.session.trim() : undefined;

      if (sessionKey && !UUID_RE.test(sessionKey)) {
        sendJson(res, 400, { ok: false, error: 'session must be a valid UUID' });
        return;
      }
      if (keepAlive && !sessionKey) {
        sendJson(res, 400, { ok: false, error: 'session (UUID) is required when keepAlive is true' });
        return;
      }

      const effectiveHookSession = sessionKey
        ? `${group.folder}${HOOK_SESSION_SUFFIX}-${sessionKey}`
        : hookSessionFolder;

      let prompt: string;

      // Parse optional attachments: [{ name, mimeType, data: base64 }]
      const hookAttachments: ContainerInput['attachments'] = [];
      if (Array.isArray(b.attachments)) {
        for (const att of b.attachments as Array<{ name?: unknown; mimeType?: unknown; data?: unknown }>) {
          if (typeof att.data !== 'string' || !att.data) continue;
          const mimeType = typeof att.mimeType === 'string' ? att.mimeType : 'application/octet-stream';
          const name = typeof att.name === 'string' ? att.name : 'attachment';
          const kind =
            mimeType.startsWith('image/') ? 'image' as const :
            mimeType === 'application/pdf' ? 'document' as const :
            mimeType.startsWith('text/') ? 'text' as const :
            mimeType.startsWith('audio/') ? 'audio' as const : 'file' as const;
          hookAttachments.push({ kind, mediaType: mimeType, name, data: att.data });
        }
      }

      if (typeof b.tool === 'string' && b.tool.trim()) {
        const toolName = b.tool.trim();
        if (INVOKE_DENY_LIST.has(toolName)) {
          sendJson(res, 403, { ok: false, error: `Tool not available via hooks: ${toolName}` });
          return;
        }
        const argsStr =
          b.args && typeof b.args === 'object'
            ? JSON.stringify(b.args, null, 2)
            : '{}';
        prompt = `Use the ${toolName} tool with the following arguments and respond with ONLY the raw result:\n\n${argsStr}`;
      } else if (typeof b.prompt === 'string' && b.prompt.trim()) {
        prompt = b.prompt.trim();
      } else {
        sendJson(res, 400, { ok: false, error: 'body.prompt or body.tool is required' });
        return;
      }

      const taskId = `hook:invoke:${Date.now()}:${randomBytes(3).toString('hex')}`;

      // Check for warm container
      const warm = sessionKey ? warmSessions.get(sessionKey) : undefined;
      if (warm) {
        logger.info({ groupFolder: warm.groupFolder, taskId, session: sessionKey }, 'hooks: piping invoke to warm container');

        const timeoutPromise = new Promise<{ result: null; error: string }>((resolve) =>
          setTimeout(() => resolve({ result: null, error: 'invoke timeout exceeded' }), INVOKE_TIMEOUT_MS),
        );
        const resultPromise = pipeToWarmContainer(warm, prompt, {
          attachments: hookAttachments.length > 0 ? hookAttachments : undefined,
          timeoutMs: INVOKE_TIMEOUT_MS,
        });

        const result = await Promise.race([resultPromise, timeoutPromise]);

        if (group.tenantId) logUsage(group.tenantId, warm.groupFolder, 0);
        if (result.result) {
          storeTurnPair({
            groupFolder: warm.groupFolder,
            sessionId: deps.sessions[warm.groupFolder]?.id ?? warm.groupFolder,
            userContent: prompt,
            agentContent: result.result,
          });
        }

        if (result.error) {
          const status = result.error === 'invoke timeout exceeded' ? 504 : 500;
          sendJson(res, status, { ok: false, error: result.error });
        } else {
          sendJson(res, 200, { ok: true, result: result.result });
        }
        return;
      }

      // No warm container — resolve session folder and spawn fresh
      const effectiveFolder = sessionKey && group.tenantId
        ? hookGroupFolder(group.tenantId, sessionKey)
        : effectiveHookSession;
      if (sessionKey && group.tenantId) {
        ensureHookGroupDirs(effectiveFolder);
      }

      // When keepAlive + session, launch a warm container so subsequent calls
      // can pipe into it instead of spawning fresh containers each time.
      if (keepAlive && sessionKey && group.tenantId) {
        logger.info({ groupFolder: effectiveFolder, taskId, session: sessionKey }, 'hooks: launching warm container for invoke keepAlive');
        const newWarm = launchWarmContainer(sessionKey, group, effectiveFolder, deps);

        const timeoutPromise = new Promise<{ result: null; error: string }>((resolve) =>
          setTimeout(() => resolve({ result: null, error: 'invoke timeout exceeded' }), INVOKE_TIMEOUT_MS),
        );
        const resultPromise = pipeToWarmContainer(newWarm, prompt, {
          attachments: hookAttachments.length > 0 ? hookAttachments : undefined,
          timeoutMs: INVOKE_TIMEOUT_MS,
        });

        const result = await Promise.race([resultPromise, timeoutPromise]);

        if (group.tenantId) logUsage(group.tenantId, newWarm.groupFolder, 0);
        if (result.result) {
          storeTurnPair({
            groupFolder: newWarm.groupFolder,
            sessionId: deps.sessions[newWarm.groupFolder]?.id ?? newWarm.groupFolder,
            userContent: prompt,
            agentContent: result.result,
          });
        }

        if (result.error) {
          const status = result.error === 'invoke timeout exceeded' ? 504 : 500;
          sendJson(res, status, { ok: false, error: result.error });
        } else {
          sendJson(res, 200, { ok: true, result: result.result });
        }
        return;
      }

      // Deferred promise resolved by the queued task
      let resolveInvoke!: (r: { result: string | null; error?: string }) => void;
      const invokePromise = new Promise<{ result: string | null; error?: string }>(
        (resolve) => {
          resolveInvoke = resolve;
        },
      );

      const concurrencyKey = sessionKey ? `${group.jid}:${sessionKey}` : group.jid;
      let earlyResolved = false;
      runHookConcurrent(concurrencyKey, async () => {
        logger.info({ groupFolder: effectiveFolder, taskId, session: sessionKey }, 'hooks: running /hooks/invoke');
        await runHookAgent(group, prompt, deps, effectiveFolder, (earlyResult) => {
          if (!earlyResolved) {
            earlyResolved = true;
            resolveInvoke({ result: earlyResult });
          }
        }, hookAttachments.length > 0 ? hookAttachments : undefined, undefined, keepAlive);
        if (!earlyResolved) {
          earlyResolved = true;
          resolveInvoke({ result: null, error: 'agent produced no result' });
        }
      });

      const timeoutPromise = new Promise<{ result: null; error: string }>((resolve) =>
        setTimeout(
          () => resolve({ result: null, error: 'invoke timeout exceeded' }),
          INVOKE_TIMEOUT_MS,
        ),
      );

      const result = await Promise.race([invokePromise, timeoutPromise]);

      if (result.error) {
        const status = result.error === 'invoke timeout exceeded' ? 504 : 500;
        sendJson(res, status, { ok: false, error: result.error });
        return;
      }

      sendJson(res, 200, { ok: true, result: result.result });
      return;
    }

    case '/hooks/stream': {
      let body: unknown;
      try {
        body = await readJsonBody(req, 20 * 1024 * 1024);
      } catch {
        sendJson(res, 400, { ok: false, error: 'invalid request body' });
        return;
      }
      const b = body as Record<string, unknown>;
      const keepAlive = b.keepAlive === true;
      const streamSessionKey = typeof b.session === 'string' ? (b.session as string).trim() : undefined;

      if (streamSessionKey && !UUID_RE.test(streamSessionKey)) {
        sendJson(res, 400, { ok: false, error: 'session must be a valid UUID' });
        return;
      }
      if (keepAlive && !streamSessionKey) {
        sendJson(res, 400, { ok: false, error: 'session (UUID) is required when keepAlive is true' });
        return;
      }

      const effectiveStreamSession = streamSessionKey
        ? `${group.folder}${HOOK_SESSION_SUFFIX}-${streamSessionKey}`
        : hookSessionFolder;

      let prompt: string;
      if (typeof b.tool === 'string' && b.tool.trim()) {
        const toolName = b.tool.trim();
        if (INVOKE_DENY_LIST.has(toolName)) {
          sendJson(res, 403, { ok: false, error: `Tool not available via hooks: ${toolName}` });
          return;
        }
        const argsStr =
          b.args && typeof b.args === 'object'
            ? JSON.stringify(b.args, null, 2)
            : '{}';
        prompt = `Use the ${toolName} tool with the following arguments and respond with ONLY the raw result:\n\n${argsStr}`;
      } else if (typeof b.prompt === 'string' && b.prompt.trim()) {
        prompt = b.prompt.trim();
      } else {
        sendJson(res, 400, { ok: false, error: 'body.prompt or body.tool is required' });
        return;
      }

      const hookAttachments: ContainerInput['attachments'] = [];
      if (Array.isArray(b.attachments)) {
        for (const att of b.attachments as Array<{ name?: unknown; mimeType?: unknown; data?: unknown }>) {
          if (att && att.data && att.mimeType) {
            const name = typeof att.name === 'string' ? att.name : 'file';
            const mimeType = typeof att.mimeType === 'string' ? att.mimeType : 'application/octet-stream';
            const kind =
              mimeType.startsWith('image/') ? 'image' as const :
              mimeType === 'application/pdf' ? 'document' as const :
              mimeType.startsWith('text/') ? 'text' as const :
              mimeType.startsWith('audio/') ? 'audio' as const : 'file' as const;
            hookAttachments.push({ kind, mediaType: mimeType, name, data: String(att.data) });
          }
        }
      }

      // Set SSE headers and send immediately — no timeout
      res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'X-Accel-Buffering': 'no',
      });

      const sendEvent = (event: string, data: object) => {
        try {
          res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
        } catch {
          // client disconnected
        }
      };

      sendEvent('open', { ok: true });

      const taskId = `hook:stream:${Date.now()}:${randomBytes(3).toString('hex')}`;
      let resultSent = false;

      // Check for warm container
      const warm = streamSessionKey ? warmSessions.get(streamSessionKey) : undefined;
      if (warm) {
        logger.info({ groupFolder: warm.groupFolder, taskId, session: streamSessionKey }, 'hooks: piping stream to warm container');

        pipeToWarmContainer(warm, prompt, {
          onResult: (r) => {
            if (!resultSent) {
              resultSent = true;
              sendEvent('result', { ok: true, result: r });
            }
          },
          onChunk: (chunk) => sendEvent(chunk.type, { text: chunk.text }),
          attachments: hookAttachments.length > 0 ? hookAttachments : undefined,
        }).then((pipeResult) => {
          if (group.tenantId) logUsage(group.tenantId, warm.groupFolder, 0);
          if (pipeResult.result) {
            storeTurnPair({
              groupFolder: warm.groupFolder,
              sessionId: deps.sessions[warm.groupFolder]?.id ?? warm.groupFolder,
              userContent: prompt,
              agentContent: pipeResult.result,
            });
          }
          if (!resultSent) {
            if (pipeResult.error) {
              sendEvent('error', { ok: false, error: pipeResult.error });
            } else {
              sendEvent('error', { ok: false, error: 'agent produced no result' });
            }
          }
          try { res.end(); } catch { /* already closed */ }
        });

        req.on('close', () => {
          logger.debug({ groupFolder: warm.groupFolder, taskId }, 'hooks: stream client disconnected');
        });

        return;
      }

      // No warm container — resolve session folder and spawn fresh
      const effectiveFolder = streamSessionKey && group.tenantId
        ? hookGroupFolder(group.tenantId, streamSessionKey)
        : effectiveStreamSession;
      if (streamSessionKey && group.tenantId) {
        ensureHookGroupDirs(effectiveFolder);
      }

      // When keepAlive + session, launch a warm container so subsequent calls
      // can pipe into it instead of spawning fresh containers each time.
      if (keepAlive && streamSessionKey && group.tenantId) {
        logger.info({ groupFolder: effectiveFolder, taskId, session: streamSessionKey }, 'hooks: launching warm container for stream keepAlive');
        const newWarm = launchWarmContainer(streamSessionKey, group, effectiveFolder, deps);

        pipeToWarmContainer(newWarm, prompt, {
          onResult: (r) => {
            if (!resultSent) {
              resultSent = true;
              sendEvent('result', { ok: true, result: r });
            }
          },
          onChunk: (chunk) => sendEvent(chunk.type, { text: chunk.text }),
          attachments: hookAttachments.length > 0 ? hookAttachments : undefined,
        }).then((pipeResult) => {
          if (group.tenantId) logUsage(group.tenantId!, newWarm.groupFolder, 0);
          if (pipeResult.result) {
            storeTurnPair({
              groupFolder: newWarm.groupFolder,
              sessionId: deps.sessions[newWarm.groupFolder]?.id ?? newWarm.groupFolder,
              userContent: prompt,
              agentContent: pipeResult.result,
            });
          }
          if (!resultSent) {
            if (pipeResult.error) {
              sendEvent('error', { ok: false, error: pipeResult.error });
            } else {
              sendEvent('error', { ok: false, error: 'agent produced no result' });
            }
          }
          try { res.end(); } catch { /* already closed */ }
        });

        req.on('close', () => {
          logger.debug({ groupFolder: effectiveFolder, taskId }, 'hooks: stream client disconnected');
        });

        return;
      }

      const streamConcurrencyKey = streamSessionKey ? `${group.jid}:${streamSessionKey}` : group.jid;
      runHookConcurrent(streamConcurrencyKey, async () => {
        logger.info({ groupFolder: effectiveFolder, taskId, session: streamSessionKey }, 'hooks: running /hooks/stream');
        await runHookAgent(
          group,
          prompt,
          deps,
          effectiveFolder,
          (earlyResult) => {
            if (!resultSent) {
              resultSent = true;
              sendEvent('result', { ok: true, result: earlyResult });
            }
          },
          hookAttachments.length > 0 ? hookAttachments : undefined,
          (chunk) => sendEvent(chunk.type, { text: chunk.text }),
          keepAlive,
        );
        if (!resultSent) {
          sendEvent('error', { ok: false, error: 'agent produced no result' });
        }
        try { res.end(); } catch { /* already closed */ }
      });

      req.on('close', () => {
        logger.debug({ groupFolder: effectiveFolder, taskId }, 'hooks: stream client disconnected');
      });

      return;
    }

    default:
      sendJson(res, 404, { ok: false, error: `unknown hook endpoint: ${pathname}` });
  }
}

export function startHooksServer(deps: HooksServerDeps): http.Server {
  const server = http.createServer((req, res) => {
    handleHooksRequest(req, res, deps).catch((err) => {
      logger.error({ err }, 'hooks: unhandled error in request handler');
      if (!res.headersSent) {
        sendJson(res, 500, { ok: false, error: 'internal error' });
      }
    });
  });

  server.listen(HOOKS_PORT, '127.0.0.1', () => {
    logger.info({ port: HOOKS_PORT }, 'Hooks HTTP server started');
  });

  return server;
}

/**
 * Creates an IPC task handler for hooks_agent, hooks_wake, hooks_invoke.
 * Processes tasks queued by SDK-enabled apps via the filesystem IPC layer.
 * No bearer token required — IPC is already scoped to the tenant's group.
 */
/**
 * Creates an IPC task handler for hooks_agent, hooks_wake, hooks_invoke.
 * Processes tasks queued by SDK-enabled apps via the filesystem IPC layer.
 * No bearer token required — IPC is already scoped to the tenant's group.
 */
export function createHookIpcHandler(deps: HooksServerDeps) {
  return function onHookTask(sourceGroup: string, task: HookIpcTask): void {
    const groups = deps.registeredGroups();
    const group = Object.values(groups).find((g) => g.folder === sourceGroup) as
      | (RegisteredGroup & { jid: string })
      | undefined;

    const ipcDir = resolveGroupIpcPath(sourceGroup);
    const resultPath = `${ipcDir}/hooks-result-${task.requestId}.json`;

    function writeResult(data: object): void {
      try {
        const tmp = `${resultPath}.tmp`;
        fs.writeFileSync(tmp, JSON.stringify({ ...data, timestamp: new Date().toISOString() }, null, 2));
        fs.renameSync(tmp, resultPath);
      } catch (err) {
        logger.warn({ err, sourceGroup, requestId: task.requestId }, 'hooks IPC: failed to write result file');
      }
    }

    if (!group) {
      logger.warn({ sourceGroup, taskType: task.type }, 'hooks IPC: group not found');
      writeResult({ ok: false, error: 'group not found' });
      return;
    }

    if (group.tenantId) {
      const limitCheck = checkUsageLimit(group.tenantId);
      if (!limitCheck.allowed) {
        writeResult({ ok: false, error: `Usage limit reached: ${limitCheck.reason}` });
        return;
      }
    }

    const hookSessionFolder = `${group.folder}${HOOK_SESSION_SUFFIX}`;

    if (task.type === 'hooks_agent') {
      const prompt = task.prompt?.trim() ?? '';
      if (!prompt) {
        writeResult({ ok: false, error: 'prompt is required' });
        return;
      }
      const taskId = `hook:ipc:agent:${Date.now()}:${task.requestId}`;
      runHookConcurrent(group.jid, async () => {
        logger.info({ sourceGroup, taskId }, 'hooks IPC: running hooks_agent');
        await runHookAgent(group, prompt, deps, hookSessionFolder);
      });
      writeResult({ ok: true, queued: true });
      return;
    }

    if (task.type === 'hooks_wake') {
      const groupDir = resolveGroupFolderPath(group.folder);
      const heartbeatPath = path.join(groupDir, 'HEARTBEAT.md');
      let heartbeatContent: string;
      try {
        heartbeatContent = fs.readFileSync(heartbeatPath, 'utf-8').trim();
      } catch {
        writeResult({ ok: false, error: 'HEARTBEAT.md not found in group workspace' });
        return;
      }
      if (!heartbeatContent) {
        writeResult({ ok: false, error: 'HEARTBEAT.md is empty' });
        return;
      }
      const taskId = `hook:ipc:wake:${Date.now()}:${task.requestId}`;
      runHookConcurrent(group.jid, async () => {
        logger.info({ sourceGroup, taskId }, 'hooks IPC: running hooks_wake');
        await runHookAgent(group, heartbeatContent, deps, hookSessionFolder);
      });
      writeResult({ ok: true, queued: true });
      return;
    }

    if (task.type === 'hooks_invoke') {
      let prompt: string;
      if (task.tool?.trim()) {
        const toolName = task.tool.trim();
        if (INVOKE_DENY_LIST.has(toolName)) {
          writeResult({ ok: false, error: `Tool not available via hooks: ${toolName}` });
          return;
        }
        const argsStr =
          task.args && typeof task.args === 'object'
            ? JSON.stringify(task.args, null, 2)
            : '{}';
        prompt = `Use the ${toolName} tool with the following arguments and respond with ONLY the raw result:\n\n${argsStr}`;
      } else if (task.prompt?.trim()) {
        prompt = task.prompt.trim();
      } else {
        writeResult({ ok: false, error: 'prompt or tool is required' });
        return;
      }

      const taskId = `hook:ipc:invoke:${Date.now()}:${task.requestId}`;
      runHookConcurrent(group.jid, async () => {
        logger.info({ sourceGroup, taskId }, 'hooks IPC: running hooks_invoke');
        let resultWritten = false;
        const r = await runHookAgent(group, prompt, deps, hookSessionFolder, (earlyResult) => {
          resultWritten = true;
          writeResult({ ok: true, result: earlyResult });
        });
        if (!resultWritten) {
          writeResult(r.error ? { ok: false, error: r.error } : { ok: true, result: r.result });
        }
      });
    }
  };
}
