wip(agent): P0进度快照(会话历史/AskUser/取消)

migrations/20260623000001_agent_session_transcript.sql

@@ -0,0 +1,9 @@
+-- Agent Workbench conversation history.
+--
+-- Adds a `transcript` column holding the full multi-turn message history of an
+-- agent session as a JSON array (the SDK's `Vec<Message>`, serde-serialized).
+-- Each `/api/agent/chat` round-trip loads this transcript into the agent before
+-- the turn and writes the updated transcript back when the turn completes, so a
+-- session resumed by id sees everything that happened in prior turns. NULL/empty
+-- means a fresh session with no history yet.
+ALTER TABLE agent_sessions ADD COLUMN transcript TEXT;

src/routes/agent.rs

@@ -24,6 +24,7 @@
 use crate::auth::jwt::validate_jwt;
 use crate::services::agent_workbench::{
     approval::{self, ApprovalOutcome, SettledState},
+    ask::{self, AskOutcome},
     bash_sandbox::SandboxedBashTool,
     permission::{classify_tool_risk, decide_permission, PermissionAction, PermissionMode},
     sandbox,
@@ -39,12 +40,14 @@ use axum::{
 };
 use futures::stream::Stream;
 use open_agent_sdk::api::openai::OpenAIProvider;
-use open_agent_sdk::types::{ContentBlock, PermissionDecision};
+use open_agent_sdk::types::{ContentBlock, Message, PermissionDecision};
-use open_agent_sdk::{Agent, AgentOptions, LLMProvider, SDKMessage};
+use open_agent_sdk::{Agent, AgentOptions, CancellationToken, LLMProvider, SDKMessage};
 use serde::Deserialize;
 use serde_json::{json, Value};
 use std::convert::Infallible;
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 use tokio_stream::wrappers::UnboundedReceiverStream;
@@ -54,6 +57,8 @@ use uuid::Uuid;
 const MAX_TURNS: u32 = 20;
 /// How long a tool approval waits for the user before auto-denying.
 const APPROVAL_TIMEOUT: Duration = Duration::from_secs(300);
+/// How long an AskUserQuestion waits for the user before giving up.
+const ASK_TIMEOUT: Duration = Duration::from_secs(300);
 /// Body limit for a chat request (images/attachments can be large).
 const AGENT_CHAT_BODY_LIMIT_BYTES: usize = 16 * 1024 * 1024;
 
@@ -67,6 +72,7 @@ pub fn routes() -> Router<AppState> {
     Router::new()
         .route("/api/agent/chat", post(chat_handler))
         .route("/api/agent/approve", post(approve_handler))
+        .route("/api/agent/answer", post(answer_handler))
         .route("/api/agent/status", get(status_handler))
         .layer(DefaultBodyLimit::max(AGENT_CHAT_BODY_LIMIT_BYTES))
 }
@@ -92,6 +98,13 @@ struct ApproveRequest {
     approved: bool,
 }
 
+#[derive(Debug, Deserialize)]
+struct AnswerRequest {
+    request_id: String,
+    /// The user's free-text reply to an `AskUserQuestion` prompt.
+    answer: String,
+}
+
 /// Resolved caller identity for an agent request.
 #[derive(Debug, Clone)]
 struct AgentAuth {
@@ -237,6 +250,53 @@ fn build_can_use_tool(
     })
 }
 
+/// Build the SDK `ask_fn` for the `AskUserQuestion` tool. When the agent needs
+/// the user to clarify something, this emits an `ask` SSE event and blocks the
+/// tool call until the user answers via `POST /api/agent/answer` (or the wait
+/// times out, which surfaces a cancellation back to the agent). Reuses the same
+/// SSE round-trip pattern as the approval flow.
+fn build_ask_fn(
+    sse_tx: UnboundedSender<Result<Event, Infallible>>,
+) -> open_agent_sdk::tools::askuser::AskUserFn {
+    use open_agent_sdk::tools::askuser::AskUserRequest;
+    Arc::new(move |req: AskUserRequest| {
+        let sse_tx = sse_tx.clone();
+        Box::pin(async move {
+            let request_id = Uuid::new_v4().to_string();
+            // `_guard` removes the registry entry when this future ends (normal
+            // completion or drop on session abort) — no leak.
+            let (entry, _guard) = ask::register(&request_id);
+            send_sse(
+                &sse_tx,
+                json!({
+                    "type": "ask",
+                    "data": {
+                        "request_id": request_id,
+                        "question": req.question,
+                        "options": req.options,
+                    }
+                }),
+            );
+
+            let outcome = ask::wait_for_answer(&entry, ASK_TIMEOUT).await;
+            // Tell the client the prompt is resolved so it can dismiss the modal.
+            send_sse(
+                &sse_tx,
+                json!({
+                    "type": "ask_resolved",
+                    "data": { "request_id": request_id }
+                }),
+            );
+            match outcome {
+                AskOutcome::Answered(text) => Ok(text),
+                AskOutcome::Cancelled => {
+                    Err("User did not answer the question in time".to_string())
+                }
+            }
+        })
+    })
+}
+
 #[allow(clippy::type_complexity)]
 async fn chat_handler(
     State(state): State<AppState>,
@@ -259,6 +319,18 @@ async fn chat_handler(
         .filter(|s| !s.is_empty())
         .map(ToOwned::to_owned)
         .unwrap_or_else(|| Uuid::new_v4().to_string());
+    // Refuse to resume a session owned by another tenant (which would leak/
+    // overwrite their transcript); fall back to a fresh id instead of erroring.
+    let session_id = match session_store::session_is_claimable(&state.db, &session_id, auth.user_id)
+        .await
+    {
+        Ok(true) => session_id,
+        Ok(false) => Uuid::new_v4().to_string(),
+        Err(e) => {
+            tracing::warn!(%e, "failed to check agent session ownership");
+            Uuid::new_v4().to_string()
+        }
+    };
     let mode = PermissionMode::from_wire(body.permission_mode.as_deref().unwrap_or("default"));
     let model = body
         .model
@@ -292,6 +364,26 @@ async fn chat_handler(
         tracing::warn!(%e, "failed to ensure agent session row");
     }
 
+    // Load any prior conversation for this session so the agent sees the full
+    // multi-turn history (scoped to the caller — never another tenant's).
+    let history: Vec<Message> = match session_store::load_transcript(
+        &state.db,
+        &session_id,
+        auth.user_id,
+    )
+    .await
+    {
+        Ok(Some(raw)) => serde_json::from_str(&raw).unwrap_or_else(|e| {
+            tracing::warn!(%e, "failed to parse agent transcript; starting fresh");
+            Vec::new()
+        }),
+        Ok(None) => Vec::new(),
+        Err(e) => {
+            tracing::warn!(%e, "failed to load agent transcript");
+            Vec::new()
+        }
+    };
+
     // Route the agent's LLM calls back through BoosAPI's own aggregation using
     // the built-in downstream key.
     let base_url = boosapi_base(&state);
@@ -299,9 +391,16 @@ async fn chat_handler(
 
     let (tx, rx) = unbounded_channel::<Result<Event, Infallible>>();
 
+    // Cancellation: the token is handed to the agent loop (so a long LLM call or
+    // tool can be interrupted mid-flight); a guard wrapping the SSE stream fires
+    // it when the client disconnects (stream dropped). See `CancelGuard`.
+    let cancel = CancellationToken::new();
+
     let db = state.db.clone();
+    let user_id = auth.user_id;
     let prompt = body.message;
     let session_for_event = session_id.clone();
+    let agent_cancel = cancel.clone();
 
     tokio::spawn(async move {
         // First event: hand the (possibly newly minted) session id to the client.
@@ -318,6 +417,7 @@ async fn chat_handler(
         ));
 
         let can_use_tool = build_can_use_tool(tx.clone(), cwd.clone(), mode);
+        let ask_fn = build_ask_fn(tx.clone());
 
         // Replace the SDK's built-in Bash (a bare subprocess) with an
         // OS-sandboxed one confined to this session's directory. Registered as
@@ -333,6 +433,8 @@ async fn chat_handler(
             system_prompt: Some(SYSTEM_PROMPT.to_string()),
             max_turns: Some(MAX_TURNS),
             can_use_tool: Some(can_use_tool),
+            ask_fn: Some(ask_fn),
+            abort_signal: Some(agent_cancel),
             custom_tools: vec![sandboxed_bash],
             session_id: Some(session_id.clone()),
             ..Default::default()
@@ -348,22 +450,84 @@ async fn chat_handler(
             }
         };
 
+        // Seed the agent with the prior conversation so this turn continues it.
+        agent.messages = history;
+
         let (mut rx_msg, handle) = agent.query(&prompt).await;
 
+        // The SDK's Result event carries the full updated message history; we
+        // capture it here and persist it after the stream drains so the next
+        // turn (same session id) resumes from it.
+        let mut latest_transcript: Option<Vec<Message>> = None;
+
         while let Some(event) = rx_msg.recv().await {
             // Stop early if the client disconnected.
             if tx.is_closed() {
                 break;
             }
+            if let SDKMessage::Result { messages, .. } = &event {
+                latest_transcript = Some(messages.clone());
+            }
             forward_sdk_event(&tx, &db, &session_id, event).await;
         }
         let _ = handle.await;
         agent.close().await;
 
+        // Persist the updated transcript (best-effort). Skipped if the turn was
+        // cancelled before producing a Result, leaving the prior history intact.
+        if let Some(messages) = latest_transcript {
+            match serde_json::to_string(&messages) {
+                Ok(raw) => {
+                    if let Err(e) =
+                        session_store::save_transcript(&db, &session_id, user_id, &raw).await
+                    {
+                        tracing::warn!(%e, "failed to save agent transcript");
+                    }
+                }
+                Err(e) => tracing::warn!(%e, "failed to serialize agent transcript"),
+            }
+        }
+
         send_sse(&tx, json!({ "type": "done" }));
     });
 
-    Ok(Sse::new(UnboundedReceiverStream::new(rx)))
+    // Wrap the stream so dropping it (client disconnect) cancels the agent loop.
+    Ok(Sse::new(CancelGuard::new(
+        UnboundedReceiverStream::new(rx),
+        cancel,
+    )))
+}
+
+/// Wraps the SSE body stream and cancels a [`CancellationToken`] when the stream
+/// is dropped — i.e. when the client disconnects and Axum tears down the
+/// response. This propagates cancellation into the agent loop's `abort_signal`,
+/// so a long-running LLM call or tool execution is interrupted promptly instead
+/// of running to completion on a server nobody is listening to.
+struct CancelGuard<S> {
+    inner: S,
+    cancel: CancellationToken,
+}
+
+impl<S> CancelGuard<S> {
+    fn new(inner: S, cancel: CancellationToken) -> Self {
+        Self { inner, cancel }
+    }
+}
+
+impl<S: Stream + Unpin> Stream for CancelGuard<S> {
+    type Item = S::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.inner).poll_next(cx)
+    }
+}
+
+impl<S> Drop for CancelGuard<S> {
+    fn drop(&mut self) {
+        // Idempotent: a normal end-of-stream also drops this, harmlessly
+        // cancelling an already-finished loop.
+        self.cancel.cancel();
+    }
 }
 
 /// Translate one SDK message into SSE event(s), and persist usage on Result.
