diff --git a/DEV.md b/DEV.md
index aedb389..155cef2 100644
--- a/DEV.md
+++ b/DEV.md
@@ -22,9 +22,9 @@ The language boundary has two levels:
 - `Language` supplies the parent `LanguageSession` and creates independent child sessions.
 - `LanguageSession` supplies kernel metadata, execution, completion, inspection, completeness, history, comms, debugging, and shutdown.
 
-Each shell session is driven by one scheduler object which owns its queue, active executions, hold, lock, and interruption state. Shared transport and language handles live in its services object. Output from executions and comm handlers uses the same event pump, so stream, display, buffer, flush, and parent-routing behavior cannot diverge between the two paths.
+Each shell session is driven by one scheduler object which owns its queue, active execution, hold, and interruption state. Shared transport and language handles live in its services object. Output from executions and comm handlers uses the same event pump, so stream, display, buffer, flush, and parent-routing behavior cannot diverge between the two paths.
 
-An execute receives an `ExecutionContext`. It emits streams and displays, requests stdin, publishes arbitrary messages, observes or registers for interruption, releases the execution queue with `unlock()`, and opens temporary subshell routes. The engine converts these events into correctly parented Jupyter messages.
+An execute receives an `ExecutionContext`. It emits streams and displays, requests stdin, publishes arbitrary messages, observes or registers for interruption, and opens subshell routes. The engine converts these events into correctly parented Jupyter messages.
 
 `run_kernel` installs Tokio SIGINT handling. `run_kernel_with_interrupter` lets an embedding host supply its own `KernelInterrupter`.
 
@@ -69,7 +69,7 @@ Two execute metadata extensions are supported:
 
 `KERNMINI_HOLD_TIMEOUT` is the hold backstop in seconds and defaults to 3600.
 
-Python code can call `kernmini.unlock()` to release its queue baton while the current cell continues, or use `kernmini.subshell()` to route later requests from that client session through a temporary child.
+An explicit `subshell_id` on a shell request creates that named subshell when missing, then routes the request there. `create_subshell_request` also accepts an optional `subshell_id`; a supplied ID makes explicit creation idempotent. Python code can use `kernmini.subshell()` to route later requests from its client session through a temporary child, or `kernmini.sidecar()` to route through the persistent named `sidecar` subshell.
 
 ## Output and stdin
 
diff --git a/README.md b/README.md
index 48a23b5..fc41c8d 100644
--- a/README.md
+++ b/README.md
@@ -35,7 +35,7 @@ run_kernel(sys.argv[-1], EchoShell, own_process_group=True)
 
 `run_kernel` creates a persistent asyncio event loop and runs the Rust engine until shutdown. It uses loopmini when installed and the standard asyncio loop otherwise; `loop_factory=` can select one explicitly. The factory is also used to create independent language sessions for JEP 91 subshells. Standalone executables can request process-group ownership, while embedded kernels leave their host process group unchanged by default.
 
-Rust language implementations use the `Language` and `LanguageSession` traits directly. `ExecutionContext` provides stream, display, stdin, interrupt, unlock, and temporary-subshell access without exposing Jupyter transport details.
+Rust language implementations use the `Language` and `LanguageSession` traits directly. `ExecutionContext` provides stream, display, stdin, interrupt, and subshell routing without exposing Jupyter transport details.
 
 `DapClient` is the optional language-neutral debugger transport: framed TCP, request correlation, timeouts, asynchronous events, and shutdown. Language adapters retain debugger startup, request policy, source mapping, and variable semantics.
 
diff --git a/kernmini/__init__.py b/kernmini/__init__.py
index 4933a53..df503ce 100644
--- a/kernmini/__init__.py
+++ b/kernmini/__init__.py
@@ -2,7 +2,7 @@
 
 import asyncio
 
-from .concur import unlock, subshell
+from .concur import sidecar, subshell
 from .kernelspec import install_kernelspec, install_kernelspec_dir
 
 
diff --git a/kernmini/_bridge.py b/kernmini/_bridge.py
index 1cd0ad6..9324346 100644
--- a/kernmini/_bridge.py
+++ b/kernmini/_bridge.py
@@ -1,6 +1,6 @@
 import asyncio, contextvars
 from contextlib import nullcontext
