import { createHash, randomUUID } from "node:crypto";
import { access, mkdir, readFile, readdir, rename, rm, writeFile } from "node:fs/promises";
import { join, relative, resolve } from "node:path";
import { corsHeaders, sendEvent } from "./http.js";
import { INTERNAL_GENERATED_IMAGE_ITEM, publicCapabilitySummary, requestCodexAppServer, startCodexRuntime, startCodexThreadWatcher } from "./codex-runtime.js";
import { createSessionStore } from "./session-store.js";
import { readAccountThread, threadEvents, workspaceForThread } from "./codex-history.js";
import { readWorkspaceChanges, readWorkspaceDiff } from "./workspace-diff.js";
import { publicTokenUsage } from "./usage-status.js";
import { projectPublicDiffData } from "./public-diff.js";
import { sanitizePublicData, sanitizePublicText } from "./public-sanitizer.js";
import { safeChangeFiles } from "./safe-change-paths.js";
import { projectGeneratedImageItem, projectGeneratedImages } from "./generated-images.js";
import { getWorkspaceArtifact as readWorkspaceArtifact, projectWorkspaceArtifacts } from "./workspace-artifacts.js";
import { workspaceDisplayLabel } from "./workspaces.js";
import { assertMessagePermissionsAllowed, publicPermissionProfilesWithinCeiling } from "./permission-ceiling.js";
import { createActionReceiptStore } from "./action-receipt-store.js";
import { writeSafeLog } from "./safe-logger.js";

const sessions = new Map();
const sessionListClients = new Set();
let sessionStore;
let actionReceiptStore;
let saveQueue = Promise.resolve();
let saveRequested = false;
let saveRunning = false;
let persistWaiters = [];
let persistAllRequested = false;
let dirtySessionIds = new Set();
let deferredPersistTimer;
let deferredPersistSessionIds = new Set();
let sessionMaintenanceTimer;
let sessionMaintenanceGeneration = 0;
let sessionMaintenancePromise;
let runtimeConfig;
let startThreadWatcher = startCodexThreadWatcher;
let startRuntime = startCodexRuntime;
let generateRuntimeTitle;
let requestRuntimeCapability = requestCodexAppServer;
let readDiff = readWorkspaceDiff;
let getAccountExecutionAccess = () => ({ allowed: true, entitlementLocked: false });
const DEFAULT_HANDOFF_CONTEXT_MAX_CHARS = 12000;
const MAX_LIVE_ACTIVITY_OUTPUT_BYTES = 32 * 1024;
const LIVE_EVENT_PERSIST_DEBOUNCE_MS = 250;
const LIVE_SSE_BATCH_MS = 25;
const MAX_ATTACHMENT_BYTES = 8 * 1024 * 1024;
const MAX_MESSAGE_ATTACHMENT_COUNT = 8;
const MAX_MESSAGE_ATTACHMENT_BYTES = 32 * 1024 * 1024;
const ABANDONED_ATTACHMENT_TTL_MS = 24 * 60 * 60 * 1000;
const MAX_ATTACHMENT_CLEANUP_PER_START = 128;
const TERMINAL_SESSION_TTL_MS = 30 * 24 * 60 * 60 * 1000;
const MAX_SESSION_CLEANUP_PER_START = 64;
const MAX_SESSION_CLEANUP_PER_MAINTENANCE = 8;
const SESSION_MAINTENANCE_INTERVAL_MS = 60 * 60 * 1000;
const approvalDecisionLocks = new WeakSet();
const privateApprovalData = new WeakMap();
const attachmentMutationQueues = new WeakMap();
const queueMutationQueues = new WeakMap();
const queuedDispatchClaims = new WeakMap();

export async function initializeSessions(config, store = createSessionStore(config.dataDir, config), dependencies = {}) {
  cancelSessionMaintenance();
  clearTimeout(deferredPersistTimer);
  deferredPersistTimer = undefined;
  deferredPersistSessionIds = new Set();
  for (const session of sessions.values()) discardPendingSessionEvent(session);
  runtimeConfig = config;
  startThreadWatcher = dependencies.startThreadWatcher ?? startCodexThreadWatcher;
  startRuntime = dependencies.startRuntime ?? startCodexRuntime;
  generateRuntimeTitle = dependencies.generateConversationTitle;
  requestRuntimeCapability = dependencies.requestCodexAppServer ?? requestCodexAppServer;
  readDiff = dependencies.readWorkspaceDiff ?? readWorkspaceDiff;
  getAccountExecutionAccess = dependencies.getAccountExecutionAccess ?? (() => ({ allowed: true, entitlementLocked: false }));
  sessionStore = store;
  actionReceiptStore = dependencies.actionReceiptStore ?? createActionReceiptStore();
  await actionReceiptStore.initialize();
  saveQueue = Promise.resolve();
  saveRequested = false;
  saveRunning = false;
  persistWaiters = [];
  persistAllRequested = false;
  dirtySessionIds = new Set();
  sessions.clear();
  for (const client of sessionListClients) client.end?.();
  sessionListClients.clear();

  for (const stored of await sessionStore.load()) {
    const status = ["starting", "running", "waiting_approval"].includes(stored.status)
      ? "interrupted"
      : stored.status;
    const approvals = sanitizePublicData(stored.approvals ?? [], config).map(expireRestoredCodexApproval);
    const attachments = restoreAttachments(config.dataDir, stored);
    const workspace = (config.workspaces ?? []).find((item) => item.id === stored.workspaceId);
    const events = safePublicData((stored.events ?? []).filter((item) => !isInternalEvent(item.event)), config);
    sessions.set(stored.id, {
      ...stored,
      projectLabel: workspaceDisplayLabel(workspace) ?? stored.projectLabel,
      status,
      conversationTitle: cleanConversationTitle(stored.conversationTitle) ?? firstUserMessage(events),
      events,
      handoffBaseEvents: safePublicData(stored.handoffBaseEvents, config),
      handoffPromptText: sanitizePublicText(stored.handoffPromptText, config) || undefined,
      handoffUserText: sanitizePublicText(stored.handoffUserText, config) || undefined,
      approvals: new Map(approvals.map((item) => [item.id, item])),
      files: stored.files ?? [],
      attachments,
      queuedMessages: restoreQueuedMessages(stored.queuedMessages, attachments, stored.accountId),
      queueAdvanceBlocked: true,
      threadRefs: stored.threadRefs ?? threadRefsFor(stored),
      clients: new Set(),
      runtime: null
    });
  }
  const orphanSessionDirectories = await recoverSessionDirectoryQuarantines(config);
  await cleanupExpiredTerminalSessions(config, MAX_SESSION_CLEANUP_PER_START);
  await cleanupOrphanSessionDirectories(orphanSessionDirectories, MAX_SESSION_CLEANUP_PER_START);
  const orphanQuarantines = await recoverAttachmentQuarantines(config);
  const cleanupBudget = await cleanupExpiredAbandonedAttachments(config, MAX_ATTACHMENT_CLEANUP_PER_START);
  await cleanupOrphanQuarantines(orphanQuarantines, cleanupBudget);
  await schedulePersist();
  scheduleSessionMaintenance();
}

function restoreAttachments(dataDir, session) {
  if (!dataDir) return storedAttachments(session.attachments);
  return storedAttachments(session.attachments).map((attachment) => ({
    ...attachment,
    path: attachment.kind === "artifact"
      ? join(dataDir, "artifacts", session.id, attachment.id, ...(attachment.resourcePath ?? attachment.name).split("/"))
      : join(dataDir, "attachments", session.id, `${attachment.id}${imageExtension(attachment.mimeType, attachment.name) ?? ""}`)
  }));
}

function storedAttachments(attachments = []) {
  return attachments.filter((item) => item?.id && ["image", "artifact"].includes(item.kind));
}

async function recoverSessionDirectoryQuarantines(config) {
  if (!config?.dataDir) return [];
  const orphanPaths = [];
  for (const kind of ["attachments", "artifacts"]) {
    const root = resolve(config.dataDir, kind);
    const entries = await readdir(root, { withFileTypes: true }).catch(() => []);
    for (const entry of entries) {
      if (!entry.isDirectory()) continue;
      const markerIndex = entry.name.indexOf(".deleting-");
      if (markerIndex <= 0) continue;
      const sessionId = entry.name.slice(0, markerIndex);
      const quarantinePath = resolve(root, entry.name);
      const canonicalPath = resolve(root, sessionId);
      if (!safeChildPath(root, quarantinePath) || !safeChildPath(root, canonicalPath)) continue;
      if (!sessions.has(sessionId)) {
        orphanPaths.push(quarantinePath);
        continue;
      }
      try {
        await access(canonicalPath);
        orphanPaths.push(quarantinePath);
      } catch {
        await rename(quarantinePath, canonicalPath).catch(() => {});
      }
    }
  }
  return orphanPaths;
}

async function cleanupExpiredTerminalSessions(config, budget, now = Date.now()) {
  if (!config?.dataDir || budget <= 0) return 0;
  const candidates = [...sessions.values()]
    .filter((session) => isExpiredIdleTerminalSession(session, now))
    .sort((left, right) => Date.parse(left.updatedAt) - Date.parse(right.updatedAt) || left.id.localeCompare(right.id))
    .slice(0, budget);

  const removed = [];
  for (const session of candidates) {
    if (!isExpiredIdleTerminalSession(session, now)) continue;
    const directories = await quarantineSessionDirectories(config, session.id);
    if (!directories) continue;
    if (!isExpiredIdleTerminalSession(session, now)) {
      await restoreQuarantinedDirectories(directories);
      continue;
    }
    sessions.delete(session.id);
    removed.push({ session, directories });
  }
  if (removed.length === 0) return 0;

  try {
    await schedulePersist();
  } catch {
    for (const item of removed) sessions.set(item.session.id, item.session);
    await Promise.all(removed.flatMap((item) => item.directories.map(({ canonicalPath, quarantinePath }) => quarantinePath
      ? rename(quarantinePath, canonicalPath).catch(() => {})
      : Promise.resolve())));
    return 0;
  }

  await Promise.all(removed.flatMap((item) => item.directories.map(({ quarantinePath }) => quarantinePath
    ? rm(quarantinePath, { recursive: true, force: true }).catch(() => {})
    : Promise.resolve())));
  return removed.length;
}

function isExpiredIdleTerminalSession(session, now) {
  const updatedAt = Date.parse(session.updatedAt);
  return ["completed", "failed"].includes(session.status)
    && !session.runtime
    && (session.clients?.size ?? 0) === 0
    && (session.queuedMessages ?? []).length === 0
    && ![...(session.approvals?.values?.() ?? [])].some((approval) => approval.status === "pending")
    && Number.isFinite(updatedAt)
    && now - updatedAt >= TERMINAL_SESSION_TTL_MS;
}

export function runSessionMaintenance({ now = Date.now(), budget = MAX_SESSION_CLEANUP_PER_MAINTENANCE } = {}) {
  if (sessionMaintenancePromise) return sessionMaintenancePromise;
  const task = cleanupExpiredTerminalSessions(runtimeConfig, budget, now).then((removed) => {
    if (removed > 0) broadcastSessionListSnapshot();
    return removed;
  });
  const running = task.finally(() => {
    if (sessionMaintenancePromise === running) sessionMaintenancePromise = undefined;
  });
  sessionMaintenancePromise = running;
  return running;
}

function scheduleSessionMaintenance() {
  if (!runtimeConfig?.dataDir) return;
  const generation = sessionMaintenanceGeneration;
  sessionMaintenanceTimer = setTimeout(async () => {
    sessionMaintenanceTimer = undefined;
    try {
      await runSessionMaintenance();
    } catch (error) {
      writeSafeLog(console, "warn", "session_maintenance_failed", { error }, runtimeConfig);
    } finally {
      if (generation === sessionMaintenanceGeneration) scheduleSessionMaintenance();
    }
  }, SESSION_MAINTENANCE_INTERVAL_MS);
  sessionMaintenanceTimer.unref?.();
}

function cancelSessionMaintenance() {
  sessionMaintenanceGeneration += 1;
  clearTimeout(sessionMaintenanceTimer);
  sessionMaintenanceTimer = undefined;
}

async function quarantineSessionDirectories(config, sessionId) {
  const directories = [];
  for (const kind of ["attachments", "artifacts"]) {
    const root = resolve(config.dataDir, kind);
    const canonicalPath = resolve(root, sessionId);
    if (!safeChildPath(root, canonicalPath)) {
      await restoreQuarantinedDirectories(directories);
      return undefined;
    }
    const quarantinePath = `${canonicalPath}.deleting-${randomUUID()}`;
    try {
      await rename(canonicalPath, quarantinePath);
      directories.push({ canonicalPath, quarantinePath });
    } catch (error) {
      if (error.code === "ENOENT") {
        directories.push({ canonicalPath, quarantinePath: undefined });
        continue;
      }
      await restoreQuarantinedDirectories(directories);
      return undefined;
    }
  }
  return directories;
}

function restoreQuarantinedDirectories(directories) {
  return Promise.all(directories.map(({ canonicalPath, quarantinePath }) => quarantinePath
    ? rename(quarantinePath, canonicalPath).catch(() => {})
    : Promise.resolve()));
}

