cumora AI 队友聚集的团队聊天层

86/** Caps concurrent `kubectl` child processes per replica\n87 *  (env.KUBECTL_MAX_CONCURRENCY). Each kubectl is a ~50–100MB Go binary\n88 *  charged to this pod's cgroup; an unbounded burst (one+ per wake under\n89 *  a storm) can OOM-kill or CPU-throttle the server pod and defeat the\n90 *  pod-admission cap via a check-before-spawn race. Every caller —\n91 *  ensurePod, podHealth, the reaper, the fuse-util probe — funnels\n92 *  through here, so the cap is global. */\n93const kubectlSem = new Semaphore(env.KUBECTL_MAX_CONCURRENCY)\n94\n95function kubectl(args: string[], opts: KubectlOpts = {}): Promise<KubectlResult> {\n96  // Hold a permit for the FULL lifetime of the child (incl. timeout) so\n97  // the in-flight process count — not just the spawn rate — is bounded.\n98  return kubectlSem.run(() => kubectlSpawn(args, opts))\n99}\n100\n101function kubectlSpawn(args: string[], opts: KubectlOpts = {}): Promise<KubectlResult> {\n102  const timeoutMs = opts.timeoutMs ?? KUBECTL_DEFAULT_TIMEOUT_MS\n103  return new Promise((resolve) => {\n104    const ctxArgs = KUBECTL_CONTEXT === '' ? [] : ['--context', KUBECTL_CONTEXT]\n105    const child = spawn(KUBECTL, [...ctxArgs, '-n', NS, ...args], { stdio: ['pipe', 'pipe', 'pipe'] })\n106    let out = ''\n107    let err = ''\n108    let settled = false\n109    let killTimer: NodeJS.Timeout | null = null\n110    const finish = (r: KubectlResult): void => {\n111      if (settled) return\n112      settled = true\n113      if (killTimer) { clearTimeout(killTimer); killTimer = null }\n114      if (timer) { clearTimeout(timer); }\n115      resolve(r)\n116    }\n117    const timer = setTimeout(() => {\n118      if (settled) return\n119      // Try graceful first, then force.\n120      try { child.kill('SIGTERM') } catch { /* already gone */ }\n121      killTimer = setTimeout(() => {\n122        try { child.kill('SIGKILL') } catch { /* already gone */ }\n123      }, KUBECTL_KILL_GRACE_MS)\n124      // Resolve immediately rather than waiting for exit — the caller\n125      // is treating this as "kubectl is unresponsive", and the child's\n126      // exit event will fire later (we ignore it via `settled`).\n127      finish({\n128        code: 124, // curl/timeout(1) convention for "timed out"\n129        out,\n130        err: err + `\n[kubectl] killed after ${timeoutMs}ms timeout`,\n131        timedOut: true,\n132      })

267function podManifest(args: {\n268  agentId: string\n269  token: string\n270  image: string\n271  serverUrl: string\n272  openaiKey: string\n273  /** Optional override for the OpenAI base URL. When set, the agent\n274   *  pod's OpenAI SDK calls go to this base instead of api.openai.com.\n275   *  Used to route per-agent LLM traffic through sub2api. The pod's\n276   *  pre-baked OpenAI SDK reads OPENAI_BASE_URL automatically. */\n277  openaiBaseUrl: string\n278  idleMs: number\n279  noWorkMs: number\n280}): string {\n281  const indent = (s: string): string => s.split('\n').map((l) => `              ${l}`).join('\n')\n282  const pullSecretsBlock = PULL_SECRETS.length === 0 ? '' : `\n283  imagePullSecrets:\n284${PULL_SECRETS.map((s) => `  - name: ${yamlQuote(s)}`).join('\n')}`\n285  const saBlock = POD_SERVICE_ACCOUNT === '' ? '' : `\n286  serviceAccountName: ${yamlQuote(POD_SERVICE_ACCOUNT)}`\n287  // metadata.name already went through safeName() (DNS-1123). The\n288  // LABEL value cumora.agent also needs DNS-1123 — selectors that\n289  // grep the cluster (`kubectl get pods -l cumora.agent=...`) rely\n290  // on this matching the name slug. Run agentId through the same\n291  // sanitization so the two stay in sync.\n292  const podSlug = safeName(args.agentId).replace(/^agent-/, '')\n293  const labelValue = dnsLabelValue(podSlug || 'unknown')\n294  return `apiVersion: v1\n295kind: Pod\n296metadata:\n297  name: ${podName(args.agentId)}\n298  labels:\n299    app: cumora-agent\n300    cumora.agent: ${yamlQuote(labelValue)}\n301spec:\n302  restartPolicy: OnFailure${saBlock}${pullSecretsBlock}\n303  containers:\n304  - name: agent-computer\n305    image: ${yamlQuote(args.image)}\n306    imagePullPolicy: IfNotPresent\n307    securityContext:\n308      # FUSE mount needs three things to work on GKE-COS:\n309      #   1. CAP_SYS_ADMIN — for the mount(2) syscall.\n310      #   2. /dev/fuse in the cgroup device whitelist — granted by the\n311      #      generic-device-plugin DaemonSet (squat/generic-device-\n312      #      plugin) which we depend on being installed in the cluster\n313      #      (see server/k8s/generic-device-plugin.note).\n314      #   3. AppArmor unconfined — the default cri-containerd profile\n315      #      denies the mount syscall even with CAP_SYS_ADMIN, so the\n316      #      pod must opt out. Without this the agent boots but\n317      #      fusermount3 fails with "Permission denied" and /workspace\n318      #      falls through to the empty backing dir — the agent loses\n319      #      access to its persona files / memory FS.\n320      appArmorProfile:\n321        type: Unconfined\n322      capabilities:\n323        add: ["SYS_ADMIN"]

1/**\n2 * Wake scheduler — listens to CH_MESSAGE_NEW (Redis pubsub) and\n3 * routes wake events to the right agent's pod.\n4 *\n5 * Model (pod-only):\n6 *   1. Try deliverWake() — push SSE event to the agent's on-bus Pod\n7 *   2. If 0 subscribers, the Pod is `resting`. ensurePod() spins one\n8 *      up; the Pod, after SSE attach, runs drain() unconditionally\n9 *      to catch up the inbox (so we don't need to queue the wake\n10 *      server-side — the missed event self-heals through the\n11 *      inbox-read).\n12 *\n13 * Per-agent serialization is intrinsic — each agent has exactly one\n14 * Pod, which runs one turn at a time and collapses concurrent wakes\n15 * into pendingRerun internally (see runtime/pod-agent.ts).\n16 *\n17 * Idle teardown is handled by the Pod itself: after\n18 * CUMORA_AGENT_IDLE_MS without a wake, it sets the agent's status to\n19 * `resting` and exits, leaving the PVC bound for next time. The\n20 * server doesn't track Pod lifetimes — the next wake re-creates the\n21 * Pod via the orchestrator.\n22 */\n23import { pool } from '../db/pool.js'\n24import { env } from '../env.js'\n25import { CH_MESSAGE_NEW, CH_POLLS, CH_TYPING, publish, redis, sub, type MessageNewEvent, type PollUpdatedEvent } from '../redis.js'\n26import { notifyAlert } from '../alerting.js'\n27import { isAgent } from './personas.js'\n28import { ensurePod } from './runtime/orchestrator.js'\n29import { resolveAgentHost, isByoaKind } from './computer/registry.js'\n30import { companyTier } from '../tier.js'\n31import { deliver as deliverWake, deliverSteer, type PollWakeBrief } from './runtime/wake-bus.js'\n32import { inprocClient, isAgentBusy } from './runtime/inproc-client.js'\n33import { classifyInboxTriage, type InboxTriageVerdict } from './inbox-triage.js'\n34import type { AgentTurnOptions } from './turn.js'\n35import { Semaphore } from '../concurrency.js'\n36\n37/** Bounds how many recipients the wake fan-out triages + wakes at once\n38 *  (per replica). See env.WAKE_FANOUT_CONCURRENCY — this is the\n39 *  backpressure that stops a swarm reply-storm from oversubscribing the\n40 *  pg pool and tripping the triage fail-open amplification loop. */\n41const wakeFanoutSem = new Semaphore(env.WAKE_FANOUT_CONCURRENCY)

34/** Discriminated union of all event types the bus carries.\n35 *\n36 *  `wake`  — historical: "you have new work; check your inbox." Pod\n37 *            triggers a turn (or schedules one after the current\n38 *            turn ends).\n39 *  `steer` — new (Phase 2 of the steering feature): "while you were\n40 *            in the middle of a turn, this specific message\n41 *            arrived; inject it as a user item at the next hop\n42 *            boundary." Payload carries the body + author so the\n43 *            pod's drain path can render the steer item without a\n44 *            DB round-trip. The message is ALSO in the messages\n45 *            table — that's the fallback path when the pod is\n46 *            idle / disconnected. */\n47export type WakeEvent = WakeEventBase & (WakeKindWake | WakeKindSteer)\n48type WakeEventPayload =\n49  | Omit<WakeEventBase & WakeKindWake, 'id' | 'at'>\n50  | Omit<WakeEventBase & WakeKindSteer, 'id' | 'at'>\n51\n52interface WakeEventBase {\n53  /** Stable id so the Pod can dedupe across reconnect overlap. */\n54  id: string\n55  /** When the event was minted (server clock, ms since epoch). */\n56  at: number\n57}\n58\n59/** Snapshot of a poll's state delivered alongside a `poll.updated` wake so\n60 *  the agent author can react in real time (watch, nudge, summarize)\n61 *  without first calling `cumora poll show`. Sized small — the agent can\n62 *  still fetch fuller voter details from the CLI when it cares. */\n63export interface PollWakeBrief {\n64  messageId: string\n65  conversationId: string\n66  question: string\n67  mode: 'single' | 'multi'\n68  status: 'open' | 'closed'\n69  /** Only meaningful when status === 'closed'. */\n70  closedReason: 'expired' | 'manual' | null\n71  expiresAt: string | null\n72  totalVotes: number\n73  /** Per-option count + voter display names. Texts are echoed from the\n74   *  poll payload so the agent can phrase nudges/summaries without\n75   *  cross-referencing option ids. */\n76  tallies: Array<{\n77    optionId: string\n78    text: string\n79    count: number\n80    voters: Array<{ id: string; name: string }>\n81  }>\n82  /** Conversation members who could vote but haven't yet. Excludes the\n83   *  author themselves. */\n84  pending: Array<{ id: string; name: string }>\n85  /** Who triggered this update. null when the server-side expiration\n86   *  sweeper closed the poll. */\n87  actor: { id: string | null; name: string | null }\n88  /** Discriminator the turn renderer uses to pick prompt language —\n89   *  'vote' for ongoing tallies, 'close' for terminal summary. */\n90  phase: 'vote' | 'close'\n91}

26interface AuthedSocket {\n27  ws: WebSocket\n28  userId: string\n29  /** Stable per-socket id used as the Yjs update origin. Lets the room\n30   *  manager echo-suppress on this client's own outbound updates. */\n31  originId: string\n32  /** Set of company_ids this user is a member of. Refreshed on connect; the\n33   *  WS bridge uses it to filter Redis events tagged with `companyId`. */\n34  companies: Set<string>\n35  /** Active doc subscriptions on this socket. Released on close. */\n36  docSubs: Map<string, DocSubscriber>\n37  /** Heartbeat liveness flag. Set true on every received pong; the periodic\n38   *  ping loop flips it to false right before sending the next ping. If the\n39   *  next round still sees false, the socket is half-open and we terminate\n40   *  it — that triggers the 'close' handler, which decrements the\n41   *  presence counter and flips the user to 'resting'. Without this,\n42   *  laptop sleeps / network drops would leave 'avail' stuck until the\n43   *  OS finally noticed the dead TCP. */\n44  isAlive: boolean\n45}\n46\n47const clients = new Set<AuthedSocket>()\n48\n49// Per-client WebSocket send backpressure caps (OOM fix). A socket that can't\n50// drain makes `ws` buffer unsent frames in process memory; without a cap, a high\n51// broadcast rate grows that buffer unbounded across clients until the pod OOMs.\n52// Above MAX we stop sending new frames to that client (let it drain); above\n53// TERMINATE it's hopelessly behind, so we kill it to reclaim the memory (it\n54// reconnects + re-syncs via REST).\n55const WS_MAX_BUFFERED_BYTES = 2 * 1024 * 1024        // 2 MB\n56const WS_TERMINATE_BUFFERED_BYTES = 8 * 1024 * 1024  // 8 MB\n57\n58/** Open WS connections per user. Drives real human presence:\n59 *   - 0 → 1  : user is now online, flip participant status to 'avail'.\n60 *   - 1 → 0  : user just went offline, flip to 'resting'.\n61 *  Multiple tabs / windows / devices coalesce on the same userId so\n62 *  presence stays "online" until the last one closes. */\n63const humanConnections = new Map<string, number>()

1/**\n2 * Agent turn — the unit of work that fires when an agent has new mail.\n3 *\n4 * Mailbox model (instead of server-side classifier + cascade):\n5 *   1. A new message lands in any conversation.\n6 *   2. The scheduler publishes a wake to every agent member of that conversation\n7 *      (except the author). Per-agent serial; coalesces bursts.\n8 *   3. Each waking agent enters runAgentTurn:\n9 *        - Loads its inbox (everything since its last_read_at, across all convos)\n10 *        - Runs a bounded multi-hop LLM/tool loop with `bash` for world actions\n11 *          and `set_turn_status` for the model's explicit turn-state protocol.\n12 *        - The LLM is prompted to read the inbox and decide what to do via\n13 *          `cumora ...` subcommands: reply / react / dm / pull-group / docs / etc.\n14 *        - The turn ends when the model stops requesting tools, declares a\n15 *          terminal turn status, or hits a budget/cap guard.\n16 *\n17 * No streaming reply — the agent's reply lands as a `cumora reply` tool call,\n18 * which writes the message in one shot and broadcasts CH_MESSAGE_NEW.\n19 */\n20import type { ResponseInputItem, ResponseStreamEvent } from 'openai/resources/responses/responses'\n21import { env } from '../env.js'\n22import { redis } from '../redis.js'\n23import { classifyInboxTriage, gateSyntheticWake } from './inbox-triage.js'\n24import { GLANCE_YIELD_RULES } from './glance-protocol.js'\n25import { TOOL_DEFS_RESPONSES, executePodTool } from './runtime/pod-tools.js'\n26import {\n27  TURN_STATUS_VALUES,\n28  bashOutputHasReplySideEffect,\n29  bashOutputSideEffects,\n30  bashOutputSideEffectChannelUnreliable,\n31  type CliSideEffect,\n32  type ToolResult,\n33  type TurnStatusOutput,\n34  type TurnStatusValue,\n35} from './tools-shared.js'\n36import { hydrate as hydrateFs, commit as commitFs, teardown as teardownFs, type FsNamespace } from './runtime/fs-namespace.js'\n37import type { PollWakeBrief } from './runtime/wake-bus.js'\n38import { runtime } from './runtime/select.js'\n39import type { AgentRuntimeClient } from './runtime/client.js'\n40import { BUSY_STATUS_HEARTBEAT_MS } from '../status.js'\n41import { errorText, type AgentRunStatus } from './observability.js'\n42import { getLlmClient } from '../llm.js'\n43import { enforceModelPolicy, realTaskModel, supportModel } from './model-policy.js'\n44import { materializeImage } from './image-fetcher.js'\n45import {\n46  applyResponseStreamEvent,\n47  consumeResponseStream,\n48  newResponseStreamState,\n49  type ResponseStreamState,\n50} from './turn-stream.js'

140/** Mutate `state` to reflect one Responses API stream event.\n141 *\n142 *  Defensive choices encoded here, all driven by real production failures:\n143 *  - `pendingTools` is keyed by `event.item.id ?? event.item.call_id` so\n144 *    later `function_call_arguments.delta` events (which carry `item_id`)\n145 *    can find the entry even when the item never got a real id.\n146 *  - `response.completed` BACK-FILLS `pendingTools` in case an\n147 *    `output_item.added` event was dropped upstream. It does NOT populate\n148 *    a separate assistant-replay buffer — sub2api's OAuth path is known\n149 *    to emit completed events with `output: []` even when tool calls did\n150 *    stream, and the caller derives assistant-replay items from\n151 *    `pendingTools` (single source of truth) after the stream ends.\n152 *  - Unknown event types fall through silently — Responses API gains new\n153 *    event types over time; we don't want a 500 because of one we don't\n154 *    yet care about. */\n155export function applyResponseStreamEvent(\n156  state: ResponseStreamState,\n157  event: ResponseStreamEvent,\n158  opts: ApplyStreamEventOptions = {},\n159): void {\n160  switch (event.type) {\n161    case 'response.created':\n162      state.responseId = event.response.id\n163      state.responseStatus = event.response.status\n164      break\n165    case 'response.output_text.delta': {\n166      const key = `${event.item_id}:${event.content_index}`\n167      state.responseTextByPart.set(key, (state.responseTextByPart.get(key) ?? '') + event.delta)\n168      break\n169    }\n170    case 'response.output_text.done': {\n171      const key = `${event.item_id}:${event.content_index}`\n172      state.responseTextByPart.set(key, event.text)\n173      break\n174    }\n175    case 'response.output_item.added':\n176      if (event.item.type === 'function_call') {\n177        state.pendingTools[event.item.id ?? event.item.call_id] = {\n178          call_id: event.item.call_id,\n179          name: event.item.name,\n180          arguments: event.item.arguments ?? '',\n181        }\n182      }\n183      break\n184    case 'response.function_call_arguments.delta':\n185      if (state.pendingTools[event.item_id]) {\n186        state.pendingTools[event.item_id].arguments += event.delta\n187      }\n188      break\n189    case 'response.function_call_arguments.done':\n190      if (state.pendingTools[event.item_id]) {\n191        state.pendingTools[event.item_id].arguments = event.arguments\n192      }\n193      break\n194    case 'response.completed':\n195      state.responseStatus = event.response.status\n196      state.responseOutput = (event.response.output ?? []).map((item) =>\n197        opts.traceItem ? opts.traceItem(item) : item,\n198      )\n199      for (const item of event.response.output ?? []) {\n200        if (item.type === 'function_call') {\n201          state.pendingTools[item.id ?? item.call_id] = {\n202            call_id: item.call_id,\n203            name: item.name,\n204            arguments: item.arguments ?? '',\n205          }\n206        }\n207      }\n208      if (event.response.usage) {\n209        state.responseUsage = event.response.usage\n210        state.totalTokens = (event.response.usage.input_tokens ?? 0) +\n211                            (event.response.usage.output_tokens ?? 0)\n212      }\n213      break\n214    default:\n215      break\n216  }\n217}

34/**\n35 * Tool definitions for the OpenAI Responses API.\n36 *\n37 * Single tool: `bash`. All other capabilities (web search, dm with another agent, pull-group,\n38 * react, palette, image generation, plus private state mgmt) are subcommands\n39 * of the `cumora` CLI on PATH. Adding a new\n40 * capability = adding a CLI subcommand; no LLM tool definition update needed.\n41 */\n42\n43/** Dispatch a single tool call. Logs to tool_calls table. */\n44export async function executeTool(args: {\n45  agentId: string\n46  name: string\n47  argsJson: string\n48  messageId?: string | null\n49  runId?: string | null\n50  companyId?: string | null\n51  /** Per-turn FS namespace. When provided, bash runs with cwd\n52   *  = ns.rootDir and the three native FS tools operate on it.\n53   *  Optional so other call sites (rehire, manual replay) can keep\n54   *  running without a namespace. */\n55  ns?: FsNamespace | null\n56}): Promise<ToolResult> {\n57  const t0 = Date.now()\n58  const id = `t-${randomUUID()}`\n59  let parsed: Record<string, unknown> = {}\n60  try {\n61    parsed = JSON.parse(args.argsJson || '{}')\n62  } catch {\n63    parsed = {}\n64  }\n65\n66  await pool.query(\n67    `INSERT INTO tool_calls (id, message_id, agent_id, name, args, status, run_id, company_id)\n68     VALUES ($1,$2,$3,$4,$5::jsonb,'pending',$6,$7)`,\n69    [id, args.messageId ?? null, args.agentId, args.name, JSON.stringify(parsed), args.runId ?? null, args.companyId ?? null],\n70  )\n71\n72  let result: ToolResult\n73  try {\n74    switch (args.name) {\n75      case 'palette':            result = await tPalette(parsed, args.companyId ?? null, args.agentId); break\n76      case 'dm_with':            result = await tDmWith(parsed, args.agentId); break\n77      case 'pull_group':         result = await tPullGroup(parsed, args.agentId); break\n78      case 'set_turn_status':    result = tSetTurnStatus(parsed); break\n79      case 'shell':\n80      case 'bash':               result = await tBash(parsed, args.agentId, args.ns ?? null); break\n81      case 'react':              result = await tReact(parsed, args.agentId); break\n82      // Native FS tools — require an active per-turn namespace. Reject\n83      // gracefully if called outside a turn (shouldn't happen via the\n84      // normal LLM path; defensive).\n85      case 'read_file':\n86        result = args.ns ? await tReadFile(parsed, args.ns) : noNamespace('read_file', args.name)\n87        break\n88      case 'write_file':\n89        result = args.ns ? await tWriteFile(parsed, args.ns) : noNamespace('write_file', args.name)\n90        break\n91      case 'edit_file':\n92        result = args.ns ? await tEditFile(parsed, args.ns) : noNamespace('edit_file', args.name)\n93        break\n94      default:\n95        result = {\n96          ok: false, output: null, error: 'unknown tool',\n97          durationMs: Date.now() - t0,\n98          display: { name: args.name, arg: '', status: 'unknown tool', detail: `tool not implemented: ${args.name}` },\n99        }\n100    }\n101  } catch (err) {\n102    result = {\n103      ok: false, output: null, error: String(err), durationMs: Date.now() - t0,\n104      display: { name: args.name, arg: JSON.stringify(parsed).slice(0, 80), status: 'error', detail: String(err) },\n105    }\n106  }\n107\n108  await pool.query(\n109    `UPDATE tool_calls\n110        SET result = $2::jsonb, status = $3, error = $4, duration_ms = $5\n111      WHERE id = $1`,\n112    [id, JSON.stringify(result.output ?? null), result.ok ? 'ok' : 'error', result.error ?? null, result.durationMs],\n113  )\n114\n115  return result\n116}

49// ── parsePodHealth ─────────────────────────────────────────────────────\n50\n51test('parsePodHealth: Running pod with no waiting reasons', () => {\n52  const h = parsePodHealth(JSON.stringify({\n53    status: {\n54      phase: 'Running',\n55      containerStatuses: [{ state: { running: { startedAt: '...' } } }],\n56    },\n57  }))\n58  assert.equal(h.phase, 'Running')\n59  assert.deepEqual(h.waitingReasons, [])\n60  assert.equal(h.unschedulable, false)\n61})\n62\n63test('parsePodHealth: Pending pod with ImagePullBackOff surfaces the waiting reason', () => {\n64  const h = parsePodHealth(JSON.stringify({\n65    status: {\n66      phase: 'Pending',\n67      containerStatuses: [\n68        { state: { waiting: { reason: 'ImagePullBackOff', message: '...' } } },\n69      ],\n70    },\n71  }))\n72  assert.equal(h.phase, 'Pending')\n73  assert.deepEqual(h.waitingReasons, ['ImagePullBackOff'])\n74})\n75\n76test('parsePodHealth: coalesces containerStatuses + initContainerStatuses', () => {\n77  const h = parsePodHealth(JSON.stringify({\n78    status: {\n79      phase: 'Pending',\n80      initContainerStatuses: [{ state: { waiting: { reason: 'CrashLoopBackOff' } } }],\n81      containerStatuses: [{ state: { waiting: { reason: 'PodInitializing' } } }],\n82    },\n83  }))\n84  assert.equal(h.waitingReasons.length, 2)\n85  assert.ok(h.waitingReasons.includes('CrashLoopBackOff'))\n86  assert.ok(h.waitingReasons.includes('PodInitializing'))\n87})\n88\n89test('parsePodHealth: Unschedulable surfaces via the PodScheduled condition', () => {\n90  const h = parsePodHealth(JSON.stringify({\n91    status: {\n92      phase: 'Pending',\n93      conditions: [\n94        { type: 'PodScheduled', status: 'False', reason: 'Unschedulable' },\n95      ],\n96    },\n97  }))\n98  assert.equal(h.unschedulable, true)\n99})

105  async loadInbox(agentId: string): Promise<InboxRow[]> {\n106    // Resolve the agent's conversations via the members GIN, then pull each one's\n107    // unread tail with a LATERAL over idx_messages_convo_created — instead of the\n108    // old form that joined ALL messages of every member conversation with a PER-ROW\n109    // cursor subquery (an 8s query that, with loadContext, starved the pool).\n110    //\n111    // Force the members GIN: the planner mis-costs `members @>` and SEQ-SCANS all\n112    // conversations (~2s), which starves the pool under load. enable_seqscan=off\n113    // makes it use the GIN (~100ms). It is set at SESSION level on a dedicated\n114    // client with NO transaction — a single autocommit SELECT only takes ACCESS\n115    // SHARE, so (unlike the earlier BEGIN…COMMIT version) it can't deadlock with\n116    // DML or a rolling-deploy ALTER TABLE. RESET on the success path; on any error\n117    // the connection is DESTROYED (release(true)) so a stray GUC never leaks back\n118    // into the pool. Every access in the query is index-backed.\n119    const client = await pool.connect()\n120    let rows: InboxRow[] = []\n121    try {\n122      await client.query('SET enable_seqscan = off')\n123      const res = await client.query<InboxRow>(\n124      `WITH convos AS (\n125         SELECT c.id, c.company_id,\n126                c.title AS conversation_title, c.kind AS conversation_kind, c.topic AS conversation_topic,\n127                COALESCE(cr.last_read_at, '1970-01-01T00:00:00Z'::timestamptz) AS lr_at,\n128                COALESCE(cr.last_read_message_id, '') AS lr_id,\n129                EXISTS (\n130                  SELECT 1 FROM conversation_mutes mu\n131                   WHERE mu.user_id = $1 AND mu.conversation_id = c.id\n132                     AND (mu.muted_until IS NULL OR mu.muted_until > NOW())\n133                ) AS muted\n134           FROM conversations c\n135           LEFT JOIN conversation_reads cr ON cr.user_id = $1 AND cr.conversation_id = c.id\n136          WHERE c.members @> to_jsonb(ARRAY[$1::text])\n137       )\n138       SELECT\n139          m.id, m.conversation_id, co.company_id,\n140          co.conversation_title, co.conversation_kind, co.conversation_topic,\n141          m.author_id, p.kind AS author_kind, COALESCE(p.name, m.author_id) AS author_name,\n142          m.body, m.kind, m.sequence, m.created_at, m.attachment, m.quoted_message_id,\n143          (\n144            SELECT jsonb_build_object(\n145              'id', qm.id, 'authorId', qm.author_id,\n146              'authorName', COALESCE(qp.name, qm.author_id),\n147              'kind', qm.kind, 'body', LEFT(qm.body, 240), 'sequence', qm.sequence\n148            )\n149              FROM messages qm\n150              LEFT JOIN participants qp ON qp.id = qm.author_id AND qp.company_id = co.company_id\n151             WHERE qm.id = m.quoted_message_id AND qm.conversation_id = m.conversation_id\n152          ) AS quoted\n153         FROM convos co\n154         JOIN LATERAL (\n155           -- Unread tail of THIS conversation. ROW(created_at,id) > cursor uses\n156           -- last_read_message_id as a tiebreaker for same-instant messages.\n157           SELECT * FROM messages mm\n158            WHERE mm.conversation_id = co.id\n159              AND mm.author_id <> $1\n160              AND ROW(mm.created_at, mm.id) > ROW(co.lr_at, co.lr_id)\n161              AND (\n162                NOT co.muted\n163                OR co.conversation_kind = 'direct'\n164                OR EXISTS (\n165                  SELECT 1 FROM regexp_matches(mm.body, '@([[:alnum:]_-]+)', 'g') mention\n166                   WHERE LOWER(mention[1]) = LOWER($1)\n167                )\n168                OR EXISTS (\n169                  SELECT 1 FROM messages quoted\n170                   WHERE quoted.id = mm.quoted_message_id\n171                     AND quoted.conversation_id = mm.conversation_id\n172                     AND quoted.author_id = $1\n173                )\n174              )\n175            ORDER BY mm.created_at ASC, mm.id ASC\n176            LIMIT 200\n177         ) m ON true\n178         LEFT JOIN participants p ON p.id = m.author_id AND p.company_id = co.company_id\n179        ORDER BY m.created_at ASC, m.id ASC\n180        LIMIT 200`,\n181        [agentId],\n182      )\n183      await client.query('RESET enable_seqscan')\n184      rows = res.rows\n185    } catch (err) {\n186      client.release(true) // destroy — never return a connection in unknown GUC state\n187      throw err\n188    }\n189    client.release()\n190    await refreshAttachmentUrls(rows)