2026-07-06 01:52:14 +00:00
13 changed files with 476 additions and 29 deletions
@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT tokens_in, tokens_out, credits FROM usage_events\n               WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "tokens_in",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "tokens_out",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 2,
+        "name": "credits",
+        "type_info": "Numeric"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "47c1cee8591250819d22e87058c6011bebb77eaf93c1cfa2535db42221d8ecf3"
+}
@@ -3,9 +3,10 @@
 //! it by construction.
 //!
 //! Configuration (environment):
-//! - `CLAWMATES_DATABASE__URL`   Postgres connection string (required)
-//! - `CLAWMATES_BROKER_SOCKET`   unix socket path (default /tmp/clawmates-broker.sock)
-//! - `CLAWMATES_BROKER_KEY_FILE` master key file; generated on first boot
+//! - `CLAWMATES_DATABASE__URL`     Postgres connection string (required)
+//! - `CLAWMATES_BROKER_SOCKET`     unix socket path (default /tmp/clawmates-broker.sock)
+//! - `CLAWMATES_BROKER_KEY_FILE`   master key file; generated on first boot
+//! - `CLAWMATES_BROKER_POOL_SIZE`  max Postgres connections (default 8)

 use std::path::PathBuf;
 use std::process::ExitCode;
@@ -45,7 +46,13 @@ async fn run() -> Result<(), String> {
    }
    let key = FileKey::load(&key_path).map_err(|e| format!("key load failed: {e}"))?;

-    let pool = cm_db::connect(&database_url, 5)
+    // Pool size defaults to 8: one broker connection per concurrent door
+    // execution before we start queueing. Tune via env when the fleet grows.
+    let pool_size: u32 = std::env::var("CLAWMATES_BROKER_POOL_SIZE")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(8);
+    let pool = cm_db::connect(&database_url, pool_size)
        .await
        .map_err(|e| format!("database connection failed: {e}"))?;

@@ -134,13 +134,38 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
        std::thread::sleep(Duration::from_secs(900));
    });

-    // Liveness: the server pings every 15s. If nothing inbound arrives for 40s
-    // the socket is dead — return so main() reconnects.
+    // Liveness has two independent failure modes to catch, both of which we
+    // hit on architect during the Jul 5 2026 outage:
+    //   (a) READ-side stall: server never sends anything (or the socket goes
+    //       half-open on read). Caught by last_rx / 40s idle window below.
+    //   (b) WRITE-side stall: peer's TCP stack has died but our OS buffer is
+    //       still soaking heartbeat writes. write.send().await blocks
+    //       indefinitely inside the select! branch — tokio::select doesn't
+    //       preempt a running future, so the whole loop freezes; idle_tick
+    //       never gets to fire. Wrapping the send in a timeout is the fix.
+    //
+    // WRITE_DEADLINE is short enough (10s) that a stuck send is caught before
+    // it can outlast the 40s read-idle threshold and leave the daemon spinning
+    // silently for hours (which is what happened pre-patch).
+    const WRITE_DEADLINE: Duration = Duration::from_secs(10);
    let mut idle_tick = tokio::time::interval(Duration::from_secs(5));
    let mut last_rx = std::time::Instant::now();
    loop {
        tokio::select! {
-            Some(frame) = out_rx.recv() => { write.send(Message::Text(frame.into())).await?; }
+            Some(frame) = out_rx.recv() => {
+                match tokio::time::timeout(
+                    WRITE_DEADLINE,
+                    write.send(Message::Text(frame.into())),
+                ).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(e)) => return Err(e.into()),
+                    Err(_) => {
+                        eprintln!("write.send timeout > {}s — socket dead, reconnecting",
+                            WRITE_DEADLINE.as_secs());
+                        return Ok(());
+                    }
+                }
+            }
            _ = idle_tick.tick() => {
                if last_rx.elapsed() > Duration::from_secs(40) {
                    return Ok(());
@@ -158,7 +183,16 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
                        let text = t.to_string();
                        tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; });
                    }
-                    Some(Ok(Message::Ping(p))) => write.send(Message::Pong(p)).await?,
+                    Some(Ok(Message::Ping(p))) => {
+                        match tokio::time::timeout(WRITE_DEADLINE, write.send(Message::Pong(p))).await {
+                            Ok(Ok(())) => {}
+                            Ok(Err(e)) => return Err(e.into()),
+                            Err(_) => {
+                                eprintln!("pong write timeout — socket dead, reconnecting");
+                                return Ok(());
+                            }
+                        }
+                    }
                    Some(Ok(Message::Close(_))) | None => return Ok(()),
                    Some(Err(e)) => return Err(e.into()),
                    _ => {}
