Backend lanes and hourly memory runs

.env.example

@@ -105,18 +105,22 @@
 # PERSISTENT_QUEUE_JOB_MESSAGE_CLEANUP_MINUTES=30
 # PERSISTENT_QUEUE_CONTROL_WORKERS=2
 # PERSISTENT_QUEUE_TEXT_WORKERS=4
-# Deprecated: only the fallback worker count when the routing table has no
-# dialog route; the live count derives from the capacity pools.
+# Deprecated: only the fallback worker count when the dialog primary has no
+# lane pool; the live count is the primary's lanes times the per-lane count.
 # PERSISTENT_QUEUE_DIALOG_AIFARM_WORKERS=2
 # PERSISTENT_QUEUE_DIALOG_WORKERS_CAP=24
-# PERSISTENT_QUEUE_DIALOG_UNPOOLED_SHARE=2
+# PERSISTENT_QUEUE_DIALOG_WORKERS_PER_LANE=2
 # PERSISTENT_QUEUE_DIALOG_AIFARM_FALLBACK_HIGH_WATERMARK=30
 # PERSISTENT_QUEUE_DIALOG_AIFARM_FALLBACK_LOW_WATERMARK=20
 # PERSISTENT_QUEUE_DIALOG_AIFARM_FALLBACK_POLL_INTERVAL_SECONDS=1
 # PERSISTENT_QUEUE_IMAGE_REGULAR_WORKERS=1
 # PERSISTENT_QUEUE_IMAGE_VIP_WORKERS=1
 # PERSISTENT_QUEUE_MUSIC_VIP_WORKERS=1
+# Minimum memory workers; the live count is the memory primary's lanes times
+# the per-lane count, up to the cap. 0 disables memory workers.
 # PERSISTENT_QUEUE_MEMORY_CONSOLIDATION_WORKERS=1
+# PERSISTENT_QUEUE_MEMORY_WORKERS_CAP=32
+# PERSISTENT_QUEUE_MEMORY_WORKERS_PER_LANE=2
 # PERSISTENT_QUEUE_PLACEHOLDER_CLEANUP_INTERVAL_SECONDS=3600
 # PERSISTENT_QUEUE_PLACEHOLDER_MAX_AGE_SECONDS=7200
 # LLM_JOB_MAX_ATTEMPTS=5
@@ -192,8 +196,6 @@
 # DIALOG_AIFARM_POOL_MODELS=
 # DIALOG_AIFARM_POOL_BASE_URLS=
 # DIALOG_AIFARM_POOL_API_KEY=
-# How long the dialog router waits for a busy primary pool (Gemma) slot before falling back; ms.
-# DIALOG_AIFARM_POOL_PRIMARY_CAPACITY_WAIT_MS=12000
 # DIALOG_AIFARM_POOL_REASONING_MAX_TOKENS=8192
 # DIALOG_VMLX_URL=
 # DIALOG_VMLX_API_KEY=
@@ -288,8 +290,8 @@
 # MEMORY_CONSOLIDATION_PROVIDER=aifarm
 # MEMORY_CONSOLIDATION_MODEL=qwen3.8-27b
 # MEMORY_CONSOLIDATION_TIMEOUT_SECONDS=600
-# MEMORY_DAILY_SCHEDULE=03:17
-# MEMORY_DAILY_WINDOW_HOURS=24
+# 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.
 # MEMORY_WORKER_CONCURRENCY=1
 # MEMORY_MAX_MESSAGES_PER_RUN=200
 # MEMORY_MIN_MESSAGES_PER_RUN=20

crates/openplotva-app/src/dialog_runtime.rs

@@ -41,7 +41,6 @@ pub type DialogProviderHandle = Arc<dyn ChatProvider>;
 
 /// Workflow key the dialog worker routes through.
 const DIALOG_WORKFLOW_KEY: &str = "dialog";
-const INTERACTIVE_CAPACITY_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
 
 /// Builds and caches per-provider transport clients from DB provider rows, so
 /// a provider added in the admin panel is immediately routable without a
