evanspearman · Jul 6, 2026 · Jun 27, 2026
@@ -465,6 +465,13 @@ impl sessions::core::hooks::PolicyHooks for AbortOnUnapproved {
 /// over the [`SubmitVerdict`] RPC; unwraps the daemon's terminal
 /// [`SessionStep::Active`] reply to the finalized session id.
 ///
+/// If Phase 3 gating fails (user cancels, policy hook aborts, var
+/// resolution fails, etc.), the client owes the daemon an
+/// `AbortSession` — otherwise the on-disk `Pending` record and its
+/// stash slot leak until daemon restart. Fires it as a
+/// fire-and-report step: the abort's own failure is logged but
+/// doesn't override the original gating error surfaced to the user.
+///
 /// [`UserPolicy::default`]: sessions::core::policy::UserPolicy::default
 /// [`SubmitVerdict`]: minimald_rpc::SubmitVerdict
 async fn drive_pending_to_active(
@@ -477,16 +484,26 @@ async fn drive_pending_to_active(
     use sessions::core::policy::UserPolicy;
     use sessions::wire::request::SessionStep;
 
+    // Capture the session id before consuming `response` — we need it
+    // to send AbortSession on any Phase 3 failure below.
+    let session_id = response.session_id;
+
     let hooks = AbortOnUnapproved;
-    let verdict = handle_response(
+    let verdict = match handle_response(
         response,
         &[],
         UserPolicy::default(),
         &hooks,
         ComposeOptions::default(),
         &|name| std::env::var(name),
-    )
-    .map_err(|e| eprintln!("Composition gating failed: {e}"))?;
+    ) {
+        Ok(v) => v,
+        Err(e) => {
+            eprintln!("Composition gating failed: {e}");
+            send_abort(client, session_id).await;
+            return Err(());
+        }
+    };
 
     let resp = client
         .oneshot_rpc::<SubmitVerdict>(verdict)
@@ -501,22 +518,38 @@ async fn drive_pending_to_active(
     };
     match step {
         SessionStep::Active { id } => Ok(id),
-        SessionStep::Response { .. } => {
-            // The daemon's terminal reply today is always Active
-            // (single-round flow). Multi-round Response would be a
-            /
8000
/ future extension and the client isn't wired for it.
-            eprintln!(
-                "SubmitVerdict returned a follow-up Response, which this client cannot drive"
-            );
-            Err(())
-        }
         SessionStep::Fault { error } => {
+            // `Fault` means the daemon already tore down its side
+            // (`resume_from_verdict` failure destroys the record;
+            // `UnknownSessionId`/`WrongState` means there was nothing
+            // for us to abort). No AbortSession needed here.
             eprintln!("SubmitVerdict faulted: {error}");
             Err(())
         }
     }
 }
 
+/// Fire an `AbortSession` at the daemon for a `Pending` session the
+/// client has decided not to finalize. A best-effort teardown: if
+/// the abort itself fails (network, daemon crash, or a state race)
+/// we log and move on — the daemon's startup reap pass is the
+/// backstop for the leaked `Pending` record.
+async fn send_abort(client: &mut client::Client, session_id: sessions::SessionId) {
+    use minimald_rpc::AbortSession;
+    match client
+        .oneshot_rpc::<AbortSession>(minimald_rpc::AbortSessionRequest { id: session_id })
+        .await
+    {
+        Ok(minimald_rpc::Errorable::Ok(_)) => {}
+        Ok(minimald_rpc::Errorable::Err { error }) => {
+            eprintln!("AbortSession failed: {error}");
+        }
+        Err(e) => {
+            eprintln!("AbortSession RPC failed: {e}");
+        }
+    }
+}
+
 /// Create a new session via the `CreateSession` RPC.
 async fn cmd_activate(global: &GlobalArgs, args: ActivateArgs) -> Result<(), ()> {
     if let Err(e) = autospawn::ensure_daemon_running(global.minvmd, global.minimal_dir.as_deref()) {

@@ -331,34 +331,16 @@ impl OneshotSshRpc for Shutdown {
     type Response = ShutdownResponse;
 }
 
-// ---------------------------------------------------------------------------
-// Session-creation flow (multi-round contribution composition).
-//
-// Distinct from the simpler [`CreateSession`] above: that one takes a
-// fully-formed [`sessions::Record`]; the flow below composes the record
-// by walking client contributions and daemon-side closures across one or
-// more rounds. Each call returns a [`SessionStep`]: either the next round
-// of pending items or a protocol-level fault.
-//
-// TODO: these three RPCs are the building blocks for what is eventually
-// going to subsume `CreateSession` — once the multi-round flow lands on
-// the daemon, the terminal `SessionStep` will assemble a `sessions::Record`
-// and the single-shot `CreateSession` becomes redundant. Keeping them
-// separate for now so the existing `CreateSession` callers stay working
-// while the contribution flow is built out.
-
-/// An RPC to open a new session and receive the first round of items
-/// the client must resolve.
-pub struct SessionCreate;
-
-impl OneshotSshRpc for SessionCreate {
-    const NAME: &'static str = constcat::concat!(RPC_SUBSYSTEM_PREFIX, "SessionCreate");
-    type Request<'a> = sessions::wire::request::SessionCreateRequest;
-    type Response = Errorable<sessions::wire::request::SessionStep>;
-}
-
-/// An RPC to submit the client's verdicts for one round and receive the
-/// next round (or a `complete` signal in [`sessions::wire::request::ContributionResponse`]).
+/// Resume a `Pending` session with the client's per-item
+/// [`ContributionVerdict`]. Terminal — the daemon promotes the
+/// record `Pending → Active` and replies with
+/// [`SessionStep::Active`](sessions::wire::request::SessionStep::Active).
+/// A `Fault` reply carries a structured
+/// [`WireError`](sessions::wire::errors::WireError) —
+/// `UnknownSessionId` for a verdict against no stashed session,
+/// `WrongState` if the record isn't `Pending`, or an
+/// `InvalidContribution` / `Internal` reflecting a failed
+/// `resume_from_verdict`.
 pub struct SubmitVerdict;
 
 impl OneshotSshRpc for SubmitVerdict {
@@ -367,13 +349,33 @@ impl OneshotSshRpc for SubmitVerdict {
     type Response = Errorable<sessions::wire::request::SessionStep>;
 }
 
-/// An RPC to abort an in-flight session-creation flow.
-pub struct SessionAbort;
+/// Abort a `Pending` session before its `SubmitVerdict`.
+///
+/// Called by the client when its Phase 3 gating produces no verdict —
+/// user cancelled at a prompt, policy hooks returned `Abort`, or an
+/// upstream resolution / expansion failed. Drops the daemon's stash
+/// entry and deletes the on-disk `Pending` record so the session name
+/// is freed and the stash slot isn't burned.
+///
+/// Refuses non-`Pending` records: an unknown id or a record already
+/// promoted to `Active` (destroy that via [`DestroySession`]) surfaces
+/// as an `Errorable::Err`.
+pub struct AbortSession;
+
+/// The request for an [`AbortSession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AbortSessionRequest {
+    pub id: SessionId,
+}
+
+/// The response for an [`AbortSession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct AbortSessionResponse;
 
-impl OneshotSshRpc for SessionAbort {
-    const NAME: &'static str = constcat::concat!(RPC_SUBSYSTEM_PREFIX, "SessionAbort");
-    type Request<'a> = sessions::wire::request::Abort;
-    type Response = Errorable<()>;
+impl OneshotSshRpc for AbortSession {
+    const NAME: &'static str = constcat::concat!(RPC_SUBSYSTEM_PREFIX, "AbortSession");
+    type Request<'a> = AbortSessionRequest;
+    type Response = Errorable<AbortSessionResponse>;
 }
 
 // ---------------------------------------------------------------------------

@@ -1,12 +1,12 @@
 #[cfg(feature = "networking-proxy")]
 use minimald_rpc::IssueClientCertResponse;
 use minimald_rpc::{
-    CreateSession, DestroySession, DestroySessionResponse, Errorable, GetMeshStatus,
-    GetSessionPolicy, GetSessionPolicyRequest, GetSessionRecord, GetSessionRecordRequest,
-    GetSessionRecordResponse, GetVersion, GetVersionResponse, IssueClientCert,
-    IssueClientCertRequest, ListSessions, ListSessionsEntry, ListSessionsResponse, OneshotSshRpc,
-    RPC_SUBSYSTEM_PREFIX, RenameSession, RenameSessionResponse, Shutdown, ShutdownRequest,
-    ShutdownResponse, SubmitVerdict,
+    AbortSession, AbortSessionResponse, CreateSession, DestroySession, DestroySessionResponse,
+    Errorable, GetMeshStatus, GetSessionPolicy, GetSessionPolicyRequest, GetSessionRecord,
+    GetSessionRecordRequest, GetSessionRecordResponse, GetVersion, GetVersionResponse,
+    IssueClientCert, IssueClientCertRequest, ListSessions, ListSessionsEntry, ListSessionsResponse,
+    OneshotSshRpc, RPC_SUBSYSTEM_PREFIX, RenameSession, RenameSessionResponse, Shutdown,
+    ShutdownRequest, ShutdownResponse, SubmitVerdict,
 };
 use russh::{
     Channel as RuChannel, ChannelId,
@@ -261,6 +261,26 @@ async fn serve_shutdown(s: ServerStateHandle, c: RuChannel<Msg>) {
     }
 }
 
+/// `AbortSession`: drop a `Pending` session's stash entry and delete
+/// its on-disk record. See the manager arm for the actor-side rules
+/// (refuses `Active` records and unknown ids).
+async fn serve_abort_session(s: ServerStateHandle, c: RuChannel<Msg>) {
+    let res = AbortSession
+        .handle_channel(c, async |req| {
+            let res = s.sessions_manager().await.abort_session(req.id).await;
+            match res {
+                Ok(()) => Ok(Errorable::Ok(AbortSessionResponse)),
+                Err(e) => Ok(Errorable::Err {
+                    error: e.to_string(),
+                }),
+            }
+        })
+        .await;
+    if let Err(e) = res {
+        tracing::warn!("RPC handler for {} failed: {}", AbortSession::NAME, e);
+    }
+}
+
 async fn serve_get_session_policy(s: ServerStateHandle, c: RuChannel<Msg>) {
     let res = GetSessionPolicy
         .handle_channel(c, async |req| {
@@ -421,6 +441,7 @@ pub async fn handle_ssh_rpc(
         | RenameSession::NAME
         | DestroySession::NAME
         | Shutdown::NAME
+        | AbortSession::NAME
         | GetSessionPolicy::NAME
         | GetMeshStatus::NAME
         | STREAM_WORKSPACE_FILES
@@ -458,6 +479,7 @@ pub async fn handle_ssh_rpc(
         RenameSession::NAME => drop(spawn(serve_rename_session(s, channel))),
         DestroySession::NAME => drop(spawn(serve_destroy_session(s, channel))),
         Shutdown::NAME => drop(spawn(serve_shutdown(s, channel))),
+        AbortSession::NAME => drop(spawn(serve_abort_session(s, channel))),
         GetSessionPolicy::NAME => drop(spawn(serve_get_session_policy(s, channel))),
         GetMeshStatus::NAME => drop(spawn(serve_get_mesh_status(s, channel))),
         STREAM_WORKSPACE_FILES => drop(spawn(serve_stream_workspace_files(s, config, channel))),

@@ -154,6 +154,7 @@ enum ManagerMessage {
     SubmitVerdict(Box<SubmitVerdictMsg>),
     RenameSession(SessionId, String, Responder<()>),
     DestroySession(SessionId, Responder<()>),
+    AbortSession(SessionId, Responder<()>),
     Shutdown(bool, Responder<Result<(), ()>>),
 }
 
@@ -680,6 +681,44 @@ impl<L: Loader> Manager<L> {
                 })
                 .await
             }
+            // Abort a `Pending` session: drop its stash entry and
+            // delete the on-disk record. Refuses `Active` (use
+            // `DestroySession`) or unknown ids. Never touches
+            // `running`: a Pending record can't be running.
+            ManagerMessage::AbortSession(id, r) => {
+                r.handle(async {
+                    if self.in_shutdown {
+                        return Err(SessionsError::new(
+                            std::io::ErrorKind::ConnectionRefused,
+                            "in shutdown",
+                        ));
+                    }
+                    let k = self.store.find_by_id(&id)?.ok_or_else(|| {
+                        std::io::Error::new(
+                            NotFound,
+                            format!("no session with ID `{}`", id.as_ref()),
+                        )
+                    })?;
+                    let record = self.store.get(&k)?.record().clone();
+                    if record.status != sessions::SessionStatus::Pending {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidInput,
+                            format!(
+                                "cannot abort session `{}`: status is {:?}, expected Pending",
+                                id.as_ref(),
+                                record.status,
+                            ),
+                        ));
+                    }
+                    // Stash may be absent (post-restart orphan reached
+                    // us before the reap pass, or the pending path was
+                    // never populated). Delete the record either way.
+                    self.pending.remove(&id);
+                    self.store.delete(&k)?;
+                    Ok(())
+                })
+                .await
+            }
             // Destroys a session: tears down its running host and actor (if
             // any), then removes its on-disk record.
             ManagerMessage::DestroySession(id, r) => {
@@ -897,6 +936,23 @@ impl ManagerHandle {
         recv.await.expect("corresponding sessions manager is dead")
     }
 
+    /// Aborts a `Pending` session: drops the daemon's stash entry and
+    /// deletes the on-disk record. Used by the client when its
+    /// Phase 3 gating produces no verdict (e.g. user cancelled at a
+    /// prompt).
+    ///
+    /// `NotFound` if the id is unknown; `InvalidInput` if the record
+    /// exists but its status isn't `Pending`
+    /// (use [`Self::destroy_session`] on an `Active` session).
+    pub async fn abort_session(&self, id: SessionId) -> Result<(), SessionsError> {
+        let (send, recv) = Responder::channel();
+        let _ = self
+            .sender
+            .send(ManagerMessage::AbortSession(id, send))
+            .await;
+        recv.await.expect("corresponding sessions manager is dead")
+    }
+
     /// Shuts down all sessions for process termination. If force is true, live sessions are killed.
     /// If force is false but there are live sessions, an error is returned.
     pub async fn shutdown(&self, force: bool) -> Result<(), ()> {
@@ -1108,6 +1164,42 @@ mod tests {
         );
     }
 
+    /// `AbortSession` on an unknown id → `NotFound`, mirrors
+    /// `DestroySession`'s error shape so clients can uniformly
+    /// handle "no such session".
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn abort_unknown_id_errors() {
+        let (_state, _cache, mngr) = manager().await;
+        let err = mngr
+            .abort_session(SessionId::nil())
+            .await
+            .expect_err("aborting an unknown id should error");
+        assert_eq!(err.kind(), ErrorKind::NotFound);
+    }
+
+    /// `AbortSession` refuses `Active` records — abort is for
+    /// `Pending` only; `DestroySession` handles `Active`. Guards
+    /// against a client accidentally tearing down a running session
+    /// via the wrong RPC.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn abort_refuses_active_session() {
+        let (_state, _cache, mngr) = manager().await;
+        let id = create_and_unwrap_id(&mngr).await;
+        let err = mngr
+            .abort_session(id)
+            .await
+            .expect_err("aborting an Active session should error");
+        assert_eq!(err.kind(), ErrorKind::InvalidInput);
+        // Record is untouched.
+        assert!(
+            mngr.get_record(SessionKeyPredicate::Id(id))
+                .await
+                .unwrap()
+                .is_some(),
+            "the Active record should survive a refused abort"
+        );
+    }
+
     /// `SubmitVerdict` for an id with no stash entry → terminal
     /// `Fault::UnknownSessionId`. The stash is the single source of
     /// truth for in-flight sessions; a verdict for an id that was

@@ -28,11 +28,12 @@ sequenceDiagram
     Note over C: Loadout → Contribution (Loadout::contribute, resolve vars, tag provenance)
     Note over C: Contribution → Composition (UserComposer::compose runs the policy gate)
     Note over C: Composition → WireContribution (composition_to_wire)
-    C->>D: SessionCreate carrying WireContribution
+    C->>D: CreateSession carrying WireContribution
     Note over D: Phase 2 — collect project + package contributions, collect pending items
-    D-->>C: SessionStep::Response carrying ContributionResponse
+    D-->>C: CreateSessionResponse::Pending carrying ContributionResponse
     Note over C: Phase 3 — policy gate pending items, produce verdicts
     C->>D: SubmitVerdict carrying ContributionVerdict
+    D-->>C: SessionStep::Active { id }
     Note over D: Phase 4 — apply verdicts, assemble Composition, hand to apply layer
 ```
 

@@ -57,8 +57,16 @@ use crate::wire::request::{ContributionResponse, ContributionVerdict};
 ///
 /// See [`ComposeError`]. In particular:
 /// - [`ComposeError::Aborted`] when a hook returns
-///   [`HookResult::Abort`]; the caller should send a `SessionAbort`
-///   RPC with [`AbortReason::UserCancelled`](crate::wire::request::AbortReason::UserCancelled).
+///   [`HookResult::Abort`]. The caller should surface the abort to
+///   the user, skip `SubmitVerdict`, and instead send `AbortSession`
+///   (`minimald_rpc::AbortSession`) so the daemon drops its stash
+///   entry and deletes the on-disk `Pending` record. Without that,
+///   the record and stash slot leak until the next daemon restart:
+///   the in-memory stash is cleared by the restart itself, and the
+///   on-disk record is cleared by `reap_orphan_pending` at startup.
+///   The stash is bounded by `MAX_PENDING_SESSIONS`; further
+///   `CreateSession` requests that would produce a Pending outcome
+///   are refused with `ResourceBusy` once the cap is reached.
 /// - [`ComposeError::VarResolution`] when an `Inherit`-shaped
 ///   pending var can't be resolved against the env.
 /// - [`ComposeError::Expansion`] / [`ComposeError::PatchWalk`] for