@@ -60,8 +60,10 @@ struct NodeConn {
    tx: mpsc::UnboundedSender<String>,
    pending: Mutex<HashMap<u64, oneshot::Sender<ExecOutput>>>,
    /// Live terminal sessions: sid → byte sink for the browser bridge (WS-relay
-    /// PTY output).
-    pty_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<Vec<u8>>>>,
+    /// PTY output). Bounded: a runaway PTY (say `cat /var/log/huge`) with a
+    /// stalled browser must not accumulate megabytes here. On overflow the
+    /// session is closed instead of holding output indefinitely.
+    pty_sinks: Mutex<HashMap<u64, mpsc::Sender<Vec<u8>>>>,
    /// WebRTC signaling: sid → text sink delivering the daemon's answer/ICE to
    /// the browser bridge.
    signal_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<String>>>,
@@ -187,12 +189,13 @@ impl NodeHub {
        id: NodeId,
    ) -> Option<(
        u64,
-        mpsc::UnboundedReceiver<Vec<u8>>,
+        mpsc::Receiver<Vec<u8>>,
        mpsc::UnboundedReceiver<String>,
    )> {
        let conn = self.get(id).await?;
        let sid = conn.next_id.fetch_add(1, Ordering::Relaxed);
-        let (ptx, prx) = mpsc::unbounded_channel();
+        // 256 × ~4KB PTY frames = ~1 MB per stalled session before we close it.
+        let (ptx, prx) = mpsc::channel(256);
        let (stx, srx) = mpsc::unbounded_channel();
        conn.pty_sinks.lock().await.insert(sid, ptx);
        conn.signal_sinks.lock().await.insert(sid, stx);
@@ -462,7 +465,21 @@ pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socke
                        if let Ok(bytes) = B64.decode(&data) {
                            let sink = conn.pty_sinks.lock().await.get(&sid).cloned();
                            if let Some(s) = sink {
-                                let _ = s.send(bytes);
+                                // try_send so a stalled browser can't grow the
+                                // per-session buffer without bound. On Full, the
+                                // session is torn down: drop both sinks and tell
+                                // the node to close its side, preventing an
+                                // orphan PTY.
+                                if let Err(err) = s.try_send(bytes) {
+                                    if matches!(err, mpsc::error::TrySendError::Full(_)) {
+                                        conn.pty_sinks.lock().await.remove(&sid);
+                                        conn.signal_sinks.lock().await.remove(&sid);
+                                        let _ = conn.tx.send(
+                                            json!({ "t": "pty_close", "sid": sid })
+                                                .to_string(),
+                                        );
+                                    }
+                                }
                            }
                        }
                    }
@@ -13,6 +13,10 @@ use crate::{ApiError, AppState, Authed};
 pub struct Quota {
    pub max_agents: i64,
    pub max_live_containers: i64,
+    /// Ceiling on `queued` + `running` topology runs at once. Prevents one
+    /// workspace flooding the shared queue (a single team run also spawns a
+    /// tier-tree of children, so the practical cap grows with the topology).
+    pub max_active_runs: i64,
 }

 /// Per-plan limits. Unknown plans fall back to the free tier.
@@ -21,14 +25,17 @@ pub fn plan_quota(plan: &str) -> Quota {
        "team" => Quota {
            max_agents: 50,
            max_live_containers: 50,
+            max_active_runs: 100,
        },
        "pro" => Quota {
            max_agents: 20,
            max_live_containers: 20,
+            max_active_runs: 25,
        },
        _ => Quota {
            max_agents: 3,
            max_live_containers: 3,
+            max_active_runs: 5,
        },
    }
 }
@@ -61,6 +68,23 @@ pub async fn enforce_new_agent(
    Ok(())
 }

