fix(server): review-fix sweep

crates/preloop-runner-server/src/bootstrap.rs

@@ -515,10 +515,16 @@ fn build_operational_snapshot_sync(
             max_lease_age_seconds: None,
         },
         pool: pool_snapshot,
-        vms: VmFleetSnapshot {
-            source: VmSource::Unavailable,
-            sample_age_seconds: None,
-            ..Default::default()
+        vms: {
+            // Host sampler is stubbed until the cgroup parser lands;
+            // the registry is the source of truth for configured counts
+            // and will be populated by RunnerPool on create/fork.
+            let caps = std::collections::HashMap::new();
+            preloop_observability::vm_telemetry::build_fleet_snapshot(
+                observability.vm_registry(),
+                None,
+                caps,
+            )
         },
         store: StoreSnapshot::default(),
         storage: {
@@ -533,11 +539,11 @@ fn build_operational_snapshot_sync(
                 state_dir: state_dir.display().to_string(),
                 state_fs_free_bytes: None,
                 state_fs_free_ratio: None,
-                components: vec![
-                    component("database", state_dir.join("preloop.db")),
-                    component("cache", state_dir.join("cache")),
-                    component("artifacts", state_dir.join("artifacts")),
-                ],
+                // Only the database is a single file. `cache` and `artifacts`
+                // are directories, whose `metadata().len()` is the inode
+                // size, not the contents size; they need the recursive walk
+                // on the 60s cadence.
+                components: vec![component("database", state_dir.join("preloop.db"))],
                 last_gc_at: None,
             }
         },
@@ -644,15 +650,16 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
     {
         // One decorator around the private `Store` trait — never per-backend
         // duplication. The backend label is bounded to sqlite|postgres.
-        let backend = if config
+        // Mirror `open_store` precedence: the explicit URL wins, and the
+        // environment is consulted only when none was supplied. Otherwise a
+        // stale `PRELOOP_STORE_URL=postgres://…` would mislabel every
+        // SQLite operation.
+        let effective_url = config
             .store_url
-            .as_deref()
-            .map(|u| u.contains("postgres"))
-            .unwrap_or(false)
-            || std::env::var("PRELOOP_STORE_URL")
-                .map(|v| v.contains("postgres"))
-                .unwrap_or(false)
-        {
+            .clone()
+            .or_else(|| std::env::var("PRELOOP_STORE_URL").ok())
+            .unwrap_or_default();
+        let backend = if effective_url.contains("postgres") {
             "postgres"
         } else {
             "sqlite"
@@ -768,33 +775,33 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
             shutdown: shutdown.clone(),
         });
         let scheduler_clone = scheduler.clone();
-        // Spawn a holder task for scheduler_scan heartbeat — keep handle alive for lifetime.
-        let hb_for_holder = scheduler_heartbeat.clone();
-        let shutdown_for_holder = shutdown.clone();
-        tokio::spawn(async move {
-            let _handle = hb_for_holder.register(
-                "scheduler_scan",
-                preloop_observability::Criticality::Critical,
-            );
-            hb_for_holder.beat("scheduler_scan");
-            let mut int = tokio::time::interval(Duration::from_secs(10));
-            int.tick().await;
-            while !shutdown_for_holder.is_cancelled() {
-                tokio::select! {
-                    _ = int.tick() => hb_for_holder.beat("scheduler_scan"),
-                    _ = shutdown_for_holder.cancelled() => break,
-                }
-            }
-        });
+        // The scheduler heartbeat must prove the startup scan progressed, not
+        // that an unrelated timer is awake. The scan tasks beat it per
+        // workflow file and deregister on completion: a scan that hangs or
+        // panics stops beating and `/readyz` goes 503; a completed scan is
+        // not a stale critical task.
+        let scan_hb = scheduler_heartbeat.clone();
         if let Some(workspace) = state.local_workspace.clone() {
+            let shared_for_scan = shared_for_scan.clone();
             tokio::spawn(async move {
+                let handle = scan_hb.register(
+                    "scheduler_scan",
+                    preloop_observability::Criticality::Critical,
+                );
                 scheduler_clone
-                    .scan_workspace(&workspace, shared_for_scan)
+                    .scan_workspace(&workspace, shared_for_scan, Some(handle))
                     .await;
             });
         } else {
+            let shared_for_scan = shared_for_scan.clone();
             tokio::spawn(async move {
-                scheduler_clone.scan_remote(shared_for_scan).await;
+                let handle = scan_hb.register(
+                    "scheduler_scan",
+                    preloop_observability::Criticality::Critical,
+                );
+                scheduler_clone
+                    .scan_remote(shared_for_scan, Some(handle))
+                    .await;
             });
         }
     }

crates/preloop-runner-server/src/broker.rs

@@ -271,6 +271,7 @@ pub(crate) async fn next_message_broker_ref(
             .queue_depth
             .store(inner.queue.len(), std::sync::atomic::Ordering::Release);
         runtime_scheduling::sync_next_job_labels(&inner, &shared.state.next_job_runs_on);
+        record_claim_queue_wait(&shared, &claimed);
         let Some(queued) = claimed else {
             drop(inner);
             if wait_seconds == 0 {
@@ -397,6 +398,28 @@ pub(crate) async fn broker_session_root(
     ))
 }
 
+/// Record how long a claimed job sat in the ready queue. `enqueued_at` is
+/// stamped when the job enters the queue and survives requeues, so a job
+/// that bounced off a purged runner still measures total queue time. Jobs
+/// restored from a snapshot without the field (`0`) are not recorded.
+fn record_claim_queue_wait(shared: &Arc<SharedState>, claimed: &Option<QueuedJob>) {
+    let Some(queued) = claimed else { return };
+    if queued.enqueued_at_unix_nanos <= 0 {
+        return;
+    }
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_nanos() as i64)
+        .unwrap_or(0);
+    let elapsed = std::time::Duration::from_nanos((now - queued.enqueued_at_unix_nanos) as u64);
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_queue_wait("claimed", elapsed);
+}
+
 pub(crate) async fn broker_delete_session_root(
     State(shared): State<Arc<SharedState>>,
     Query(params): Query<std::collections::HashMap<String, String>>,
@@ -409,13 +432,15 @@ pub(crate) async fn broker_delete_session_root(
     if let Some(session_id) = header_session.or_else(|| params.get("sessionId").map(String::as_str))
     {
         remove_broker_session(&shared, session_id, runner_id).await?;
+        shared
+            .state
+            .observability
+            .metrics()
+            .lifecycle
+            .record_session_transition("delete", "ok");
     }
-    shared
-        .state
-        .observability
-        .metrics()
-        .lifecycle
-        .record_session_transition("delete", "ok");
+    // No session id present: nothing was deleted, so no transition is
+    // recorded — a 204 with no-op must not count as a successful delete.
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -598,6 +623,7 @@ pub(crate) async fn next_message_broker_ref_root(
                     .queue_depth
                     .store(inner.queue.len(), std::sync::atomic::Ordering::Release);
                 runtime_scheduling::sync_next_job_labels(&inner, &shared.state.next_job_runs_on);
+                record_claim_queue_wait(&shared, &claimed);
                 if let Some(queued) = claimed {
                     if let Some(run) = inner.runs.get_mut(&queued.run_id) {
                         run.status = ExecutionStatus::InProgress;
@@ -873,13 +899,11 @@ pub(crate) async fn broker_acquire_job(
     message.request_id = 0;
     let payload = serde_json::to_value(&message)
         .map_err(|error| ApiError::internal(format!("serialize broker job payload: {error}")))?;
-    // Queue wait and broker poll outcomes — bounded, exactly one per successful claim.
-    shared
-        .state
-        .observability
-        .metrics()
-        .lifecycle
-        .record_queue_wait("claimed", std::time::Duration::from_secs(1));
+    // Broker poll outcome — bounded, exactly one per successful claim. Queue
+    // wait is recorded at the claim sites in `next_message_broker_ref` /
+    // `next_message_disttask`, where the enqueue timestamp is still on the
+    // job; by the time the acquire payload is built the queue position has
+    // been lost.
     shared
         .state
         .observability

crates/preloop-runner-server/src/scheduler.rs

@@ -301,14 +301,22 @@ impl Scheduler {
     /// Scan a local workspace directory and register all schedule workflows.
     ///
     /// Called once on server startup when `--enable-scheduler` is active.
-    pub async fn scan_workspace(self: &Arc<Self>, workspace: &PathBuf, shared: Arc<SharedState>) {
+    pub async fn scan_workspace(
+        self: &Arc<Self>,
+        workspace: &PathBuf,
+        shared: Arc<SharedState>,
+        heartbeat: Option<preloop_observability::HeartbeatHandle>,
+    ) {
         let wf_dir = workspace.join(".github").join("workflows");
         let entries = match std::fs::read_dir(&wf_dir) {
             Ok(e) => e,
             Err(_) => return,
         };
         for entry in entries.flatten() {
             let path = entry.path();
+            if let Some(hb) = &heartbeat {
+                hb.beat();
+            }
             if !matches!(
                 path.extension().and_then(|e| e.to_str()),
                 Some("yml") | Some("yaml")
@@ -375,7 +383,14 @@ impl Scheduler {
     /// Fetch and install schedules for a remote-backed server at startup.
     /// The repository and token use `PRELOOP_GITHUB_REPOSITORY` and
     /// `PRELOOP_GITHUB_TOKEN`, the same explicit remote workflow configuration.
-    pub async fn scan_remote(self: &Arc<Self>, shared: Arc<SharedState>) {
+    pub async fn scan_remote(
+        self: &Arc<Self>,
+        shared: Arc<SharedState>,
+        heartbeat: Option<preloop_observability::HeartbeatHandle>,
+    ) {
+        if let Some(hb) = &heartbeat {
+            hb.beat();
+        }
         let (Ok(repository), Ok(token)) = (
             std::env::var("PRELOOP_GITHUB_REPOSITORY"),
             std::env::var("PRELOOP_GITHUB_TOKEN"),

crates/preloop-runner-server/src/state.rs

@@ -607,12 +607,16 @@ fn bounded_termination_reason(value: &str) -> &'static str {
     //   - the external-host check, where the server has no runner of that
     //     platform class at all (`no {platform} runner is registered with
     //     this server, so `runs-on: …` cannot be scheduled`).
-    if value.contains("runner is registered with this server") {
-        return "no_platform_runner";
-    }
+    // The starvation prose interpolates workflow-controlled `runs-on`
+    // labels, so the anchored prefix MUST be checked before the substring:
+    // a crafted label containing the platform phrase must not flip a
+    // starvation reason into `no_platform_runner`.
     if value.starts_with("no runner is registered for") {
         return "no_runner";
     }
+    if value.contains("runner is registered with this server") {
+        return "no_platform_runner";
+    }
     if value.starts_with("job exceeded its timeout")
         || value.starts_with("timed out")
         || value.contains("timeout-minutes")
@@ -628,6 +632,25 @@ fn bounded_termination_reason(value: &str) -> &'static str {
     "unrecognized"
 }
 
+/// Bound free-form reason prose for export as a telemetry attribute. The
+/// prose interpolates workflow input (e.g. `runs-on` labels), so one job
+/// must not emit an arbitrarily large attribute. Truncation cuts on a
+/// character boundary — byte slicing panics on multi-byte input.
+fn bounded_reason_detail(detail: &str) -> String {
+    const DETAIL_MAX: usize = 512;
+    let mut detail = detail.to_string();
+    if detail.len() > DETAIL_MAX {
+        let cut = detail
+            .char_indices()
+            .map(|(i, _)| i)
+            .take_while(|&i| i <= DETAIL_MAX)
+            .last()
+            .unwrap_or(0);
+        detail.truncate(cut);
+    }
+    detail
+}
+
 impl AppState {
     pub async fn new(state_dir: PathBuf) -> anyhow::Result<Self> {
         let config_path = crate::config::config_path();
@@ -961,27 +984,17 @@ impl AppState {
                         // it an `unrecognized` classification is a dead end —
                         // you cannot tell which path produced it.
                         if let Some(detail) = reason.as_deref() {
-                            attributes.push(("reason.detail".to_string(), detail.to_string()));
+                            attributes
+                                .push(("reason.detail".to_string(), bounded_reason_detail(detail)));
                         }
                         attributes
                     },
                 );
             }
-            NdjsonEvent::JobCompleted { status, .. } if status.is_terminal() => {
-                let conclusion = execution_conclusion(*status);
-                self.observability
-                    .metrics()
-                    .lifecycle
-                    .record_job_completed(conclusion, "completed");
-                self.observability.export_log(
-                    "INFO",
-                    "job.completed",
-                    vec![
-                        ("event.name".to_string(), "job.completed".to_string()),
-                        ("conclusion".to_string(), conclusion.to_string()),
-                    ],
-                );
-            }
+            // `NdjsonEvent::JobCompleted` has no constructor anywhere in the
+            // workspace — the terminal transition is reported as a terminal
+            // `JobStatus`, which the arm above records. Keeping a counter
+            // call here would make the record look double-sourced.
             _ => {}
         }
         // Capture the projection under the lock, then persist after releasing
@@ -1479,6 +1492,29 @@ mod termination_reason_tests {
         }
     }
 
+    #[test]
+    fn reason_detail_is_bounded_on_a_char_boundary() {
+        use super::bounded_reason_detail;
+        // 4-byte characters: 300 of them is 1200 bytes, way over the cap.
+        let long = "w".repeat(300);
+        let bounded = bounded_reason_detail(&long);
+        assert!(bounded.len() <= 512);
+        assert!(bounded.is_char_boundary(bounded.len()));
+        // Short prose passes through untouched.
+        assert_eq!(bounded_reason_detail("short"), "short");
+    }
+
+    #[test]
+    fn crafted_runs_on_label_cannot_flip_the_classification() {
+        // The starvation prose interpolates `runs-on` labels verbatim. A
+        // label containing the platform phrase must still classify as
+        // `no_runner` — the anchored prefix is checked first.
+        let starved = "no runner is registered for `runs-on: self-hosted, \
+                       runner is registered with this server` and none \
+                       appeared within 120s, so the job cannot be scheduled";
+        assert_eq!(bounded_termination_reason(starved), "no_runner");
+    }
+
     #[test]
     fn platform_and_starvation_do_not_collide() {
         let starved = "no runner is registered for `runs-on: self-hosted, Linux, ARM64` and none \