function safeChildPath(root, target) {
  const relation = relative(root, target);
  return Boolean(relation) && !relation.startsWith("..") && resolve(root, relation) === target;
}

async function cleanupOrphanSessionDirectories(paths, budget) {
  for (const path of paths.slice(0, Math.max(0, budget))) {
    await rm(path, { recursive: true, force: true }).catch(() => {});
  }
}

async function recoverAttachmentQuarantines(config) {
  if (!config?.dataDir) return [];
  const root = join(config.dataDir, "attachments");
  const directories = await readdir(root, { withFileTypes: true }).catch((error) => error.code === "ENOENT" ? [] : []);
  const orphanPaths = [];
  for (const directory of directories) {
    if (!directory.isDirectory()) continue;
    const session = sessions.get(directory.name);
    const attachmentDir = join(root, directory.name);
    const entries = await readdir(attachmentDir, { withFileTypes: true }).catch(() => []);
    for (const entry of entries) {
      if (!entry.isFile()) continue;
      const markerIndex = entry.name.indexOf(".deleting-");
      if (markerIndex <= 0) continue;
      const quarantinePath = join(attachmentDir, entry.name);
      const canonicalPath = join(attachmentDir, entry.name.slice(0, markerIndex));
      const owner = (session?.attachments ?? []).find((attachment) => attachment.kind === "image" && attachment.path === canonicalPath);
      if (!owner) {
        orphanPaths.push(quarantinePath);
        continue;
      }
      try {
        await access(canonicalPath);
        orphanPaths.push(quarantinePath);
      } catch {
        await rename(quarantinePath, canonicalPath).catch(() => {});
      }
    }
  }
  return orphanPaths;
}

async function cleanupExpiredAbandonedAttachments(config, budget) {
  if (!config?.dataDir || budget <= 0) return budget;
  const now = Date.now();
  const candidates = [];
  for (const session of sessions.values()) {
    const referencedIds = referencedAttachmentIds(session);
    for (const attachment of session.attachments ?? []) {
      const createdAt = Date.parse(attachment.createdAt);
      if (attachment.kind !== "image"
        || attachment.id.startsWith("generated-")
        || referencedIds.has(attachment.id)
        || !Number.isFinite(createdAt)
        || now - createdAt < ABANDONED_ATTACHMENT_TTL_MS) continue;
      candidates.push({ session, attachment, createdAt });
    }
  }
  candidates.sort((left, right) => left.createdAt - right.createdAt || left.attachment.id.localeCompare(right.attachment.id));

  const quarantined = [];
  const previousAttachments = new Map();
  for (const candidate of candidates.slice(0, budget)) {
    const quarantinePath = `${candidate.attachment.path}.deleting-${randomUUID()}`;
    try {
      await rename(candidate.attachment.path, quarantinePath);
      quarantined.push({ ...candidate, quarantinePath });
    } catch (error) {
      if (error.code !== "ENOENT") continue;
      quarantined.push({ ...candidate, quarantinePath: undefined });
    }
    if (!previousAttachments.has(candidate.session)) previousAttachments.set(candidate.session, candidate.session.attachments);
    candidate.session.attachments = candidate.session.attachments.filter((item) => item.id !== candidate.attachment.id);
  }
  if (quarantined.length === 0) return budget;

  try {
    await schedulePersist();
  } catch {
    for (const [session, attachments] of previousAttachments) session.attachments = attachments;
    await Promise.all(quarantined.map(({ attachment, quarantinePath }) => quarantinePath
      ? rename(quarantinePath, attachment.path).catch(() => {})
      : Promise.resolve()));
    return budget;
  }

  await Promise.all(quarantined.map(({ quarantinePath }) => quarantinePath
    ? rm(quarantinePath, { force: true }).catch(() => {})
    : Promise.resolve()));
  return budget - quarantined.length;
}

function referencedAttachmentIds(session) {
  const knownIds = new Set((session.attachments ?? []).filter((item) => item.kind === "image").map((item) => item.id));
  const referenced = new Set();
  for (const message of session.queuedMessages ?? []) {
    for (const attachment of message.attachments ?? []) {
      if (knownIds.has(attachment.id)) referenced.add(attachment.id);
    }
  }
  collectAttachmentReferences(session.events ?? [], knownIds, referenced);
  return referenced;
}

function collectAttachmentReferences(value, knownIds, referenced) {
  if (Array.isArray(value)) {
    for (const item of value) collectAttachmentReferences(item, knownIds, referenced);
    return;
  }
  if (!value || typeof value !== "object") return;
  if (knownIds.has(value.id)) referenced.add(value.id);
  for (const item of Object.values(value)) collectAttachmentReferences(item, knownIds, referenced);
}

async function cleanupOrphanQuarantines(paths, budget) {
  for (const path of paths.slice(0, Math.max(0, budget))) {
    await rm(path, { force: true }).catch(() => {});
  }
}

function restoreQueuedMessages(messages = [], attachments = [], fallbackExecutionAccountId) {
  const attachmentsById = new Map(attachments.map((item) => [item.id, item]));
  return messages.map((item) => ({
    ...item,
    executionAccountId: item.executionAccountId ?? fallbackExecutionAccountId,
    attachments: (item.attachments ?? []).map((attachment) => attachmentsById.get(attachment.id) ?? attachment)
  }));
}

function expireRestoredCodexApproval(approval) {
  if (approval.status !== "pending" || approval.runtimeRequestId == null) return approval;
  return {
    ...approval,
    status: "expired",
    title: "Approval expired after backend restart",
    resolvedAt: new Date().toISOString()
  };
}

export async function createSession(config, input) {
  input = input && typeof input === "object" ? input : {};
  const account = config.accounts.find((item) => item.id === input.accountId);
  const workspace = input.workspaceId
    ? config.workspaces.find((item) => item.id === input.workspaceId)
    : undefined;

  assertAccountAvailableForExecution(account);

  if (!account || (input.workspaceId && !workspace)) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }

  const sessionId = input.clientSessionId
    ? normalizedClientActionId(input.clientSessionId, "invalid_client_session_id")
    : randomUUID();
  const existing = sessions.get(sessionId);
  if (existing) {
    if (existing.accountId !== account.id || existing.workspaceId !== workspace?.id || Boolean(existing.projectless) !== !workspace) {
      throw idempotencyError("idempotency_conflict", 409);
    }
    return publicSession(existing);
  }

  const session = {
    id: sessionId,
    accountId: account.id,
    workspaceId: workspace?.id,
    projectLabel: workspaceDisplayLabel(workspace),
    projectless: !workspace,
    status: "idle",
    createdAt: new Date().toISOString(),
    updatedAt: new Date().toISOString(),
    events: [],
    clients: new Set(),
    approvals: new Map(),
    files: [],
    queuedMessages: [],
    runtime: null
  };

  sessions.set(session.id, session);
  await emit(session, "session.created", publicSession(session));
  return publicSession(session);
}

export async function openThread(
  config,
  accountId,
  threadId,
  input = {},
  reader = readAccountThread
) {
  const account = config.accounts.find((item) => item.id === accountId);
  if (!account) {
    const error = new Error("account_not_found");
    error.status = 404;
    throw error;
  }

  const result = await reader(config, accountId, threadId);
  const thread = result.thread;
  if (!thread || thread.id !== threadId) {
    const error = new Error("thread_not_found");
    error.status = 404;
    throw error;
  }

  const explicitWorkspace = input.workspaceId
    ? config.workspaces.find((item) => item.id === input.workspaceId)
    : null;
  if (input.workspaceId && !explicitWorkspace) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }
  const workspace = workspaceForThread(config.workspaces, thread.cwd) ?? explicitWorkspace;
  let session = [...sessions.values()].find((item) =>
    item.accountId === accountId && item.codexThreadId === threadId
  );

  if (!session) {
    session = {
      id: randomUUID(),
      accountId,
      workspaceId: workspace?.id,
      projectLabel: workspaceDisplayLabel(workspace) ?? safeProjectLabel(input.projectLabel),
      codexThreadId: threadId,
      conversationTitle: threadConversationTitle(thread),
      threadRefs: [{ accountId, threadId }],
      status: publicThreadStatus(thread.status),
      createdAt: timestampFromSeconds(thread.createdAt) ?? new Date().toISOString(),
      updatedAt: timestampFromSeconds(thread.updatedAt) ?? new Date().toISOString(),
      events: [],
      clients: new Set(),
      approvals: new Map(),
      files: [],
      runtime: null
    };
    session.events = projectSessionThreadEvents(session, thread);
    session.externalSignature = JSON.stringify(session.events);
    sessions.set(session.id, session);
  } else if (!session.runtime) {
    session.workspaceId = workspace?.id ?? session.workspaceId;
    session.projectLabel = workspaceDisplayLabel(workspace) ?? safeProjectLabel(input.projectLabel) ?? session.projectLabel;
    syncThreadConversationTitle(session, thread);
    session.events = mergeSessionThreadEvents(session.events, projectSessionThreadEvents(session, thread));
    session.externalSignature = JSON.stringify(session.events);
    session.status = publicThreadStatus(thread.status);
    session.updatedAt = timestampFromSeconds(thread.updatedAt) ?? session.updatedAt;
  }

  await schedulePersist(session);
  return { session: publicSession(session), events: session.events };
}

export function getSession(id) {
  const session = sessions.get(id);
  if (!session) {
    const error = new Error("session_not_found");
    error.status = 404;
    throw error;
  }
  return session;
}

export function publicSession(session) {
  return safePublicData({
    id: session.id,
    accountId: session.accountId,
    workspaceId: session.workspaceId,
    projectLabel: session.projectLabel,
    projectless: Boolean(session.projectless),
    codexThreadId: session.codexThreadId ?? session.runtime?.threadId,
    conversationTitle: session.conversationTitle,
    threadRefs: session.threadRefs ?? threadRefsFor(session),
    status: session.status,
    createdAt: session.createdAt,
    updatedAt: session.updatedAt,
    tokenUsage: session.tokenUsage ?? null,
    queuedMessages: (session.queuedMessages ?? []).map(publicQueuedMessage)
  });
}

export function listSessions() {
  return [...sessions.values()].map(publicSession);
}

export async function prepareSessionsForApplicationDataReset() {
  const inUse = [...sessions.values()].some((session) =>
    ["starting", "running", "waiting_approval"].includes(session.status)
    || (session.queuedMessages ?? []).length > 0
    || [...(session.approvals?.values?.() ?? [])].some((approval) => approval.status === "pending")
  );
  if (inUse) {
    const error = new Error("application_data_in_use");
    error.status = 409;
    error.safeLogCode = "application_data_in_use";
    throw error;
  }

  await schedulePersist();
  cancelSessionMaintenance();
  clearTimeout(deferredPersistTimer);
  deferredPersistTimer = undefined;
  deferredPersistSessionIds = new Set();
  sessionStore = undefined;
  for (const session of sessions.values()) {
    discardPendingSessionEvent(session);
    stopExternalWatcher(session);
    session.runtime?.stop?.();
    session.runtime = null;
    for (const client of session.clients ?? []) client.end?.();
    session.clients?.clear?.();
  }
  for (const client of sessionListClients) client.end?.();
  sessionListClients.clear();
}

export async function updateThreadSessionTitle(accountId, threadId, value) {
  const title = cleanConversationTitle(value);
  if (!title) return { updatedSessions: [] };
  const updatedSessions = [];
  const saves = [];
  for (const session of sessions.values()) {
    const referencesThread = (session.threadRefs ?? threadRefsFor(session)).some((ref) =>
      ref.accountId === accountId && ref.threadId === threadId
    );
    if (!referencesThread || session.conversationTitle === title) continue;
    session.conversationTitle = title;
    updatedSessions.push(session.id);
    saves.push(schedulePersist(session));
    broadcastSessionSnapshot(session);
    broadcastSessionUpdate(session);
  }
  await Promise.all(saves);
  return { updatedSessions };
}

export function attachSessionListEventStream(res) {
  res.publicSanitizerConfig ??= runtimeConfig;
  res.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store",
    connection: "keep-alive",
    ...(res.corsHeaders ?? corsHeaders())
  });

  sessionListClients.add(res);
  sendEvent(res, "sessions.snapshot", { sessions: listSessions() });
  res.on("close", () => sessionListClients.delete(res));
}

function threadRefsFor(session) {
  const threadId = session.codexThreadId ?? session.runtime?.threadId;
  return session.accountId && threadId ? [{ accountId: session.accountId, threadId }] : [];
}

export async function listCapabilities(config, accountId, workspaceId, requester = requestCodexAppServer) {
  const account = config.accounts.find((item) => item.id === accountId);
  if (!account) {
    const error = new Error("account_not_found");
    error.status = 404;
    throw error;
  }
  assertAccountAvailableForExecution(account);
  const workspace = workspaceId ? config.workspaces.find((item) => item.id === workspaceId) : null;
  if (workspaceId && !workspace) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }

  if (config.engine !== "codex") {
    return publicCapabilitySummary({});
  }

  const common = { codexCliPath: config.codexCliPath, codexHome: account.codexHome };
  const [models, permissions, plugins, skills] = await Promise.allSettled([
    requester({ ...common, method: "model/list", params: { includeHidden: false } }),
    requester({ ...common, method: "permissionProfile/list", params: { cwd: workspace?.path } }),
    requester({ ...common, method: "plugin/list", params: { cwds: workspace ? [workspace.path] : [] } }),
    requester({ ...common, method: "skills/list", params: { cwds: workspace ? [workspace.path] : [] } })
  ]);

  const summary = publicCapabilitySummary({
      models: fulfilledValue(models),
      permissions: fulfilledValue(permissions),
      plugins: fulfilledValue(plugins),
      skills: fulfilledValue(skills)
    });
  return {
    ...summary,
    permissions: publicPermissionProfilesWithinCeiling(summary.permissions, config.permissionCeiling),
    errors: [
      rejectedCapability("models", models),
      rejectedCapability("permissions", permissions),
      rejectedCapability("plugins", plugins),
      rejectedCapability("skills", skills)
    ].filter(Boolean)
  };
}