+/// Reject enqueueing another topology run if the workspace is at its plan cap.
+pub async fn enforce_new_run(
+    state: &AppState,
+    workspace_id: WorkspaceId,
+) -> Result<(), ApiError> {
+    let plan = plan_of(state, workspace_id).await?;
+    let quota = plan_quota(&plan);
+    let used = cm_db::repo::topology_runs::count_active(&state.pool, workspace_id).await?;
+    if used >= quota.max_active_runs {
+        return Err(ApiError::Quota(format!(
+            "active-run limit reached ({} on the {plan} plan) — wait for a run to finish or upgrade",
+            quota.max_active_runs
+        )));
+    }
+    Ok(())
+}
+
 /// Reject spinning up another container if the workspace is at its plan cap.
 pub async fn enforce_new_container(
    state: &AppState,
@@ -86,6 +110,8 @@ pub struct QuotaUsage {
    max_agents: i64,
    containers_used: i64,
    max_live_containers: i64,
+    active_runs: i64,
+    max_active_runs: i64,
 }

 /// `GET /api/quota` — the caller's workspace usage + limits (for the UI).
@@ -98,11 +124,15 @@ pub async fn get_quota(
    let agents_used = cm_db::repo::agents::count_active(&state.pool, user.workspace_id).await?;
    let containers_used =
        cm_db::repo::agent_containers::count_for_workspace(&state.pool, user.workspace_id).await?;
+    let active_runs =
+        cm_db::repo::topology_runs::count_active(&state.pool, user.workspace_id).await?;
    Ok(Json(QuotaUsage {
        plan,
        agents_used,
        max_agents: quota.max_agents,
        containers_used,
        max_live_containers: quota.max_live_containers,
+        active_runs,
+        max_active_runs: quota.max_active_runs,
    }))
 }
@@ -266,6 +266,7 @@ pub async fn run_company(
    Json(body): Json<RunCompanyRequest>,
 ) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
    let company = cm_db::repo::companies::get(&state.pool, id, user.workspace_id).await?;
+    crate::quota::enforce_new_run(&state, user.workspace_id).await?;
    let run_id = Uuid::now_v7();
    cm_db::repo::topology_runs::enqueue_run_tier(
        &state.pool,
@@ -209,6 +209,7 @@ pub async fn run_org(
    Json(body): Json<RunOrgRequest>,
 ) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
    let org = cm_db::repo::orgs::get(&state.pool, id, user.workspace_id).await?;
+    crate::quota::enforce_new_run(&state, user.workspace_id).await?;
    let run_id = Uuid::now_v7();
    cm_db::repo::topology_runs::enqueue_run_tier(
        &state.pool,
@@ -371,6 +371,7 @@ pub async fn run_team(
    Json(body): Json<RunTeamRequest>,
 ) -> Result<(StatusCode, Json<RunAccepted>), ApiError> {
    let team = cm_db::repo::teams::get_team(&state.pool, id, user.workspace_id).await?;
+    crate::quota::enforce_new_run(&state, user.workspace_id).await?;
    let run_id = Uuid::now_v7();
    cm_db::repo::topology_runs::enqueue_run(
        &state.pool,
@@ -99,6 +99,12 @@ pub async fn trigger_hook(
        .filter(|t| !t.trim().is_empty())
        .or_else(|| (!default_task.trim().is_empty()).then_some(default_task))
        .unwrap_or_else(|| "webhook trigger".to_string());
+    // Webhooks are unauthenticated public endpoints — the enforce_new_run
+    // guard is what stops a leaked token from being weaponized into a queue
+    // flood. 429 (not 402) so external callers can back off.
+    if crate::quota::enforce_new_run(&state, ws).await.is_err() {
+        return StatusCode::TOO_MANY_REQUESTS;
+    }
    let run_id = Uuid::now_v7();
    if cm_db::repo::topology_runs::enqueue_run(&state.pool, run_id, ws, &task, &team.graph)
        .await
@@ -83,7 +83,7 @@ const railIcon: Record<Tier, React.ReactNode> = {
  infra: (<svg width="20" height="20" viewBox="0 0 20 20"><rect x="3.5" y="4" width="13" height="5" rx="1.4" stroke="currentColor" strokeWidth="1.4" fill="none" /><rect x="3.5" y="11" width="13" height="5" rx="1.4" stroke="currentColor" strokeWidth="1.4" fill="none" /><circle cx="6.4" cy="6.5" r="1" fill="currentColor" /><circle cx="6.4" cy="13.5" r="1" fill="currentColor" /></svg>),
 };
 const TIER_TABS: { key: Tier; label: string }[] = [
-  { key: "world", label: "WORLD" }, { key: "claw", label: "AGENT" }, { key: "infra", label: "INFRA" },
+  { key: "world", label: "VIZ" }, { key: "claw", label: "AGENT" }, { key: "infra", label: "INFRA" },
 ];
 const companyIcon = (size: number) => (
  <svg width={size} height={size} viewBox="0 0 20 20"><rect x="3.5" y="5" width="6" height="11" rx="1" stroke="currentColor" strokeWidth="1.4" fill="none" /><rect x="10.5" y="2.5" width="6" height="13.5" rx="1" stroke="currentColor" strokeWidth="1.4" fill="none" /></svg>
@@ -452,6 +452,26 @@ export function Dashboard({ user, orgs, claws }: { user?: { display_name?: strin
  const allAgents = orgs.flatMap((o) => o.companies.flatMap((c) => c.teams.flatMap((t) => t.agents)));
  // World: the full expandable org→company→team→agent forest. Agents page: a flat list.
  const worldRoots: TreeItem[] = orgs.map(orgNode);
+  // World canvas: when a node is selected, prune to just the branch that
+  // contains it — sibling orgs/companies/teams disappear so the visualization
+  // frames the selection alone. The left tree keeps the full forest.
+  const narrowRoots = (roots: TreeItem[], sel: string | null): TreeItem[] => {
+    if (!sel) return roots;
+    const prune = (n: TreeItem): TreeItem | null => {
+      if (n.id === sel) return n;
+      for (const c of n.children ?? []) {
+        const found = prune(c);
+        if (found) return { ...n, children: [found] };
+      }
+      return null;
+    };
+    for (const r of roots) {
+      const found = prune(r);
+      if (found) return [found];
+    }
+    return roots;
+  };
+  const worldCanvasRoots: TreeItem[] = narrowRoots(worldRoots, worldSel);
  const treeRoots: TreeItem[] = isClaw ? allAgents.map(clawNode) : worldRoots;
  const treeActiveId = isClaw ? agentId : worldSel;
  const treeAutoExpand: string[] = [];
@@ -496,7 +516,7 @@ export function Dashboard({ user, orgs, claws }: { user?: { display_name?: strin
        </Link>
        <div style={{ width: 1, height: 22, background: "rgba(255,255,255,.08)" }} />
        <div style={{ display: "flex", alignItems: "center", gap: 6, fontFamily: mono, fontSize: 12 }}>
-          <span style={crumbStyle(isWorld)} onClick={() => setTier("world")}>Large World</span>
+          <span style={crumbStyle(isWorld)} onClick={() => setTier("world")}>Visualizations</span>
          <span style={{ color: "#3a3a40" }}>/</span>
          <span style={crumbStyle(isClaw)} onClick={() => setTier("claw")}>Agents</span>
          <span style={{ color: "#3a3a40" }}>/</span>
@@ -540,7 +560,7 @@ export function Dashboard({ user, orgs, claws }: { user?: { display_name?: strin
            <div style={{ padding: "16px 16px 12px", borderBottom: "1px solid rgba(255,255,255,.06)", display: "flex", alignItems: "flex-start", gap: 8 }}>
              <div style={{ flex: 1, minWidth: 0 }}>
                <div style={{ fontFamily: mono, fontSize: 10, letterSpacing: ".12em", color: "#5a5a62", marginBottom: 6 }}>{orgs.length} ORG{orgs.length === 1 ? "" : "S"} · {allAgents.length} AGENTS</div>
-                <div style={{ fontSize: 18, fontWeight: 700, color: "#f3f3f5", letterSpacing: "-.01em" }}>Large World</div>
+                <div style={{ fontSize: 18, fontWeight: 700, color: "#f3f3f5", letterSpacing: "-.01em" }}>Visualizations</div>
              </div>
              <button type="button" onClick={() => { setSelectMode((v) => { if (v) setSelectedAgents(new Set()); return !v; }); }} title="Select to manage" aria-label="Select to manage" style={{ flex: "none", width: 34, height: 34, borderRadius: 9, border: `1px solid ${selectMode ? "rgba(255,111,97,.5)" : "rgba(255,255,255,.12)"}`, background: selectMode ? "rgba(255,111,97,.12)" : "transparent", color: selectMode ? "#ff6f61" : "#9a9aa2", cursor: "pointer", display: "inline-flex", alignItems: "center", justifyContent: "center" }}><Wrench aria-hidden size={16} /></button>
            </div>
@@ -589,7 +609,7 @@ export function Dashboard({ user, orgs, claws }: { user?: { display_name?: strin
            <>
              {/* Graph stage — condensed by the right slide-out's width. */}
              <div style={{ position: "absolute", top: 0, bottom: 0, left: 0, right: "var(--world-width, 0px)", background: "radial-gradient(120% 90% at 55% 38%, #0e0e13 0%, #08080a 70%)", transition: "right var(--duration-normal) var(--ease-app)" }}>
-                <WorldCanvas roots={worldRoots} expanded={expanded} onToggleExpand={toggleExpand} selectedId={worldSel} onSelect={onWorldSelect} onOpenRuns={() => setRunsOpen(true)} />
+                <WorldCanvas roots={worldCanvasRoots} expanded={expanded} onToggleExpand={toggleExpand} selectedId={worldSel} onSelect={onWorldSelect} onOpenRuns={() => setRunsOpen(true)} />
                {/* Open the slide-out (top-right) when it's closed. */}
                {!worldPanelOpen ? (
                  <button type="button" aria-label="Open panel" title="Panel" onClick={() => setWorldPanelOpen(true)} style={{ position: "absolute", top: 14, right: 16, zIndex: 50, width: 38, height: 38, borderRadius: "50%", border: "1px solid rgba(255,111,97,.4)", background: "rgba(255,111,97,.08)", color: "#ff6f61", cursor: "pointer", display: "flex", alignItems: "center", justifyContent: "center" }}><PanelRight aria-hidden size={19} /></button>
@@ -16,6 +16,33 @@ import {
  type UiMessage,
 } from "./transcript";

+// Auto-reconnect budget for transport failures (Cloudflare/Traefik culling an
+// idle SSE stream, transient network flap). The reducer dedupes replayed
+// events by seq, so re-attaching with resumeFrom is safe even if the server
+// replays events we already saw.
+const RECONNECT_MAX_ATTEMPTS = 5;
+const RECONNECT_BASE_MS = 500;
+const RECONNECT_MAX_MS = 8000;
+
+type StreamOutcome = "clean" | "aborted" | "error";
+
+// Sleep that resolves early if the abort signal fires. Returns true if the
+// wait was cut short by an abort so callers can bail out of retry loops.
+function sleepUnlessAborted(ms: number, signal: AbortSignal): Promise<boolean> {
+  if (signal.aborted) return Promise.resolve(true);
+  return new Promise((resolve) => {
+    const t = setTimeout(() => {
+      signal.removeEventListener("abort", onAbort);
+      resolve(false);
+    }, ms);
+    const onAbort = () => {
+      clearTimeout(t);
+      resolve(true);
+    };
+    signal.addEventListener("abort", onAbort, { once: true });
+  });
+}
+
 export interface ChatHandle {
  state: TranscriptState;
  send: (text: string) => Promise<void>;
@@ -43,14 +70,14 @@ export function useChat(
  const lastSeqRef = useRef(0);
  lastSeqRef.current = state.lastSeq;

-  const readStream = useCallback(
+  // One pass through the SSE stream. Never mutates abortRef — the retry loop
+  // in readStream owns lifecycle so it can share a controller across attempts.
+  const openStream = useCallback(
    async (
      body: Record<string, unknown>,
+      controller: AbortController,
      emit: (action: TranscriptAction) => void,
-    ) => {
-      abortRef.current?.abort();
-      const controller = new AbortController();
-      abortRef.current = controller;
+    ): Promise<StreamOutcome> => {
      try {
        const response = await fetch(
          `/api/gateway?clawId=${encodeURIComponent(clawId)}`,
@@ -61,12 +88,17 @@ export function useChat(
            signal: controller.signal,
          },
        );
-        if (!response.ok || !response.body) {
+        // 4xx from the gateway is definitive (auth, not-found, bad request).
+        // Retrying won't help — surface it to the reducer immediately.
+        if (response.status >= 400 && response.status < 500) {
          emit({
            kind: "transport_error",
            message: `gateway returned ${response.status}`,
          });
-          return;
+          return "clean";
+        }
+        if (!response.ok || !response.body) {
+          return "error";
        }
        const parser = createSseParser();
        const reader = response.body.getReader();
@@ -80,18 +112,55 @@ export function useChat(
            });
          }
          if (done) {
-            break;
+            return "clean";
          }
        }
-      } catch (error) {
-        if (!controller.signal.aborted) {
-          emit({ kind: "transport_error", message: String(error) });
-        }
+      } catch (_error) {
+        return controller.signal.aborted ? "aborted" : "error";
      }
    },
    [clawId, sessionKey],
  );

+  const readStream = useCallback(
+    async (
+      body: Record<string, unknown>,
+      emit: (action: TranscriptAction) => void,
+    ) => {
+      abortRef.current?.abort();
+      const controller = new AbortController();
+      abortRef.current = controller;
+      let attempt = 0;
+      let currentBody = body;
+      for (;;) {
+        const outcome = await openStream(currentBody, controller, emit);
+        if (outcome !== "error") {
+          return;
+        }
+        attempt += 1;
+        if (attempt > RECONNECT_MAX_ATTEMPTS) {
+          emit({
+            kind: "transport_error",
+            message: `stream failed after ${RECONNECT_MAX_ATTEMPTS} reconnect attempts`,
+          });
+          return;
+        }
+        const delay = Math.min(
+          RECONNECT_MAX_MS,
+          RECONNECT_BASE_MS * 2 ** (attempt - 1),
+        );
+        const aborted = await sleepUnlessAborted(delay, controller.signal);
+        if (aborted) {
+          return;
+        }
+        // Every retry resumes from the last event we saw — the reducer's
+        // seq dedup makes overlap harmless.
+        currentBody = { resumeFrom: lastSeqRef.current };
+      }
+    },
+    [openStream],
+  );
+
  const send = useCallback(
    async (text: string) => {
      dispatch({ kind: "send", text });
@@ -0,0 +1,201 @@
+-- Backfill ON DELETE clauses on FKs from 0001-0006 and 0026. The pattern
+-- (CASCADE for tenant-scoped children, SET NULL for historical references)
+-- was learned after v1 shipped and applied consistently from 0010 onward;
+-- this migration retrofits the earlier tables so DELETEs succeed instead of
+-- either failing with a FK violation or leaving orphans.
+--
+-- Two FKs stay as NO ACTION deliberately:
+--   * audit_log.workspace_id — audit is append-only and must outlive any
+--     workspace delete (the trigger blocks UPDATE/DELETE on audit rows).
+--   * thread_messages.from_agent — history stays attributable via the
+--     agents.deleted_at soft-delete already in schema.
+--
+-- Two FKs stay NOT NULL and become explicit RESTRICT rather than SET NULL:
+--   * agents.managed_by — the Rust domain type is `UserId` (not Option).
+--     Making it nullable would ripple through cm-domain/cm-db/cm-api/tests.
+--     RESTRICT preserves current behavior (delete blocked) but documents it.
+--   * installed_skills.installed_by — same reasoning.
+-- To hard-delete a user in either case, reassign or hard-delete the
+-- dependent rows first — which matches real org-ownership semantics.
+
+-- ---- 0001_init: workspaces cascade -----------------------------------------
+ALTER TABLE users
+    DROP CONSTRAINT IF EXISTS users_workspace_id_fkey,
+    ADD  CONSTRAINT users_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE agents
+    DROP CONSTRAINT IF EXISTS agents_workspace_id_fkey,
+    ADD  CONSTRAINT agents_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE agents
+    DROP CONSTRAINT IF EXISTS agents_managed_by_fkey,
+    ADD  CONSTRAINT agents_managed_by_fkey
+         FOREIGN KEY (managed_by) REFERENCES users (id) ON DELETE RESTRICT;
+
+ALTER TABLE sessions
+    DROP CONSTRAINT IF EXISTS sessions_agent_id_fkey,
+    ADD  CONSTRAINT sessions_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS sessions_workspace_id_fkey,
+    ADD  CONSTRAINT sessions_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE messages
+    DROP CONSTRAINT IF EXISTS messages_session_id_fkey,
+    ADD  CONSTRAINT messages_session_id_fkey
+         FOREIGN KEY (session_id) REFERENCES sessions (id) ON DELETE CASCADE;
+
+ALTER TABLE steps
+    DROP CONSTRAINT IF EXISTS steps_message_id_fkey,
+    ADD  CONSTRAINT steps_message_id_fkey
+         FOREIGN KEY (message_id) REFERENCES messages (id) ON DELETE CASCADE;
+
+ALTER TABLE agent_runs
+    DROP CONSTRAINT IF EXISTS agent_runs_session_id_fkey,
+    ADD  CONSTRAINT agent_runs_session_id_fkey
+         FOREIGN KEY (session_id) REFERENCES sessions (id) ON DELETE CASCADE;
+
+ALTER TABLE approvals
+    DROP CONSTRAINT IF EXISTS approvals_workspace_id_fkey,
+    ADD  CONSTRAINT approvals_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS approvals_run_id_fkey,
+    ADD  CONSTRAINT approvals_run_id_fkey
+         FOREIGN KEY (run_id) REFERENCES agent_runs (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS approvals_requested_by_agent_fkey,
+    ADD  CONSTRAINT approvals_requested_by_agent_fkey
+         FOREIGN KEY (requested_by_agent) REFERENCES agents (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS approvals_decided_by_fkey,
+    ADD  CONSTRAINT approvals_decided_by_fkey
+         FOREIGN KEY (decided_by) REFERENCES users (id) ON DELETE SET NULL;
+
+-- Single-use execution grants: deleting the approval should drop the grant.
+ALTER TABLE execution_grants
+    DROP CONSTRAINT IF EXISTS execution_grants_approval_id_fkey,
+    ADD  CONSTRAINT execution_grants_approval_id_fkey
+         FOREIGN KEY (approval_id) REFERENCES approvals (id) ON DELETE CASCADE;
+
+-- audit_log.workspace_id — intentionally left as NO ACTION.
+
+ALTER TABLE skills
+    DROP CONSTRAINT IF EXISTS skills_workspace_id_fkey,
+    ADD  CONSTRAINT skills_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE installed_skills
+    DROP CONSTRAINT IF EXISTS installed_skills_skill_id_fkey,
+    ADD  CONSTRAINT installed_skills_skill_id_fkey
+         FOREIGN KEY (skill_id) REFERENCES skills (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS installed_skills_installed_by_fkey,
+    ADD  CONSTRAINT installed_skills_installed_by_fkey
+         FOREIGN KEY (installed_by) REFERENCES users (id) ON DELETE RESTRICT;
+
+ALTER TABLE secrets
+    DROP CONSTRAINT IF EXISTS secrets_workspace_id_fkey,
+    ADD  CONSTRAINT secrets_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE app_connections
+    DROP CONSTRAINT IF EXISTS app_connections_workspace_id_fkey,
+    ADD  CONSTRAINT app_connections_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS app_connections_agent_id_fkey,
+    ADD  CONSTRAINT app_connections_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS app_connections_secret_ref_fkey,
+    ADD  CONSTRAINT app_connections_secret_ref_fkey
+         FOREIGN KEY (secret_ref) REFERENCES secrets (id) ON DELETE SET NULL;
+
+ALTER TABLE threads
+    DROP CONSTRAINT IF EXISTS threads_workspace_id_fkey,
+    ADD  CONSTRAINT threads_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE thread_participants
+    DROP CONSTRAINT IF EXISTS thread_participants_agent_id_fkey,
+    ADD  CONSTRAINT thread_participants_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE;
+
+-- thread_messages.thread_id → CASCADE; from_agent stays NO ACTION.
+ALTER TABLE thread_messages
+    DROP CONSTRAINT IF EXISTS thread_messages_thread_id_fkey,
+    ADD  CONSTRAINT thread_messages_thread_id_fkey
+         FOREIGN KEY (thread_id) REFERENCES threads (id) ON DELETE CASCADE;
+
+ALTER TABLE file_nodes
+    DROP CONSTRAINT IF EXISTS file_nodes_workspace_id_fkey,
+    ADD  CONSTRAINT file_nodes_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS file_nodes_agent_id_fkey,
+    ADD  CONSTRAINT file_nodes_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE;
+
+ALTER TABLE credit_lots
+    DROP CONSTRAINT IF EXISTS credit_lots_workspace_id_fkey,
+    ADD  CONSTRAINT credit_lots_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE;
+
+ALTER TABLE usage_events
+    DROP CONSTRAINT IF EXISTS usage_events_workspace_id_fkey,
+    ADD  CONSTRAINT usage_events_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS usage_events_agent_id_fkey,
+    ADD  CONSTRAINT usage_events_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE SET NULL,
+    DROP CONSTRAINT IF EXISTS usage_events_run_id_fkey,
+    ADD  CONSTRAINT usage_events_run_id_fkey
+         FOREIGN KEY (run_id) REFERENCES agent_runs (id) ON DELETE SET NULL;
+
+-- ---- 0002_auth_sessions ---------------------------------------------------
+ALTER TABLE auth_sessions
+    DROP CONSTRAINT IF EXISTS auth_sessions_user_id_fkey,
+    ADD  CONSTRAINT auth_sessions_user_id_fkey
+         FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE;
+
+-- ---- 0003_run_events ------------------------------------------------------
+ALTER TABLE run_events
+    DROP CONSTRAINT IF EXISTS run_events_run_id_fkey,
+    ADD  CONSTRAINT run_events_run_id_fkey
+         FOREIGN KEY (run_id) REFERENCES agent_runs (id) ON DELETE CASCADE;
+
+-- ---- 0004_outbox ----------------------------------------------------------
+ALTER TABLE outbox
+    DROP CONSTRAINT IF EXISTS outbox_workspace_id_fkey,
+    ADD  CONSTRAINT outbox_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS outbox_agent_id_fkey,
+    ADD  CONSTRAINT outbox_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE;
+
+-- ---- 0005_oauth_states ----------------------------------------------------
+ALTER TABLE oauth_states
+    DROP CONSTRAINT IF EXISTS oauth_states_workspace_id_fkey,
+    ADD  CONSTRAINT oauth_states_workspace_id_fkey
+         FOREIGN KEY (workspace_id) REFERENCES workspaces (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS oauth_states_user_id_fkey,
+    ADD  CONSTRAINT oauth_states_user_id_fkey
+         FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE,
+    DROP CONSTRAINT IF EXISTS oauth_states_agent_id_fkey,
+    ADD  CONSTRAINT oauth_states_agent_id_fkey
+         FOREIGN KEY (agent_id) REFERENCES agents (id) ON DELETE CASCADE;
+
+-- ---- 0006_promo_codes -----------------------------------------------------
+ALTER TABLE promo_codes
+    DROP CONSTRAINT IF EXISTS promo_codes_redeemed_by_fkey,
+    ADD  CONSTRAINT promo_codes_redeemed_by_fkey
+         FOREIGN KEY (redeemed_by) REFERENCES workspaces (id) ON DELETE SET NULL;
+
+-- ---- 0026_group_rooms -----------------------------------------------------
+-- The threads.created_by / thread_participants.added_by columns were added
+-- nullable by 0026, so SET NULL is legal.
+ALTER TABLE threads
+    DROP CONSTRAINT IF EXISTS threads_created_by_fkey,
+    ADD  CONSTRAINT threads_created_by_fkey
+         FOREIGN KEY (created_by) REFERENCES agents (id) ON DELETE SET NULL;
+
+ALTER TABLE thread_participants
+    DROP CONSTRAINT IF EXISTS thread_participants_added_by_fkey,
+    ADD  CONSTRAINT thread_participants_added_by_fkey
+         FOREIGN KEY (added_by) REFERENCES agents (id) ON DELETE SET NULL;
@@ -0,0 +1,26 @@
+-- Targeted indexes for hot queries whose current indexes don't match the
+-- filter. Each one was chosen after reading the actual sqlx call sites; we
+-- didn't add "just in case" indexes on tables whose only access pattern is
+-- a PK lookup.
+
+-- Outbox drainer: `WHERE status = 'queued' ORDER BY created_at LIMIT $1`
+-- (crates/cm-db/src/repo/outbox.rs). The existing outbox_workspace_idx is
+-- `(workspace_id, created_at DESC)` — no help for the workspace-agnostic
+-- drainer that pops the oldest queued row. Partial index keeps it tiny.
+CREATE INDEX IF NOT EXISTS outbox_queued_idx
+    ON outbox (created_at)
+    WHERE status = 'queued';
+
+-- Audit log rate limiting: `WHERE workspace_id = $1 AND event_type = $2
+-- AND created_at > now() - interval '1 hour'` fires on every door tool
+-- call (cm-api/src/mcp_door.rs) and every A2A invocation. Descending
+-- created_at because count()s scan the recent tail.
+CREATE INDEX IF NOT EXISTS audit_log_workspace_event_idx
+    ON audit_log (workspace_id, event_type, created_at DESC);
+
+-- Node rules eval loop: `SELECT ... FROM node_rules WHERE enabled`
+-- (crates/cm-db/src/repo/node_rules.rs:90) runs periodically. Partial
+-- index avoids indexing disabled rules.
+CREATE INDEX IF NOT EXISTS node_rules_enabled_idx
+    ON node_rules (workspace_id)
+    WHERE enabled;