fix(memory/sources): size ingest RPC budget for multi-window sessions

app/src/services/memorySourcesService.test.ts

@@ -230,11 +230,19 @@ describe('memorySourcesService', () => {
 
     const result = await ingestCodingSessions(false, 25);
 
+    // Clamped to the batch max (5), and the per-call timeout stays under the RPC
+    // client's hard 600s ceiling: 120s + 5*90s + 15s = 585s.
     expect(mockedCall).toHaveBeenCalledWith({
       method: 'openhuman.memory_sources_ingest_coding_sessions',
-      params: { backfill: false, max_sessions: 15 },
+      params: { backfill: false, max_sessions: 5 },
       timeoutMs: 585_000,
     });
+    // Guard the cross-wire contract: the per-call timeout must stay under
+    // coreRpcClient's PER_CALL_TIMEOUT_MAX_MS (600s), or the override is clamped
+    // and silently unreachable (the #5509 gap). Raising BATCH_MAX / per-session
+    // past this must fail here.
+    const CORE_RPC_PER_CALL_TIMEOUT_MAX_MS = 10 * 60 * 1_000;
+    expect(585_000).toBeLessThanOrEqual(CORE_RPC_PER_CALL_TIMEOUT_MAX_MS);
     expect(result.sessions_processed).toBe(2);
   });
 
@@ -269,7 +277,7 @@ describe('memorySourcesService', () => {
     expect(mockedCall).toHaveBeenCalledTimes(3);
     // Every pass stays bounded to the timeout-safe per-call maximum.
     expect(mockedCall).toHaveBeenLastCalledWith(
-      expect.objectContaining({ params: { backfill: false, max_sessions: 15 } })
+      expect.objectContaining({ params: { backfill: false, max_sessions: 5 } })
     );
     expect(result.passes).toBe(3);
     expect(result.sessionsProcessed).toBe(40); // 15 + 15 + 10

app/src/services/memorySourcesService.ts

@@ -222,21 +222,25 @@ export interface CodingSessionIngestResult {
   pack_path?: string | null;
 }
 
-// A single ingest RPC is bounded so it stays under the core RPC client's
-// ten-minute ceiling: 120s + 15 * 30s + 15s ≈ 585s. This is a per-call bound,
-// not a per-run one — large histories drain across repeated bounded passes via
-// `drainCodingSessions`, driven by the response `budget_hit` flag. Raising this
-// to the backend's 1,000-session max would blow the RPC timeout on the first
-// call, so the cap stays and the loop does the scaling.
-const CODING_SESSION_BATCH_MAX = 15;
+// A single ingest RPC is bounded so it fits under the core RPC client's hard
+// per-call ceiling (`PER_CALL_TIMEOUT_MAX_MS` = 600s in `coreRpcClient.ts`, which
+// clamps every override — a larger request timeout is silently unreachable).
+// The bound mirrors the core's `ingest_budget` (`memory/sources/rpc.rs`): the
+// server sizes the same call at `120s + N*90s` and the client adds a 15s grace
+// so the server's structured timeout fires first. Each multi-window session
+// drives several sequential LLM calls (~90s/session), so the batch is kept small
+// (`120s + 5*90s + 15s = 585s < 600s`) and large histories drain across repeated
+// bounded passes via `drainCodingSessions`, driven by the response `budget_hit`
+// flag — never by widening a single call past the reachable ceiling.
+const CODING_SESSION_BATCH_MAX = 5;
 const CODING_SESSION_BASE_TIMEOUT_MS = 120_000;
-const CODING_SESSION_PER_SESSION_TIMEOUT_MS = 30_000;
+const CODING_SESSION_PER_SESSION_TIMEOUT_MS = 90_000;
 const CODING_SESSION_RPC_GRACE_MS = 15_000;
 // Hard safety cap on drain passes so a stuck backlog can never spin forever.
-// Sized well above the largest realistic history: at 15 sessions/pass this
-// covers ~30k sessions in a single run, so the target ~7,800-file case drains
-// fully rather than exiting capped. The `moreRemaining` flag still lets the UI
-// report an honest "paused" state if the cap is ever reached.
+// Sized well above the largest realistic history: at 5 sessions/pass this covers
+// ~10k sessions in a single run, so the target ~7,800-file case drains fully
+// rather than exiting capped. The `moreRemaining` flag still lets the UI report
+// an honest "paused" state if the cap is ever reached.
 const CODING_SESSION_MAX_DRAIN_PASSES = 2000;
 
 export async function getCodingSessionStatus(): Promise<CodingSessionSourceStatus[]> {

src/openhuman/memory/sources/rpc.rs

@@ -35,6 +35,56 @@ pub async fn coding_session_status_rpc() -> Result<RpcOutcome<CodingSessionStatu
     ))
 }
 