export function attachEventStream(session, res) {
  res.publicSanitizerConfig ??= runtimeConfig;
  res.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store",
    connection: "keep-alive",
    ...(res.corsHeaders ?? corsHeaders())
  });

  flushPendingSessionEvent(session);
  session.clients.add(res);
  sendEvent(res, "session.snapshot", safePublicData({
    session: publicSession(session),
    events: session.events
  }));
  void ensureExternalWatcher(session);

  res.on("close", () => {
    session.clients.delete(res);
    if (session.clients.size === 0) {
      discardPendingSessionEvent(session);
      stopExternalWatcher(session);
      if (!["starting", "running", "waiting_approval"].includes(session.status)) {
        releaseRuntime(session, { resumeWatcher: false });
      }
    }
  });
}

export async function sendMessage(session, input) {
  const previousSend = session.sendMessageLock ?? Promise.resolve();
  let releaseSend;
  const currentSend = new Promise((resolve) => { releaseSend = resolve; });
  const sendTail = previousSend.catch(() => {}).then(() => currentSend);
  session.sendMessageLock = sendTail;
  await previousSend.catch(() => {});
  try {
    return await sendMessageUnlocked(session, input);
  } finally {
    releaseSend();
    if (session.sendMessageLock === sendTail) session.sendMessageLock = null;
  }
}

async function sendMessageUnlocked(session, input) {
  input = input && typeof input === "object" ? input : {};
  const text = typeof input.text === "string" ? input.text : "";
  const hasAttachmentInput = Array.isArray(input.attachments) && input.attachments.length > 0;
  if (text.trim().length === 0 && !hasAttachmentInput) {
    const error = new Error("invalid_message");
    error.status = 400;
    throw error;
  }

  const requestClientMessageId = normalizedClientMessageId(input.clientMessageId ?? randomUUID());
  const payloadHash = messagePayloadHash(input, text);
  const previousReceipt = await actionReceiptStore.findMessage({
    sessionId: session.id,
    clientMessageId: requestClientMessageId,
    payloadHash
  });
  if (previousReceipt) return replayMessageDelivery(previousReceipt);

  assertMessagePermissionsAllowed(input.options, runtimeConfig?.permissionCeiling);

  if (session.status === "waiting_approval") {
    const error = new Error("session_busy");
    error.status = 409;
    throw error;
  }
  if (session.stopping) {
    const error = new Error("session_busy");
    error.status = 409;
    throw error;
  }

  const executionAccount = executionAccountFor(input.executionAccountId);
  if (runtimeConfig?.engine === "codex") assertSessionContextAvailable(session, executionAccount);
  const attachments = sanitizeAttachments(session, input.attachments);
  const skills = sanitizeSkills(input.skills);
  if (text.trim().length === 0 && attachments.length === 0) {
    const error = new Error("invalid_message");
    error.status = 400;
    throw error;
  }

  if (session.status === "running") {
    const queuedMessageId = randomUUID();
    const delivery = { disposition: "queued", queuedMessageId };
    const reservation = await actionReceiptStore.reserveMessage({
      sessionId: session.id,
      clientMessageId: requestClientMessageId,
      payloadHash,
      delivery,
      state: "accepted"
    });
    if (!reservation.created) return replayMessageDelivery(reservation.receipt);
    const queuedMessage = enqueueMessage(session, {
      ...input,
      id: queuedMessageId,
      clientMessageId: requestClientMessageId,
      text,
      attachments,
      skills,
      executionAccountId: executionAccount?.id ?? session.accountId
    });
    try {
      await schedulePersist(session);
    } catch {
      session.queueAdvanceBlocked = true;
      await actionReceiptStore.updateMessage({
        sessionId: session.id,
        clientMessageId: requestClientMessageId,
        state: "indeterminate"
      }).catch(() => {});
      throw idempotencyError("idempotency_store_unavailable", 503);
    }
    return { disposition: "queued", queuedMessageId: queuedMessage.id };
  }

  const delivery = { disposition: "started" };
  const reservation = await actionReceiptStore.reserveMessage({
    sessionId: session.id,
    clientMessageId: requestClientMessageId,
    payloadHash,
    delivery,
    state: "accepted"
  });
  if (!reservation.created) return replayMessageDelivery(reservation.receipt);

  if (!session.conversationTitle?.trim()) {
    session.conversationTitle = firstUserMessage(handoffBaseEventsFor(session)) ?? cleanConversationTitle(text);
  }
  emit(session, "message.accepted", {
    sessionId: session.id,
    clientMessageId: requestClientMessageId,
    text,
    ...(attachments.length ? { attachmentCount: attachments.length, attachments: publicMessageAttachments(attachments) } : {})
  });

  session.activeAttachmentIds = new Set(attachments.map((item) => item.id));
  stopExternalWatcher(session);
  let runtimeText = text;
  try {
    if (runtimeConfig?.engine === "codex" && executionAccount && executionAccount.id !== session.accountId) {
      runtimeText = await startHandoffRuntime(session, executionAccount, text, requestClientMessageId);
    } else if (!session.runtime && runtimeConfig?.engine === "codex") {
      await resumeRuntime(session);
    }
  } catch (error) {
    session.activeAttachmentIds = new Set();
    throw error;
  }

  if (session.status === "interrupted") return { disposition: "started" };

  setStatus(session, "starting");

  const previousQueueAdvanceBlocked = session.queueAdvanceBlocked;
  session.queueAdvanceBlocked = false;
  if (session.runtime) {
    const runtime = session.runtime;
    try {
      const resolvedSkills = await resolveRuntimeSkills(session, skills);
      if (session.runtime !== runtime || session.status !== "starting") return { disposition: "started" };
      await runtime.sendMessage({
        text: runtimeText,
        options: sanitizeMessageOptions(input.options),
        attachments,
        skills: resolvedSkills
      });
      if (session.runtime === runtime && session.status === "starting") setStatus(session, "running");
    } catch (error) {
      session.queueAdvanceBlocked = previousQueueAdvanceBlocked;
      if (["turn_stop_unconfirmed", "turn_start_outcome_unknown"].includes(error.message)) {
        session.queueAdvanceBlocked = true;
      } else if (error.message !== "turn_start_canceled") {
        emit(session, "error", { message: publicTurnStartError(error) });
        setStatus(session, "failed");
        if (session.runtime === runtime) releaseRuntime(session, { resumeWatcher: false });
      }
    }
    return { disposition: "started" };
  }

  runFakeTurn(session, text);
  return { disposition: "started" };
}

export async function steerQueuedMessage(session, queuedMessageId) {
  const previousReceipt = await actionReceiptStore.findQueuedAction({
    sessionId: session.id,
    queuedMessageId,
    action: "steer"
  });
  if (previousReceipt) return replayQueuedAction(previousReceipt);
  const item = queuedMessage(session, queuedMessageId);
  if (runtimeConfig?.engine === "codex") assertSessionContextAvailable(session);
  if (session.status === "waiting_approval") {
    const error = new Error("session_busy");
    error.status = 409;
    throw error;
  }
  if (session.status !== "running") {
    const error = new Error("steering_unavailable");
    error.status = 409;
    throw error;
  }
  if ((item.attachments ?? []).length > 0 || (item.skills ?? []).length > 0) {
    const error = new Error("steering_rich_input_unsupported");
    error.status = 409;
    throw error;
  }
  if (!session.runtime && runtimeConfig?.engine === "codex") {
    stopExternalWatcher(session);
    await resumeRuntime(session);
    setStatus(session, "running");
  }
  if (typeof session.runtime?.steerMessage !== "function") {
    const error = new Error("steering_unavailable");
    error.status = 409;
    throw error;
  }
  const delivery = { queuedMessageId, action: "steer" };
  const reservation = await actionReceiptStore.reserveQueuedAction({
    sessionId: session.id,
    queuedMessageId,
    action: "steer",
    delivery,
    state: "dispatching"
  });
  if (!reservation.created) return replayQueuedAction(reservation.receipt);
  let turnId;
  const previousQueueAdvanceBlocked = session.queueAdvanceBlocked;
  session.queueAdvanceBlocked = false;
  try {
    turnId = await session.runtime.steerMessage({ text: item.text });
  } catch {
    session.queueAdvanceBlocked = previousQueueAdvanceBlocked;
    await actionReceiptStore.updateQueuedAction({
      sessionId: session.id,
      queuedMessageId,
      state: "indeterminate"
    }).catch(() => {});
    const error = new Error("steering_failed");
    error.status = 409;
    throw error;
  }
  session.queuedMessages = session.queuedMessages.filter((queued) => queued.id !== queuedMessageId);
  setStatus(session, session.status);
  emit(session, "message.accepted", {
    sessionId: session.id,
    text: item.text,
    queuedMessageId,
    turnId,
    steering: true
  });
  try {
    await schedulePersist(session);
    await actionReceiptStore.updateQueuedAction({
      sessionId: session.id,
      queuedMessageId,
      state: "completed"
    });
  } catch {
    await actionReceiptStore.updateQueuedAction({
      sessionId: session.id,
      queuedMessageId,
      state: "indeterminate"
    }).catch(() => {});
    throw idempotencyError("idempotency_store_unavailable", 503);
  }
  return delivery;
}

export function deleteQueuedMessage(session, queuedMessageId) {
  return withQueueMutation(session, () => deleteQueuedMessageUnlocked(session, queuedMessageId));
}

async function deleteQueuedMessageUnlocked(session, queuedMessageId) {
  const previousReceipt = await actionReceiptStore.findQueuedAction({
    sessionId: session.id,
    queuedMessageId,
    action: "delete"
  });
  if (previousReceipt) return replayQueuedAction(previousReceipt);
  queuedMessage(session, queuedMessageId);
  if (queuedDispatchClaims.get(session) === queuedMessageId) {
    const error = new Error("queued_message_dispatching");
    error.status = 409;
    throw error;
  }
  const delivery = { queuedMessageId, action: "delete" };
  const reservation = await actionReceiptStore.reserveQueuedAction({
    sessionId: session.id,
    queuedMessageId,
    action: "delete",
    delivery,
    state: "dispatching"
  });
  if (!reservation.created) return replayQueuedAction(reservation.receipt);
  const previousQueuedMessages = session.queuedMessages;
  session.queuedMessages = session.queuedMessages.filter((item) => item.id !== queuedMessageId);
  session.updatedAt = new Date().toISOString();
  try {
    await schedulePersist(session);
    await actionReceiptStore.updateQueuedAction({
      sessionId: session.id,
      queuedMessageId,
      state: "completed"
    });
  } catch {
    session.queuedMessages = previousQueuedMessages;
    await actionReceiptStore.updateQueuedAction({
      sessionId: session.id,
      queuedMessageId,
      state: "indeterminate"
    }).catch(() => {});
    throw idempotencyError("idempotency_store_unavailable", 503);
  }
  setStatus(session, session.status);
  return delivery;
}

export async function stopSession(session) {
  if (["completed", "failed", "interrupted"].includes(session.status)) return;
  if (!["starting", "running", "waiting_approval"].includes(session.status)) {
    const error = new Error("stop_unavailable");
    error.status = 409;
    throw error;
  }

  const turnId = latestStartedTurnId(session.events);
  const runtime = session.runtime;
  if (session.status === "starting") {
    if (session.startAttempt) session.startAttempt.canceled = true;
    await runtime?.cancelPendingStart?.();
    if (session.runtime === runtime) releaseRuntime(session, { resumeWatcher: false });
    emit(session, "turn.completed", {
      sessionId: session.id,
      ...(turnId ? { turnId } : {}),
      status: "interrupted",
      completedAt: Date.now()
    });
    setStatus(session, "interrupted");
    return;
  }
  if (!runtime?.interrupt) {
    const error = new Error("stop_unavailable");
    error.status = 409;
    throw error;
  }
  session.stopping = true;
  try {
    await runtime.interrupt();
  } catch (error) {
    session.stopping = false;
    throw error;
  }
  if (session.runtime === runtime) releaseRuntime(session, { resumeWatcher: false });
  emit(session, "turn.completed", {
    sessionId: session.id,
    ...(turnId ? { turnId } : {}),
    status: "interrupted",
    completedAt: Date.now()
  });
  session.stopping = false;
  setStatus(session, "interrupted");
}

function queuedMessage(session, queuedMessageId) {
  const item = (session.queuedMessages ?? []).find((queued) => queued.id === queuedMessageId);
  if (!item) {
    const error = new Error("queued_message_not_found");
    error.status = 404;
    throw error;
  }
  return item;
}

