From 8db1e2aa11283e2019062bf027d3dc12518600ca Mon Sep 17 00:00:00 2001 From: kosmos Date: Wed, 11 Nov 2026 16:04:23 +0000 Subject: [PATCH] feat(claw): delete a session and its turn history over http MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A session the human is done with has to be removable, and removing it must mean forgetting it: `DELETE /v1/sessions/{key}` drops the dispatch record and the turn history recorded under the key, so re-creating the same name starts an empty chat instead of replaying someone else's turns. The route refuses a leased key, because the running turn's own save would either fail against the missing record or write the session back after the delete; the inventory is not narrowed to the `interactive:` prefix, so a forge key deletes on the same route. The history purge rides the hub, which owns the turn table, and it lands after the record does: the purge is the one step this side cannot undo, so it never runs for a delete the dispatcher refused. Purging on the no-record path as well is what makes a half-done delete repairable — a retried DELETE finishes the purge instead of 404-ing with the old turns still under the key, where a session created again under the same name would replay them. The sessions routes stay off the agent token's allowlist in tools/contract: the agent must not be able to delete the conversation it is serving. --- crates/rhema-claw/README.md | 1 + crates/rhema-claw/src/http.rs | 96 ++++++++++++- crates/rhema-claw/src/hub.rs | 19 ++- crates/rhema-claw/src/turn_store.rs | 46 ++++++ crates/rhema-claw/tests/http.rs | 215 ++++++++++++++++++++++++++++ 5 files changed, 375 insertions(+), 2 deletions(-) diff --git a/crates/rhema-claw/README.md b/crates/rhema-claw/README.md index c5d8654..ee90653 100644 --- a/crates/rhema-claw/README.md +++ b/crates/rhema-claw/README.md @@ -126,6 +126,7 @@ POST /v1/sessions {"name"} — create an empty interactive se POST /v1/sessions/{key}/messages {"text": "..."} -> 202 GET /v1/sessions/{key}/events SSE turn events (see below); ?since= resumes a cursor GET /v1/sessions/{key}/history one page of older turns: ?before=&limit= +DELETE /v1/sessions/{key} drop a session and its turn history GET /v1/watches POST /v1/watches DELETE /v1/watches/{forge}/{repo} GET /v1/bootstrap {"soul": bool, "user": bool} POST /v1/ingest {"source","key","payload"} -> 202 diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index 2cb2e4b..7b97e54 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -47,7 +47,8 @@ use rhema_claw_tools_contract::CONTRACT_HEADER; use rhema_claw_tools_contract::CONTRACT_VERSION; use rhema_core::ImageMediaType; use rhema_rsd::{ - DispatchError, Dispatcher, IngestResult, KeyRecord, KeyState, ResourceKey, SqliteStore, Store, + Conflict, DispatchError, Dispatcher, IngestResult, KeyRecord, KeyState, ResourceKey, + SqliteStore, Store, }; use serde::{Deserialize, Deserializer}; use serde_json::{Value, json}; @@ -69,6 +70,21 @@ pub(crate) const MAX_DRAIN_PASSES: usize = 10; /// How often the SSE keep-alive comment goes out. pub(crate) const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); +/// Why a delete stopped short: the dispatcher refused it (a live lease holds +/// the key, or the store failed), or the turn-history purge did. The two +/// answers the route derives itself — deleted, or never there — ride the +/// `Ok(bool)`, so this carries the failure seams and nothing else. +#[derive(Debug, thiserror::Error)] +pub enum SessionForgetError { + /// The dispatcher refused the delete, or the store under it failed. + #[error(transparent)] + Dispatch(#[from] DispatchError), + /// The turn-history table failed, so a transcript survived the record + /// it belonged to. + #[error(transparent)] + History(#[from] crate::turn_store::TurnStoreError), +} + /// Everything the routes touch, shared behind `Arc`s. Built by the /// daemon's assembly; [`ClawState::new`] is its only constructor. #[derive(Clone)] @@ -198,6 +214,30 @@ impl ClawState { Ok(sessions) } + /// Delete one session: the dispatch record — through the dispatcher, so + /// the decision and the delete are one gated transition — and the turn + /// history recorded under the key, so a key re-created under the same + /// name starts empty. + /// + /// `Ok(true)` when a record existed and is gone, `Ok(false)` when there + /// was none; the history is purged either way. A live lease refuses the + /// delete ([`Conflict::LeaseHeld`]) — the running turn's own checkpoint + /// would otherwise write to a vanished record. Any key with a record + /// deletes: the inventory is not narrowed to the `interactive:` prefix. + pub fn delete_session(&self, key: &ResourceKey) -> Result { + // The record goes first, gated inside the dispatcher, so a refusal + // leaves the transcript alone: the purge below is the one step this + // side cannot undo, and it must never run for a session the 409 kept. + let existed = self.dispatcher.forget(key)?; + // Then the history, unconditionally — including the `Ok(false)` path, + // where a record already gone left the transcript behind. Deleting + // that transcript is the repair: a retried DELETE finishes a half-done + // delete instead of 404-ing with the old turns still under the key, + // where a session created again under the same name would replay them. + self.hub.forget(key)?; + Ok(existed) + } + /// The interactive dispatcher: message ingest and key status. pub fn dispatcher(&self) -> &Dispatcher { &self.dispatcher @@ -256,6 +296,7 @@ impl ClawState { pub fn router(state: ClawState) -> Router { let v1 = Router::new() .route("/sessions", get(sessions).post(sessions_create)) + .route("/sessions/{key}", delete(session_delete)) .route( "/sessions/{key}/messages", post(messages).layer(axum::extract::DefaultBodyLimit::max(MAX_MESSAGE_BODY_BYTES)), @@ -528,6 +569,59 @@ async fn sessions_create( } } +/// `DELETE /v1/sessions/{key}`: drop one session — its dispatch record +/// and the turn history recorded under the key — so a key re-created +/// under the same name starts empty. 409 while a turn holds the key, 404 +/// when there is no such session, 200 with `{"key", "deleted"}` once both +/// tables are clean. Any key with a record is deletable: the route does +/// not narrow the inventory to the `interactive:` prefix. +/// +/// One segment is the whole key: a key carrying `/` — `github:acme/widgets#7` +/// — arrives intact only percent-encoded (`%2F`), which is what an +/// `encodeURIComponent` client sends, because a raw `/` splits the path and +/// the route does not match. +async fn session_delete(State(state): State, Path(key): Path) -> Response { + let key = key.trim(); + if key.is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "key is required"})), + ) + .into_response(); + } + let key = ResourceKey::from(key); + match state.delete_session(&key) { + Ok(true) => ( + StatusCode::OK, + Json(json!({"key": key.as_str(), "deleted": true})), + ) + .into_response(), + Ok(false) => ( + StatusCode::NOT_FOUND, + Json(json!({"error": format!("no session {key:?}")})), + ) + .into_response(), + // A live turn is the one refusal: the body spells the holder and the + // lease the way the dispatcher recorded them. + Err(SessionForgetError::Dispatch(DispatchError::Conflict(Conflict::LeaseHeld { + lease_id, + holder, + }))) => { + let held = Conflict::LeaseHeld { lease_id, holder }; + ( + StatusCode::CONFLICT, + Json(json!({"error": format!("session {key:?}: {held}")})), + ) + .into_response() + } + Err(err) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": err.to_string()})), + ) + .into_response(), + } +} + /// The setup badge: which of SOUL.md and USER.md exist, read live — a /// UI badges first-run setup from it while the onboarding offer is /// still open. diff --git a/crates/rhema-claw/src/hub.rs b/crates/rhema-claw/src/hub.rs index 4f646a4..db38b14 100644 --- a/crates/rhema-claw/src/hub.rs +++ b/crates/rhema-claw/src/hub.rs @@ -32,7 +32,7 @@ use rhema_rsd::ResourceKey; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; -use crate::turn_store::TurnStore; +use crate::turn_store::{TurnStore, TurnStoreError}; /// How many events a subscriber may trail by before the oldest drop. /// Generous: a turn's deltas arrive faster than any viewer, but rarely @@ -501,6 +501,23 @@ impl EventHub { let _ = entry.sender.send(event); } + /// Forgets `key`'s recorded stream: the store's rows go, and so does + /// the in-memory record an assembly without one keeps, so a key + /// re-created under the same name replays nothing from before. The + /// live channel stays — it belongs to the process, not to the deleted + /// session's history. A failed purge is the caller's to report: the + /// delete is not done until the history is. + pub fn forget(&self, key: &ResourceKey) -> Result<(), TurnStoreError> { + let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(entry) = entries.get_mut(key) { + entry.history.clear(); + } + match self.record_store.as_ref() { + Some(store) => store.forget(key.as_str()), + None => Ok(()), + } + } + /// The map entry for `key`, creating it (and its channel) on first /// use. Entries live for the process lifetime once created — the /// live channel is the point; with a store attached the record diff --git a/crates/rhema-claw/src/turn_store.rs b/crates/rhema-claw/src/turn_store.rs index fa7e712..73b6e6e 100644 --- a/crates/rhema-claw/src/turn_store.rs +++ b/crates/rhema-claw/src/turn_store.rs @@ -220,6 +220,18 @@ impl TurnStore { ) .map_err(TurnStoreError::from) } + + /// Removes `key`'s whole recorded stream, durable before it returns: + /// every row for the key goes, and the next append starts the stream + /// over at its own sequence, so a key re-created under the same name + /// replays nothing from before. A key with no rows is not an error — + /// deleting a session purges whatever history it has, including none. + pub fn forget(&self, key: &str) -> Result<(), TurnStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + conn.execute("DELETE FROM turn_events WHERE key = ?1", params![key]) + .map_err(TurnStoreError::from)?; + Ok(()) + } } /// Moves a legacy `turn_history` table (one JSON-array blob per key, @@ -404,6 +416,40 @@ mod tests { assert!(err.to_string().contains("interactive:web"), "{err}"); } + #[test] + fn forget_removes_one_keys_rows_and_leaves_the_rest() { + let store = TurnStore::in_memory().expect("store opens"); + store + .append("interactive:web", &[text_delta("a"), text_delta("b")]) + .expect("appends"); + store.append("task:2", &[text_delta("c")]).expect("appends"); + + store.forget("interactive:web").expect("forgets"); + + assert_eq!( + store.events("interactive:web").expect("reads"), + vec![], + "the deleted key's stream is gone" + ); + assert_eq!( + store.events("task:2").expect("reads"), + vec![text_delta("c")], + "another key's stream is untouched" + ); + // The stream starts over: a re-created key replays nothing from + // before, and appends land at sequence 0 again. + store + .append("interactive:web", &[text_delta("fresh")]) + .expect("appends"); + assert_eq!( + store.events("interactive:web").expect("reads"), + vec![text_delta("fresh")], + "the re-created stream starts empty" + ); + // A key with no rows is not an error. + store.forget("interactive:absent").expect("quiet"); + } + /// The one-time migration: a pre-window blob table opens into the /// append-only shape, its rows gone. #[test] diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index 5e68397..29172b6 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -3293,3 +3293,218 @@ async fn a_restart_releases_a_dead_lease_and_the_key_answers() { read_until(&mut stream, &mut seen, "back, and whole.").await; read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; } + +/// Listing the inventory, as a client reads it. +async fn inventory(client: &reqwest::Client, base: &str) -> Vec { + client + .get(format!("{base}/v1/sessions")) + .send() + .await + .expect("sessions reads") + .json() + .await + .expect("sessions parses") +} + +/// `DELETE /v1/sessions/{key}` is the whole story of a session's life +/// ending: the record leaves the inventory, its turn history goes with +/// it, and the key is absent again — so re-creating the same name starts +/// a chat that streams nothing from before. +#[tokio::test] +async fn deleting_a_session_drops_the_record_and_its_history() { + let provider = Arc::new(MockProvider::new([reply("first words"), reply("second words")])); + let dir = tempfile::tempdir().expect("tempdir"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(provider), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + // One session with one recorded turn. + let created = client + .post(format!("{base}/v1/sessions")) + .json(&json!({"name": "web"})) + .send() + .await + .expect("the create posts"); + assert_eq!(created.status(), 201); + let events = client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"); + let mut stream = sse_body(events).await; + let mut seen = String::new(); + client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "hello"})) + .send() + .await + .expect("the message posts") + .error_for_status() + .expect("the message accepts"); + read_until(&mut stream, &mut seen, "first words").await; + read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; + // The subscriber goes away; the transcript stays in the table. + drop(stream); + assert_eq!(inventory(&client, &base).await.len(), 1); + + // The delete answers the key it removed. + let deleted = client + .delete(format!("{base}/v1/sessions/interactive:web")) + .send() + .await + .expect("the delete goes out"); + assert_eq!(deleted.status(), 200); + assert_eq!( + deleted.json::().await.expect("the body parses"), + json!({"key": "interactive:web", "deleted": true}) + ); + + // Gone from the inventory... + assert!(inventory(&client, &base).await.is_empty(), "the key is gone"); + + // ...and gone from the store, so the create precondition holds again. + let recreated = client + .post(format!("{base}/v1/sessions")) + .json(&json!({"name": "web"})) + .send() + .await + .expect("the second create posts"); + assert_eq!(recreated.status(), 201, "a deleted key can be created again"); + + // The re-created key replays nothing from before: its stream opens + // with the new turn only — one user message, the new reply, and no + // trace of the deleted transcript. + let events = client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"); + let mut stream = sse_body(events).await; + let mut seen = String::new(); + client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "a fresh start"})) + .send() + .await + .expect("the message posts") + .error_for_status() + .expect("the message accepts"); + read_until(&mut stream, &mut seen, "second words").await; + read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; + assert!(!seen.contains("first words"), "no old turn replays: {seen}"); + assert_eq!( + seen.matches(r#""type":"user_message""#).count(), + 1, + "only the new turn replays: {seen}" + ); +} + +/// A delete of a key with no record is a 404 that names the key — and a +/// key with a record is deletable whatever its prefix: the route never +/// narrows the inventory to `interactive:`. +#[tokio::test] +async fn deleting_an_unknown_session_is_a_404_and_any_key_deletes() { + let provider = Arc::new(MockProvider::new([reply("unused")])); + let dir = tempfile::tempdir().expect("tempdir"); + let db = dir.path().join("dispatch.db"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(provider), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + let unknown = client + .delete(format!("{base}/v1/sessions/interactive:nope")) + .send() + .await + .expect("the delete goes out"); + assert_eq!(unknown.status(), 404); + let body: Value = unknown.json().await.expect("the body parses"); + assert!( + body["error"].as_str().expect("an error").contains("interactive:nope"), + "the refusal names the key: {body}" + ); + + // A forge key's record — written by the dispatcher in every real + // deployment — deletes on the same route. + let store = SqliteStore::open(&db).expect("the store opens"); + let forge = ResourceKey::from("github:acme/widgets#7"); + store.save(rhema_rsd::KeyRecord::fresh(forge.clone(), 1), None).expect("the row seeds"); + let deleted = client + .delete(format!("{base}/v1/sessions/github%3Aacme%2Fwidgets%237")) + .send() + .await + .expect("the delete goes out"); + assert_eq!(deleted.status(), 200); + assert_eq!( + store.load(&forge).expect("the store reads"), + None, + "no prefix is special: the forge key's record is gone" + ); +} + +/// A delete never races a running turn: while a lease holds the key the +/// route answers 409 and leaves the session exactly as it was — the +/// turn's own save would otherwise fail on the missing record, or write +/// the session back after the delete. +#[tokio::test] +async fn a_leased_session_is_not_deletable() { + let provider = Arc::new(MockProvider::new([reply("still working")])); + let dir = tempfile::tempdir().expect("tempdir"); + let db = dir.path().join("dispatch.db"); + let daemon = Daemon::build( + &config(&dir, None), + mock_registry(provider), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + // A turn in flight, modelled the way the store sees one: a dispatcher + // over the same file holds the key's lease with hours left on it. + { + use rhema_rsd::DispatcherConfig; + let store = Arc::new(SqliteStore::open(&db).expect("the store opens")); + let dispatcher = rhema_rsd::Dispatcher::new( + DispatcherConfig { + debounce_window_millis: 0, + lease_ttl_millis: 3_600_000, + ..DispatcherConfig::default() + }, + Arc::new(rhema_rsd::SystemClock), + store, + ); + let (lease, _) = dispatcher + .acquire(ResourceKey::from("interactive:web"), "a-running-turn") + .expect("the running turn takes its lease"); + assert_eq!(lease.holder, "a-running-turn"); + } + + let refused = client + .delete(format!("{base}/v1/sessions/interactive:web")) + .send() + .await + .expect("the delete goes out"); + assert_eq!(refused.status(), 409, "a running turn blocks the delete"); + let body: Value = refused.json().await.expect("the body parses"); + let error = body["error"].as_str().expect("an error"); + assert!( + error.contains("leased") && error.contains("a-running-turn"), + "the refusal names the lease and its holder: {body}" + ); + + // Untouched: still listed, still leased. + let sessions = inventory(&client, &base).await; + assert_eq!(sessions.len(), 1, "{sessions:?}"); + assert_eq!(sessions[0]["key"], "interactive:web"); + assert_eq!(sessions[0]["state"], "leased", "{sessions:?}"); +}