+/// Wall-clock ceiling for one `ingest_coding_sessions` RPC, sized to the number
+/// of sessions the caller asked to backfill and hard-capped at the ceiling the
+/// frontend can actually wait for.
+///
+/// The original formula (`120 + N*30`) assumed **one LLM call per session**.
+/// That premise is false: TinyCortex's persona pipeline splits an oversized
+/// session into windows (`WINDOW_CHARS`-sized chunks of evidence) and issues one
+/// LLM call *per window*, so a multi-window session drives several sequential
+/// calls. A dense backfill therefore blew the old budget — 15 sessions hit the
+/// exact 570 s ceiling (`120 + 15*30`) and were killed mid-flight.
+///
+/// The per-session allowance is therefore sized for *multiple* windows, not one
+/// call: `PER_SESSION_SECS` budgets ~3 sequential per-window LLM calls at the
+/// windows' observed 20–45 s span (#5509). It is a deliberate flat estimate, not
+/// a per-session window count.
+///
+/// The result is hard-capped at `HARD_CAP_SECS` for two reasons that are really
+/// one. First, this is the true reachable ceiling: the frontend RPC client
+/// clamps every per-call timeout to `PER_CALL_TIMEOUT_MAX_MS = 600 s`
+/// (`app/src/services/coreRpcClient.ts`), so a server budget above that can never
+/// be observed — the client aborts first. Second, that cap also bounds the
+/// blocking-pool worker this budget guards: `max_sessions` is untrusted (an
+/// advertised programmatic RPC, `platform/about_app/catalog_data.rs`), and
+/// without the cap a caller passing 1000 would pin a thread for ~33 h. Capping
+/// the resulting `Duration` — not the multiplier — makes both true at once.
+///
+/// Because a single pass is bounded, large histories drain across repeated passes
+/// (client `drainCodingSessions`); the per-pass batch is sized so `BASE + N*PER`
+/// stays under the cap for the UI's `CODING_SESSION_BATCH_MAX`, keeping the
+/// server budget the *tighter* of the two so it returns a clean structured
+/// timeout before the client's fetch aborts. This is a *ceiling to catch a wedged
+/// run*, not a latency target.
+fn ingest_budget(max_sessions: usize) -> std::time::Duration {
+    /// Fixed overhead allowance (config load, discovery, process warm-up) added
+    /// on top of the per-session budget.
+    const BASE_SECS: u64 = 120;
+    /// Per-session allowance, sized for ~3 sequential per-window LLM calls at the
+    /// 20–45 s/window span observed in #5509 rather than the single call the old
+    /// formula assumed.
+    const PER_SESSION_SECS: u64 = 90;
+    /// Hard ceiling on the whole budget. Mirrors the frontend's
+    /// `PER_CALL_TIMEOUT_MAX_MS` (600 s) — a larger budget is unreachable because
+    /// the client aborts first — and bounds the blocking worker against an
+    /// untrusted `max_sessions`.
+    const HARD_CAP_SECS: u64 = 600;
+
+    let scaled = BASE_SECS.saturating_add((max_sessions as u64).saturating_mul(PER_SESSION_SECS));
+    std::time::Duration::from_secs(scaled.min(HARD_CAP_SECS))
+}
+
 pub async fn ingest_coding_sessions_rpc(
     req: crate::openhuman::memory::tinycortex::CodingSessionIngestRequest,
 ) -> Result<RpcOutcome<crate::openhuman::memory::tinycortex::CodingSessionIngestResponse>, String> {
@@ -49,12 +99,9 @@ pub async fn ingest_coding_sessions_rpc(
     let runtime = tokio::runtime::Handle::current();
     // Wall-clock ceiling so a stalled provider call or a wedged session step
     // can't keep the RPC (and its blocking worker) waiting indefinitely (#4863
-    // review). Scale to the requested budget — each session drives at most one
-    // LLM call — so a large backfill isn't killed mid-flight while a genuine
-    // infinite hang still terminates. `max_sessions` is untrusted, so cap the
-    // multiplier before computing the budget.
-    let ingest_timeout =
-        std::time::Duration::from_secs(120 + (req.max_sessions.min(1_000) as u64) * 30);
+    // review), sized to the requested backfill so a legitimate large run isn't
+    // killed mid-flight while a genuine infinite hang still terminates.
+    let ingest_timeout = ingest_budget(req.max_sessions);
     let response = tokio::task::spawn_blocking(move || {
         runtime.block_on(async move {
             tokio::time::timeout(
@@ -932,3 +979,66 @@ mod supported_toolkits_tests {
         );
     }
 }
+
+#[cfg(test)]
+mod budget_tests {
+    use super::*;
+
+    /// The frontend's `CODING_SESSION_BATCH_MAX` (`app/src/services/memorySourcesService.ts`).
+    /// Mirrored here so the cross-wire invariant below is checkable Rust-side; the
+    /// two must move together.
+    const CLIENT_BATCH_MAX: usize = 5;
+    /// The frontend's `PER_CALL_TIMEOUT_MAX_MS` (`app/src/services/coreRpcClient.ts`),
+    /// in seconds — the ceiling the client can actually wait for.
+    const CLIENT_HARD_CAP_SECS: u64 = 600;
+    /// The frontend's `CODING_SESSION_RPC_GRACE_MS`, in seconds.
+    const CLIENT_GRACE_SECS: u64 = 15;
+
+    /// The formula is `min(120 + N * 90, 600)` seconds.
+    #[test]
+    fn budget_scales_per_session_for_multiple_windows() {
+        // Zero sessions → base overhead only.
+        assert_eq!(ingest_budget(0).as_secs(), 120);
+        // One session carries a full per-session (multi-window) allowance.
+        assert_eq!(ingest_budget(1).as_secs(), 120 + 90);
+        // The UI batch (5 sessions) gets 570s — sized so a pass fits under the
+        // 600s reachable ceiling while a 15-session backlog drains across passes.
+        assert_eq!(ingest_budget(CLIENT_BATCH_MAX).as_secs(), 120 + 5 * 90);
+    }
+
+    /// The budget is hard-capped at the reachable ceiling (600s), so an untrusted
+    /// `max_sessions` cannot pin the blocking worker beyond it — and cannot
+    /// overflow.
+    #[test]
+    fn budget_is_capped_at_the_reachable_ceiling() {
+        assert_eq!(ingest_budget(1_000).as_secs(), CLIENT_HARD_CAP_SECS);
+        // Anything at or above the cap yields exactly the ceiling, no overflow.
+        assert_eq!(ingest_budget(usize::MAX).as_secs(), CLIENT_HARD_CAP_SECS);
+        assert_eq!(ingest_budget(5_000).as_secs(), CLIENT_HARD_CAP_SECS);
+    }
+
+    /// The invariant #5509 actually needs, pinned across the wire: for the UI's
+    /// batch size the server budget must (a) stay under the client's hard cap so
+    /// the pass is reachable, and (b) be the *tighter* of the two — i.e. below the
+    /// client's own timeout for the same batch — so the server returns a clean
+    /// structured timeout before the client's fetch aborts. This is the guard
+    /// that would have caught the server-only fix moving the ceiling from 570s to
+    /// an unreachable 1920s.
+    #[test]
+    fn server_budget_is_reachable_and_tighter_than_the_client() {
+        let server = ingest_budget(CLIENT_BATCH_MAX).as_secs();
+        let client = 120 + (CLIENT_BATCH_MAX as u64) * 90 + CLIENT_GRACE_SECS;
+        assert!(
+            server <= CLIENT_HARD_CAP_SECS,
+            "server budget {server}s must be reachable (<= {CLIENT_HARD_CAP_SECS}s client cap)"
+        );
+        assert!(
+            client <= CLIENT_HARD_CAP_SECS,
+            "client budget {client}s must stay under its own {CLIENT_HARD_CAP_SECS}s clamp"
+        );
+        assert!(
+            server < client,
+            "server budget {server}s must fire before the client's {client}s abort"
+        );
+    }
+}