function replayQueuedAction(receipt) {
  if (receipt.state === "dispatching") {
    const error = new Error("queued_action_pending");
    error.status = 409;
    throw error;
  }
  if (receipt.state !== "completed") {
    throw idempotencyError("idempotency_outcome_unknown", 503);
  }
  return { ...receipt.delivery };
}

function latestStartedTurnId(events = []) {
  for (let index = events.length - 1; index >= 0; index -= 1) {
    const event = events[index];
    if (event.event === "turn.completed") return undefined;
    if (event.event === "turn.started") return event.data?.turnId;
  }
  return undefined;
}

function hasInterruptedTurn(events = []) {
  return events.some((event) => event.event === "turn.completed" && event.data?.status === "interrupted");
}
export function createAttachment(session, input) {
  return withAttachmentMutation(session, () => createAttachmentUnlocked(session, input));
}

async function createAttachmentUnlocked(session, input) {
  if (!runtimeConfig?.dataDir) {
    const error = new Error("attachment_storage_unavailable");
    error.status = 500;
    throw error;
  }

  input = input && typeof input === "object" ? input : {};
  const data = decodeAttachmentData(input.dataBase64);
  const detectedType = detectImageType(data);
  if (!detectedType) throw attachmentError("unsupported_attachment_type", 400);

  const dir = join(runtimeConfig.dataDir, "attachments", session.id);
  await mkdir(dir, { recursive: true });
  const attachmentId = randomUUID();
  const attachment = {
    id: attachmentId,
    kind: "image",
    name: attachmentDisplayName(input.name, detectedType.extension),
    mimeType: detectedType.mimeType,
    size: data.length,
    path: join(dir, `${attachmentId}${detectedType.extension}`),
    createdAt: new Date().toISOString()
  };
  await writeFile(attachment.path, data);
  session.attachments = [...(session.attachments ?? []), attachment];
  schedulePersist(session);
  return attachment;
}

export function deleteAttachment(session, attachmentId) {
  return withAttachmentMutation(session, () => deleteAttachmentUnlocked(session, attachmentId));
}

async function deleteAttachmentUnlocked(session, attachmentId) {
  const attachment = (session.attachments ?? []).find((item) => item.id === attachmentId && item.kind === "image");
  if (!attachment?.path) throw attachmentError("attachment_not_found", 404);
  if (attachmentIsInUse(session, attachmentId)) throw attachmentError("attachment_in_use", 409);

  const quarantinePath = `${attachment.path}.deleting-${randomUUID()}`;
  try {
    await rename(attachment.path, quarantinePath);
  } catch {
    throw attachmentError("attachment_delete_failed", 503);
  }

  const previousAttachments = session.attachments;
  const previousEvents = session.events;
  session.attachments = previousAttachments.filter((item) => item.id !== attachmentId);
  session.events = markAttachmentDeleted(previousEvents, attachment);
  session.updatedAt = new Date().toISOString();
  try {
    await schedulePersist(session);
  } catch {
    session.attachments = previousAttachments;
    session.events = previousEvents;
    await rename(quarantinePath, attachment.path).catch(() => {});
    throw attachmentError("attachment_delete_failed", 503);
  }

  await rm(quarantinePath, { force: true }).catch(() => {});
  broadcastSessionSnapshot(session);
  broadcastSessionUpdate(session);
  return { attachmentId, deleted: true };
}

export async function getAttachment(session, attachmentId) {
  const attachment = (session.attachments ?? []).find((item) => item.id === attachmentId && item.kind === "image");
  if (!attachment?.path) {
    const error = new Error("attachment_not_found");
    error.status = 404;
    throw error;
  }
  return {
    data: await readFile(attachment.path),
    mimeType: attachment.mimeType || "application/octet-stream",
    name: attachment.name || "image"
  };
}

export async function getArtifact(session, artifactId, resourcePath) {
  return readWorkspaceArtifact({ session, artifactId, resourcePath, dataDir: runtimeConfig?.dataDir });
}

export async function deleteThreadLocalData(config, accountId, threadId) {
  const matches = [...sessions.values()].filter((session) =>
    (session.accountId === accountId && session.codexThreadId === threadId)
    || (session.threadRefs ?? []).some((ref) => ref.accountId === accountId && ref.threadId === threadId)
  );
  for (const session of matches) {
    discardPendingSessionEvent(session);
    releaseRuntime(session, { resumeWatcher: false });
    stopExternalWatcher(session);
    for (const client of session.clients ?? []) client.end?.();
    session.clients?.clear?.();
    sessions.delete(session.id);
    await deleteSessionAttachments(config, session.id);
  }
  await schedulePersist();
  return { removedSessions: matches.map((session) => session.id) };
}

export function syncExternalThread(session, thread) {
  if ((session.runtime && ["starting", "running", "waiting_approval"].includes(session.status)) || thread.id !== session.codexThreadId) return false;
  const titleChanged = syncThreadConversationTitle(session, thread);
  const projectedEvents = projectSessionThreadEvents(session, thread);
  if (hasMissingCompletedActivity(session, projectedEvents)) {
    if (titleChanged) session.updatedAt = timestampFromSeconds(thread.updatedAt) ?? session.updatedAt;
    return titleChanged;
  }
  const projectedStatus = externalThreadStatus(thread.status, projectedEvents);
  if (session.status === "interrupted" && projectedStatus === "running" && hasInterruptedTurn(session.events)) {
    if (titleChanged) session.updatedAt = timestampFromSeconds(thread.updatedAt) ?? session.updatedAt;
    return titleChanged;
  }
  const events = mergeSessionThreadEvents(session.events, projectedEvents);
  const signature = JSON.stringify(events);
  if (signature === session.externalSignature) {
    if (titleChanged) session.updatedAt = timestampFromSeconds(thread.updatedAt) ?? session.updatedAt;
    return titleChanged;
  }
  const nextStatus = externalThreadStatus(thread.status, events);
  session.externalSignature = signature;
  session.events = events;
  session.activityUpdateEvents?.clear();
  const wasRunning = session.status === "running";
  session.status = nextStatus;
  if (wasRunning && session.status !== "running" && (session.queuedMessages ?? []).length > 0) scheduleQueueAdvance(session);
  session.updatedAt = timestampFromSeconds(thread.updatedAt) ?? session.updatedAt;
  return true;
}

function hasMissingCompletedActivity(session, projectedEvents) {
  if (!["completed", "failed", "interrupted"].includes(session.status)) return false;
  const projectedItems = new Set(projectedEvents
    .filter(isTechnicalActivityEvent)
    .map(activityEventKey));
  return (session.events ?? [])
    .filter(isTechnicalActivityEvent)
    .some((event) => !projectedItems.has(activityEventKey(event)));
}

function isTechnicalActivityEvent(event) {
  return event.event?.startsWith("activity.")
    && !["agentMessage", "userMessage", "reasoning", "plan"].includes(event.data?.kind)
    && Boolean(event.data?.itemId);
}

function activityEventKey(event) {
  return `${event.data?.turnId ?? ""}:${event.data.itemId}`;
}

function mergeSessionThreadEvents(previousEvents = [], projectedEvents = []) {
  const previousActivity = activitySnapshots(previousEvents);
  const projectedActivityKeys = new Set(projectedEvents.filter(isActivityEvent).map(activityEventKey));
  const projectedPlanTurns = new Set(projectedEvents
    .filter((event) => event.event === "plan.updated")
    .map((event) => event.data?.turnId ?? ""));
  const previousTerminal = new Map(previousEvents
    .filter((event) => event.event === "turn.completed" && event.data?.turnId)
    .map((event) => [event.data.turnId, event]));

  const merged = projectedEvents.map((event) => {
    if (event.event.startsWith("activity.") && event.data?.itemId) {
      const snapshot = previousActivity.get(activityEventKey(event));
      if (snapshot) return { ...event, data: mergeActivityData(snapshot, event.data) };
    }
    if (event.event === "turn.completed" && event.data?.turnId) {
      const terminal = previousTerminal.get(event.data.turnId);
      if (terminal) {
        const preserveTerminal = ["failed", "interrupted"].includes(terminal.data?.status) && event.data.status === "completed";
        const error = event.data.error ?? terminal.data?.error;
        return { ...event, data: { ...event.data, ...(preserveTerminal ? terminal.data : {}), ...(error ? { error } : {}) } };
      }
    }
    return event;
  });

  for (const event of previousEvents) {
    if (event.event === "error") {
      if (!merged.some((candidate) => candidate.event === "error" && JSON.stringify(candidate.data) === JSON.stringify(event.data))) merged.push(event);
      continue;
    }
    if (event.event === "plan.updated" && !projectedPlanTurns.has(event.data?.turnId ?? "")) {
      merged.push(event);
      continue;
    }
    if (isActivityEvent(event) && event.data?.itemId && !projectedActivityKeys.has(activityEventKey(event))) {
      merged.push(event);
      continue;
    }
    if (event.event === "turn.completed" && event.data?.turnId
      && !merged.some((candidate) => candidate.event === "turn.completed" && candidate.data?.turnId === event.data.turnId)) {
      merged.push(event);
    }
  }
  return merged;
}

function isActivityEvent(event) {
  return event.event?.startsWith("activity.");
}

function activitySnapshots(events) {
  const snapshots = new Map();
  for (const event of events) {
    if (!isActivityEvent(event) || !event.data?.itemId) continue;
    const key = activityEventKey(event);
    const current = snapshots.get(key) ?? {};
    const text = event.data.text
      ? event.event === "activity.completed" ? event.data.text : `${current.text ?? ""}${event.data.text}`
      : current.text;
    snapshots.set(key, { ...current, ...event.data, ...(text ? { text } : {}) });
  }
  return snapshots;
}

function mergeActivityData(previous, projected) {
  const durable = Object.fromEntries(["text", "error", "exitCode", "images", "toolCall"]
    .filter((key) => projected[key] == null && previous[key] != null)
    .map((key) => [key, previous[key]]));
  return { ...projected, ...durable };
}

function projectSessionThreadEvents(session, thread) {
  const projected = preserveMessageAttachments(session, threadEvents(thread, {
    ...runtimeConfig,
    projectAssistantText: (text) => projectAssistantContent(session, text),
    projectGeneratedImageItem: (item) => projectGeneratedImageActivity(session, item)
  })).map((event) => {
    if (
      event.event === "message.accepted"
      && event.data?.text === session.handoffPromptText
      && session.handoffUserText
    ) {
      return { ...event, data: { ...event.data, text: sanitizePublicText(session.handoffUserText, runtimeConfig) } };
    }
    return event;
  });
  return [...(session.handoffBaseEvents ?? []), ...projected];
}

function preserveMessageAttachments(session, projectedEvents) {
  const previous = new Map();
  for (const event of session.events ?? []) {
    if (event.event !== "message.accepted" || !event.data?.attachments?.length) continue;
    const key = messageAttachmentKey(event.data);
    previous.set(key, [...(previous.get(key) ?? []), event.data]);
  }
  return projectedEvents.map((event) => {
    if (event.event !== "message.accepted" || !(event.data?.attachmentCount > 0)) return event;
    const match = previous.get(messageAttachmentKey(event.data))?.shift();
    if (!match) return event;
    return {
      ...event,
      data: {
        ...event.data,
        sessionId: session.id,
        attachments: publicMessageAttachments(match.attachments)
      }
    };
  });
}

function messageAttachmentKey(data) {
  return `${data?.text ?? ""}\u0000${data?.attachmentCount ?? 0}`;
}

async function ensureExternalWatcher(session) {
  if (session.watcher || session.watcherStarting || !session.codexThreadId || session.clients.size === 0) return;
  const account = runtimeConfig?.accounts?.find((item) => item.id === session.accountId);
  if (!account) return;
  const generation = session.watcherGeneration ?? 0;
  session.watcherStarting = startThreadWatcher({
    codexCliPath: runtimeConfig.codexCliPath,
    codexHome: account.codexHome,
    threadId: session.codexThreadId,
    onThread: (thread) => {
      if (!syncExternalThread(session, thread)) return;
      broadcastSessionSnapshot(session);
      broadcastSessionUpdate(session);
      schedulePersist(session);
    },
    onEvent: (event, data) => {
      if (event === "process.exited") return;
      emitRuntimeEvent(session, event, data);
    }
  });
  try {
    const watcher = await session.watcherStarting;
    session.watcherStarting = null;
    if (session.clients.size === 0 || generation !== (session.watcherGeneration ?? 0)) {
      watcher.stop();
      return;
    }
    session.watcher = watcher;
  } catch {
    session.watcherStarting = null;
  }
}

function stopExternalWatcher(session) {
  session.watcherGeneration = (session.watcherGeneration ?? 0) + 1;
  session.watcher?.stop();
  session.watcher = null;
}

function releaseRuntime(session, { resumeWatcher = true } = {}) {
  if (!session.runtime) return;
  const runtime = session.runtime;
  session.runtime = null;
  try {
    void runtime.stop?.();
  } catch {
    // Runtime already closed. External sync can still continue.
  }
  if (resumeWatcher) void ensureExternalWatcher(session);
}