@@ -271,20 +270,12 @@ impl RouterChatProvider {
         factory: Arc<ChatClientFactory>,
     ) -> Self {
         Self {
-            walker: RoutedAttemptWalker::new(handle, breakers, triggers, pools)
-                .with_max_slot_wait(INTERACTIVE_CAPACITY_WAIT_TIMEOUT),
+            walker: RoutedAttemptWalker::new(handle, breakers, triggers, pools),
             factory,
             provider_name: "router".to_owned(),
         }
     }
 
-    /// Bounded wait for a busy primary pool before the fallback tail is used.
-    #[must_use]
-    pub fn with_primary_slot_wait(mut self, primary_slot_wait: std::time::Duration) -> Self {
-        self.walker = self.walker.with_primary_slot_wait(primary_slot_wait);
-        self
-    }
-
     /// Extra samples from the same model after a validator rejects its output.
     #[must_use]
     pub fn with_model_output_retries(mut self, retries: usize) -> Self {
@@ -505,16 +496,6 @@ pub fn router_dialog_provider(
         aifarm,
         prompt_store,
     ));
-    // Zero disables the wait on purpose; a negative value is a config slip and
-    // keeps the documented default instead of silently disabling it.
-    let configured_wait_ms = config.llm.dialog.aifarm_pool_primary_capacity_wait_ms;
-    let primary_wait_ms = if configured_wait_ms < 0 {
-        openplotva_config::DEFAULT_DIALOG_PRIMARY_CAPACITY_WAIT_MS
-    } else {
-        configured_wait_ms
-    };
-    let primary_slot_wait =
-        std::time::Duration::from_millis(u64::try_from(primary_wait_ms).unwrap_or(0));
     // A negative value is a config slip, not a way to disable the re-sampling;
     // zero disables it on purpose.
     let configured_retries = config.llm.dialog.model_output_retries;
@@ -524,7 +505,6 @@ pub fn router_dialog_provider(
         configured_retries
     };
     let provider = RouterChatProvider::new(handle, breakers, triggers, pools, factory)
-        .with_primary_slot_wait(primary_slot_wait)
         .with_model_output_retries(usize::try_from(model_output_retries).unwrap_or(0));
     let provider = match routing_events {
         Some(reporter) => provider.with_routing_event_reporter(reporter),
@@ -1018,7 +998,7 @@ mod tests {
     }
 
     #[tokio::test(start_paused = true)]
-    async fn interactive_router_stops_waiting_for_a_busy_gpu_after_one_virtual_second() {
+    async fn interactive_router_waits_for_a_busy_gpu_until_the_turn_deadline() {
         let mut snapshot = routed_dialog_snapshot();
         snapshot.models[0].pool_id = Some(1);
         snapshot.pools.push(PoolRecord {
@@ -1048,14 +1028,23 @@ mod tests {
         )
         .with_routing_event_reporter(reporter.clone());
         let started = tokio::time::Instant::now();
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(200);
 
-        let error = provider
-            .run_chat_step(default_step_request())
+        let error = crate::dialog_turn::TURN_DEADLINE
+            .scope(
+                Some(deadline),
+                provider.run_chat_step(default_step_request()),
+            )
             .await
-            .expect_err("a permanently busy interactive route must fail fast");
+            .expect_err("a permanently busy interactive route fails at the turn deadline");
 
         assert!(error.to_string().contains("capacity unavailable"));
-        assert_eq!(started.elapsed(), INTERACTIVE_CAPACITY_WAIT_TIMEOUT);
+        let waited = started.elapsed();
+        assert!(
+            waited > std::time::Duration::from_millis(150)
+                && waited <= std::time::Duration::from_millis(200),
+            "{waited:?}"
+        );
         assert!(
             reporter
                 .buffer()
@@ -1925,6 +1914,8 @@ mod tests {
                 retry_max_hops: 1,
                 retry_wall_ms: 60_000,
                 enabled: true,
+                lane_priority: 0,
+                fallback_reserve_ms: 30_000,
             }],
             assignments: vec![AssignmentRecord {
                 id: 100,