From dc9fb786b0a608f4c2118ff7682ca35cc21ffa03 Mon Sep 17 00:00:00 2001 From: shanu Date: Wed, 12 Aug 2026 18:33:42 +0530 Subject: [PATCH 1/2] fix(memory/sources): size ingest RPC budget for multi-window sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `ingest_coding_sessions` RPC computed its wall-clock ceiling as `120 + min(max_sessions, 1000) * 30` on the premise that "each session drives at most one LLM call". That premise is false: TinyCortex's persona pipeline splits an oversized session into `WINDOW_CHARS`-sized windows and issues one LLM call per window, so a multi-window session drives several sequential calls. A dense backfill blew the ceiling — 15 sessions hit the exact 570 s budget (`120 + 15*30`) and were killed mid-flight. Extract the computation into a pure `ingest_budget(max_sessions)` and size the per-session allowance for multiple windows (`PER_SESSION_SECS = 120`, ~4 sequential calls) rather than one, so a legitimate backfill runs to completion while a genuine infinite hang still terminates. The untrusted `max_sessions` cap (1000) is preserved, and the false comment is corrected. Adds unit tests for the new formula, including the 15-session regression and the cap. `ingest_budget` takes `usize` to match the request field type. This is the RPC-timeout half of #5509. The digest-truncation half lives in tinycortex (retain observations when a digest is truncated); the vendored tinycortex submodule pointer is bumped in a follow-up once that lands. Part of #5509 --- src/openhuman/memory/sources/rpc.rs | 78 ++++++++++++++++++++++++++--- 1 file changed, 72 insertions(+), 6 deletions(-) diff --git a/src/openhuman/memory/sources/rpc.rs b/src/openhuman/memory/sources/rpc.rs index a29516bd8b..5422b2258c 100644 --- a/src/openhuman/memory/sources/rpc.rs +++ b/src/openhuman/memory/sources/rpc.rs @@ -35,6 +35,42 @@ pub async fn coding_session_status_rpc() -> Result 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 several sequential per-window LLM calls + /// rather than the single call the old formula assumed. + const PER_SESSION_SECS: u64 = 120; + /// Cap on the untrusted `max_sessions` multiplier so a hostile/garbage value + /// can't inflate the ceiling without bound. + const MAX_SESSIONS_FOR_BUDGET: usize = 1_000; + + let sessions = max_sessions.min(MAX_SESSIONS_FOR_BUDGET) as u64; + std::time::Duration::from_secs(BASE_SECS + sessions * PER_SESSION_SECS) +} + pub async fn ingest_coding_sessions_rpc( req: crate::openhuman::memory::tinycortex::CodingSessionIngestRequest, ) -> Result, String> { @@ -49,12 +85,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 +965,36 @@ mod supported_toolkits_tests { ); } } + +#[cfg(test)] +mod budget_tests { + use super::*; + + /// The formula is `120 + min(N, 1000) * 120` 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 + 120); + // The 15-session backfill that used to die at the old 570s ceiling now + // gets 1920s — proof the per-session allowance was raised to cover + // multi-window sessions (old formula: 120 + 15*30 = 570). + assert_eq!(ingest_budget(15).as_secs(), 120 + 15 * 120); + assert!( + ingest_budget(15) > std::time::Duration::from_secs(570), + "the multi-window budget must exceed the old flat 570s ceiling" + ); + } + + /// `max_sessions` is untrusted, so the multiplier is capped at 1000 — an + /// inflated request cannot turn the ceiling into an unbounded wait. + #[test] + fn budget_caps_untrusted_max_sessions() { + let at_cap = ingest_budget(1_000).as_secs(); + assert_eq!(at_cap, 120 + 1_000 * 120); + // Anything above the cap yields the same ceiling as the cap. + assert_eq!(ingest_budget(usize::MAX).as_secs(), at_cap); + assert_eq!(ingest_budget(5_000).as_secs(), at_cap); + } +} From 69199c82978cbb31b490f6480a8f84a0118b35a2 Mon Sep 17 00:00:00 2001 From: shanu Date: Thu, 13 Aug 2026 19:46:19 +0530 Subject: [PATCH 2/2] fix(memory/sources): bound coding-session ingest to the reachable RPC ceiling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on #5531: the server-only budget raise did not reach the client. The frontend RPC client clamps every per-call timeout to PER_CALL_TIMEOUT_MAX_MS = 600s (coreRpcClient.ts), and memorySourcesService bounds the same call at 120s + 15*30s + 15s = 585s — a mirror of the *old* server formula. So raising the server ceiling to 1920s only moved the binding constraint from the server's 570s to the client's 585s (~15s gain), and any budget above 600s is unreachable by construction. Fix the real constraint on both sides of the wire so a pass fits under the 600s ceiling and large histories drain across passes: - Server (rpc.rs): size ingest_budget at 120 + N*90s (an honest multi-window estimate matching the 20-45s/window observed in #5509) and hard-cap the Duration at 600s. The Duration cap replaces the multiplier cap, so an untrusted max_sessions=1000 can no longer pin the blocking worker for ~33h. - Client (memorySourcesService.ts): drop CODING_SESSION_BATCH_MAX 15 -> 5 and raise per-session 30s -> 90s, so a pass is 120 + 5*90 + 15 = 585s < 600s; drainCodingSessions already iterates passes (2000-pass cap covers ~10k sessions). Correct the stale "120s + 15*30s" comment. - Keep the server budget the tighter of the two (570s < client 585s) so the server returns a clean structured timeout before the client's fetch aborts. Tests: server_budget_is_reachable_and_tighter_than_the_client pins the cross-wire invariant (server <= client <= 600s) that would have caught the server-only gap; budget_is_capped_at_the_reachable_ceiling pins the cap; the client test guards timeoutMs <= 600s. Part of #5509 --- app/src/services/memorySourcesService.test.ts | 12 +- app/src/services/memorySourcesService.ts | 28 ++-- src/openhuman/memory/sources/rpc.rs | 122 ++++++++++++------ 3 files changed, 109 insertions(+), 53 deletions(-) diff --git a/app/src/services/memorySourcesService.test.ts b/app/src/services/memorySourcesService.test.ts index 10d559f1f5..2ad168be44 100644 --- a/app/src/services/memorySourcesService.test.ts +++ b/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 diff --git a/app/src/services/memorySourcesService.ts b/app/src/services/memorySourcesService.ts index 86eba4e3e1..c33f5e8fbe 100644 --- a/app/src/services/memorySourcesService.ts +++ b/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 { diff --git a/src/openhuman/memory/sources/rpc.rs b/src/openhuman/memory/sources/rpc.rs index 5422b2258c..a2acc643b2 100644 --- a/src/openhuman/memory/sources/rpc.rs +++ b/src/openhuman/memory/sources/rpc.rs @@ -36,39 +36,53 @@ pub async fn coding_session_status_rpc() -> Result 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 several sequential per-window LLM calls - /// rather than the single call the old formula assumed. - const PER_SESSION_SECS: u64 = 120; - /// Cap on the untrusted `max_sessions` multiplier so a hostile/garbage value - /// can't inflate the ceiling without bound. - const MAX_SESSIONS_FOR_BUDGET: usize = 1_000; - - let sessions = max_sessions.min(MAX_SESSIONS_FOR_BUDGET) as u64; - std::time::Duration::from_secs(BASE_SECS + sessions * PER_SESSION_SECS) + /// 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( @@ -970,31 +984,61 @@ mod supported_toolkits_tests { mod budget_tests { use super::*; - /// The formula is `120 + min(N, 1000) * 120` seconds. + /// 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 + 120); - // The 15-session backfill that used to die at the old 570s ceiling now - // gets 1920s — proof the per-session allowance was raised to cover - // multi-window sessions (old formula: 120 + 15*30 = 570). - assert_eq!(ingest_budget(15).as_secs(), 120 + 15 * 120); - assert!( - ingest_budget(15) > std::time::Duration::from_secs(570), - "the multi-window budget must exceed the old flat 570s ceiling" - ); + 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); } - /// `max_sessions` is untrusted, so the multiplier is capped at 1000 — an - /// inflated request cannot turn the ceiling into an unbounded wait. + /// 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 budget_caps_untrusted_max_sessions() { - let at_cap = ingest_budget(1_000).as_secs(); - assert_eq!(at_cap, 120 + 1_000 * 120); - // Anything above the cap yields the same ceiling as the cap. - assert_eq!(ingest_budget(usize::MAX).as_secs(), at_cap); - assert_eq!(ingest_budget(5_000).as_secs(), at_cap); + 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" + ); } }