diff --git a/.env.example b/.env.example index 5f367221..481a6aab 100644 --- a/.env.example +++ b/.env.example @@ -291,12 +291,11 @@ # MEMORY_CONSOLIDATION_MODEL=qwen3.8-27b # MEMORY_CONSOLIDATION_TIMEOUT_SECONDS=600 # Each chat hour with enough messages becomes a memory run at HH:02 UTC, after -# the hour completes; the daily cap counts runs created since midnight UTC. +# the hour completes; MEMORY_MAX_QUEUED_RUNS bounds the active queue. # MEMORY_WORKER_CONCURRENCY=1 # MEMORY_MAX_MESSAGES_PER_RUN=200 # MEMORY_MIN_MESSAGES_PER_RUN=20 # MEMORY_MAX_QUEUED_RUNS=5000 -# MEMORY_MAX_DAILY_ENQUEUED_RUNS=2000 # MEMORY_MAX_INPUT_TOKENS=10000 # MEMORY_TOKENIZER_MODEL=google/gemma-4-26B-A4B-it # MEMORY_TOKENIZER_FILE= diff --git a/crates/openplotva-app/src/lib.rs b/crates/openplotva-app/src/lib.rs index d18a2e94..fc317479 100644 --- a/crates/openplotva-app/src/lib.rs +++ b/crates/openplotva-app/src/lib.rs @@ -4372,7 +4372,6 @@ fn admin_memory_policy_json(routes: &StaticWebRoutes) -> serde_json::Value { serde_json::json!({ "min_messages_per_run": routes.memory_enqueue_policy.min_messages_per_run, "max_queued_runs": routes.memory_enqueue_policy.max_queued_runs, - "max_daily_enqueued_runs": routes.memory_enqueue_policy.max_daily_enqueued_runs, "max_messages_per_run": routes.memory_max_messages_per_run, "max_input_tokens": routes.memory_max_input_tokens, "token_estimator": routes.memory_token_estimator_source.as_ref(), @@ -16624,7 +16623,7 @@ mod tests { assert!(value["runs"].is_null()); assert_eq!(value["policy"]["min_messages_per_run"], 20); assert_eq!(value["policy"]["max_queued_runs"], 5000); - assert_eq!(value["policy"]["max_daily_enqueued_runs"], 2000); + assert!(value["policy"].get("max_daily_enqueued_runs").is_none()); assert_eq!(value["policy"]["token_estimator"], "heuristic"); assert!(value["enqueue_rollups"].is_null()); diff --git a/crates/openplotva-app/src/memory_runtime.rs b/crates/openplotva-app/src/memory_runtime.rs index 7dc5c016..82d87480 100644 --- a/crates/openplotva-app/src/memory_runtime.rs +++ b/crates/openplotva-app/src/memory_runtime.rs @@ -2525,7 +2525,7 @@ const MEMORY_ENQUEUE_MINUTE: u8 = 2; /// short outage still lands its hours. pub const MEMORY_RECENT_LOOKBACK: Duration = Duration::from_secs(2 * 60 * 60); /// UTC hour whose tick rescans the whole retention window for hours a longer -/// outage or the daily cap left without a run. +/// outage or a full queue left without a run. const MEMORY_FULL_SCAN_HOUR: u8 = 3; /// History span one enqueue tick scans: the whole retention window on startup @@ -4675,7 +4675,6 @@ pub fn memory_service_worker_config_from_memory_config( enqueue_policy: openplotva_memory::MemoryRunEnqueuePolicy { min_messages_per_run: config.min_messages_per_run, max_queued_runs: config.max_queued_runs, - max_daily_enqueued_runs: config.max_daily_enqueued_runs, }, process: MemoryRunProcessConfig { lease: positive_seconds(config.consolidation_timeout_seconds), @@ -5667,7 +5666,6 @@ mod tests { let config = AppConfig::from_raw(openplotva_config::RawConfig { memory_min_messages_per_run: Some("21".to_owned()), memory_max_queued_runs: Some("4321".to_owned()), - memory_max_daily_enqueued_runs: Some("1234".to_owned()), ..openplotva_config::RawConfig::default() }) .expect("config"); @@ -5676,7 +5674,6 @@ mod tests { assert_eq!(cfg.enqueue_policy.min_messages_per_run, 21); assert_eq!(cfg.enqueue_policy.max_queued_runs, 4321); - assert_eq!(cfg.enqueue_policy.max_daily_enqueued_runs, 1234); } #[test] diff --git a/crates/openplotva-config/src/lib.rs b/crates/openplotva-config/src/lib.rs index d5ce7f68..dff061d9 100644 --- a/crates/openplotva-config/src/lib.rs +++ b/crates/openplotva-config/src/lib.rs @@ -383,8 +383,6 @@ pub const DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN: i32 = 20; pub const DEFAULT_MEMORY_MAX_QUEUED_RUNS: i32 = 5_000; -pub const DEFAULT_MEMORY_MAX_DAILY_ENQUEUED_RUNS: i32 = 2_000; - pub const DEFAULT_MEMORY_TOKENIZER_MODEL: &str = "google/gemma-4-26B-A4B-it"; /// Discovery service name the embedder registers under on the AI Farm. @@ -1047,7 +1045,6 @@ pub struct MemoryConfig { pub max_messages_per_run: i32, pub min_messages_per_run: i32, pub max_queued_runs: i32, - pub max_daily_enqueued_runs: i32, pub max_input_tokens: i32, pub tokenizer_model: String, pub tokenizer_file: String, @@ -1574,8 +1571,6 @@ pub struct RawConfig { pub memory_min_messages_per_run: Option, /// `MEMORY_MAX_QUEUED_RUNS`. pub memory_max_queued_runs: Option, - /// `MEMORY_MAX_DAILY_ENQUEUED_RUNS`. - pub memory_max_daily_enqueued_runs: Option, /// `MEMORY_MAX_INPUT_TOKENS`. pub memory_max_input_tokens: Option, /// `MEMORY_TOKENIZER_MODEL`. @@ -2900,14 +2895,6 @@ impl AppConfig { DEFAULT_MEMORY_MAX_QUEUED_RUNS, )?, )?, - max_daily_enqueued_runs: validate_positive_i32( - "MEMORY_MAX_DAILY_ENQUEUED_RUNS", - parse_i32( - "MEMORY_MAX_DAILY_ENQUEUED_RUNS", - raw.memory_max_daily_enqueued_runs, - DEFAULT_MEMORY_MAX_DAILY_ENQUEUED_RUNS, - )?, - )?, max_input_tokens: parse_i32( "MEMORY_MAX_INPUT_TOKENS", raw.memory_max_input_tokens, @@ -3392,7 +3379,6 @@ impl RawConfig { memory_max_messages_per_run: env("MEMORY_MAX_MESSAGES_PER_RUN"), memory_min_messages_per_run: env("MEMORY_MIN_MESSAGES_PER_RUN"), memory_max_queued_runs: env("MEMORY_MAX_QUEUED_RUNS"), - memory_max_daily_enqueued_runs: env("MEMORY_MAX_DAILY_ENQUEUED_RUNS"), memory_max_input_tokens: env("MEMORY_MAX_INPUT_TOKENS"), memory_tokenizer_model: env("MEMORY_TOKENIZER_MODEL"), memory_tokenizer_file: env("MEMORY_TOKENIZER_FILE"), @@ -4210,10 +4196,6 @@ mod tests { config.memory.max_queued_runs, super::DEFAULT_MEMORY_MAX_QUEUED_RUNS ); - assert_eq!( - config.memory.max_daily_enqueued_runs, - super::DEFAULT_MEMORY_MAX_DAILY_ENQUEUED_RUNS - ); assert_eq!(config.memory.max_input_tokens, 10_000); assert_eq!(config.memory.tokenizer_model, "google/gemma-4-26B-A4B-it"); assert_eq!(config.memory.embedder_service_name, "embedder"); @@ -4923,7 +4905,6 @@ mod tests { memory_max_messages_per_run: Some("55".to_owned()), memory_min_messages_per_run: Some("20".to_owned()), memory_max_queued_runs: Some("5000".to_owned()), - memory_max_daily_enqueued_runs: Some("2000".to_owned()), memory_max_input_tokens: Some("6000".to_owned()), memory_tokenizer_model: Some("tokenizer".to_owned()), memory_tokenizer_file: Some("/tmp/tokenizer.json".to_owned()), @@ -4961,7 +4942,6 @@ mod tests { assert_eq!(config.memory.max_messages_per_run, 55); assert_eq!(config.memory.min_messages_per_run, 20); assert_eq!(config.memory.max_queued_runs, 5000); - assert_eq!(config.memory.max_daily_enqueued_runs, 2000); assert_eq!(config.memory.max_input_tokens, 6000); assert_eq!(config.memory.tokenizer_model, "tokenizer"); assert_eq!(config.memory.tokenizer_file, "/tmp/tokenizer.json"); @@ -4999,7 +4979,6 @@ mod tests { let config = AppConfig::from_raw(RawConfig::default())?; assert_eq!(config.memory.min_messages_per_run, 20); assert_eq!(config.memory.max_queued_runs, 5000); - assert_eq!(config.memory.max_daily_enqueued_runs, 2000); let error = AppConfig::from_raw(RawConfig { memory_min_messages_per_run: Some("0".to_owned()), @@ -5027,19 +5006,6 @@ mod tests { }) )); - let error = AppConfig::from_raw(RawConfig { - memory_max_daily_enqueued_runs: Some("0".to_owned()), - ..RawConfig::default() - }) - .err(); - assert!(matches!( - error, - Some(super::ConfigError::NonPositiveInteger { - name: "MEMORY_MAX_DAILY_ENQUEUED_RUNS", - value: 0, - }) - )); - Ok(()) } diff --git a/crates/openplotva-memory/src/lib.rs b/crates/openplotva-memory/src/lib.rs index 7330fab7..60afa9db 100644 --- a/crates/openplotva-memory/src/lib.rs +++ b/crates/openplotva-memory/src/lib.rs @@ -20,7 +20,6 @@ pub const PURPOSE: &str = "memory"; pub const DEFAULT_MEMORY_MAX_OUTPUT_TOKENS: i32 = 4000; pub const DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN: i32 = 20; pub const DEFAULT_MEMORY_MAX_QUEUED_RUNS: i32 = 5_000; -pub const DEFAULT_MEMORY_MAX_DAILY_ENQUEUED_RUNS: i32 = 2_000; pub const DISCOVERY_PRIORITY_MEMORY: i32 = 10; pub const DEFAULT_MEMORY_CONSOLIDATION_MODEL: &str = "Gemma 4 26B Heretic"; pub const PROMPT_VERSION: &str = "chat_memory_daily_v7"; @@ -179,7 +178,6 @@ pub struct RetrievalScope { pub struct MemoryRunEnqueuePolicy { pub min_messages_per_run: i32, pub max_queued_runs: i32, - pub max_daily_enqueued_runs: i32, } impl Default for MemoryRunEnqueuePolicy { @@ -187,7 +185,6 @@ impl Default for MemoryRunEnqueuePolicy { Self { min_messages_per_run: DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN, max_queued_runs: DEFAULT_MEMORY_MAX_QUEUED_RUNS, - max_daily_enqueued_runs: DEFAULT_MEMORY_MAX_DAILY_ENQUEUED_RUNS, } } } @@ -207,11 +204,6 @@ impl MemoryRunEnqueuePolicy { } else { self.max_queued_runs }, - max_daily_enqueued_runs: if self.max_daily_enqueued_runs <= 0 { - fallback.max_daily_enqueued_runs - } else { - self.max_daily_enqueued_runs - }, } } } diff --git a/crates/openplotva-storage/src/lib.rs b/crates/openplotva-storage/src/lib.rs index f62b61a1..f230966a 100644 --- a/crates/openplotva-storage/src/lib.rs +++ b/crates/openplotva-storage/src/lib.rs @@ -553,13 +553,8 @@ fn memory_message_count_bucket(message_count: i32) -> &'static str { fn memory_enqueue_remaining_capacity( policy: openplotva_memory::MemoryRunEnqueuePolicy, active_queue_depth: i64, - enqueued_today: u64, ) -> Option { - let policy = policy.normalized(); - let enqueued_today = i64::try_from(enqueued_today).unwrap_or(i64::MAX); - let queue_capacity = i64::from(policy.max_queued_runs) - active_queue_depth; - let daily_capacity = i64::from(policy.max_daily_enqueued_runs) - enqueued_today; - let remaining = queue_capacity.min(daily_capacity); + let remaining = i64::from(policy.normalized().max_queued_runs) - active_queue_depth; (remaining > 0).then(|| i32::try_from(remaining).unwrap_or(i32::MAX)) } @@ -1216,14 +1211,6 @@ pub const SQL_RETRY_FAILED_MEMORY_RUNS: &str = "UPDATE memory_runs SET status = pub const SQL_COUNT_ACTIVE_MEMORY_RUNS: &str = "SELECT count(*)::bigint FROM memory_runs WHERE prompt_version = $1 AND status IN ('queued', 'processing') AND range_end_at >= CURRENT_TIMESTAMP - make_interval(secs => $2::double precision)"; -pub const SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE: &str = r#"SELECT count(*)::bigint -FROM memory_runs -WHERE prompt_version = $1 - AND created_at >= $2 - AND cursor_after_at = '1970-01-01 00:00:00+00'::timestamptz - AND cursor_after_message_id = 0 - AND cursor_after_entry_id = ''"#; - pub const SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE: &str = r#"WITH dims AS ( SELECT jsonb_build_object( 'decision', 'backpressure', @@ -1250,8 +1237,7 @@ SELECT 'prompt_version', $1::text, 'queue_depth', $4::bigint, 'min_messages_per_run', $5::int, - 'max_queued_runs', $6::int, - 'max_daily_enqueued_runs', $7::int + 'max_queued_runs', $6::int ) FROM dims ON CONFLICT (source, kind, granularity, bucket_start, dimensions_hash) DO UPDATE SET @@ -1318,7 +1304,7 @@ classified AS ( WHEN message_count < $5 THEN 'below_threshold' WHEN already_enqueued THEN 'already_enqueued' WHEN enqueue_rank <= $8 THEN 'enqueued' - ELSE 'daily_cap' + ELSE 'queue_cap' END AS decision, CASE WHEN canonical_thread_id = 0 THEN 'chat' ELSE 'thread' END AS scope, CASE @@ -1350,8 +1336,7 @@ prepared AS ( 'queue_depth', $6::bigint, 'min_messages_per_run', $5::int, 'max_queued_runs', $7::int, - 'max_daily_enqueued_runs', $9::int, - 'effective_daily_limit', $8::int + 'effective_limit', $8::int ) AS metrics FROM classified GROUP BY decision, scope, message_count_bucket @@ -5820,7 +5805,7 @@ impl PostgresMemoryStore { /// Enqueue a run for every chat hour that completed before `now`, started /// within the last `lookback` (at most `retention`), and has enough - /// messages: newest first, within the queue and daily caps. The enqueue + /// messages: newest first, while the active queue has room. The enqueue /// rollups still describe whole UTC days of the scanned range. pub async fn ensure_hourly_runs( &self, @@ -5844,10 +5829,7 @@ impl PostgresMemoryStore { let mut window_end = completed_hours_end; let mut total = 0; let active_queue_depth = self.active_memory_run_count().await?; - let base_runs_enqueued_today = self.base_memory_run_count_created_since(day_start).await?; - if memory_enqueue_remaining_capacity(policy, active_queue_depth, base_runs_enqueued_today) - .is_none() - { + if memory_enqueue_remaining_capacity(policy, active_queue_depth).is_none() { self.record_memory_enqueue_backpressure( earliest_day_start, completed_hours_end, @@ -5868,7 +5850,6 @@ impl PostgresMemoryStore { let Some(limit) = memory_enqueue_remaining_capacity( policy, active_queue_depth.saturating_add(i64::try_from(total).unwrap_or(i64::MAX)), - base_runs_enqueued_today.saturating_add(total), ) else { break; }; @@ -5898,18 +5879,6 @@ impl PostgresMemoryStore { .await?) } - async fn base_memory_run_count_created_since( - &self, - created_since: OffsetDateTime, - ) -> Result { - let count: i64 = sqlx::query_scalar(SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE) - .bind(openplotva_memory::PROMPT_VERSION) - .bind(created_since) - .fetch_one(&self.pool) - .await?; - Ok(u64::try_from(count).unwrap_or(u64::MAX)) - } - async fn record_memory_enqueue_backpressure( &self, window_start: OffsetDateTime, @@ -5924,7 +5893,6 @@ impl PostgresMemoryStore { .bind(active_queue_depth) .bind(policy.min_messages_per_run) .bind(policy.max_queued_runs) - .bind(policy.max_daily_enqueued_runs) .execute(&self.pool) .await?; Ok(()) @@ -5937,7 +5905,7 @@ impl PostgresMemoryStore { earliest: OffsetDateTime, active_queue_depth: i64, policy: openplotva_memory::MemoryRunEnqueuePolicy, - daily_limit: i32, + limit: i32, ) -> Result<(), StorageError> { sqlx::query(SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS) .bind(openplotva_memory::PROMPT_VERSION) @@ -5947,8 +5915,7 @@ impl PostgresMemoryStore { .bind(policy.min_messages_per_run) .bind(active_queue_depth) .bind(policy.max_queued_runs) - .bind(daily_limit) - .bind(policy.max_daily_enqueued_runs) + .bind(limit) .execute(&self.pool) .await?; Ok(()) @@ -5960,7 +5927,7 @@ impl PostgresMemoryStore { window_end: OffsetDateTime, earliest: OffsetDateTime, policy: openplotva_memory::MemoryRunEnqueuePolicy, - daily_limit: i32, + limit: i32, ) -> Result { Ok(sqlx::query(SQL_ENSURE_HOURLY_MEMORY_RUNS) .bind(openplotva_memory::PROMPT_VERSION) @@ -5968,7 +5935,7 @@ impl PostgresMemoryStore { .bind(window_end) .bind(earliest) .bind(policy.min_messages_per_run) - .bind(daily_limit) + .bind(limit) .execute(&self.pool) .await? .rows_affected()) @@ -11645,18 +11612,10 @@ mod tests { .contains("WHEN COALESCE(c.is_forum, false) AND e.thread_id <> 0") ); assert!(super::SQL_COUNT_ACTIVE_MEMORY_RUNS.contains("status IN ('queued', 'processing')")); - assert!(super::SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE.contains("created_at >= $2")); - assert!( - super::SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE - .contains("cursor_after_at = '1970-01-01 00:00:00+00'::timestamptz") - ); - assert!( - super::SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE.contains("cursor_after_message_id = 0") - ); - assert!( - super::SQL_COUNT_BASE_MEMORY_RUNS_CREATED_SINCE.contains("cursor_after_entry_id = ''") - ); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("message_count_bucket")); + assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'queue_cap'")); + assert!(!super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("daily")); + assert!(!super::SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE.contains("daily")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'below_threshold'")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'already_enqueued'")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE.contains("'backpressure'")); @@ -11791,26 +11750,18 @@ mod tests { let policy = openplotva_memory::MemoryRunEnqueuePolicy { min_messages_per_run: 20, max_queued_runs: 5000, - max_daily_enqueued_runs: 2000, }; assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 4_900, 0), + super::memory_enqueue_remaining_capacity(policy, 4_900), Some(100) ); assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 4_900, 100), - Some(100) - ); - assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 0, 1_999), - Some(1) - ); - assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 0, 2_000), - None + super::memory_enqueue_remaining_capacity(policy, 0), + Some(5_000), + "only the active queue bounds the enqueue; there is no daily cap" ); assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 5_000, 0), + super::memory_enqueue_remaining_capacity(policy, 5_000), None ); } @@ -12076,7 +12027,6 @@ mod tests { let policy = openplotva_memory::MemoryRunEnqueuePolicy { min_messages_per_run: 3, max_queued_runs: 1_000_000, - max_daily_enqueued_runs: 1_000_000, }; let retention = Duration::from_secs(7 * 24 * 60 * 60); let windows = || async { diff --git a/docs/superpowers/specs/2026-10-12-backend-lanes-design.md b/docs/superpowers/specs/2026-10-12-backend-lanes-design.md index a5003b26..e0a55e45 100644 --- a/docs/superpowers/specs/2026-10-12-backend-lanes-design.md +++ b/docs/superpowers/specs/2026-10-12-backend-lanes-design.md @@ -150,9 +150,11 @@ Memory runs already cover one chat hour each, but they were enqueued once a day UTC day, which put a whole day of memory work into a morning burst. The scheduler now ticks at `HH:02` UTC and enqueues every chat hour that completed since the last tick. An hourly tick scans the last two completed hours. The startup tick and the `03:02` tick scan the whole retention -window, for hours that a longer outage or the daily cap left without a run. The daily cap still -counts runs created since midnight UTC, and the enqueue rollups still describe whole UTC days. -`MEMORY_DAILY_SCHEDULE` and `MEMORY_DAILY_WINDOW_HOURS` are removed. +window, for hours that a longer outage or a full queue left without a run. Only +`MEMORY_MAX_QUEUED_RUNS` bounds the enqueue; the daily cap (`MEMORY_MAX_DAILY_ENQUEUED_RUNS`) +was removed on 2026-10-13 after it stopped all memory work from about 04:00 UTC for the rest of +the day. The enqueue rollups still describe whole UTC days. `MEMORY_DAILY_SCHEDULE` and +`MEMORY_DAILY_WINDOW_HOURS` are removed. ## Migrations