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 a29516bd8b..a2acc643b2 100644 --- a/src/openhuman/memory/sources/rpc.rs +++ b/src/openhuman/memory/sources/rpc.rs @@ -35,6 +35,56 @@ 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 ~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, 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" + ); + } +}