async function resumeRuntime(session) {
  const account = runtimeConfig.accounts.find((item) => item.id === session.accountId);
  const workspace = runtimeConfig.workspaces.find((item) => item.id === session.workspaceId);
  assertAccountAvailableForExecution(account);
  if (!account || (session.workspaceId && !workspace) || (!session.workspaceId && !session.projectless)) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }

  const previousStatus = session.status;
  const previousRuntime = session.runtime;
  const previousThreadId = session.codexThreadId;
  const attempt = { canceled: false };
  session.startAttempt = attempt;
  setStatus(session, "starting");
  let runtime;
  try {
    runtime = await startRuntime({
      accountId: account.id,
      codexCliPath: runtimeConfig.codexCliPath,
      codexHome: account.codexHome,
      cwd: workspace?.path,
      threadId: session.codexThreadId,
      emit: (event, data) => emitRuntimeEvent(session, event, data)
    });
  } catch (error) {
    if (session.startAttempt === attempt) session.startAttempt = undefined;
    if (attempt.canceled || session.status === "interrupted") return false;
    session.runtime = previousRuntime ?? null;
    session.codexThreadId = previousThreadId;
    setStatus(session, previousStatus);
    throw error;
  }
  if (session.startAttempt === attempt) session.startAttempt = undefined;
  if (attempt.canceled || session.status === "interrupted") {
    runtime.stop?.();
    return false;
  }
  session.runtime = runtime;
  session.codexThreadId = runtime.threadId;
  schedulePersist(session);
  return true;
}

async function startHandoffRuntime(session, account, userText, currentClientMessageId) {
  const workspace = runtimeConfig.workspaces.find((item) => item.id === session.workspaceId);
  if ((session.workspaceId && !workspace) || (!session.workspaceId && !session.projectless)) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }

  const snapshot = {
    accountId: session.accountId,
    codexThreadId: session.codexThreadId,
    conversationTitle: session.conversationTitle,
    threadRefs: session.threadRefs,
    handoffBaseEvents: session.handoffBaseEvents,
    handoffPromptText: session.handoffPromptText,
    handoffUserText: session.handoffUserText,
    externalSignature: session.externalSignature,
    status: session.status,
    runtime: session.runtime
  };
  releaseRuntime(session, { resumeWatcher: false });
  const baseEvents = handoffBaseEventsFor(session).filter((event) => !(
    currentClientMessageId
    && event.event === "message.accepted"
    && event.data?.clientMessageId === currentClientMessageId
  ));
  const previousRef = session.accountId && session.codexThreadId
    ? { accountId: session.accountId, threadId: session.codexThreadId }
    : null;
  const prompt = handoffPrompt(baseEvents, userText);
  const attempt = { canceled: false };
  session.startAttempt = attempt;
  setStatus(session, "starting");
  let runtime;
  try {
    runtime = await startRuntime({
      accountId: account.id,
      codexCliPath: runtimeConfig.codexCliPath,
      codexHome: account.codexHome,
      cwd: workspace?.path,
      emit: (event, data) => emitRuntimeEvent(session, event, data)
    });
  } catch (error) {
    if (session.startAttempt === attempt) session.startAttempt = undefined;
    if (attempt.canceled || session.status === "interrupted") {
      Object.assign(session, snapshot, { runtime: null, status: "interrupted" });
      schedulePersist(session);
      return null;
    }
    Object.assign(session, snapshot);
    schedulePersist(session);
    throw error;
  }
  if (session.startAttempt === attempt) session.startAttempt = undefined;
  if (attempt.canceled || session.status === "interrupted") {
    runtime.stop?.();
    Object.assign(session, snapshot, { runtime: null, status: "interrupted" });
    schedulePersist(session);
    return null;
  }
  session.runtime = runtime;
  session.accountId = account.id;
  session.codexThreadId = undefined;
  session.conversationTitle ??= firstUserMessage(baseEvents);
  session.handoffBaseEvents = baseEvents;
  session.handoffPromptText = prompt;
  session.handoffUserText = sanitizePublicText(userText, runtimeConfig);
  session.externalSignature = undefined;
  session.codexThreadId = session.runtime.threadId;
  session.threadRefs = appendThreadRefs(session.threadRefs, previousRef, {
    accountId: account.id,
    threadId: session.runtime.threadId
  });
  schedulePersist(session);
  return prompt;
}

function handoffBaseEventsFor(session) {
  const events = session.events ?? [];
  const pendingQueuedIds = new Set((session.queuedMessages ?? []).map((item) => item.id));
  const deliveredQueuedIds = new Set(events
    .filter((event) => event.data?.queuedMessageId && event.data?.turnId)
    .map((event) => event.data.queuedMessageId));
  return events.filter((event) => {
    const queuedMessageId = event.data?.queuedMessageId;
    if (!queuedMessageId) return true;
    if (pendingQueuedIds.has(queuedMessageId)) return false;
    return !deliveredQueuedIds.has(queuedMessageId) || Boolean(event.data?.turnId);
  });
}

function appendThreadRefs(existing = [], ...refs) {
  const unique = new Map();
  for (const ref of [...existing, ...refs]) {
    if (ref?.accountId && ref?.threadId) unique.set(`${ref.accountId}:${ref.threadId}`, ref);
  }
  return [...unique.values()];
}

function firstUserMessage(events = []) {
  return events.find((event) => event.event === "message.accepted" && event.data?.text)?.data.text.trim() || undefined;
}

function executionAccountFor(accountId) {
  if (!accountId) return null;
  const account = runtimeConfig?.accounts?.find((item) => item.id === accountId);
  assertAccountAvailableForExecution(account);
  if (account) return account;
  const error = new Error("account_not_found");
  error.status = 404;
  throw error;
}

function assertSessionContextAvailable(session, executionAccount = null) {
  const account = runtimeConfig?.accounts?.find((item) => item.id === session.accountId);
  const workspace = runtimeConfig?.workspaces?.find((item) => item.id === session.workspaceId);
  if ((session.accountId && !account) || (session.workspaceId && !workspace) || (!session.workspaceId && !session.projectless)) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }
  if (account && (!executionAccount || executionAccount.id === session.accountId)) assertAccountAvailableForExecution(account);
}

function assertAccountAvailableForExecution(account) {
  if (!account) return;
  if (account.historyOnly) {
    const error = new Error("account_not_available_for_execution");
    error.status = 403;
    throw error;
  }
  const access = getAccountExecutionAccess(account.id);
  if (access?.allowed !== false || !access.entitlementLocked) return;
  const error = new Error(access.message || "Pro is required to use this account.");
  error.status = access.status ?? 402;
  error.code = access.code ?? "pro_required";
  throw error;
}

function handoffPrompt(events, userText) {
  const context = limitedHandoffContext(events);
  return [
    "Continue this Code Remote Manager work in a new account-owned thread.",
    "Use the previous visible conversation as context. Do not expose hidden reasoning, credentials, local auth data, or private paths.",
    "",
    "Previous visible conversation:",
    context,
    "",
    "Current user request:",
    sanitizePublicText(userText, runtimeConfig)
  ].join("\n");
}

function limitedHandoffContext(events) {
  const maxChars = handoffContextMaxChars(runtimeConfig);
  const lines = visibleConversationLines(events)
    .map((line) => sanitizePublicText(line, runtimeConfig).trim())
    .filter(Boolean);
  const selected = [];
  let length = 0;
  for (let index = lines.length - 1; index >= 0; index -= 1) {
    const line = lines[index];
    const nextLength = length + line.length + (selected.length ? 1 : 0);
    if (nextLength > maxChars) continue;
    selected.unshift(line);
    length = nextLength;
  }
  return selected.join("\n").trim() || "No prior visible conversation.";
}

function handoffContextMaxChars(config) {
  const value = Number(config?.handoffContextMaxChars);
  return Number.isFinite(value) && value > 0 ? value : DEFAULT_HANDOFF_CONTEXT_MAX_CHARS;
}

function visibleConversationLines(events = []) {
  const lines = [];
  const assistantLines = new Map();
  let unscopedGroup = 0;
  for (const event of events) {
    if (event.event === "message.accepted" && event.data?.text) {
      lines.push(`User: ${event.data.text}`);
      unscopedGroup += 1;
      continue;
    }
    if (event.event !== "session.output" || !["assistant_delta", "assistant_snapshot"].includes(event.data?.kind) || !event.data?.text) continue;
    const key = event.data.itemId ?? event.data.turnId ?? `unscoped-${unscopedGroup}`;
    let index = assistantLines.get(key);
    if (index == null) {
      index = lines.length;
      assistantLines.set(key, index);
      lines.push("Assistant: ");
    }
    const text = event.data.kind === "assistant_snapshot"
      ? event.data.text
      : `${lines[index].slice("Assistant: ".length)}${event.data.text}`;
    lines[index] = `Assistant: ${text}`;
  }
  return lines;
}

export async function resolveApproval(session, approvalId, input) {
  const decision = input?.decision;
  const approved = ["approved", "approved_for_session", "approved_always"].includes(decision);
  const rejected = decision === "rejected";
  if (!approved && !rejected) {
    const error = new Error("invalid_approval_decision");
    error.status = 400;
    throw error;
  }

  const existingReceipt = await actionReceiptStore.findApprovalDecision({
    sessionId: session.id,
    approvalId,
    decision
  });
  if (existingReceipt) return replayApprovalDecision(session, approvalId, existingReceipt);

  const approval = session.approvals.get(approvalId);
  if (!approval || approval.status !== "pending") {
    const error = new Error("approval_not_found");
    error.status = 404;
    throw error;
  }
  const sessionScopeUnavailable = decision === "approved_for_session"
    && !["command", "file_change", "permissions"].includes(approval.kind);
  const permanentScopeUnavailable = decision === "approved_always"
    && (approval.kind !== "command" || !privateApprovalData.get(approval)?.proposedExecpolicyAmendment);
  if (sessionScopeUnavailable || permanentScopeUnavailable) {
    const error = new Error("approval_scope_unavailable");
    error.status = 400;
    throw error;
  }
  if (approvalDecisionLocks.has(approval)) {
    const error = new Error("approval_decision_pending");
    error.status = 409;
    throw error;
  }

  approvalDecisionLocks.add(approval);
  const delivery = { approvalId, status: approved ? "approved" : "rejected" };
  try {
    const reservation = await actionReceiptStore.reserveApprovalDecision({
      sessionId: session.id,
      approvalId,
      decision,
      delivery,
      state: "dispatching"
    });
    if (!reservation.created) return replayApprovalDecision(session, approvalId, reservation.receipt);

    try {
      if (session.runtime) {
        await session.runtime.resolveApproval(approval.runtimeRequestId, decision, {
          ...approval,
          ...privateApprovalData.get(approval)
        });
      } else {
        completeFakeApproval(session, approved);
      }
      await actionReceiptStore.updateApprovalDecision({ sessionId: session.id, approvalId, state: "resolved" });
    } catch (error) {
      await actionReceiptStore.updateApprovalDecision({
        sessionId: session.id,
        approvalId,
        state: "indeterminate"
      }).catch(() => {});
      throw error;
    }

    approval.status = delivery.status;
    approval.resolvedAt = new Date().toISOString();
    emit(session, "approval.resolved", publicApproval(approval));
    if (session.runtime) {
      setStatus(session, "running");
    } else if (!approved) {
      setStatus(session, "failed");
    }
    return delivery;
  } finally {
    approvalDecisionLocks.delete(approval);
  }
}

function replayApprovalDecision(session, approvalId, receipt) {
  if (receipt.state === "dispatching") {
    const error = new Error("approval_decision_pending");
    error.status = 409;
    throw error;
  }
  if (receipt.state !== "resolved") {
    const error = new Error("idempotency_outcome_unknown");
    error.status = 503;
    error.safeLogCode = "idempotency_outcome_unknown";
    throw error;
  }
  const approval = session.approvals.get(approvalId);
  if (approval && approval.status !== receipt.delivery.status) {
    approval.status = receipt.delivery.status;
    approval.resolvedAt ??= new Date().toISOString();
    void schedulePersist(session);
  }
  return { ...receipt.delivery };
}

function completeFakeApproval(session, approved) {
  if (approved) {
    emit(session, "session.output", {
      sessionId: session.id,
      text: "Approval accepted. Fake Codex continues."
    });
    completeFakeTurn(session);
    return;
  }
  emit(session, "session.output", {
    sessionId: session.id,
    text: "Approval rejected. Fake Codex stopped."
  });
}

export function listApprovals(session) {
  return {
    approvals: [...session.approvals.values()].map(publicApproval)
  };
}

export async function listFiles(session) {
  return readWorkspaceChanges(workspaceFor(session).path);
}

export async function getDiff(session, filePath) {
  if (typeof filePath !== "string" || filePath.length === 0) {
    const error = new Error("file_path_required");
    error.status = 400;
    throw error;
  }
  return safePublicData(await readWorkspaceDiff(workspaceFor(session).path, filePath));
}

function workspaceFor(session) {
  const workspace = runtimeConfig?.workspaces?.find((item) => item.id === session.workspaceId);
  if (!workspace) {
    const error = new Error("invalid_session_context");
    error.status = 400;
    throw error;
  }
  return workspace;
}

function runFakeTurn(session, text) {
  setTimeout(() => {
    emit(session, "session.output", {
      sessionId: session.id,
      text: "Fake Codex received message."
    });
  }, 250);

  setTimeout(() => {
    if (needsApproval(text)) {
      requestApproval(session, text);
      return;
    }
    completeFakeTurn(session);
  }, 800);
}