@@ -448,6 +612,18 @@ async fn approve_handler(
     Ok(Json(json!({ "ok": delivered })))
 }
 
+async fn answer_handler(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(body): Json<AnswerRequest>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    // Auth required; the pending question is keyed by an unguessable uuid, so any
+    // authenticated caller may answer their own running session's prompt.
+    resolve_auth(&headers, &state).await?;
+    let delivered = ask::answer(&body.request_id, body.answer);
+    Ok(Json(json!({ "ok": delivered })))
+}
+
 async fn status_handler(
     State(state): State<AppState>,
     headers: HeaderMap,

src/routes/auth.rs

@@ -1047,6 +1047,7 @@ mod tests {
                     template_proxy_url: None,
                 },
                 comfyui: Default::default(),
+                pixelle_video: Default::default(),
                 payment: Default::default(),
                 jimeng: Default::default(),
             },

src/services/agent_workbench/ask.rs

@@ -0,0 +1,207 @@
+//! Pending-question registry for the Agent Workbench's `AskUserQuestion` tool.
+//!
+//! Mirrors [`super::approval`] but carries a free-text answer instead of a
+//! boolean decision. When the agent calls `AskUserQuestion`, the SDK invokes the
+//! configured `ask_fn`, which registers a pending question, emits an `ask`
+//! SSE event, and awaits the user's reply (or a timeout). The
+//! `POST /api/agent/answer` endpoint resolves the pending entry with the user's
+//! text.
+//!
+//! ## Single source of truth
+//!
+//! Each entry carries an [`AtomicU8`] state machine (`Pending → Answered /
+//! Cancelled`) plus a [`Notify`] and a slot holding the answer. The answer is
+//! published *before* the state transitions to `Answered`, so any waiter that
+//! observes `Answered` is guaranteed to read the answer. Exactly one
+//! `compare_exchange` from `Pending` wins; later answers observe a losing CAS
+//! and no-op, so the user can never answer the same question twice.
+
+use once_cell::sync::Lazy;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::{Arc, Mutex};
+use tokio::sync::Notify;
+
+const PENDING: u8 = 0;
+const ANSWERED: u8 = 1;
+const CANCELLED: u8 = 2;
+
+/// Settled state of a pending question.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AskOutcome {
+    /// The user replied with this text.
+    Answered(String),
+    /// No answer arrived before the timeout (or the session ended).
+    Cancelled,
+}
+
+/// Shared question entry: an atomic state machine, the answer slot, and a wakeup.
+#[derive(Debug)]
+pub struct AskEntry {
+    state: AtomicU8,
+    answer: Mutex<Option<String>>,
+    notify: Notify,
+}
+
+impl AskEntry {
+    fn new() -> Self {
+        Self {
+            state: AtomicU8::new(PENDING),
+            answer: Mutex::new(None),
+            notify: Notify::new(),
+        }
+    }
+
+    fn outcome(&self) -> Option<AskOutcome> {
+        match self.state.load(Ordering::SeqCst) {
+            ANSWERED => Some(AskOutcome::Answered(
+                self.answer.lock().unwrap().clone().unwrap_or_default(),
+            )),
+            CANCELLED => Some(AskOutcome::Cancelled),
+            _ => None,
+        }
+    }
+}
+
+static REGISTRY: Lazy<Mutex<HashMap<String, Arc<AskEntry>>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));
+
+/// RAII guard that removes its entry from the registry when dropped, so a
+/// pending question is cleaned up whether `ask_fn` finishes normally or its
+/// future is dropped (session aborted / client disconnected). No leak on any
+/// path.
+pub struct PendingGuard {
+    request_id: String,
+}
+
+impl Drop for PendingGuard {
+    fn drop(&mut self) {
+        REGISTRY.lock().unwrap().remove(&self.request_id);
+    }
+}
+
+/// Register a new pending question. Returns the shared entry (await
+/// [`wait_for_answer`]) plus a [`PendingGuard`] the caller must hold for the
+/// lifetime of the wait.
+pub fn register(request_id: &str) -> (Arc<AskEntry>, PendingGuard) {
+    let entry = Arc::new(AskEntry::new());
+    REGISTRY
+        .lock()
+        .unwrap()
+        .insert(request_id.to_string(), entry.clone());
+    (
+        entry,
+        PendingGuard {
+            request_id: request_id.to_string(),
+        },
+    )
+}
+
+/// Wait until the entry is answered or `timeout` elapses. On timeout this claims
+/// `Cancelled` (if still pending) so the result is always terminal.
+pub async fn wait_for_answer(
+    entry: &AskEntry,
+    timeout: std::time::Duration,
+) -> AskOutcome {
+    // Arm the notified future before checking state to avoid a lost wakeup.
+    let notified = entry.notify.notified();
+
+    if let Some(o) = entry.outcome() {
+        return o;
+    }
+
+    if tokio::time::timeout(timeout, notified).await.is_err() {
+        // Timer fired first — claim Cancelled if nobody answered meanwhile.
+        let _ = entry
+            .state
+            .compare_exchange(PENDING, CANCELLED, Ordering::SeqCst, Ordering::SeqCst);
+    }
+
+    entry.outcome().unwrap_or(AskOutcome::Cancelled)
+}
+
+/// Deliver the user's answer to a pending question.
+///
+/// Returns `true` iff this call's CAS won — i.e. it is the call that settled the
+/// question. Returns `false` for an unknown id (already cleaned up) or when the
+/// timeout / another answer already claimed the entry.
+pub fn answer(request_id: &str, text: String) -> bool {
+    let entry = REGISTRY.lock().unwrap().get(request_id).cloned();
+    let Some(entry) = entry else {
+        return false;
+    };
+    // Publish the answer *before* flipping state so a waiter that sees ANSWERED
+    // always reads the value.
+    *entry.answer.lock().unwrap() = Some(text);
+    if entry
+        .state
+        .compare_exchange(PENDING, ANSWERED, Ordering::SeqCst, Ordering::SeqCst)
+        .is_ok()
+    {
+        entry.notify.notify_one();
+        true
+    } else {
+        false
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn answer_before_wait_is_observed() {
+        let (entry, _guard) = register("q1");
+        assert!(answer("q1", "hello".to_string()));
+        assert_eq!(
+            wait_for_answer(&entry, Duration::from_secs(5)).await,
+            AskOutcome::Answered("hello".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn answer_wakes_waiter() {
+        let (entry, _guard) = register("q2");
+        let id = "q2".to_string();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            answer(&id, "later".to_string());
+        });
+        assert_eq!(
+            wait_for_answer(&entry, Duration::from_secs(5)).await,
+            AskOutcome::Answered("later".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn timeout_cancels_and_late_answer_loses() {
+        let (entry, _guard) = register("q3");
+        assert_eq!(
+            wait_for_answer(&entry, Duration::from_millis(20)).await,
+            AskOutcome::Cancelled
+        );
+        // A late answer must report false: the CAS already went to Cancelled.
+        assert!(!answer("q3", "too late".to_string()));
+    }
+
+    #[tokio::test]
+    async fn first_answer_wins_second_loses() {
+        let (_entry, _guard) = register("q4");
+        assert!(answer("q4", "first".to_string()));
+        assert!(!answer("q4", "second".to_string()));
+    }
+
+    #[tokio::test]
+    async fn answer_unknown_id_is_false() {
+        assert!(!answer("nope", "x".to_string()));
+    }
+
+    #[tokio::test]
+    async fn guard_drop_removes_entry() {
+        {
+            let (_entry, _guard) = register("q5");
+        }
+        assert!(!answer("q5", "x".to_string()));
+    }
+}

src/services/agent_workbench/mod.rs

@@ -7,12 +7,14 @@
 //! - [`sandbox`]: per-session isolated working directory + path validation.
 //! - [`bash_sandbox`]: OS-level (bubblewrap) confinement for the Bash tool.
 //! - [`approval`]: process-global registry bridging the SSE approval round-trip.
+//! - [`ask`]: process-global registry bridging the SSE AskUserQuestion round-trip.
 //! - [`session_store`]: persistence of per-session token/cost accounting.
 //!
 //! The HTTP surface (SSE chat, approval callback, status) lives in
 //! `crate::routes::agent`.
 
 pub mod approval;
+pub mod ask;
 pub mod bash_sandbox;
 pub mod permission;
 pub mod sandbox;

src/services/agent_workbench/session_store.rs

@@ -93,6 +93,70 @@ pub async fn record_turn(
     Ok(())
 }
 
+/// Whether `user_id` may use `session_id` for a new turn.
+///
+/// Returns `Ok(true)` when the id is free (no row yet) or already owned by this
+/// caller, and `Ok(false)` only when a row exists under a *different* owner —
+/// so a caller can never resume (and thus read/overwrite) another tenant's
+/// session by guessing its id. Admin callers pass `None`, which owns the
+/// NULL-`user_id` rows.
+pub async fn session_is_claimable(
+    db: &SqlitePool,
+    session_id: &str,
+    user_id: Option<i64>,
+) -> AppResult<bool> {
+    let row: Option<(Option<i64>,)> =
+        sqlx::query_as("SELECT user_id FROM agent_sessions WHERE id = ?")
+            .bind(session_id)
+            .fetch_optional(db)
+            .await
+            .map_err(AppError::Database)?;
+    Ok(match row {
+        None => true,                       // free — first use of this id
+        Some((owner,)) => owner == user_id, // already ours?
+    })
+}
+
+/// Load the raw transcript JSON for a session, scoped to `user_id` so a caller
+/// can never read another tenant's conversation. `Ok(None)` when the session is
+/// new, owned by someone else, or has no transcript yet — all of which mean
+/// "start this turn with empty history".
+pub async fn load_transcript(
+    db: &SqlitePool,
+    session_id: &str,
+    user_id: Option<i64>,
+) -> AppResult<Option<String>> {
+    let row: Option<(Option<String>,)> =
+        sqlx::query_as("SELECT transcript FROM agent_sessions WHERE id = ? AND user_id IS ?")
+            .bind(session_id)
+            .bind(user_id)
+            .fetch_optional(db)
+            .await
+            .map_err(AppError::Database)?;
+    // Outer None: no such (owned) session. Inner None / empty: no transcript.
+    Ok(row
+        .and_then(|(t,)| t)
+        .filter(|s| !s.trim().is_empty()))
+}
+
+/// Persist the full transcript JSON for a session, scoped to `user_id` so a
+/// caller can only overwrite their own session's history.
+pub async fn save_transcript(
+    db: &SqlitePool,
+    session_id: &str,
+    user_id: Option<i64>,
+    transcript: &str,
+) -> AppResult<()> {
+    sqlx::query("UPDATE agent_sessions SET transcript = ? WHERE id = ? AND user_id IS ?")
+        .bind(transcript)
+        .bind(session_id)
+        .bind(user_id)
+        .execute(db)
+        .await
+        .map_err(AppError::Database)?;
+    Ok(())
+}
+
 /// Fetch a single session by id (scoped to `user_id` when provided, so a user
 /// cannot read another tenant's session). Admin callers pass `None`.
 pub async fn get_session(
@@ -138,6 +202,12 @@ mod tests {
         .execute(&pool)
         .await
         .unwrap();
+        sqlx::query(include_str!(
+            "../../../migrations/20260623000001_agent_session_transcript.sql"
+        ))
+        .execute(&pool)
+        .await
+        .unwrap();
         pool
     }
 
@@ -198,4 +268,54 @@ mod tests {
         // Owner can.
         assert!(get_session(&db, "s3", Some(1)).await.unwrap().is_some());
     }
+
+    #[tokio::test]
+    async fn transcript_round_trips_for_owner() {
+        let db = pool().await;
+        ensure_session(&db, "t1", Some(5), "m", "default").await.unwrap();
+        // Fresh session has no transcript.
+        assert!(load_transcript(&db, "t1", Some(5)).await.unwrap().is_none());
+
+        save_transcript(&db, "t1", Some(5), r#"[{"role":"user"}]"#)
+            .await
+            .unwrap();
+        assert_eq!(
+            load_transcript(&db, "t1", Some(5)).await.unwrap().as_deref(),
+            Some(r#"[{"role":"user"}]"#)
+        );
+    }
+
+    #[tokio::test]
+    async fn transcript_is_user_scoped() {
+        let db = pool().await;
+        ensure_session(&db, "t2", Some(1), "m", "default").await.unwrap();
+        save_transcript(&db, "t2", Some(1), "history").await.unwrap();
+
+        // Another tenant can neither read nor overwrite it.
+        assert!(load_transcript(&db, "t2", Some(2)).await.unwrap().is_none());
+        save_transcript(&db, "t2", Some(2), "evil").await.unwrap();
+        assert_eq!(
+            load_transcript(&db, "t2", Some(1)).await.unwrap().as_deref(),
+            Some("history")
+        );
+    }
+
+    #[tokio::test]
+    async fn claimable_is_true_when_free_or_owned() {
+        let db = pool().await;
+        // Free id is claimable by anyone.
+        assert!(session_is_claimable(&db, "c1", Some(9)).await.unwrap());
+
+        ensure_session(&db, "c1", Some(9), "m", "default").await.unwrap();
+        // Owner can re-claim.
+        assert!(session_is_claimable(&db, "c1", Some(9)).await.unwrap());
+        // Another tenant cannot.
+        assert!(!session_is_claimable(&db, "c1", Some(8)).await.unwrap());
+        assert!(!session_is_claimable(&db, "c1", None).await.unwrap());
+
+        ensure_session(&db, "c2", None, "m", "default").await.unwrap();
+        // Admin (None) owns NULL-user rows.
+        assert!(session_is_claimable(&db, "c2", None).await.unwrap());
+        assert!(!session_is_claimable(&db, "c2", Some(9)).await.unwrap());
+    }
 }