feat(runtime-api): serve running-work accounting at GET /v1/threads/running (#6180)

Background-capable clients (quit, backgrounding; codewhale-apps#573)
can enumerate owned in-flight work in one call: threads with queued
or in-progress turns, each with thread_id/model/title plus active
turn ids+statuses for targeting. One turns scan grouped by thread
(never per-thread scans, #3757); archive state ignored since
archiving has no quiescence gate.

New e2e test lists-then-clears 1/1; runtime_api suite 263/263
(excluding configured_model_api_tests, which stack-overflows in this
shared tree with and without this change). Docs in RUNTIME_API.md.

crates/tui/src/runtime_api.rs

@@ -1198,6 +1198,7 @@ pub fn build_router(state: RuntimeApiState) -> Router {
         .route("/v1/jobs", get(jobs::list_jobs))
         .route("/v1/threads", get(list_threads).post(create_thread))
         .route("/v1/threads/summary", get(list_threads_summary))
+        .route("/v1/threads/running", get(list_running_threads))
         .route("/v1/threads/{id}", get(get_thread).patch(update_thread))
         .route(
             "/v1/threads/{id}/jobs",
@@ -1746,6 +1747,19 @@ async fn list_threads(
     Ok(Json(threads))
 }
 
+/// Threads with queued or in-progress turns, for quit/background
+/// accounting (#6180). One call, no inference from latest-turn status.
+async fn list_running_threads(
+    State(state): State<RuntimeApiState>,
+) -> Result<Json<Vec<crate::runtime_threads::RunningThread>>, ApiError> {
+    let running = state
+        .runtime_threads
+        .running_threads()
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    Ok(Json(running))
+}
+
 async fn list_threads_summary(
     State(state): State<RuntimeApiState>,
     Query(query): Query<ThreadSummaryQuery>,

crates/tui/src/runtime_api/tests.rs

@@ -17422,3 +17422,83 @@ async fn agent_mail_cancel_withdraws_queued_mail() -> Result<()> {
     handle.abort();
     Ok(())
 }
+
+/// Running-work accounting (#6180): a watch-only client enumerates owned
+/// in-flight work in one call, with turn identity for targeting, and the
+/// listing clears when the work settles.
+#[tokio::test]
+async fn threads_running_lists_active_turns_and_clears_on_settle() -> Result<()> {
+    let Some((addr, manager, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let base = format!("http://{addr}");
+
+    let idle: serde_json::Value = client
+        .get(format!("{base}/v1/threads/running"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(idle, serde_json::json!([]));
+
+    let created: serde_json::Value = client
+        .post(format!("{base}/v1/threads"))
+        .json(&serde_json::json!({ "model": "test-model" }))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let thread_id = created["id"].as_str().context("missing thread id")?;
+
+    client
+        .post(format!("{base}/v1/threads/{thread_id}/turns"))
+        .json(&serde_json::json!({ "prompt": "do the thing" }))
+        .send()
+        .await?
+        .error_for_status()?;
+
+    // Pin the turn in-flight deterministically: a test server may settle a
+    // started turn on its own, which would make the listing below racy.
+    let mut started = manager
+        .test_store()
+        .list_turns_for_thread(thread_id)?
+        .pop()
+        .context("missing started turn")?;
+    started.status = crate::runtime_threads::RuntimeTurnStatus::InProgress;
+    manager.test_store().save_turn(&started)?;
+
+    let running: serde_json::Value = client
+        .get(format!("{base}/v1/threads/running"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(running.as_array().map(Vec::len), Some(1));
+    assert_eq!(running[0]["thread_id"], thread_id);
+    let active = running[0]["active_turns"]
+        .as_array()
+        .context("missing active_turns")?;
+    assert_eq!(active.len(), 1);
+    assert_eq!(active[0]["status"], "in_progress");
+    let turn_id = active[0]["turn_id"].as_str().context("missing turn id")?;
+
+    let mut turn = manager.test_store().load_turn(turn_id)?;
+    turn.status = crate::runtime_threads::RuntimeTurnStatus::Completed;
+    manager.test_store().save_turn(&turn)?;
+
+    let settled: serde_json::Value = client
+        .get(format!("{base}/v1/threads/running"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(settled, serde_json::json!([]));
+
+    handle.abort();
+    Ok(())
+}

crates/tui/src/runtime_threads.rs

@@ -3598,6 +3598,27 @@ pub struct UsageTotals {
     pub turns: u64,
 }
 
+/// One in-flight or queued turn, for running-work accounting. Served by
+/// `GET /v1/threads/running` so background-capable clients (quit,
+/// backgrounding) can enumerate owned work without inferring it from
+/// per-thread latest-turn status (#6180, codewhale-apps#573).
+#[derive(Debug, Clone, Serialize)]
+pub struct ActiveTurn {
+    pub turn_id: String,
+    pub status: RuntimeTurnStatus,
+}
+
+/// One thread with at least one [`ActiveTurn`]. Archive state is ignored:
+/// archiving is a plain flag with no quiescence gate, so an archived thread
+/// can still carry live work and quit-accounting must count it.
+#[derive(Debug, Clone, Serialize)]
+pub struct RunningThread {
+    pub thread_id: String,
+    pub model: String,
+    pub title: Option<String>,
+    pub active_turns: Vec<ActiveTurn>,
+}
+
 #[derive(Debug, Clone, Default, Serialize)]
 pub struct UsageBucket {
     pub key: String,
@@ -7203,6 +7224,43 @@ impl RuntimeThreadManager {
         Ok(threads)
     }
 
+    /// Threads with at least one queued or in-progress turn, for
+    /// running-work accounting (#6180). One turns scan grouped by thread
+    /// (never a scan per thread, #3757), joined against the thread rows in
+    /// store order. Archive state is ignored: see [`RunningThread`].
+    pub async fn running_threads(&self) -> Result<Vec<RunningThread>> {
+        let mut active_by_thread: std::collections::BTreeMap<String, Vec<ActiveTurn>> =
+            std::collections::BTreeMap::new();
+        for turn in self.store.list_all_turns()? {
+            if matches!(
+                turn.status,
+                RuntimeTurnStatus::Queued | RuntimeTurnStatus::InProgress
+            ) {
+                active_by_thread.entry(turn.thread_id.clone()).or_default().push(
+                    ActiveTurn {
+                        turn_id: turn.id.clone(),
+                        status: turn.status,
+                    },
+                );
+            }
+        }
+        if active_by_thread.is_empty() {
+            return Ok(Vec::new());
+        }
+        let mut out = Vec::with_capacity(active_by_thread.len());
+        for thread in self.store.list_threads()? {
+            if let Some(active_turns) = active_by_thread.remove(&thread.id) {
+                out.push(RunningThread {
+                    thread_id: thread.id.clone(),
+                    model: thread.model.clone(),
+                    title: thread.title.clone(),
+                    active_turns,
+                });
+            }
+        }
+        Ok(out)
+    }
+
     /// Whether `/v1/threads/summary?search=` should keep this thread.
     ///
     /// Matches fields already on the thread record (`id`, explicit `title`,

docs/RUNTIME_API.md

@@ -679,6 +679,7 @@ and live state comes only from a resumed thread's SSE stream.
 **Threads** (durable runtime data model)
 - `GET /v1/threads?limit=50&include_archived=false&archived_only=false`
 - `GET /v1/threads/summary?limit=50&search=<optional>&include_archived=false&archived_only=false`
+- `GET /v1/threads/running`
 - `POST /v1/threads`
 - `GET /v1/threads/{id}`
 - `PATCH /v1/threads/{id}` (see body shape below)
@@ -745,6 +746,13 @@ Thread list and summary responses remain flat in v0.8.40, so clients that need
 a graph should reconstruct it from events instead of assuming list order is a
 complete tree.
 
+`GET /v1/threads/running` is the running-work accounting surface
+(#6180): threads with at least one queued or in-progress turn, each with
+`thread_id`, `model`, `title`, and `active_turns` (`turn_id` + `status`).
+Background-capable clients use it for quit/background decisions — one call,
+no inference from latest-turn status. Archive state is ignored (archiving
+has no quiescence gate); an empty array means no owned work is live.
+
 `archived_only=true` returns archived threads only (mutually overrides
 `include_archived`). Default behavior is unchanged: `include_archived=false`
 and `archived_only=false` returns active threads. Added in v0.8.10 (#563).