function completeFakeTurn(session) {
  session.files = [
    {
      path: "README.md",
      status: "modified",
      additions: 1,
      deletions: 0
    }
  ];
  emit(session, "files.changed", { sessionId: session.id, files: session.files });
  emit(session, "session.output", {
    sessionId: session.id,
    text: "Fake Codex turn completed."
  });
  setStatus(session, "completed");
}

function requestApproval(session, text) {
  const approval = {
    id: randomUUID(),
    sessionId: session.id,
    status: "pending",
    kind: "generic",
    title: "Approval required",
    reason: "Message looks like install, network, delete, or outside-workspace action.",
    reasonKey: "approval.reason.generic",
    commandPreview: text,
    createdAt: new Date().toISOString()
  };

  session.approvals.set(approval.id, approval);
  setStatus(session, "waiting_approval");
  emit(session, "approval.requested", publicApproval(approval));
}

function publicApproval(approval) {
  return sanitizePublicData({
    id: approval.id,
    sessionId: approval.sessionId,
    status: approval.status,
    title: approval.title,
    kind: approval.kind,
    reason: approval.reason,
    reasonKey: approval.reasonKey,
    commandPreview: approval.commandPreview,
    commandPreviewKey: approval.commandPreviewKey,
    commandPreviewValues: approval.commandPreviewValues,
    cwd: approval.cwd,
    createdAt: approval.createdAt,
    resolvedAt: approval.resolvedAt,
    canApproveForSession: ["command", "file_change", "permissions"].includes(approval.kind),
    canApproveAlways: Boolean(privateApprovalData.get(approval)?.proposedExecpolicyAmendment)
  }, runtimeConfig);
}

function needsApproval(text) {
  return /\b(npm install|pip install|cargo install|winget|delete|remove|network|outside|deploy|push)\b/i.test(text);
}

function normalizedClientMessageId(value) {
  return normalizedClientActionId(value, "invalid_client_message_id");
}

function normalizedClientActionId(value, errorCode) {
  if (typeof value !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value)) {
    throw idempotencyError(errorCode, 400);
  }
  return value.toLowerCase();
}

function messagePayloadHash(input, text) {
  const normalized = {
    text,
    executionAccountId: typeof input.executionAccountId === "string" ? input.executionAccountId : null,
    options: sanitizeMessageOptions(input.options),
    attachments: Array.isArray(input.attachments)
      ? input.attachments.map((item) => typeof item?.id === "string" ? item.id : null)
      : [],
    skills: sanitizeSkills(input.skills).map((item) => item.name)
  };
  return createHash("sha256").update(JSON.stringify(normalized)).digest("hex");
}

function replayMessageDelivery(receipt) {
  if (["accepted", "dispatched", "steered", "deleted"].includes(receipt.state)) {
    return { ...receipt.delivery };
  }
  throw idempotencyError("message_delivery_indeterminate", 409);
}

function idempotencyError(code, status) {
  const error = new Error(code);
  error.status = status;
  error.safeLogCode = code;
  return error;
}

function publicTurnStartError(error) {
  return /turn\/start timeout/i.test(error?.message ?? "") ? "turn_start_timeout" : "turn_start_failed";
}

function enqueueMessage(session, input) {
  const item = {
    id: input.id ?? randomUUID(),
    clientMessageId: input.clientMessageId,
    text: input.text,
    executionAccountId: input.executionAccountId,
    options: sanitizeMessageOptions(input.options),
    attachments: input.attachments ?? sanitizeAttachments(session, input.attachments),
    skills: input.skills ?? sanitizeSkills(input.skills),
    createdAt: new Date().toISOString()
  };
  session.queueAdvanceBlocked = false;
  session.queuedMessages = [...(session.queuedMessages ?? []), item];
  emit(session, "message.accepted", {
    sessionId: session.id,
    clientMessageId: item.clientMessageId,
    queuedMessageId: item.id,
    text: item.text,
    ...((item.attachments ?? []).length ? { attachmentCount: item.attachments.length, attachments: publicMessageAttachments(item.attachments) } : {}),
    queued: true
  });
  setStatus(session, session.status);
  return item;
}

function publicQueuedMessage(item) {
  return {
    id: item.id,
    clientMessageId: item.clientMessageId,
    text: sanitizePublicText(item.text, runtimeConfig),
    executionAccountId: sanitizePublicText(item.executionAccountId, runtimeConfig) || undefined,
    attachments: (item.attachments ?? []).map(({ id, kind, name, mimeType, createdAt }) => compactObject({ id, kind, name, mimeType, createdAt })),
    skills: (item.skills ?? []).map(({ name }) => ({ name: sanitizePublicText(name, runtimeConfig) })),
    createdAt: item.createdAt
  };
}

function publicMessageAttachments(attachments = []) {
  return attachments.map(({ id, kind, name, mimeType, createdAt }) =>
    compactObject({ id, kind, name, mimeType, createdAt })
  );
}

function sanitizeMessageOptions(options = {}) {
  return compactObject({
    model: cleanString(options.model),
    effort: cleanString(options.effort),
    serviceTier: cleanString(options.serviceTier),
    permissionProfile: cleanCapabilityId(options.permissionProfile),
    approvalPolicy: ["untrusted", "on-failure", "on-request", "never"].includes(options.approvalPolicy) ? options.approvalPolicy : undefined,
    sandbox: ["read-only", "workspace-write", "danger-full-access"].includes(options.sandbox) ? options.sandbox : undefined,
    personality: ["none", "friendly", "pragmatic"].includes(options.personality) ? options.personality : undefined,
    planMode: Boolean(options.planMode)
  });
}

function sanitizeAttachments(session, attachments = []) {
  if (!Array.isArray(attachments)) return [];
  if (attachments.length > MAX_MESSAGE_ATTACHMENT_COUNT) {
    throw attachmentError("attachment_count_exceeded", 413);
  }
  const storedById = new Map((session.attachments ?? []).map((item) => [item.id, item]));
  const sanitized = attachments
    .map((item) => storedById.get(item?.id))
    .filter((item) => item?.kind === "image" && typeof item.path === "string")
    .map((item) => compactObject({
      id: item.id,
      kind: "image",
      name: item.name,
      mimeType: item.mimeType,
      path: item.path,
      size: item.size,
      createdAt: item.createdAt
    }));
  const totalBytes = sanitized.reduce((total, item) => total + (Number.isSafeInteger(item.size) && item.size > 0 ? item.size : 0), 0);
  if (totalBytes > MAX_MESSAGE_ATTACHMENT_BYTES) {
    throw attachmentError("attachment_total_size_exceeded", 413);
  }
  return sanitized;
}

function sanitizeSkills(skills = []) {
  const names = new Set();
  for (const item of skills ?? []) {
    const name = cleanCapabilityId(item?.name);
    if (name) names.add(name);
  }
  return [...names].map((name) => ({ name }));
}

async function resolveRuntimeSkills(session, skills = []) {
  const requested = sanitizeSkills(skills);
  if (!requested.length || runtimeConfig?.engine !== "codex") return requested;
  const account = runtimeConfig?.accounts?.find((item) => item.id === session.accountId);
  const workspace = runtimeConfig?.workspaces?.find((item) => item.id === session.workspaceId);
  if (!account || !workspace) return [];
  const result = await requestRuntimeCapability({
    codexCliPath: runtimeConfig.codexCliPath,
    codexHome: account.codexHome,
    method: "skills/list",
    params: { cwds: [workspace.path] }
  });
  const available = new Map();
  for (const entry of result?.data ?? []) {
    for (const skill of entry.skills ?? []) {
      if (skill?.enabled !== false && typeof skill.name === "string" && typeof skill.path === "string") {
        available.set(skill.name, { name: skill.name, path: skill.path });
      }
    }
  }
  return requested.map((skill) => available.get(skill.name)).filter(Boolean);
}

function cleanCapabilityId(value) {
  if (typeof value !== "string") return undefined;
  const name = value.trim();
  if (!name || name.length > 160 || /[\\/]/.test(name) || /^[A-Za-z]:/.test(name)) return undefined;
  return name;
}

function cleanString(value) {
  return typeof value === "string" && value.trim() ? value.trim() : undefined;
}

function imageExtension(mimeType) {
  if (mimeType === "image/png") return ".png";
  if (mimeType === "image/jpeg") return ".jpg";
  if (mimeType === "image/webp") return ".webp";
  if (mimeType === "image/gif") return ".gif";
  return "";
}

function decodeAttachmentData(value) {
  if (typeof value !== "string" || !value || value.length % 4 !== 0 || !hasCanonicalBase64Characters(value)) {
    throw attachmentError("invalid_attachment_data", 400);
  }
  const data = Buffer.from(value, "base64");
  if (data.length === 0 || data.toString("base64") !== value) {
    throw attachmentError("invalid_attachment_data", 400);
  }
  if (data.length > MAX_ATTACHMENT_BYTES) throw attachmentError("attachment_too_large", 413);
  return data;
}

function hasCanonicalBase64Characters(value) {
  const padding = value.endsWith("==") ? 2 : value.endsWith("=") ? 1 : 0;
  for (let index = 0; index < value.length - padding; index += 1) {
    const code = value.charCodeAt(index);
    const allowed = (code >= 0x41 && code <= 0x5a)
      || (code >= 0x61 && code <= 0x7a)
      || (code >= 0x30 && code <= 0x39)
      || code === 0x2b
      || code === 0x2f;
    if (!allowed) return false;
  }
  for (let index = value.length - padding; index < value.length; index += 1) {
    if (value[index] !== "=") return false;
  }
  return true;
}

function detectImageType(data) {
  if (data.length >= 8 && data.subarray(0, 8).equals(Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]))) {
    return { mimeType: "image/png", extension: ".png" };
  }
  if (data.length >= 3 && data[0] === 0xff && data[1] === 0xd8 && data[2] === 0xff) {
    return { mimeType: "image/jpeg", extension: ".jpg" };
  }
  if (data.length >= 6 && ["GIF87a", "GIF89a"].includes(data.subarray(0, 6).toString("ascii"))) {
    return { mimeType: "image/gif", extension: ".gif" };
  }
  if (data.length >= 12 && data.subarray(0, 4).toString("ascii") === "RIFF" && data.subarray(8, 12).toString("ascii") === "WEBP") {
    return { mimeType: "image/webp", extension: ".webp" };
  }
  return undefined;
}

function attachmentDisplayName(name, extension) {
  const safeName = safeAttachmentName(name);
  const baseName = safeName.replace(/\.[^.]*$/, "") || "image";
  return `${baseName.slice(0, 120 - extension.length)}${extension}`;
}

function attachmentError(code, status) {
  const error = new Error(code);
  error.status = status;
  error.safeLogCode = code;
  return error;
}

function attachmentIsInUse(session, attachmentId) {
  if ((session.queuedMessages ?? []).some((item) => (item.attachments ?? []).some((attachment) => attachment.id === attachmentId))) {
    return true;
  }
  return ["starting", "running", "waiting_approval"].includes(session.status)
    && session.activeAttachmentIds?.has(attachmentId);
}

function markAttachmentDeleted(events = [], attachment) {
  return events.map((event) => {
    if (!Array.isArray(event.data?.attachments) || !event.data.attachments.some((item) => item?.id === attachment.id)) return event;
    return {
      ...event,
      data: {
        ...event.data,
        attachments: event.data.attachments.map((item) => item?.id === attachment.id
          ? { id: attachment.id, kind: "image", name: attachment.name, deleted: true }
          : item)
      }
    };
  });
}

function withAttachmentMutation(session, action) {
  const previous = attachmentMutationQueues.get(session) ?? Promise.resolve();
  const current = previous.catch(() => {}).then(action);
  attachmentMutationQueues.set(session, current);
  return current.finally(() => {
    if (attachmentMutationQueues.get(session) === current) attachmentMutationQueues.delete(session);
  });
}

function withQueueMutation(session, action) {
  const previous = queueMutationQueues.get(session) ?? Promise.resolve();
  const current = previous.catch(() => {}).then(action);
  queueMutationQueues.set(session, current);
  return current.finally(() => {
    if (queueMutationQueues.get(session) === current) queueMutationQueues.delete(session);
  });
}

function safeAttachmentName(name) {
  const value = typeof name === "string" && name.trim() ? name.trim() : "image";
  return value.replace(/[\p{Cc}\p{Cf}\\/:*?"<>|]+/gu, "_").slice(0, 120);
}

async function deleteSessionAttachments(config, sessionId) {
  if (!config?.dataDir || typeof sessionId !== "string" || !sessionId) return;
  for (const directory of ["attachments", "artifacts"]) {
    const root = resolve(config.dataDir, directory);
    const target = resolve(root, sessionId);
    const relation = relative(root, target);
    if (!relation || relation.startsWith("..") || resolve(root, relation) !== target) continue;
    await rm(target, { recursive: true, force: true });
  }
}

function compactObject(value) {
  return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined && item !== null));
}

function safePublicData(value, config = runtimeConfig) {
  return sanitizePublicData(projectPublicDiffData(value), config);
}

function fulfilledValue(result) {
  return result.status === "fulfilled" ? result.value : {};
}