-from .concur import _release, _subshell, subshell, unlock
+from .concur import _subshell, sidecar, subshell
 
 _current = contextvars.ContextVar("kernmini.execution", default=None)
 
@@ -13,8 +13,8 @@ def send(self, msg_type, parent=None, content=None, metadata=None, ident=None, b
 
 class NativeKernel:
     def __init__(self, target): self.target,self.iopub = target,_IOPub()
-    def unlock(self): return unlock()
     def subshell(self): return subshell()
+    def sidecar(self): return sidecar()
     def current_parent(self):
         sink = _current.get()
         return sink.parent() if sink is not None else {}
@@ -29,7 +29,6 @@ def kernel_proxy(target): return NativeKernel(target)
 async def execute(target, current, sink, code, **kwargs):
     "Run one Python execution with its task-local routing and capture context."
     token = current.set(sink)
-    release_token = _release.set(sink.unlock)
     subshell_token = _subshell.set(sink)
     sink.started(asyncio.current_task())
     try:
@@ -38,7 +37,6 @@ async def execute(target, current, sink, code, **kwargs):
         with context: return await target.execute(code, **kwargs)
     finally:
         _subshell.reset(subshell_token)
-        _release.reset(release_token)
         current.reset(token)
 
 
diff --git a/kernmini/concur.py b/kernmini/concur.py
index d4cfb5f..58bb50b 100644
--- a/kernmini/concur.py
+++ b/kernmini/concur.py
@@ -1,20 +1,11 @@
-"In-cell opt-ins for concurrent execution: unlock() and subshell()."
+"In-cell routing to temporary and persistent subshells."
 
 import contextvars
 from contextlib import contextmanager
 
-_release = contextvars.ContextVar("kernmini_release", default=None)
 _subshell = contextvars.ContextVar("kernmini_subshell", default=None)
 
 
-def unlock()->bool:
-    "Let queued shell messages run while the current cell awaits; irreversible for the rest of the cell."
-    release = _release.get()
-    if release is None: return False
-    release()
-    return True
-
-
 @contextmanager
 def subshell():
     "Run execute_requests arriving from this cell's client session in a fresh subshell while the body runs."
@@ -22,4 +13,14 @@ def subshell():
     if sub is None: raise RuntimeError("subshell() only works inside a cell running under a kernmini kernel")
     sid = sub.open_subshell()
     try: yield sid
-    finally: sub.close_subshell(sid)
+    finally: sub.close_subshell(sid, delete=True)
+
+
+@contextmanager
+def sidecar():
+    "Route execute requests from this cell's client session through the persistent sidecar."
+    sub = _subshell.get()
+    if sub is None: raise RuntimeError("sidecar() only works inside a cell running under a kernmini kernel")
+    sid = sub.open_subshell("sidecar")
+    try: yield sid
+    finally: sub.close_subshell(sid, delete=False)
diff --git a/src/engine.rs b/src/engine.rs
index 0775dc1..a4d4c9f 100644
--- a/src/engine.rs
+++ b/src/engine.rs
@@ -8,7 +8,7 @@ use crate::wire::{Message, Session};
 use bytes::Bytes;
 use serde_json::{Value, json};
 use std::cmp::Ordering;
-use std::collections::{BinaryHeap, HashMap, hash_map::Entry};
+use std::collections::{BinaryHeap, HashMap};
 use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;
@@ -114,7 +114,6 @@ struct ShellServices<L> {
     connection: Arc<Value>,
     supports_subshells: bool,
     config: KernelConfig,
-    unlocks: mpsc::UnboundedSender<String>,
     subshells: mpsc::UnboundedSender<SessionCommand>,
 }
 
@@ -122,14 +121,13 @@ impl<L: LanguageSession> ShellServices<L> {
     fn output_context(&self, request: &Message, identity: Option<Bytes>, silent: bool, interrupt: ExecutionInterrupt) -> ExecutionContext {
         let (events, mut output) = event_channel(self.config.iopub_capacity);
         let execution = identity.is_some();
-        let unlock = execution.then(|| (request.msg_id().to_owned(), self.unlocks.clone()));
         let client_session = request.header.get("session").and_then(Value::as_str).unwrap_or("").to_owned();
         let subshells = execution.then(|| (client_session, self.subshells.clone()));
         let parent = json!({
             "header": request.header, "parent_header": request.parent_header,
             "metadata": request.metadata, "content": request.content,
         });
-        let context = ExecutionContext::new(events, interrupt, unlock, subshells, parent);
+        let context = ExecutionContext::new(events, interrupt, subshells, parent);
         let iopub = self.iopub.clone();
         let session = self.session.clone();
         let request = request.clone();
@@ -492,13 +490,13 @@ async fn wait_hold(deadline: Option<tokio::time::Instant>) {
     if let Some(deadline) = deadline { tokio::time::sleep_until(deadline).await } else { std::future::pending().await }
 }
 
-fn pop_runnable(queue: &mut BinaryHeap<QueueItem>, execution_locked: bool, held: Option<&Held>) -> Option<QueueItem> {
+fn pop_runnable(queue: &mut BinaryHeap<QueueItem>, execution_active: bool, held: Option<&Held>) -> Option<QueueItem> {
     let mut parked = vec![];
     let runnable = loop {
         let Some(item) = queue.pop() else { break None };
         let execute = item.inbound.message.msg_type() == "execute_request";
         let above_hold = held.is_none_or(|hold| item.priority > hold.item.priority);
-        if !execute || (!execution_locked && above_hold) { break Some(item); }
+        if !execute || (!execution_active && above_hold) { break Some(item); }
         parked.push(item);
     };
     queue.extend(parked);
@@ -512,8 +510,6 @@ struct Shell<L: LanguageSession> {
     queue: BinaryHeap<QueueItem>,
     order: u64,
     held: Option<Held>,
-    locked: Option<String>,
-    unlocks: mpsc::UnboundedReceiver<String>,
     executions: JoinSet<anyhow::Result<ExecutionDone>>,
     active: HashMap<String, ExecutionInterrupt>,
     stopping: Option<oneshot::Sender<()>>,
@@ -539,8 +535,6 @@ impl<L: LanguageSession> Shell<L> {
         Ok(())
     }
 
-    fn unlock(&mut self, msg_id: Option<String>) { if self.locked.as_deref() == msg_id.as_deref() { self.locked = None } }
-
     async fn apply_control(&mut self, control: ShellControl) -> anyhow::Result<()> {
         match control {
             ShellControl::Release { msg_id, status: release_status, complete } => {
@@ -573,7 +567,6 @@ impl<L: LanguageSession> Shell<L> {
 
     async fn execution_done(&mut self, done: ExecutionDone) -> anyhow::Result<()> {
         self.active.remove(&done.msg_id);
-        if self.locked.as_deref() == Some(done.msg_id.as_str()) { self.locked = None }
         if done.failed && done.stop_on_error && !self.interrupting { self.abort_pending().await? }
         Ok(())
     }
@@ -608,7 +601,6 @@ impl<L: LanguageSession> Shell<L> {
             let interrupt = ExecutionInterrupt::default();
             self.active.insert(msg_id.clone(), interrupt.clone());
             self.executions.spawn(run_execution(self.services.clone(), item.inbound, interrupt));
-            self.locked = Some(msg_id);
         }
         Ok(false)
     }
@@ -617,7 +609,6 @@ impl<L: LanguageSession> Shell<L> {
         loop {
             while let Ok(inbound) = self.incoming.try_recv() { self.enqueue(inbound) }
             while let Ok(control) = self.controls.try_recv() { self.apply_control(control).await? }
-            while let Ok(msg_id) = self.unlocks.try_recv() { self.unlock(Some(msg_id)) }
             while let Some(result) = self.executions.try_join_next() { self.execution_done(result??).await? }
             if self.interrupting && self.executions.is_empty() { self.interrupting = false }
             if self.stopping.is_some() && self.queue.is_empty() && self.executions.is_empty() && self.held.is_none() {
@@ -626,7 +617,7 @@ impl<L: LanguageSession> Shell<L> {
                 return Ok(());
             }
 
-            if let Some(item) = pop_runnable(&mut self.queue, self.locked.is_some(), self.held.as_ref()) {
+            if let Some(item) = pop_runnable(&mut self.queue, !self.executions.is_empty(), self.held.as_ref()) {
                 if self.handle_item(item).await? { return Ok(()); }
                 continue;
             }
@@ -634,7 +625,6 @@ impl<L: LanguageSession> Shell<L> {
             tokio::select! {
                 message = self.incoming.recv() => self.enqueue(message.ok_or_else(|| anyhow::anyhow!("shell service ended"))?),
                 control = self.controls.recv() => self.apply_control(control.ok_or_else(|| anyhow::anyhow!("shell control ended"))?).await?,
-                unlocked = self.unlocks.recv() => self.unlock(unlocked),
                 result = self.executions.join_next(), if !self.executions.is_empty() => {
                     self.execution_done(result.expect("non-empty execution set")??).await?;
                 }
@@ -665,7 +655,6 @@ impl KernelServices {
     fn spawn_shell(&self, language: impl LanguageSession) -> ShellHandle {
         let (incoming, requests) = mpsc::channel(256);
         let (controls, shell_controls) = mpsc::channel(64);
-        let (unlock_send, unlocks) = mpsc::unbounded_channel();
         let services = ShellServices {
             language,
             iopub: self.iopub.clone(),
@@ -674,7 +663,6 @@ impl KernelServices {
             connection: self.connection.clone(),
             supports_subshells: self.supports_subshells,
             config: self.config,
-            unlocks: unlock_send,
             subshells: self.subshells.clone(),
         };
         let shell = Shell {
@@ -684,8 +672,6 @@ impl KernelServices {
             queue: BinaryHeap::new(),
             order: 0,
             held: None,
-            locked: None,
-            unlocks,
             executions: JoinSet::new(),
             active: HashMap::new(),
             stopping: None,
@@ -697,13 +683,11 @@ impl KernelServices {
 
 fn subshell_id(request: &Message) -> &str { request.header.get("subshell_id").and_then(Value::as_str).filter(|id| !id.is_empty()).unwrap_or("") }
 
-async fn reply_subshell_not_found(iopub: &Iopub, session: &Session, inbound: Inbound) -> anyhow::Result<()> {
+async fn reply_subshell_error(iopub: &Iopub, session: &Session, inbound: Inbound, ename: &str, evalue: String) -> anyhow::Result<()> {
     let request = inbound.message;
     if !request.msg_type().ends_with("_request") { return Ok(()); }
-    let id = subshell_id(&request);
     let mut content = json!({
-        "status": "error", "ename": "SubshellNotFound",
-        "evalue": format!("Unknown subshell_id {id:?}"), "traceback": [],
+        "status": "error", "ename": ename, "evalue": evalue, "traceback": [],
     });
     if request.msg_type() == "execute_request" {
         content["execution_count"] = json!(0);
@@ -718,9 +702,16 @@ async fn reply_subshell_not_found(iopub: &Iopub, session: &Session, inbound: Inb
     Ok(())
 }
 
-async fn route_shell(
+async fn reply_subshell_not_found(iopub: &Iopub, session: &Session, inbound: Inbound) -> anyhow::Result<()> {
+    let id = subshell_id(&inbound.message).to_owned();
+    reply_subshell_error(iopub, session, inbound, "SubshellNotFound", format!("Unknown subshell_id {id:?}")).await
+}
+
+async fn route_shell<L: Language>(
     inbound: Inbound,
-    shells: &HashMap<String, ShellHandle>,
+    language: &L,
+    services: &KernelServices,
+    shells: &mut HashMap<String, ShellHandle>,
     route_overrides: &HashMap<String, String>,
     iopub: &Iopub,
     session: &Session,
@@ -728,6 +719,11 @@ async fn route_shell(
     let explicit = subshell_id(&inbound.message);
     let client_session = inbound.message.header.get("session").and_then(Value::as_str).unwrap_or("");
     let id = if !explicit.is_empty() { explicit.to_owned() } else if inbound.message.msg_type() == "execute_request" { route_overrides.get(client_session).cloned().unwrap_or_default() } else { String::new() };
+    if !explicit.is_empty() && !shells.contains_key(&id) {
+        if let Err(error) = create_subshell(language, services, shells, Some(id.clone())).await {
+            return reply_subshell_error(iopub, session, inbound, "SubshellCreationError", error.to_string()).await;
+        }
+    }
     if let Some(target) = shells.get(&id) {
         if let Err(error) = target.incoming.send(inbound).await { reply_subshell_not_found(iopub, session, error.0).await? }
     }
@@ -735,14 +731,16 @@ async fn route_shell(
     Ok(())
 }
 
-async fn route_pending_shells(
+async fn route_pending_shells<L: Language>(
     shell: &mut mpsc::Receiver<Inbound>,
-    shells: &HashMap<String, ShellHandle>,
+    language: &L,
+    services: &KernelServices,
+    shells: &mut HashMap<String, ShellHandle>,
     route_overrides: &HashMap<String, String>,
     iopub: &Iopub,
     session: &Session,
 ) -> anyhow::Result<()> {
-    while let Ok(inbound) = shell.try_recv() { route_shell(inbound, shells, route_overrides, iopub, session).await? }
+    while let Ok(inbound) = shell.try_recv() { route_shell(inbound, language, services, shells, route_overrides, iopub, session).await? }
     Ok(())
 }
 
@@ -759,6 +757,20 @@ async fn interrupt_shells(stdin: &Stdin, shells: &HashMap<String, ShellHandle>)
     for shell in shells.values() { let _ = shell.controls.send(ShellControl::Interrupt).await; }
 }
 
+async fn create_subshell(
+    language: &impl Language,
+    services: &KernelServices,
+    shells: &mut HashMap<String, ShellHandle>,
+    requested_id: Option<String>,
+) -> anyhow::Result<String> {
+    let id = requested_id.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
+    if id.is_empty() { anyhow::bail!("subshell_id cannot be empty") }
+    if shells.contains_key(&id) { return Ok(id); }
+    let child = language.create_child().await?;
+    shells.insert(id.clone(), services.spawn_shell(child));
+    Ok(id)
+}
+
 pub async fn run_kernel(connection_file: impl AsRef<Path>, language: impl Language) -> anyhow::Result<()> {
     let interrupt = KernelInterrupter::default();
     let signal_interrupt = interrupt.clone();
@@ -829,33 +841,31 @@ pub async fn run_kernel_with_interrupter(connection_file: impl AsRef<Path>, lang
             }
             message = shell.recv() => {
                 let inbound = message.ok_or_else(|| anyhow::anyhow!("shell service ended"))?;
-                route_shell(inbound, &shells, &route_overrides, &iopub, &session).await?;
+                route_shell(inbound, &language, &kernel_services, &mut shells, &route_overrides, &iopub, &session).await?;
                 continue;
             }
             command = subshell_commands.recv() => {
                 match command.ok_or_else(|| anyhow::anyhow!("subshell command service ended"))? {
-                    SessionCommand::Open { client_session, complete } => {
-                        let result = match route_overrides.entry(client_session) {
-                            Entry::Occupied(_) => Err(anyhow::anyhow!("this client session already has a temporary subshell")),
-                            Entry::Vacant(route) => match language.create_child().await {
-                                Ok(child) => {
-                                    let id = uuid::Uuid::new_v4().to_string();
-                                    shells.insert(id.clone(), kernel_services.spawn_shell(child));
-                                    route.insert(id.clone());
-                                    Ok(id)
-                                }
-                                Err(error) => Err(error),
-                            },
+                    SessionCommand::Open { client_session, subshell_id, complete } => {
+                        let result = if route_overrides.contains_key(&client_session) {
+                            Err(anyhow::anyhow!("this client session already has a subshell route"))
+                        } else {
+                            create_subshell(&language, &kernel_services, &mut shells, subshell_id).await.map(|id| {
+                                route_overrides.insert(client_session, id.clone());
+                                id
+                            })
                         };
                         let _ = complete.send(result);
                     }
-                    SessionCommand::Close { client_session, subshell_id, complete } => {
+                    SessionCommand::Close { client_session, subshell_id, delete, complete } => {
                         let result = if route_overrides.get(&client_session) == Some(&subshell_id) {
                             route_overrides.remove(&client_session);
-                            if let Some(shell) = shells.remove(&subshell_id) { stop_shell(shell, false).await }
+                            if delete {
+                                if let Some(shell) = shells.remove(&subshell_id) { stop_shell(shell, false).await }
+                            }
                             Ok(())
                         } else {
-                            Err(anyhow::anyhow!("temporary subshell is not active"))
+                            Err(anyhow::anyhow!("subshell route is not active"))
                         };
                         let _ = complete.send(result);
                     }
@@ -875,7 +885,7 @@ pub async fn run_kernel_with_interrupter(connection_file: impl AsRef<Path>, lang
                 return Ok(());
             }
             "interrupt_request" => {
-                route_pending_shells(&mut shell, &shells, &route_overrides, &iopub, &session).await?;
+                route_pending_shells(&mut shell, &language, &kernel_services, &mut shells, &route_overrides, &iopub, &session).await?;
                 interrupt_shells(&stdin, &shells).await;
                 send_reply(&reply, &session, &request, "interrupt_reply", json!({"status": "ok"})).await?;
             }
@@ -893,12 +903,9 @@ pub async fn run_kernel_with_interrupter(connection_file: impl AsRef<Path>, lang
                     .await?;
                     continue;
                 }
-                let id = uuid::Uuid::new_v4().to_string();
-                let content = match language.create_child().await {
-                    Ok(child) => {
-                        shells.insert(id.clone(), kernel_services.spawn_shell(child));
-                        json!({"status": "ok", "subshell_id": id})
-                    }
+                let requested_id = request.content.get("subshell_id").and_then(Value::as_str).map(str::to_owned);
+                let content = match create_subshell(&language, &kernel_services, &mut shells, requested_id).await {
+                    Ok(id) => json!({"status": "ok", "subshell_id": id}),
                     Err(error) => json!({"status": "error", "ename": "SubshellCreationError", "evalue": error.to_string(), "traceback": []}),
                 };
                 send_reply(&reply, &session, &request, "create_subshell_reply", content).await?;
@@ -917,7 +924,7 @@ pub async fn run_kernel_with_interrupter(connection_file: impl AsRef<Path>, lang
                 send_reply(&reply, &session, &request, "delete_subshell_reply", content).await?;
             }
             "release_request" => {
-                route_pending_shells(&mut shell, &shells, &route_overrides, &iopub, &session).await?;
+                route_pending_shells(&mut shell, &language, &kernel_services, &mut shells, &route_overrides, &iopub, &session).await?;
                 let msg_id = request.content.get("msg_id").and_then(Value::as_str).unwrap_or("").to_owned();
                 let release_status = request.content.get("status").and_then(Value::as_str).unwrap_or("ok").to_owned();
                 let mut found = false;
diff --git a/src/language.rs b/src/language.rs
index 10e842d..b6ab473 100644
--- a/src/language.rs
+++ b/src/language.rs
@@ -59,8 +59,8 @@ pub(crate) enum ContextMessage {
 }
 
 pub(crate) enum SessionCommand {
-    Open { client_session: String, complete: std::sync::mpsc::SyncSender<anyhow::Result<String>> },
-    Close { client_session: String, subshell_id: String, complete: std::sync::mpsc::SyncSender<anyhow::Result<()>> },
+    Open { client_session: String, subshell_id: Option<String>, complete: std::sync::mpsc::SyncSender<anyhow::Result<String>> },
+    Close { client_session: String, subshell_id: String, delete: bool, complete: std::sync::mpsc::SyncSender<anyhow::Result<()>> },
 }
 
 pub type InterruptHandler = Arc<dyn Fn() -> anyhow::Result<()> + Send + Sync>;
@@ -106,27 +106,22 @@ impl ExecutionInterrupt {
 pub struct ExecutionContext {
     events: mpsc::Sender<ContextMessage>,
     interrupt: ExecutionInterrupt,
-    unlock: Option<Arc<Unlock>>,
     subshells: Option<Arc<SubshellAccess>>,
     parent: Arc<Value>,
 }
 
-struct Unlock { sent: AtomicBool, execution_id: String, events: mpsc::UnboundedSender<String> }
-
 struct SubshellAccess { client_session: String, commands: mpsc::UnboundedSender<SessionCommand> }
 
 impl ExecutionContext {
     pub(crate) fn new(
         events: mpsc::Sender<ContextMessage>,
         interrupt: ExecutionInterrupt,
-        unlock: Option<(String, mpsc::UnboundedSender<String>)>,
         subshells: Option<(String, mpsc::UnboundedSender<SessionCommand>)>,
         parent: Value,
     ) -> Self {
         Self {
             events,
             interrupt,
-            unlock: unlock.map(|(execution_id, events)| Arc::new(Unlock { sent: AtomicBool::new(false), execution_id, events })),
             subshells: subshells.map(|(client_session, commands)| Arc::new(SubshellAccess { client_session, commands })),
             parent: Arc::new(parent),
         }
@@ -152,23 +147,17 @@ impl ExecutionContext {
         result.recv()?
     }
 
-    pub fn unlock(&self) -> bool {
-        let Some(unlock) = &self.unlock else { return false };
-        if unlock.sent.swap(true, Ordering::AcqRel) { return false; }
-        unlock.events.send(unlock.execution_id.clone()).is_ok()
-    }
-
-    pub fn open_subshell(&self) -> anyhow::Result<String> {
+    pub fn open_subshell(&self, subshell_id: Option<String>) -> anyhow::Result<String> {
         let access = self.subshells.as_ref().ok_or_else(|| anyhow::anyhow!("subshells are not available"))?;
         let (complete, result) = std::sync::mpsc::sync_channel(1);
-        access.commands.send(SessionCommand::Open { client_session: access.client_session.clone(), complete })?;
+        access.commands.send(SessionCommand::Open { client_session: access.client_session.clone(), subshell_id, complete })?;
         result.recv()?
     }
 
-    pub fn close_subshell(&self, subshell_id: String) -> anyhow::Result<()> {
+    pub fn close_subshell(&self, subshell_id: String, delete: bool) -> anyhow::Result<()> {
         let access = self.subshells.as_ref().ok_or_else(|| anyhow::anyhow!("subshells are not available"))?;
         let (complete, result) = std::sync::mpsc::sync_channel(1);
-        access.commands.send(SessionCommand::Close { client_session: access.client_session.clone(), subshell_id, complete })?;
+        access.commands.send(SessionCommand::Close { client_session: access.client_session.clone(), subshell_id, delete, complete })?;
         result.recv()?
     }
 
diff --git a/src/python.rs b/src/python.rs
index 145031f..4c965f8 100644
--- a/src/python.rs
+++ b/src/python.rs
@@ -68,16 +68,16 @@ impl ExecutionSink {
             .map_err(|error| PyRuntimeError::new_err(error.to_string()))
     }
 
-    fn unlock(&self) -> bool { self.context.unlock() }
-
-    fn open_subshell(&self, py: Python<'_>) -> PyResult<String> {
+    #[pyo3(signature = (subshell_id=None))]
+    fn open_subshell(&self, py: Python<'_>, subshell_id: Option<String>) -> PyResult<String> {
         let context = self.context.clone();
-        py.detach(|| context.open_subshell()).map_err(|error| PyRuntimeError::new_err(error.to_string()))
+        py.detach(|| context.open_subshell(subshell_id)).map_err(|error| PyRuntimeError::new_err(error.to_string()))
     }
 
-    fn close_subshell(&self, py: Python<'_>, subshell_id: String) -> PyResult<()> {
+    #[pyo3(signature = (subshell_id, delete=true))]
+    fn close_subshell(&self, py: Python<'_>, subshell_id: String, delete: bool) -> PyResult<()> {
         let context = self.context.clone();
-        py.detach(|| context.close_subshell(subshell_id)).map_err(|error| PyRuntimeError::new_err(error.to_string()))
+        py.detach(|| context.close_subshell(subshell_id, delete)).map_err(|error| PyRuntimeError::new_err(error.to_string()))
     }
 
     fn parent(&self, py: Python<'_>) -> PyResult<Py<PyAny>> { json_to_py(py, &self.context.parent()) }
@@ -438,7 +438,11 @@ impl LanguageSession for PyLanguageSession {
 #[pyfunction]
 #[pyo3(signature = (connection_file, factory, loop_factory, own_process_group=false))]
 fn run_kernel<'py>(
-    py: Python<'py>, connection_file: String, factory: Py<PyAny>, loop_factory: Py<PyAny>, own_process_group: bool,
+    py: Python<'py>,
+    connection_file: String,
+    factory: Py<PyAny>,
+    loop_factory: Py<PyAny>,
+    own_process_group: bool,
 ) -> PyResult<Bound<'py, PyAny>> {
     #[cfg(unix)]
     let owns_process_group = own_process_group
diff --git a/tests/test_rust_ipython_kernel.py b/tests/test_rust_ipython_kernel.py
index 25441ba..d1c344c 100644
--- a/tests/test_rust_ipython_kernel.py
+++ b/tests/test_rust_ipython_kernel.py
@@ -85,27 +85,29 @@ def test_ipython_story(rust_ipython_kernel):
     result, = (m for m in msgs if m["msg_type"] == "execute_result")
     assert result["content"]["data"]["text/plain"] == "42"
 
-    created = _request(control, sess, "create_subshell_request", {}, timeout=30)
-    child = created["content"]["subshell_id"]
-    assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == [child]
+    child = "sidecar"
     child_id = _send(shell, sess, "execute_request", dict(code="x + 1"), subshell_id=child)
     (reply_id, child_reply), = _replies(shell, sess, 1)
     assert reply_id == child_id and child_reply["status"] == "ok" and child_reply["execution_count"] == 1
     msgs = _drain_iopub(sub)
     result, = (m for m in msgs if m["msg_type"] == "execute_result")
     assert result["content"]["data"]["text/plain"] == "42" and result["parent_header"]["subshell_id"] == child
+    assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == [child]
+    created = _request(control, sess, "create_subshell_request", dict(subshell_id=child), timeout=30)
+    again = _request(control, sess, "create_subshell_request", dict(subshell_id=child), timeout=30)
+    assert created["content"] == again["content"] == dict(status="ok", subshell_id=child)
     assert _request(control, sess, "delete_subshell_request", dict(subshell_id=child))["content"]["status"] == "ok"
     assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == []
 
-    caller = _send(shell, sess, "execute_request", dict(code="import asyncio\nfrom ipymini import subshell\nloop = asyncio.get_running_loop()\ngate2 = asyncio.Event()\nwith subshell():\n    print('subshell ready', flush=True)\n    await asyncio.wait_for(gate2.wait(), 5)"))
-    _wait_stream(sub, "subshell ready")
+    caller = _send(shell, sess, "execute_request", dict(code="import asyncio\nfrom ipymini import sidecar\nloop = asyncio.get_running_loop()\ngate2 = asyncio.Event()\nwith sidecar():\n    print('sidecar ready', flush=True)\n    await asyncio.wait_for(gate2.wait(), 5)"))
+    _wait_stream(sub, "sidecar ready")
     routed = _send(shell, sess, "execute_request", dict(code="loop.call_soon_threadsafe(gate2.set)"))
     replies = dict(_replies(shell, sess, 2))
     assert replies[caller]["status"] == replies[routed]["status"] == "ok"
     assert replies[routed]["execution_count"] == 1
     _drain_iopub(sub)
     _drain_iopub(sub)
-    assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == []
+    assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == ["sidecar"]
 
     input_id = _send(shell, sess, "execute_request", dict(code="print(input('Name: '))", allow_stdin=True))
     assert stdin.poll(10_000), "no input_request"
@@ -145,14 +147,6 @@ def test_ipython_story(rust_ipython_kernel):
     background, = (m for m in msgs if m["msg_type"] == "stream" and m["content"]["text"] == "background\n")
     assert background["parent_header"]["msg_id"] == task_id
 
-    waiter = _send(shell, sess, "execute_request", dict(code="from ipymini import unlock\ngate = asyncio.Event()\nassert unlock()\nprint('unlocked', flush=True)\nawait asyncio.wait_for(gate.wait(), 5)"))
-    _wait_stream(sub, "unlocked")
-    setter = _send(shell, sess, "execute_request", dict(code="gate.set()"))
-    replies = dict(_replies(shell, sess, 2))
-    assert replies[waiter]["status"] == replies[setter]["status"] == "ok"
-    _drain_iopub(sub)
-    _drain_iopub(sub)
-
     sleeper = _send(shell, sess, "execute_request", dict(code="print('sleeping', flush=True)\nawait asyncio.sleep(.3)"))
     _wait_stream(sub, "sleeping")
     completer = _send(shell, sess, "complete_request", dict(code="x.rea", cursor_pos=5))