diff --git a/.changeset/session-storage-blip-not-found.md b/.changeset/session-storage-blip-not-found.md new file mode 100644 index 00000000..229bff10 --- /dev/null +++ b/.changeset/session-storage-blip-not-found.md @@ -0,0 +1,45 @@ +--- +"@smooai/smooth-operator": patch +--- + +fix(server): a storage blip is no longer reported as `session '' not found` + +`AppState::load_session` hydrates a session from storage when the local +per-pod registry misses (th-ca579c) — the normal path for a returning visitor +whose WebSocket lands on a pod that has never seen their session. That read +collapsed `Err` into `None`, so a transient Postgres failure was +indistinguishable from a session that genuinely does not exist. Every caller +renders `None` as `session '' not found`, so a backend hiccup told a live +visitor on smoo.ai, in the chat bubble, that their conversation was gone. Seen +in production. + +`load_session` now returns `anyhow::Result>` and +`handler::scoped_session` propagates it. The three outcomes are distinct at the +user-visible boundary: + +- `Ok(Some(session))` — unchanged. +- `Ok(None)` — not found, **or** not yours: still the identical + `SESSION_NOT_FOUND` / `NO_PENDING_*` event, byte for byte, so there is no + existence oracle to enumerate other users' session ids with. +- `Err(_)` — storage could not answer: a retryable `STORAGE_ERROR` ("session + lookup is temporarily unavailable, please try again"), which is not an + existence claim and leaks nothing (a storage failure is independent of + whether the id is real or ours). The underlying error is logged server-side, + not sent to the client. + +All six session-id-taking actions route through the one chokepoint and switch +together: `get_session`, `get_conversation_messages`, `send_message` and +`verify_otp` (previously `SESSION_NOT_FOUND`), plus `confirm_tool_action` and +`submit_interaction` (previously `NO_PENDING_CONFIRMATION` / +`NO_PENDING_INTERACTION`, which for a parked turn was equally wrong — the park +is still there, and a retry still resolves it). + +`rust/smooth-operator-server/tests/session_storage_blip.rs` drives the real +dispatcher against a storage adapter whose `get_session` fails on demand and +pins both halves: a blip is never rendered as not-found on any of the six +actions, and a genuinely unknown id still is (a fix that made everything +retryable would leave clients retrying an id that will never resolve). + +**Host-facing API change**: `AppState::load_session` returns +`anyhow::Result>` instead of `Option` — hosts calling +it directly add a `?` or a `match`. diff --git a/rust/smooth-operator-server/src/handler.rs b/rust/smooth-operator-server/src/handler.rs index 068356ea..3d286c2f 100644 --- a/rust/smooth-operator-server/src/handler.rs +++ b/rust/smooth-operator-server/src/handler.rs @@ -163,11 +163,17 @@ async fn may_read_conversation( /// The **only** way a handler may turn a client-supplied `sessionId` into a /// session. It loads the session and then hides it unless the connection's -/// authenticated principal owns its conversation — returning `None`, exactly +/// authenticated principal owns its conversation — returning `Ok(None)`, exactly /// what an unknown session id returns, so every caller emits the identical /// not-found event and no caller can distinguish "not yours" from "never /// existed". /// +/// `Err` is the third outcome and is NOT an existence claim: storage could not +/// answer. Callers emit `STORAGE_ERROR` (retryable) for it instead of +/// not-found — telling a visitor their live session does not exist because +/// Postgres hiccuped is a lie the UI has no way to walk back. It leaks nothing: +/// a storage failure is independent of whether the id is real or ours. +/// /// Every sessionId-taking handler routes through here rather than calling /// [`AppState::get_session`] directly: the check lives once, at the chokepoint, /// instead of being re-derived — and forgotten — per handler. `get_session`, @@ -180,19 +186,36 @@ async fn scoped_session( session_id: &str, auth_org: Option<&str>, scope: &UserScope, -) -> Option { +) -> anyhow::Result> { // th-ca579c: hydrate from storage on a local miss. `get_session` here would // report "not found" for a session this pod simply has not seen — which with // 2+ replicas is most returning visitors. - let session = state.load_session(session_id).await?; + let Some(session) = state.load_session(session_id).await? else { + return Ok(None); + }; // The session carries its own org, so the tenant check needs no extra read // and covers every session-id-taking handler at this one chokepoint. if !same_org(auth_org, &session.organization_id) { - return None; + return Ok(None); } - may_read_conversation(state, &session.conversation_id, auth_org, scope) - .await - .then_some(session) + Ok( + may_read_conversation(state, &session.conversation_id, auth_org, scope) + .await + .then_some(session), + ) +} + +/// The `error` event for a session read that storage could not answer. Distinct +/// from the not-found event on purpose: this one says "try again", and the code +/// (`STORAGE_ERROR`, already used by the rename path) is what a client keys on +/// to retry rather than to clear its session. +fn session_storage_error(request_id: Option<&str>, session_id: &str, e: &anyhow::Error) -> Value { + tracing::warn!(error = %e, session_id, "session lookup unavailable"); + protocol::error( + request_id, + "STORAGE_ERROR", + "session lookup is temporarily unavailable, please try again", + ) } /// A spawned agent turn: its task handle, plus the flag the cancel path raises to @@ -757,7 +780,7 @@ async fn handle_get_session( }; match scoped_session(state, session_id, auth_org, scope).await { - Some(s) => { + Ok(Some(s)) => { let data = json!({ "sessionId": s.session_id, "conversationId": s.conversation_id, @@ -776,13 +799,16 @@ async fn handle_get_session( request_id, 200, "Session", data, )); } - None => { + Ok(None) => { let _ = sink.send(protocol::error( request_id, "SESSION_NOT_FOUND", &format!("session '{session_id}' not found"), )); } + Err(e) => { + let _ = sink.send(session_storage_error(request_id, session_id, &e)); + } } } @@ -814,13 +840,20 @@ async fn handle_get_conversation_messages( // same message, same shape. A distinct "forbidden" would be an existence // oracle: it would tell a caller which session ids are real, which is all // an attacker needs to enumerate other users' conversations. - let Some(session) = scoped_session(state, session_id, auth_org, scope).await else { - let _ = sink.send(protocol::error( - request_id, - "SESSION_NOT_FOUND", - &format!("session '{session_id}' not found"), - )); - return; + let session = match scoped_session(state, session_id, auth_org, scope).await { + Ok(Some(session)) => session, + Ok(None) => { + let _ = sink.send(protocol::error( + request_id, + "SESSION_NOT_FOUND", + &format!("session '{session_id}' not found"), + )); + return; + } + Err(e) => { + let _ = sink.send(session_storage_error(request_id, session_id, &e)); + return; + } }; const DEFAULT_LIMIT: usize = 50; @@ -1495,13 +1528,20 @@ async fn handle_send_message( // sending into another user's session would replay their history as context // and stream the reply back to the sender, so an unscoped write here is also // a read of their conversation. - let Some(session) = scoped_session(state, session_id, auth_org, scope).await else { - let _ = sink.send(protocol::error( - Some(request_id), - "SESSION_NOT_FOUND", - &format!("session '{session_id}' not found"), - )); - return None; + let session = match scoped_session(state, session_id, auth_org, scope).await { + Ok(Some(session)) => session, + Ok(None) => { + let _ = sink.send(protocol::error( + Some(request_id), + "SESSION_NOT_FOUND", + &format!("session '{session_id}' not found"), + )); + return None; + } + Err(e) => { + let _ = sink.send(session_storage_error(Some(request_id), session_id, &e)); + return None; + } }; // A test-injected provider (the scenario-parity corpus's `MockLlmClient`) @@ -2055,9 +2095,15 @@ async fn handle_confirm_tool_action( // Approving a write parked in ANOTHER user's turn is the same class of hole // as writing into their session. A session we may not read is reported with // the identical event an id with no pending confirmation produces. - let owned = scoped_session(state, session_id, auth_org, scope) - .await - .is_some(); + let owned = match scoped_session(state, session_id, auth_org, scope).await { + Ok(session) => session.is_some(), + Err(e) => { + // Not "no such pending confirmation" — we could not find out. The + // parked turn is still waiting; a retry can still approve it. + let _ = sink.send(session_storage_error(request_id, session_id, &e)); + return; + } + }; let Some(responder) = owned.then(|| state.take_confirmation(session_id)).flatten() else { let _ = sink.send(protocol::error( request_id, @@ -2281,9 +2327,15 @@ async fn handle_submit_interaction( // read reports the identical event an id with no pending park produces (the // submitted values would otherwise land in another user's turn, and its // identity-attach effect on their session). - let owned = scoped_session(state, session_id, auth_org, scope) - .await - .is_some(); + let owned = match scoped_session(state, session_id, auth_org, scope).await { + Ok(session) => session.is_some(), + Err(e) => { + // The park is untouched (this path only peeks), so a retry after the + // blip still resolves the same interaction. + let _ = sink.send(session_storage_error(Some(request_id), session_id, &e)); + return; + } + }; let Some(pending) = owned .then(|| state.pending_interaction(session_id)) .flatten() @@ -2500,16 +2552,22 @@ async fn handle_verify_otp( // The session must exist AND be ours (a code can't verify — or brute-force — // a session we don't track, nor one belonging to another user). - if scoped_session(state, session_id, auth_org, scope) - .await - .is_none() - { - let _ = sink.send(protocol::error( - Some(request_id), - "SESSION_NOT_FOUND", - &format!("session '{session_id}' not found"), - )); - return; + match scoped_session(state, session_id, auth_org, scope).await { + Ok(Some(_)) => {} + Ok(None) => { + let _ = sink.send(protocol::error( + Some(request_id), + "SESSION_NOT_FOUND", + &format!("session '{session_id}' not found"), + )); + return; + } + Err(e) => { + // Fail closed on the gate — no code is checked — but say why, so the + // caller retries instead of abandoning a verification in progress. + let _ = sink.send(session_storage_error(Some(request_id), session_id, &e)); + return; + } } // No host OTP service → verification is impossible. Fail closed on the diff --git a/rust/smooth-operator-server/src/state.rs b/rust/smooth-operator-server/src/state.rs index fb7b88bb..63107967 100644 --- a/rust/smooth-operator-server/src/state.rs +++ b/rust/smooth-operator-server/src/state.rs @@ -517,22 +517,24 @@ impl AppState { /// [`session_supports`](Self::session_supports)) working untouched: every /// frame that needs them passes the ownership check first, and that check is /// what calls this. - pub async fn load_session(&self, session_id: &str) -> Option { + /// `Ok(None)` is an existence claim — this session does not exist. `Err` is + /// NOT: a storage failure means the question could not be answered, and the + /// caller must surface it as retryable rather than telling a human their + /// session is gone. The two used to collapse into `None`, so a Postgres blip + /// put `session '' not found` in a live visitor's chat bubble. + pub async fn load_session(&self, session_id: &str) -> anyhow::Result> { if let Some(session) = self.get_session(session_id) { - return Some(session); + return Ok(Some(session)); } match self.storage.get_session(session_id).await { Ok(Some(session)) => { self.insert_session(session.clone()); - Some(session) + Ok(Some(session)) } - Ok(None) => None, + Ok(None) => Ok(None), Err(e) => { - // A storage failure is NOT "no such session". Saying so would - // turn a blip into an existence claim, and the caller renders - // that to a human as a not-found error. tracing::warn!(error = %e, session_id, "session lookup failed against storage"); - None + Err(e) } } } @@ -1010,7 +1012,7 @@ mod tests { "premise broken: pod B must not have it locally" ); - let loaded = pod_b.load_session("s-shared").await; + let loaded = pod_b.load_session("s-shared").await.expect("storage ok"); assert!( loaded.is_some(), "pod B must hydrate the session from storage" @@ -1030,7 +1032,11 @@ mod tests { #[tokio::test] async fn hydration_does_not_conjure_an_unknown_session() { let state = state_with(config_with_env_key(None)); - assert!(state.load_session("s-nope").await.is_none()); + assert!(state + .load_session("s-nope") + .await + .expect("storage ok") + .is_none()); } /// th-ca579c — identity verification survives a pod hop. @@ -1055,7 +1061,11 @@ mod tests { // Pod B hydrates fresh from storage and must agree. assert!( - pod_b.load_session("s-otp").await.is_some(), + pod_b + .load_session("s-otp") + .await + .expect("storage ok") + .is_some(), "pod B must see the session" ); assert!( diff --git a/rust/smooth-operator-server/tests/session_storage_blip.rs b/rust/smooth-operator-server/tests/session_storage_blip.rs new file mode 100644 index 00000000..59658c29 --- /dev/null +++ b/rust/smooth-operator-server/tests/session_storage_blip.rs @@ -0,0 +1,316 @@ +//! A storage blip is not an existence claim. +//! +//! `AppState::load_session` hydrates a session from storage on a local-registry +//! miss (th-ca579c), which is the normal path for a returning visitor whose +//! WebSocket lands on a pod that has never seen their session. That read used to +//! collapse `Err` into `None`, so a Postgres hiccup was indistinguishable from a +//! session that genuinely does not exist — and every caller renders `None` as +//! `session '' not found`, in a live visitor's chat bubble. +//! +//! These tests drive the real `handler::handle_frame` against a storage adapter +//! whose `get_session` can be made to fail on demand, and assert BOTH halves: +//! +//! - a failing `get_session` produces a retryable `STORAGE_ERROR`, and never a +//! not-found code or the words "not found"; +//! - with storage healthy, an unknown session id STILL produces +//! `SESSION_NOT_FOUND` — the fix must not turn a real not-found into a +//! retry-forever loop. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::{json, Value}; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver}; + +use smooth_operator::access_control::AccessContext; +use smooth_operator::adapter::{ + ConversationUpdate, MessagePage, MessageQuery, SessionUpdate, StorageAdapter, +}; +use smooth_operator::domain::{Conversation, Message, Participant, Session}; +use smooth_operator_adapter_memory::InMemoryStorageAdapter; +use smooth_operator_core::{CheckpointStore, KnowledgeBase}; + +use smooth_operator_server::config::{ServerConfig, StorageBackend}; +use smooth_operator_server::handler::{self, UserScope}; +use smooth_operator_server::state::AppState; + +/// A `StorageAdapter` that delegates everything to an in-memory adapter, except +/// that `get_session` fails while [`fail`](Self::fail) is raised — the transient +/// backend blip, on the one read this fix is about. +struct FlakySessionAdapter { + inner: Arc, + fail: Arc, +} + +impl FlakySessionAdapter { + fn new() -> (Self, Arc) { + let fail = Arc::new(AtomicBool::new(false)); + ( + Self { + inner: Arc::new(InMemoryStorageAdapter::new()), + fail: Arc::clone(&fail), + }, + fail, + ) + } +} + +#[async_trait] +impl StorageAdapter for FlakySessionAdapter { + async fn create_conversation( + &self, + conversation: Conversation, + ) -> anyhow::Result { + self.inner.create_conversation(conversation).await + } + async fn get_conversation(&self, id: &str) -> anyhow::Result> { + self.inner.get_conversation(id).await + } + async fn list_conversations_by_org( + &self, + organization_id: &str, + ) -> anyhow::Result> { + self.inner.list_conversations_by_org(organization_id).await + } + async fn update_conversation( + &self, + id: &str, + update: ConversationUpdate, + ) -> anyhow::Result { + self.inner.update_conversation(id, update).await + } + async fn add_participant(&self, participant: Participant) -> anyhow::Result { + self.inner.add_participant(participant).await + } + async fn get_participant(&self, id: &str) -> anyhow::Result> { + self.inner.get_participant(id).await + } + async fn list_participants_by_conversation( + &self, + conversation_id: &str, + ) -> anyhow::Result> { + self.inner + .list_participants_by_conversation(conversation_id) + .await + } + async fn resolve_participant_by_external_id( + &self, + conversation_id: &str, + external_id: &str, + ) -> anyhow::Result> { + self.inner + .resolve_participant_by_external_id(conversation_id, external_id) + .await + } + async fn append_message(&self, message: Message) -> anyhow::Result { + self.inner.append_message(message).await + } + async fn get_message(&self, id: &str) -> anyhow::Result> { + self.inner.get_message(id).await + } + async fn list_messages_by_conversation( + &self, + query: MessageQuery, + ) -> anyhow::Result { + self.inner.list_messages_by_conversation(query).await + } + async fn create_session(&self, session: Session) -> anyhow::Result { + self.inner.create_session(session).await + } + async fn get_session(&self, session_id: &str) -> anyhow::Result> { + if self.fail.load(Ordering::SeqCst) { + anyhow::bail!("connection reset by peer"); + } + self.inner.get_session(session_id).await + } + async fn update_session( + &self, + session_id: &str, + update: SessionUpdate, + ) -> anyhow::Result { + self.inner.update_session(session_id, update).await + } + async fn list_sessions_by_conversation( + &self, + conversation_id: &str, + ) -> anyhow::Result> { + self.inner + .list_sessions_by_conversation(conversation_id) + .await + } + fn checkpoints(&self) -> Arc { + self.inner.checkpoints() + } + fn knowledge(&self) -> Arc { + self.inner.knowledge() + } +} + +fn base_config() -> ServerConfig { + ServerConfig { + bind: "127.0.0.1".into(), + port: 0, + gateway_url: "https://example.invalid/v1".into(), + gateway_key: None, + model: "claude-haiku-4-5".into(), + seed_kb: false, + max_iterations: 4, + max_tokens: 128, + storage: StorageBackend::Memory, + widget_auth_strict: false, + confirm_tools: Vec::new(), + judge_model: "claude-haiku-4-5".to_string(), + } +} + +/// Drive one frame through the real dispatcher and return the first event. +/// Unscoped + no org: the ownership/tenant checks pass trivially, so the only +/// thing under test is the session read. +async fn drive(state: &AppState, frame: &Value) -> Value { + let (tx, mut rx): (_, UnboundedReceiver) = unbounded_channel(); + handler::handle_frame( + state, + &AccessContext::anonymous(), + "conn-blip", + None, + None, + &UserScope::Unscoped, + &frame.to_string(), + &tx, + ) + .await; + tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("an event should be emitted") + .expect("sink open") +} + +/// Every action that takes a client-supplied `sessionId`, with the error code it +/// emits when the session is genuinely unknown. All six route through +/// `scoped_session`, so all six must switch to `STORAGE_ERROR` on a blip. +fn session_frames() -> Vec<(Value, &'static str)> { + vec![ + ( + json!({"action": "get_session", "requestId": "r1", "sessionId": "s-ghost"}), + "SESSION_NOT_FOUND", + ), + ( + json!({"action": "get_conversation_messages", "requestId": "r2", "sessionId": "s-ghost"}), + "SESSION_NOT_FOUND", + ), + ( + json!({"action": "send_message", "requestId": "r3", "sessionId": "s-ghost", "message": "hi"}), + "SESSION_NOT_FOUND", + ), + ( + json!({"action": "verify_otp", "requestId": "r4", "sessionId": "s-ghost", "code": "123456"}), + "SESSION_NOT_FOUND", + ), + ( + json!({"action": "confirm_tool_action", "requestId": "r5", "sessionId": "s-ghost", "approved": true}), + "NO_PENDING_CONFIRMATION", + ), + ( + json!({"action": "submit_interaction", "requestId": "r6", "sessionId": "s-ghost", "interactionId": "i-1", "values": {}}), + "NO_PENDING_INTERACTION", + ), + ] +} + +/// THE regression: storage down must never be rendered as "your session does not +/// exist". A visitor told that clears their conversation; a visitor told to retry +/// keeps it. +#[tokio::test] +async fn a_storage_blip_is_never_reported_as_not_found() { + let (adapter, fail) = FlakySessionAdapter::new(); + let state = AppState::new(Arc::new(adapter), base_config()); + fail.store(true, Ordering::SeqCst); + + // Every action is checked before failing, so one run reports every handler + // that regressed — not just the first. + let mut regressed = Vec::new(); + for (frame, notfound_code) in session_frames() { + let ev = drive(&state, &frame).await; + let code = ev["error"]["code"].as_str().unwrap_or_default().to_string(); + let message = ev["error"]["message"] + .as_str() + .unwrap_or_default() + .to_string(); + if code != "STORAGE_ERROR" + || code == notfound_code + || message.to_lowercase().contains("not found") + { + regressed.push(format!("{}: {ev}", frame["action"])); + } + } + assert!( + regressed.is_empty(), + "these actions turned a storage blip into an existence claim:\n{}", + regressed.join("\n") + ); +} + +/// The other half — and the reason this needs two tests. A genuinely unknown id +/// must still be not-found; a fix that reported everything as retryable would +/// leave a client retrying an id that will never resolve. +#[tokio::test] +async fn an_unknown_session_is_still_not_found_when_storage_is_healthy() { + let (adapter, fail) = FlakySessionAdapter::new(); + let state = AppState::new(Arc::new(adapter), base_config()); + assert!(!fail.load(Ordering::SeqCst), "storage must be healthy here"); + + let mut wrong = Vec::new(); + for (frame, notfound_code) in session_frames() { + let ev = drive(&state, &frame).await; + if ev["error"]["code"].as_str().unwrap_or_default() != notfound_code { + wrong.push(format!("{} (want {notfound_code}): {ev}", frame["action"])); + } + } + assert!( + wrong.is_empty(), + "these actions stopped reporting a genuinely unknown id as not-found:\n{}", + wrong.join("\n") + ); +} + +/// A session that IS in storage still resolves through the blip-aware path — the +/// error branch must not swallow the happy one. +#[tokio::test] +async fn a_healthy_hydrate_still_serves_the_session() { + let (adapter, _fail) = FlakySessionAdapter::new(); + let adapter = Arc::new(adapter); + adapter + .create_session(Session { + session_id: "s-real".into(), + conversation_id: "conv-1".into(), + organization_id: "org-1".into(), + agent_id: None, + agent_name: "Agent".into(), + user_participant_id: "p-user".into(), + agent_participant_id: "p-agent".into(), + thread_id: "thread-1".into(), + status: None, + token_count: None, + message_count: None, + metadata: None, + created_at: None, + updated_at: None, + ended_at: None, + last_activity_at: None, + }) + .await + .expect("seed the session"); + + // A fresh pod: storage has it, the local registry has never seen it. + let state = AppState::new(adapter, base_config()); + let ev = drive( + &state, + &json!({"action": "get_session", "requestId": "r", "sessionId": "s-real"}), + ) + .await; + + assert_eq!(ev["type"], "immediate_response", "got: {ev}"); + assert_eq!(ev["data"]["sessionId"], "s-real"); +}