function rejectedCapability(kind, result) {
  return result.status === "rejected" ? { kind, message: "capability_unavailable" } : null;
}

function setStatus(session, status) {
  session.status = status;
  if (["completed", "failed", "interrupted"].includes(status)) session.activeAttachmentIds = new Set();
  session.updatedAt = new Date().toISOString();
  emit(session, "session.status", publicSession(session));
}

function emit(session, event, data) {
  const publicData = safePublicData(data);
  let item = {
    id: randomUUID(),
    event,
    data: publicData,
    createdAt: new Date().toISOString()
  };

  const previousAssistantDelta = event === "session.output" && publicData?.kind === "assistant_delta"
    ? session.events.at(-1)
    : undefined;
  const assistantDeltaKey = publicData?.itemId ?? publicData?.turnId;
  const coalescesAssistantDelta = previousAssistantDelta?.event === "session.output"
    && previousAssistantDelta.data?.kind === "assistant_delta"
    && assistantDeltaKey != null
    && (previousAssistantDelta.data?.itemId ?? previousAssistantDelta.data?.turnId)
      === assistantDeltaKey;
  if (coalescesAssistantDelta) {
    previousAssistantDelta.data = {
      ...previousAssistantDelta.data,
      ...publicData,
      text: `${previousAssistantDelta.data?.text ?? ""}${publicData.text ?? ""}`
    };
    item = previousAssistantDelta;
  }

  const activityKey = event === "activity.updated" ? liveActivityEventKey(publicData) : undefined;
  const previousActivityUpdate = activityKey
    ? session.activityUpdateEvents?.get(activityKey)
    : undefined;
  const coalescesActivityUpdate = Boolean(previousActivityUpdate);
  if (previousActivityUpdate) {
    previousActivityUpdate.data = {
      ...previousActivityUpdate.data,
      ...publicData,
      text: `${previousActivityUpdate.data?.text ?? ""}${publicData.text ?? ""}`
    };
    item = previousActivityUpdate;
  } else if (activityKey) {
    session.activityUpdateEvents ??= new Map();
    session.activityUpdateEvents.set(activityKey, item);
  }

  const snapshotKey = event === "session.output" && publicData?.kind === "assistant_snapshot"
    ? publicData.itemId ?? publicData.turnId
    : undefined;
  const replacesSnapshot = snapshotKey != null && session.events.some((candidate) => candidate.event === "session.output"
    && candidate.data?.kind === "assistant_snapshot"
    && (candidate.data?.itemId ?? candidate.data?.turnId) === snapshotKey);
  if (replacesSnapshot) {
    session.events = session.events.filter((candidate) => candidate.event !== "session.output"
      || !["assistant_delta", "assistant_snapshot"].includes(candidate.data?.kind)
      || (candidate.data?.itemId ?? candidate.data?.turnId) !== snapshotKey);
  }
  if (!coalescesAssistantDelta && !coalescesActivityUpdate) session.events.push(item);
  const persisted = replacesSnapshot
    ? Promise.resolve()
    : shouldDeferLiveEventPersistence(event, publicData)
      ? scheduleDeferredPersist(session)
      : schedulePersist(session);
  broadcastSessionEvent(session, event, publicData);
  if (event === "session.created" || event === "session.status") broadcastSessionUpdate(session);
  return persisted;
}

function broadcastSessionEvent(session, event, data) {
  const batchKey = liveSseBatchKey(event, data);
  if (!batchKey || session.clients.size === 0) {
    flushPendingSessionEvent(session);
    for (const client of session.clients) sendEvent(client, event, data);
    return;
  }

  const pending = session.pendingSseEvent;
  if (pending?.key === batchKey) {
    pending.data = {
      ...pending.data,
      ...data,
      text: `${pending.data?.text ?? ""}${data.text ?? ""}`
    };
    return;
  }

  flushPendingSessionEvent(session);
  const queued = {
    key: batchKey,
    event,
    data: { ...data }
  };
  queued.timer = setTimeout(() => flushPendingSessionEvent(session), LIVE_SSE_BATCH_MS);
  queued.timer.unref?.();
  session.pendingSseEvent = queued;
}

function flushPendingSessionEvent(session) {
  const pending = session?.pendingSseEvent;
  if (!pending) return;
  clearTimeout(pending.timer);
  session.pendingSseEvent = undefined;
  for (const client of session.clients) sendEvent(client, pending.event, pending.data);
}

function discardPendingSessionEvent(session) {
  const pending = session?.pendingSseEvent;
  if (!pending) return;
  clearTimeout(pending.timer);
  session.pendingSseEvent = undefined;
}

function liveSseBatchKey(event, data) {
  if (typeof data?.text !== "string") return undefined;
  if (event === "session.output" && data.kind !== "assistant_delta") return undefined;
  if (!["session.output", "reasoning.delta", "activity.updated"].includes(event)) return undefined;
  if (data.itemId == null && data.turnId == null) return undefined;
  return `${event}\u0000${data.turnId ?? ""}\u0000${data.itemId ?? ""}`;
}

function liveActivityEventKey(data) {
  return data?.itemId == null ? undefined : `${data.turnId ?? ""}\u0000${data.itemId}`;
}

function broadcastSessionUpdate(session) {
  const projected = publicSession(session);
  for (const client of sessionListClients) sendEvent(client, "session.updated", projected);
}

function broadcastSessionListSnapshot() {
  const snapshot = { sessions: listSessions() };
  for (const client of sessionListClients) sendEvent(client, "sessions.snapshot", snapshot);
}

function broadcastSessionSnapshot(session) {
  const snapshot = safePublicData({ session: publicSession(session), events: session.events });
  for (const client of session.clients) sendEvent(client, "session.snapshot", snapshot);
}

function cleanConversationTitle(value) {
  if (typeof value !== "string") return undefined;
  const title = sanitizePublicText(value.trim(), runtimeConfig);
  return typeof title === "string" && title.trim() ? title.trim() : undefined;
}

function threadConversationTitle(thread) {
  return cleanConversationTitle(thread?.name) ?? cleanConversationTitle(thread?.preview);
}

function syncThreadConversationTitle(session, thread) {
  const title = cleanConversationTitle(thread?.name)
    ?? (!session.conversationTitle?.trim() ? cleanConversationTitle(thread?.preview) : undefined);
  if (!title || title === session.conversationTitle) return false;
  session.conversationTitle = title;
  return true;
}

function schedulePersist(session) {
  if (!sessionStore) return Promise.resolve();
  cancelDeferredPersist(session);
  return enqueuePersist(session?.id ? new Set([session.id]) : undefined);
}

function scheduleDeferredPersist(session) {
  if (!sessionStore || !session?.id) return Promise.resolve();
  deferredPersistSessionIds.add(session.id);
  if (!deferredPersistTimer) {
    deferredPersistTimer = setTimeout(flushDeferredPersists, LIVE_EVENT_PERSIST_DEBOUNCE_MS);
    deferredPersistTimer.unref?.();
  }
  return Promise.resolve();
}

function flushDeferredPersists() {
  deferredPersistTimer = undefined;
  const dirtyIds = deferredPersistSessionIds;
  deferredPersistSessionIds = new Set();
  if (dirtyIds.size === 0) return;
  void enqueuePersist(dirtyIds);
}

function cancelDeferredPersist(session) {
  if (session?.id) deferredPersistSessionIds.delete(session.id);
  else deferredPersistSessionIds.clear();
  if (deferredPersistSessionIds.size === 0 && deferredPersistTimer) {
    clearTimeout(deferredPersistTimer);
    deferredPersistTimer = undefined;
  }
}

function enqueuePersist(requestedDirtyIds) {
  saveRequested = true;
  if (requestedDirtyIds && !persistAllRequested) {
    for (const id of requestedDirtyIds) dirtySessionIds.add(id);
  } else {
    persistAllRequested = true;
    dirtySessionIds.clear();
  }
  const request = new Promise((resolve, reject) => {
    persistWaiters.push({ resolve, reject });
  });
  if (!saveRunning) {
    saveRunning = true;
    saveQueue = drainPersistQueue().finally(() => {
      saveRunning = false;
    });
    void saveQueue.catch((error) => writeSafeLog(console, "error", "session_persistence_failed", { error }, runtimeConfig));
  }
  void request.catch(() => {});
  return request;
}

function shouldDeferLiveEventPersistence(event, data) {
  return event === "reasoning.delta"
    || event === "activity.updated"
    || (event === "session.output" && data?.kind === "assistant_delta");
}

async function drainPersistQueue() {
  while (saveRequested) {
    saveRequested = false;
    const persistAll = persistAllRequested;
    const dirtyIds = dirtySessionIds;
    persistAllRequested = false;
    dirtySessionIds = new Set();
    const waiters = persistWaiters;
    persistWaiters = [];
    try {
      await persistSessions(persistAll ? undefined : dirtyIds);
      for (const waiter of waiters) waiter.resolve();
    } catch (error) {
      for (const waiter of [...waiters, ...persistWaiters]) waiter.reject(error);
      persistWaiters = [];
      saveRequested = false;
      persistAllRequested = false;
      dirtySessionIds.clear();
      throw error;
    }
  }
}

function persistSessions(dirtyIds) {
  if (!sessionStore) return Promise.resolve();
  return sessionStore.save([...sessions.values()], { dirtySessionIds: dirtyIds });
}

function emitRuntimeEvent(session, event, data = {}) {
  if (session.codexThreadId && data.threadId && data.threadId !== session.codexThreadId) return;

  if (event === "process.stderr") {
    writeSafeLog(console, "warn", "codex_runtime_stderr", { text: data.text }, runtimeConfig);
    return;
  }

  if (event === "codex.event") {
    return;
  }

  if (event === "approval.requested") {
    return recordRuntimeApproval(session, data).catch((error) => {
      writeSafeLog(console, "warn", "approval_notification_idempotency_failed", { error }, runtimeConfig);
      emit(session, "error", { message: "idempotency_store_unavailable" });
    });
  }

  if (event === "session.output") {
    if (data.kind === "assistant_delta") {
      const itemId = data.itemId ?? data.turnId ?? "assistant";
      session.assistantSnapshots ??= new Map();
      session.artifactRegistrationCache ??= new Map();
      const snapshot = `${session.assistantSnapshots.get(itemId) ?? ""}${data.text ?? ""}`;
      session.assistantSnapshots.set(itemId, snapshot);
      if (!shouldProjectAssistantSnapshot(snapshot, data.text)) {
        emit(session, "session.output", {
          sessionId: session.id,
          turnId: data.turnId,
          itemId: data.itemId,
          text: data.text,
          kind: "assistant_delta"
        });
        if (data.text?.trim()) maybeGenerateConversationTitle(session);
        return;
      }
      emit(session, "session.output", {
        sessionId: session.id,
        turnId: data.turnId,
        itemId: data.itemId,
        text: projectAssistantContent(session, snapshot),
        kind: "assistant_snapshot"
      });
      if (data.text?.trim()) maybeGenerateConversationTitle(session);
      return;
    }
    emit(session, "session.output", {
      sessionId: session.id,
      turnId: data.turnId,
      itemId: data.itemId,
      text: data.text,
      kind: data.kind
    });
    return;
  }

  if (event === "thread.token_usage") {
    session.tokenUsage = publicTokenUsage(data.tokenUsage);
    emit(session, event, { sessionId: session.id, tokenUsage: session.tokenUsage });
    return;
  }

  if (event === "turn.completed") {
    emit(session, event, { sessionId: session.id, ...data });
    const terminalStatus = data.status === "failed" ? "failed" : data.status === "interrupted" ? "interrupted" : "completed";
    setStatus(session, terminalStatus);
    session.assistantSnapshots?.clear();
    session.artifactRegistrationCache?.clear();
    session.activityUpdateEvents?.clear();
    const hasQueuedMessages = (session.queuedMessages ?? []).length > 0;
    const shouldAdvanceQueue = terminalStatus === "completed" && hasQueuedMessages;
    if (shouldAdvanceQueue) {
      scheduleQueueAdvance(session);
    } else if (session.clients.size === 0) {
      releaseRuntime(session, { resumeWatcher: false });
    } else {
      void ensureExternalWatcher(session);
    }
    return;
  }

  if (event === "activity.updated" && typeof data.text === "string" && data.itemId) {
    const text = boundedLiveActivityDelta(session, data.itemId, data.text);
    if (!text) return;
    data = { ...data, text };
  }

  if (event.startsWith("activity.") && data[INTERNAL_GENERATED_IMAGE_ITEM]) {
    const images = projectGeneratedImageActivity(session, data[INTERNAL_GENERATED_IMAGE_ITEM]);
    data = { ...data, ...(images.length ? { images } : {}) };
  }

  if (event.startsWith("activity.") && data.kind === "fileChange") {
    void emitFileChangeActivity(session, event, data);
    return;
  }

  if (event === "activity.completed" && data.itemId) {
    session.activityOutputBytes?.delete(data.itemId);
    session.activityUpdateEvents?.delete(liveActivityEventKey(data));
  }

  if (event === "process.exited") {
    emit(session, "process.exited", data);
    if (session.status === "running" || session.status === "starting") {
      setStatus(session, "failed");
    }
    releaseRuntime(session, { resumeWatcher: false });
    return;
  }

  if (event === "error") {
    emit(session, "error", { message: data?.message ?? "runtime_error" });
    if (session.status === "running" || session.status === "starting" || session.status === "waiting_approval") {
      setStatus(session, "failed");
    }
    releaseRuntime(session, { resumeWatcher: false });
    return;
  }

  emit(session, event, data);
}

