feat(claw): delete a session and its turn history over http

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.
This commit is contained in:
kosmos authored and Kosmos committed 2026-11-11 16:11:07 +00:00
1 parent 263184ff8e
commit 8db1e2aa11
5 files changed
+375 -2

No files matched your search

+1
View File
@@ -126,6 +126,7 @@ POST /v1/sessions {"name"} — create an empty interactive se
POST /v1/sessions/{key}/messages {"text": "..."} -> 202 POST /v1/sessions/{key}/messages {"text": "..."} -> 202
GET /v1/sessions/{key}/events SSE turn events (see below); ?since=<seq> resumes a cursor GET /v1/sessions/{key}/events SSE turn events (see below); ?since=<seq> resumes a cursor
GET /v1/sessions/{key}/history one page of older turns: ?before=<seq>&limit=<n> GET /v1/sessions/{key}/history one page of older turns: ?before=<seq>&limit=<n>
DELETE /v1/sessions/{key} drop a session and its turn history
GET /v1/watches POST /v1/watches DELETE /v1/watches/{forge}/{repo} GET /v1/watches POST /v1/watches DELETE /v1/watches/{forge}/{repo}
GET /v1/bootstrap {"soul": bool, "user": bool} GET /v1/bootstrap {"soul": bool, "user": bool}
POST /v1/ingest {"source","key","payload"} -> 202 POST /v1/ingest {"source","key","payload"} -> 202
+95 -1
View File
@@ -47,7 +47,8 @@ use rhema_claw_tools_contract::CONTRACT_HEADER;
use rhema_claw_tools_contract::CONTRACT_VERSION; use rhema_claw_tools_contract::CONTRACT_VERSION;
use rhema_core::ImageMediaType; use rhema_core::ImageMediaType;
use rhema_rsd::{ 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::{Deserialize, Deserializer};
use serde_json::{Value, json}; 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. /// How often the SSE keep-alive comment goes out.
pub(crate) const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); 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 /// Everything the routes touch, shared behind `Arc`s. Built by the
/// daemon's assembly; [`ClawState::new`] is its only constructor. /// daemon's assembly; [`ClawState::new`] is its only constructor.
#[derive(Clone)] #[derive(Clone)]
@@ -198,6 +214,30 @@ impl ClawState {
Ok(sessions) 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<bool, SessionForgetError> {
// 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. /// The interactive dispatcher: message ingest and key status.
pub fn dispatcher(&self) -> &Dispatcher { pub fn dispatcher(&self) -> &Dispatcher {
&self.dispatcher &self.dispatcher
@@ -256,6 +296,7 @@ impl ClawState {
pub fn router(state: ClawState) -> Router { pub fn router(state: ClawState) -> Router {
let v1 = Router::new() let v1 = Router::new()
.route("/sessions", get(sessions).post(sessions_create)) .route("/sessions", get(sessions).post(sessions_create))
.route("/sessions/{key}", delete(session_delete))
.route( .route(
"/sessions/{key}/messages", "/sessions/{key}/messages",
post(messages).layer(axum::extract::DefaultBodyLimit::max(MAX_MESSAGE_BODY_BYTES)), 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<ClawState>, Path(key): Path<String>) -> 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 /// 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 /// UI badges first-run setup from it while the onboarding offer is
/// still open. /// still open.
+18 -1
View File
@@ -32,7 +32,7 @@ use rhema_rsd::ResourceKey;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tokio::sync::broadcast; 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. /// How many events a subscriber may trail by before the oldest drop.
/// Generous: a turn's deltas arrive faster than any viewer, but rarely /// Generous: a turn's deltas arrive faster than any viewer, but rarely
@@ -501,6 +501,23 @@ impl EventHub {
let _ = entry.sender.send(event); 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 /// The map entry for `key`, creating it (and its channel) on first
/// use. Entries live for the process lifetime once created — the /// use. Entries live for the process lifetime once created — the
/// live channel is the point; with a store attached the record /// live channel is the point; with a store attached the record
+46
View File
@@ -220,6 +220,18 @@ impl TurnStore {
) )
.map_err(TurnStoreError::from) .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, /// 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}"); 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 /// The one-time migration: a pre-window blob table opens into the
/// append-only shape, its rows gone. /// append-only shape, its rows gone.
#[test] #[test]
+215
View File
@@ -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, "back, and whole.").await;
read_until(&mut stream, &mut seen, r#""type":"turn_done""#).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<Value> {
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::<Value>().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:?}");
}