async function recordRuntimeApproval(session, data) {
  const approvalId = randomUUID();
  const reservation = await actionReceiptStore.reserveApprovalNotification({
    sessionId: session.id,
    accountId: session.accountId,
    threadId: data.threadId ?? session.runtime?.threadId ?? session.codexThreadId,
    turnId: data.turnId,
    itemId: data.itemId,
    runtimeRequestId: data.runtimeRequestId,
    approvalId
  });
  if (!reservation.created) return;
  const approval = {
    id: reservation.receipt.approvalId,
    sessionId: session.id,
    runtimeRequestId: data.runtimeRequestId,
    threadId: data.threadId ?? session.runtime?.threadId ?? session.codexThreadId,
    turnId: data.turnId,
    itemId: data.itemId,
    kind: data.kind,
    status: "pending",
    title: data.title,
    reason: data.reason,
    reasonKey: data.reasonKey,
    commandPreview: data.commandPreview,
    commandPreviewKey: data.commandPreviewKey,
    commandPreviewValues: data.commandPreviewValues,
    cwd: data.cwd,
    permissions: data.permissions,
    createdAt: new Date().toISOString()
  };
  if (data.proposedExecpolicyAmendment) {
    privateApprovalData.set(approval, { proposedExecpolicyAmendment: data.proposedExecpolicyAmendment });
  }
  session.approvals.set(approval.id, approval);
  setStatus(session, "waiting_approval");
  await emit(session, "approval.requested", publicApproval(approval));
}

function maybeGenerateConversationTitle(session) {
  if (!generateRuntimeTitle || session.autoTitleAttempted || !session.codexThreadId) return;
  const completedTurns = session.events.filter((event) =>
    event.event === "turn.completed" && event.data?.status === "completed"
  );
  if (completedTurns.length !== 0) return;
  const fallbackTitle = cleanConversationTitle(firstUserMessage(session.events));
  if (!fallbackTitle || session.conversationTitle !== fallbackTitle) return;
  const conversation = visibleConversationLines(session.events).join("\n").trim();
  if (!conversation.includes("Assistant:")) return;
  const account = runtimeConfig?.accounts?.find((item) => item.id === session.accountId);
  if (!account || account.historyOnly) return;
  session.autoTitleAttempted = true;
  void generateRuntimeTitle({
    accountId: account.id,
    codexHome: account.codexHome,
    threadId: session.codexThreadId,
    conversation
  }).then((value) => {
    const title = cleanConversationTitle(value);
    if (!title || session.conversationTitle !== fallbackTitle) return;
    session.conversationTitle = title;
    broadcastSessionSnapshot(session);
    broadcastSessionUpdate(session);
    schedulePersist(session);
  }).catch(() => {});
}

function boundedLiveActivityDelta(session, itemId, delta) {
  session.activityOutputBytes ??= new Map();
  const used = session.activityOutputBytes.get(itemId) ?? 0;
  const remaining = MAX_LIVE_ACTIVITY_OUTPUT_BYTES - used;
  if (remaining <= 0) return "";
  const bytes = Buffer.from(delta, "utf8");
  if (bytes.length <= remaining) {
    session.activityOutputBytes.set(itemId, used + bytes.length);
    return delta;
  }
  session.activityOutputBytes.set(itemId, MAX_LIVE_ACTIVITY_OUTPUT_BYTES);
  const prefix = bytes.subarray(0, remaining).toString("utf8").replace(/\uFFFD$/, "");
  return `${prefix}\n[output truncated]`;
}

function projectAssistantContent(session, text) {
  const workspace = runtimeConfig?.workspaces?.find((item) => item.id === session.workspaceId);
  const withImages = projectGeneratedImages({
    text,
    session,
    workspacePath: workspace?.path,
    dataDir: runtimeConfig?.dataDir
  });
  return projectWorkspaceArtifacts({
    text: withImages,
    session,
    workspacePath: workspace?.path,
    dataDir: runtimeConfig?.dataDir,
    registrationCache: session.artifactRegistrationCache
  });
}

function shouldProjectAssistantSnapshot(snapshot, delta) {
  if (!/[})]/.test(delta ?? "")) return false;
  return /::codex-inline-vis\{[^}]*\}/i.test(snapshot)
    || /!?\[[^\]]*\]\((?!https?:|mailto:|#|\/api\/)[^)]+\)/i.test(snapshot);
}

function projectGeneratedImageActivity(session, item) {
  return projectGeneratedImageItem({
    item,
    session,
    dataDir: runtimeConfig?.dataDir
  });
}

async function emitFileChangeActivity(session, event, data) {
  const workspace = runtimeConfig.workspaces.find((item) => item.id === session.workspaceId);
  const { changes, ...publicData } = data;
  const files = safeChangeFiles(changes, workspace?.path, workspace?.path);
  const diffs = await Promise.all(files.map(async (file) => {
    if (typeof file.diff === "string" || file.unavailableReason) return file;
    try {
      const snapshot = await readDiff(workspace.path, file.path);
      return typeof snapshot.diff === "string" || snapshot.unavailableReason ? { ...file, ...snapshot } : file;
    } catch {
      return file;
    }
  }));
  emit(session, event, {
    ...publicData,
    paths: files.map((file) => file.path),
    diffs: diffs.filter((file) => typeof file.diff === "string" || file.unavailableReason)
  });
}

function scheduleQueueAdvance(session) {
  if (session.queueAdvanceBlocked || session.queueAdvancing || (session.queuedMessages ?? []).length === 0) return;
  session.queueAdvancing = true;
  let advanceAgain = false;
  void advanceQueuedMessage(session).then((result) => {
    advanceAgain = result === "continue";
  }).finally(() => {
    session.queueAdvancing = false;
    if (advanceAgain) scheduleQueueAdvance(session);
  });
}

async function advanceQueuedMessage(session) {
  const item = session.queuedMessages?.[0];
  if (!item) return;
  let startedRuntime;
  let dispatchReceiptWritePending = false;
  const previousStatus = session.status;
  try {
    if (item.clientMessageId) {
      const receipt = await actionReceiptStore.getMessage({
        sessionId: session.id,
        clientMessageId: item.clientMessageId
      });
      if (!receipt || ["dispatching", "steering", "indeterminate"].includes(receipt.state)) {
        throw new Error("queued_message_delivery_indeterminate");
      }
      if (["dispatched", "steered", "deleted"].includes(receipt.state)) {
        session.queuedMessages = session.queuedMessages.filter((queued) => queued.id !== item.id);
        await schedulePersist(session);
        return "continue";
      }
    }
    assertMessagePermissionsAllowed(item.options, runtimeConfig?.permissionCeiling);
    await ensureQueuedAttachmentsAvailable(item);
    const executionAccount = runtimeConfig?.engine === "codex"
      ? executionAccountFor(item.executionAccountId ?? session.accountId)
      : null;
    if (runtimeConfig?.engine === "codex") {
      assertSessionContextAvailable(session, executionAccount);
    }
    stopExternalWatcher(session);
    let runtimeText = item.text;
    if (runtimeConfig?.engine === "codex" && executionAccount?.id !== session.accountId) {
      runtimeText = await startHandoffRuntime(session, executionAccount, item.text, item.clientMessageId);
    } else if (!session.runtime && runtimeConfig?.engine === "codex") {
      await resumeRuntime(session);
    }
    if (!session.runtime) {
      if (session.status === "interrupted") throw new Error("turn_start_canceled");
      return;
    }
    session.activeAttachmentIds = new Set((item.attachments ?? []).map((attachment) => attachment.id));
    setStatus(session, "starting");
    const runtime = session.runtime;
    startedRuntime = runtime;
    const resolvedSkills = await resolveRuntimeSkills(session, item.skills ?? []);
    if (session.runtime !== runtime || session.status !== "starting") throw new Error("turn_start_canceled");
    const claimed = await withQueueMutation(session, () => {
      if (!(session.queuedMessages ?? []).some((queued) => queued.id === item.id)) return false;
      queuedDispatchClaims.set(session, item.id);
      return true;
    });
    if (!claimed) {
      session.activeAttachmentIds = new Set();
      setStatus(session, previousStatus);
      return "continue";
    }
    if (item.clientMessageId) {
      await actionReceiptStore.updateMessage({
        sessionId: session.id,
        clientMessageId: item.clientMessageId,
        state: "dispatching"
      });
    }
    const turnId = await runtime.sendMessage({
      text: runtimeText,
      options: item.options ?? {},
      attachments: item.attachments ?? [],
      skills: resolvedSkills
    });
    if (session.runtime !== runtime || session.status !== "starting") throw new Error("turn_start_canceled");
    if (item.clientMessageId) {
      dispatchReceiptWritePending = true;
      await actionReceiptStore.updateMessage({
        sessionId: session.id,
        clientMessageId: item.clientMessageId,
        state: "dispatched"
      });
      dispatchReceiptWritePending = false;
    }
    session.queuedMessages = session.queuedMessages.filter((queued) => queued.id !== item.id);
    emit(session, "message.accepted", {
      sessionId: session.id,
      clientMessageId: item.clientMessageId,
      queuedMessageId: item.id,
      turnId,
      text: item.text,
      ...((item.attachments ?? []).length ? { attachmentCount: item.attachments.length, attachments: publicMessageAttachments(item.attachments) } : {}),
      queued: true
    });
    setStatus(session, "running");
  } catch (error) {
    if (dispatchReceiptWritePending) {
      session.queueAdvanceBlocked = true;
      if (item.clientMessageId) {
        await actionReceiptStore.updateMessage({
          sessionId: session.id,
          clientMessageId: item.clientMessageId,
          state: "indeterminate"
        }).catch(() => {});
      }
      try {
        await startedRuntime?.interrupt?.();
      } catch {
        if (["completed", "failed", "interrupted"].includes(session.status)) {
          const terminalStatus = session.status;
          emit(session, "error", { message: "idempotency_store_unavailable" });
          setStatus(session, terminalStatus);
          return;
        }
        emit(session, "error", { message: "turn_stop_unconfirmed" });
        setStatus(session, "starting");
        return;
      }
      if (session.runtime === startedRuntime) releaseRuntime(session, { resumeWatcher: false });
      emit(session, "error", { message: "idempotency_store_unavailable" });
      setStatus(session, "failed");
      return;
    }
    if (["turn_stop_unconfirmed", "turn_start_outcome_unknown"].includes(error.message)) {
      session.queueAdvanceBlocked = true;
      return;
    }
    if (error.message === "queued_message_delivery_indeterminate") {
      session.queueAdvanceBlocked = true;
      emit(session, "error", { message: error.message });
      setStatus(session, "failed");
      return;
    }
    if (error.message === "turn_start_canceled" || session.status === "interrupted") {
      await restoreQueuedReceipt(session, item);
      return;
    }
    if (!await restoreQueuedReceipt(session, item)) return;
    const message = error.code === "pro_required"
      ? "pro_required"
      : ["queued_attachment_unavailable", "queued_message_delivery_indeterminate"].includes(error.message)
        ? error.message
        : "queued_message_failed";
    emit(session, "error", { message });
    setStatus(session, "failed");
    releaseRuntime(session);
  } finally {
    if (queuedDispatchClaims.get(session) === item.id) queuedDispatchClaims.delete(session);
  }
}

async function restoreQueuedReceipt(session, item) {
  if (!item.clientMessageId) return true;
  try {
    await actionReceiptStore.updateMessage({
      sessionId: session.id,
      clientMessageId: item.clientMessageId,
      state: "accepted"
    });
    return true;
  } catch {
    session.queueAdvanceBlocked = true;
    emit(session, "error", { message: "idempotency_store_unavailable" });
    setStatus(session, "failed");
    return false;
  }
}

async function ensureQueuedAttachmentsAvailable(item) {
  try {
    await Promise.all((item.attachments ?? []).map((attachment) => {
      if (typeof attachment.path !== "string") throw new Error("queued_attachment_unavailable");
      return access(attachment.path);
    }));
  } catch {
    throw new Error("queued_attachment_unavailable");
  }
}
function isInternalEvent(event) {
  return event === "process.stderr" || event === "codex.event";
}

function timestampFromSeconds(value) {
  return Number.isFinite(value) ? new Date(value * 1000).toISOString() : undefined;
}

function publicThreadStatus(status) {
  const type = status?.type ?? status;
  if (type === "active") return "interrupted";
  if (type === "systemError") return "failed";
  return ["idle", "notLoaded", "completed"].includes(type) ? "completed" : "idle";
}

function externalThreadStatus(status, events) {
  for (let index = events.length - 1; index >= 0; index -= 1) {
    if (events[index].event === "turn.completed") break;
    if (events[index].event === "turn.started") return "running";
  }
  return publicThreadStatus(status);
}

function safeProjectLabel(value) {
  if (typeof value !== "string") return undefined;
  const label = value.trim();
  if (!label || label.length > 120 || /[\\/:]/.test(label)) return undefined;
  return label;
}
