Phase 2 core: ClaudeDriver over stream-json, permissions and questions on the phone

The second driver behind the same trait: claude -p with stream-json both
ways, the hidden --permission-prompt-tool stdio flag (without which no
permission ever reaches a client), text deltas streamed from raw API
events, tool_use/tool_result mapped to tool events, and can_use_tool
control requests surfaced as Question events -- plain permissions as
Allow/Deny, AskUserQuestion as one Question per sub-question with the
chosen labels sent back in updatedInput.answers keyed by question text
(wire shapes pinned by live probes against CLI 2.1.237, recorded in the
module doc). The CLI session id is persisted per session dir, so a
backend restart respawns with --resume and loses nothing. set_model
rides the control protocol and persists through the manager; the spawn
screen grows model/cwd/permission-mode fields.

Also: the dev CA now carries proper keyUsage/basicConstraints
extensions (strict verifiers reject it otherwise) -- regenerated and
re-pinned before any real phone has installed the app.

Verified: 20 unit tests + clippy clean; scripted end-to-end over the
HTTP API (AskUserQuestion round trip, Bash permission allow, streaming,
restart with --resume remembering earlier work, delete); and on the
emulator, a live haiku session asking Tea-or-coffee and acknowledging
the tapped answer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017xn8nHw1tw1R6PtiY1eEtw

app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt

@@ -111,12 +111,23 @@ fun fetchSessions(settings: ServerSettings): List<SessionSummary> =
     }
 
 /** Spawns a session and returns it as the list would show it. */
-fun spawnSession(settings: ServerSettings, kind: String, title: String): SessionSummary =
+fun spawnSession(
+    settings: ServerSettings,
+    kind: String,
+    title: String,
+    model: String? = null,
+    cwd: String? = null,
+    permissionMode: String? = null,
+): SessionSummary =
     requestFromServer(
         settings,
         "/sessions",
         method = "POST",
-        jsonBody = JSONObject().put("kind", kind).put("title", title).toString(),
+        jsonBody = JSONObject().put("kind", kind).put("title", title).apply {
+            if (!model.isNullOrBlank()) put("model", model)
+            if (!cwd.isNullOrBlank()) put("cwd", cwd)
+            if (!permissionMode.isNullOrBlank()) put("permissionMode", permissionMode)
+        }.toString(),
     ) { connection ->
         val session = JSONObject(connection.inputStream.bufferedReader().readText())
         SessionSummary(

app/androidApp/src/main/kotlin/com/example/aiapp/PinnedCert.kt

@@ -23,16 +23,16 @@ import javax.net.ssl.X509TrustManager
  * the APK, so photographing the terminal leaks only the (rotatable) token.
  */
 const val PINNED_CA_PEM = """-----BEGIN CERTIFICATE-----
-MIIBrjCCAVWgAwIBAgIUGomDFgIrkwxX9504lixc8kM83I0wCgYIKoZIzj0EAwIw
+MIIBvzCCAWWgAwIBAgIUGCxNZPJIfzGQipZGxWVZ8vMYZXgwCgYIKoZIzj0EAwIw
 LTETMBEGA1UECgwKYWktYXBwIGRldjEWMBQGA1UEAwwNYWktYXBwIGRldiBDQTAe
-Fw0yNjA4MjUwMDQ5NDBaFw0zNjA4MjIwMDQ5NDBaMC0xEzARBgNVBAoMCmFpLWFw
+Fw0yNjA4MjUwMTI2MDRaFw0zNjA4MjIwMTI2MDRaMC0xEzARBgNVBAoMCmFpLWFw
 cCBkZXYxFjAUBgNVBAMMDWFpLWFwcCBkZXYgQ0EwWTATBgcqhkjOPQIBBggqhkjO
-PQMBBwNCAARW8deDZhiVxUDo1TyGMIpOpvu45vei8Vd5rWFNgSOl80h8TQ8/v8fI
+PQMBBwNCAARE6qRKz1HeCzcvmdT6ztwTR2w4DGP97aaYJhp3z+es6dceNXdpP1qx
-tcacAGiPK0OUDOPb6iSaSMS8QEtfPB8+o1MwUTAdBgNVHQ4EFgQUFnEtKeV8et8Z
+3DlazArgYLjOcNOHTqonj4H5NwHfeP4So2MwYTAdBgNVHQ4EFgQUL9VAISmEPDhJ
-O7/ihnMOckS45qwwHwYDVR0jBBgwFoAUFnEtKeV8et8ZO7/ihnMOckS45qwwDwYD
+dHnl5iS10qLbcekwHwYDVR0jBBgwFoAUL9VAISmEPDhJdHnl5iS10qLbcekwDwYD
-VR0TAQH/BAUwAwEB/zAKBggqhkjOPQQDAgNHADBEAiBppx+X09AjEJ8X24KxHYaX
+VR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAQYwCgYIKoZIzj0EAwIDSAAwRQIh
-4NRE+8OlxqJCrkuAM7tLagIgND6JvA5K0WjlfZxomla45C5Vd91j2jdIPmM+UkEa
+AL7Z0wfT0pXD08J6GNbPVfy/PB3EoUtIwA9z9rYGPEhZAiBeLPiXCvIbWWTpsjRr
-Wb4=
+Uk5VTHJchx0xXCdTRSJo2BEh7Q==
 -----END CERTIFICATE-----
 """
 

app/androidApp/src/main/kotlin/com/example/aiapp/SpawnScreen.kt

@@ -1,5 +1,6 @@
 package com.example.aiapp
 
+import androidx.compose.foundation.layout.Arrangement
 import androidx.compose.foundation.layout.Column
 import androidx.compose.foundation.layout.Row
 import androidx.compose.foundation.layout.Spacer
@@ -7,6 +8,8 @@ import androidx.compose.foundation.layout.fillMaxSize
 import androidx.compose.foundation.layout.fillMaxWidth
 import androidx.compose.foundation.layout.height
 import androidx.compose.foundation.layout.padding
+import androidx.compose.foundation.rememberScrollState
+import androidx.compose.foundation.verticalScroll
 import androidx.compose.material3.Button
 import androidx.compose.material3.FilterChip
 import androidx.compose.material3.MaterialTheme
@@ -26,13 +29,19 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 
-// The session kinds this build can spawn. Phase 2 adds "claude" (with
+// The session kinds this build can spawn; phase 4 adds "pi". A new kind is
-// model / working directory / permission-mode fields), phase 4 "pi" --
+// another entry here plus its fields below -- never a parallel screen.
-// extending this list and the per-kind fields, not adding a parallel
+private val KINDS = listOf("claude", "echo")
-// screen.
-private val KINDS = listOf("echo")
 
-/** The spawn screen: kind, title, go. */
+// Claude Code 2.x permission modes. "manual" asks for everything (each ask
+// arrives on the phone as a question card); the others are the CLI's own
+// escalating levels of autonomy.
+private val PERMISSION_MODES = listOf("manual", "acceptEdits", "auto", "bypassPermissions", "plan")
+
+// Model shortcuts the CLI accepts; free text is also fine (full model ids).
+private val CLAUDE_MODELS = listOf("default", "fable", "opus", "sonnet", "haiku")
+
+/** The spawn screen: kind, per-kind fields, go. */
 @Composable
 fun SpawnScreen(
     settings: ServerSettings,
@@ -42,10 +51,13 @@ fun SpawnScreen(
     val scope = rememberCoroutineScope()
     var kind by remember { mutableStateOf(KINDS.first()) }
     var title by remember { mutableStateOf("") }
+    var model by remember { mutableStateOf("default") }
+    var cwd by remember { mutableStateOf("") }
+    var permissionMode by remember { mutableStateOf("manual") }
     var busy by remember { mutableStateOf(false) }
     var error by remember { mutableStateOf<String?>(null) }
 
-    Column(Modifier.fillMaxSize().padding(16.dp)) {
+    Column(Modifier.fillMaxSize().verticalScroll(rememberScrollState()).padding(16.dp)) {
         Row(verticalAlignment = Alignment.CenterVertically, modifier = Modifier.fillMaxWidth()) {
             Text(
                 "New session",
@@ -57,14 +69,13 @@ fun SpawnScreen(
         Spacer(Modifier.height(16.dp))
 
         Text("Kind", style = MaterialTheme.typography.labelLarge)
-        Row {
+        Row(horizontalArrangement = Arrangement.spacedBy(8.dp)) {
             KINDS.forEach { candidate ->
                 FilterChip(
                     selected = kind == candidate,
                     onClick = { kind = candidate },
                     label = { Text(candidate) },
                 )
-                Spacer(Modifier.height(0.dp))
             }
         }
         Spacer(Modifier.height(16.dp))
@@ -73,10 +84,49 @@ fun SpawnScreen(
             value = title,
             onValueChange = { title = it },
             label = { Text("Title") },
-            placeholder = { Text("Echo session") },
             singleLine = true,
             modifier = Modifier.fillMaxWidth(),
         )
+
+        if (kind == "claude") {
+            Spacer(Modifier.height(16.dp))
+            Text("Model", style = MaterialTheme.typography.labelLarge)
+            Row(horizontalArrangement = Arrangement.spacedBy(8.dp)) {
+                CLAUDE_MODELS.forEach { candidate ->
+                    FilterChip(
+                        selected = model == candidate,
+                        onClick = { model = candidate },
+                        label = { Text(candidate) },
+                    )
+                }
+            }
+            Spacer(Modifier.height(16.dp))
+
+            OutlinedTextField(
+                value = cwd,
+                onValueChange = { cwd = it },
+                label = { Text("Working directory") },
+                placeholder = { Text("/home/…") },
+                singleLine = true,
+                modifier = Modifier.fillMaxWidth(),
+            )
+            Spacer(Modifier.height(16.dp))
+
+            Text("Permissions", style = MaterialTheme.typography.labelLarge)
+            // Two rows rather than horizontal scroll: all the choices stay
+            // visible, and bypassPermissions shouldn't be pickable blind.
+            for (chunk in PERMISSION_MODES.chunked(3)) {
+                Row(horizontalArrangement = Arrangement.spacedBy(8.dp)) {
+                    chunk.forEach { candidate ->
+                        FilterChip(
+                            selected = permissionMode == candidate,
+                            onClick = { permissionMode = candidate },
+                            label = { Text(candidate) },
+                        )
+                    }
+                }
+            }
+        }
         Spacer(Modifier.height(24.dp))
 
         error?.let {
@@ -90,7 +140,14 @@ fun SpawnScreen(
                 scope.launch {
                     try {
                         val spawned = withContext(Dispatchers.IO) {
-                            spawnSession(settings, kind, title.trim())
+                            spawnSession(
+                                settings,
+                                kind,
+                                title.trim(),
+                                model = model.takeIf { kind == "claude" && it != "default" },
+                                cwd = cwd.takeIf { kind == "claude" },
+                                permissionMode = permissionMode.takeIf { kind == "claude" },
+                            )
                         }
                         onSpawned(spawned)
                     } catch (e: ApiException) {

gen-dev-cert.sh

@@ -61,8 +61,13 @@ if [ -f ca.pem ]; then
 else
     echo "==> Generating CA key + self-signed CA certificate"
     openssl ecparam -name prime256v1 -genkey -noout -out ca-key.pem
+    # Explicit keyUsage: strict verifiers (e.g. Python 3.14's ssl) reject a
+    # CA without it, and Android could follow -- cheap to be proper now,
+    # expensive to regenerate after phones have pinned it.
     openssl req -new -x509 -key ca-key.pem -out ca.pem -days 3650 \
-        -subj "/O=ai-app dev/CN=ai-app dev CA"
+        -subj "/O=ai-app dev/CN=ai-app dev CA" \
+        -addext "basicConstraints=critical,CA:TRUE" \
+        -addext "keyUsage=critical,keyCertSign,cRLSign"
 fi
 
 echo "==> Generating leaf key + CSR for $SERVER_IP (+ dev addresses)"
@@ -73,6 +78,9 @@ openssl req -new -key leaf-key.pem -out leaf.csr \
 echo "==> Signing leaf certificate with the dev CA"
 cat > leaf.ext <<EOF
 subjectAltName = IP:$SERVER_IP,IP:$LOOPBACK_IP,IP:$EMULATOR_HOST_IP,IP:$LAN_IP
+basicConstraints = CA:FALSE
+keyUsage = digitalSignature
+extendedKeyUsage = serverAuth
 EOF
 openssl x509 -req -in leaf.csr -CA ca.pem -CAkey ca-key.pem -CAcreateserial \
     -out leaf.pem -days 3650 -extfile leaf.ext

server/Cargo.lock

@@ -983,6 +983,16 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
+dependencies = [
+ "errno",
+ "libc",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.12"
@@ -1097,6 +1107,7 @@ dependencies = [
  "libc",
  "mio",
  "pin-project-lite",
+ "signal-hook-registry",
  "socket2",
  "tokio-macros",
  "windows-sys 0.61.2",

server/Cargo.toml

@@ -10,7 +10,7 @@ path = "src/main.rs"
 [dependencies]
 axum = { version = "0.8", features = ["json"] }
 axum-server = { version = "0.8", features = ["tls-rustls"] }
-tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time", "process", "io-util"] }
 tokio-stream = "0.1"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

server/src/config.rs

@@ -37,9 +37,8 @@ pub struct TokenEntry {
     pub sha256: String,
 }
 
-/// Which driver a session runs. Phase 2 adds `Claude`, phase 4 adds `Pi`;
+/// Which driver a session runs. Phase 4 adds `Pi`; a new kind is a new
-/// a new kind is a new driver behind the same trait, never a branch in
+/// driver behind the same trait, never a branch in shared code.
-/// shared code.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum SessionKind {
@@ -47,6 +46,8 @@ pub enum SessionKind {
     /// the whole pipe (spawn, SSE, transcript cursors, questions) with no
     /// AI involved, and stays useful as a connectivity check.
     Echo,
+    /// Claude Code over stream-json (see `session::claude`).
+    Claude,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

server/src/routes.rs

@@ -185,14 +185,12 @@ struct ModelRequest {
     model: String,
 }
 
-/// What happens is the driver's call -- a driver that can't switch in
-/// place reports how it handled it (or that it can't) as events.
 async fn set_model(
     State(manager): State<Arc<SessionManager>>,
     UrlPath(id): UrlPath<String>,
     axum::Json(body): axum::Json<ModelRequest>,
 ) -> Result<StatusCode, ApiError> {
-    lookup(&manager, &id)?.set_model(&body.model);
+    manager.set_session_model(&id, &body.model).map_err(bad_request)?;
     Ok(StatusCode::NO_CONTENT)
 }
 

server/src/session/claude.rs

@@ -0,0 +1,736 @@
+//! The Claude Code driver: `claude -p` speaking stream-json on stdio,
+//! translated into the common event model.
+//!
+//! Wire format pinned against CLI 2.1.237 by probing (2026-08-24; scripts
+//! summarized here since they live outside the repo):
+//!
+//! - Outbound: `system/init` (carries `session_id`, the `--resume` token),
+//!   `stream_event` (raw API deltas; `text_delta` is the streaming text),
+//!   consolidated `assistant` messages (their `tool_use` blocks have the
+//!   complete input), `user` messages with `tool_result` blocks, a `result`
+//!   per turn (usage + cost), `control_request` for anything needing a
+//!   human, `control_response` answering ours.
+//! - Permission prompts require the hidden `--permission-prompt-tool stdio`
+//!   flag; they arrive as `control_request{subtype:can_use_tool}` and are
+//!   answered with `{behavior:"allow",updatedInput}` or
+//!   `{behavior:"deny",message}`. `AskUserQuestion` uses the same shape,
+//!   with the chosen labels added to `updatedInput` as
+//!   `answers:{<question text>:<label>}`.
+//! - Inbound `user` messages sent mid-turn are queued and injected at the
+//!   next tool boundary (verified live: the model acknowledged a steer
+//!   between two Bash calls) -- the behavior this app exists for.
+//! - `control_request{subtype:set_model}` answers success;
+//!   `{subtype:interrupt}` stops the turn.
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::sync::{Arc, Mutex};
+
+use anyhow::{Context, Result};
+use serde_json::{Value, json};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::process::Command;
+use tokio::sync::{mpsc, oneshot};
+
+use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus};
+use crate::config::SessionConfig;
+
+/// Where the driver remembers its CLI session id between backend runs --
+/// the whole crash-recovery story: respawning with `--resume <id>` picks
+/// the conversation back up from Claude's own session files. Kept in the
+/// session directory rather than config.json so the shared schema stays
+/// free of per-driver state.
+const RESUME_FILE: &str = "claude-session.json";
+
+/// Grace period between closing stdin (the polite exit) and SIGKILL.
+const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
+
+pub struct ClaudeDriver {
+    sink: EventSink,
+    /// Lines for the child's stdin; `None` after shutdown started (taking
+    /// it closes stdin, which is the CLI's graceful exit signal).
+    to_child: Mutex<Option<mpsc::UnboundedSender<String>>>,
+    /// Fires SIGKILL if the process outlives the shutdown grace period.
+    kill: Mutex<Option<oneshot::Sender<()>>>,
+    state: Arc<Mutex<Translator>>,
+    session_dir: PathBuf,
+}
+
+impl ClaudeDriver {
+    pub fn spawn(meta: &SessionConfig, session_dir: &Path, sink: EventSink) -> Result<Self> {
+        let mut command = Command::new("claude");
+        command
+            .arg("-p")
+            .arg("--verbose")
+            .args(["--input-format", "stream-json"])
+            .args(["--output-format", "stream-json"])
+            .arg("--include-partial-messages")
+            // Hidden but load-bearing: without it the CLI resolves
+            // permissions itself and nothing ever reaches the phone.
+            .args(["--permission-prompt-tool", "stdio"]);
+        if let Some(model) = &meta.model {
+            command.args(["--model", model]);
+        }
+        if let Some(mode) = &meta.permission_mode {
+            command.args(["--permission-mode", mode]);
+        }
+        if let Some(resume) = read_resume_token(session_dir) {
+            command.args(["--resume", &resume]);
+        }
+        if let Some(cwd) = &meta.cwd {
+            command.current_dir(cwd);
+        }
+        let mut child = command
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true)
+            .spawn()
+            .context("spawn claude (is the CLI installed and on PATH?)")?;
+
+        let stdin = child.stdin.take().expect("piped stdin");
+        let stdout = child.stdout.take().expect("piped stdout");
+        let stderr = child.stderr.take().expect("piped stderr");
+        let state = Arc::new(Mutex::new(Translator::default()));
+
+        // Writer: everything for the child funnels through one channel so
+        // driver methods stay sync and writes can't interleave.
+        let (to_child, mut from_driver) = mpsc::unbounded_channel::<String>();
+        tokio::spawn(async move {
+            let mut stdin = stdin;
+            while let Some(line) = from_driver.recv().await {
+                if stdin.write_all(line.as_bytes()).await.is_err()
+                    || stdin.write_all(b"\n").await.is_err()
+                    || stdin.flush().await.is_err()
+                {
+                    break;
+                }
+            }
+            // Sender dropped/taken: stdin drops here, closing it -- the
+            // CLI's signal to finish up and exit.
+        });
+
+        tokio::spawn(read_stdout(
+            stdout,
+            Arc::clone(&state),
+            sink.clone(),
+            session_dir.to_path_buf(),
+        ));
+
+        // stderr is diagnostics only; surface it in the log, and keep the
+        // last line for the exit report below.
+        let last_stderr = Arc::new(Mutex::new(String::new()));
+        {
+            let last_stderr = Arc::clone(&last_stderr);
+            tokio::spawn(async move {
+                let mut lines = BufReader::new(stderr).lines();
+                while let Ok(Some(line)) = lines.next_line().await {
+                    tracing::warn!("claude stderr: {line}");
+                    *last_stderr.lock().unwrap() = line;
+                }
+            });
+        }
+
+        // Monitor: reports process death as an event (with stderr context
+        // when it died complaining), and carries the SIGKILL escape hatch.
+        let (kill_tx, kill_rx) = oneshot::channel::<()>();
+        {
+            let sink = sink.clone();
+            tokio::spawn(async move {
+                let status = tokio::select! {
+                    status = child.wait() => status.ok(),
+                    _ = kill_rx => {
+                        let _ = child.kill().await;
+                        None
+                    }
+                };
+                if let Some(status) = status
+                    && !status.success()
+                {
+                    let detail = last_stderr.lock().unwrap().clone();
+                    let _ = sink.send(Event::Error {
+                        message: format!(
+                            "claude exited with {status}{}",
+                            if detail.is_empty() { String::new() } else { format!(": {detail}") }
+                        ),
+                    });
+                }
+                let _ = sink.send(Event::Status { state: SessionStatus::Exited });
+            });
+        }
+
+        Ok(Self {
+            sink,
+            to_child: Mutex::new(Some(to_child)),
+            kill: Mutex::new(Some(kill_tx)),
+            state,
+            session_dir: session_dir.to_path_buf(),
+        })
+    }
+
+    fn send_line(&self, line: String) {
+        if let Some(sender) = self.to_child.lock().unwrap().as_ref() {
+            let _ = sender.send(line);
+        }
+    }
+
+    fn send_control(&self, request: Value) {
+        let id = format!("req-{}", super::now() as u64);
+        self.send_line(
+            json!({"type": "control_request", "request_id": id, "request": request}).to_string(),
+        );
+    }
+}
+
+impl Driver for ClaudeDriver {
+    fn send_user_message(&self, text: String, images: Vec<ImageRef>) {
+        let mut content = Vec::new();
+        for id in &images {
+            match attachment_block(&self.session_dir, id) {
+                Ok(block) => content.push(block),
+                Err(err) => {
+                    let _ = self.sink.send(Event::Error {
+                        message: format!("attachment {id} couldn't be sent: {err:#}"),
+                    });
+                }
+            }
+        }
+        if !text.is_empty() {
+            content.push(json!({"type": "text", "text": text}));
+        }
+        // Sent mid-turn this queues for injection at the next tool
+        // boundary; sent while idle it starts a turn.
+        let _ = self.sink.send(Event::Status { state: SessionStatus::Running });
+        self.send_line(
+            json!({"type": "user", "message": {"role": "user", "content": content}}).to_string(),
+        );
+    }
+
+    fn answer_question(&self, id: &str, answer: &str) {
+        let response = {
+            let mut state = self.state.lock().unwrap();
+            state.answer(id, answer)
+        };
+        match response {
+            AnswerOutcome::Respond(control_response) => {
+                let _ = self.sink.send(Event::Status { state: SessionStatus::Running });
+                self.send_line(control_response.to_string());
+            }
+            // A multi-question AskUserQuestion still waiting on the rest.
+            AnswerOutcome::Pending => {}
+            AnswerOutcome::Unknown => {
+                let _ = self.sink.send(Event::Error {
+                    message: format!("no question {id} is awaiting an answer"),
+                });
+            }
+        }
+    }
+
+    fn interrupt(&self) {
+        self.send_control(json!({"subtype": "interrupt"}));
+    }
+
+    fn set_model(&self, model: &str) {
+        self.send_control(json!({"subtype": "set_model", "model": model}));
+    }
+
+    fn compact(&self) {
+        // Slash commands ride the normal user-message channel.
+        self.send_line(
+            json!({"type": "user", "message": {"role": "user", "content": [
+                {"type": "text", "text": "/compact"}
+            ]}})
+            .to_string(),
+        );
+    }
+
+    fn shutdown(&self) {
+        // Closing stdin is the polite exit; the kill timer is the escape
+        // hatch for a CLI that doesn't oblige.
+        self.to_child.lock().unwrap().take();
+        if let Some(kill) = self.kill.lock().unwrap().take() {
+            tokio::spawn(async move {
+                tokio::time::sleep(SHUTDOWN_GRACE).await;
+                let _ = kill.send(());
+            });
+        }
+    }
+}
+
+async fn read_stdout(
+    stdout: tokio::process::ChildStdout,
+    state: Arc<Mutex<Translator>>,
+    sink: EventSink,
+    session_dir: PathBuf,
+) {
+    let mut lines = BufReader::new(stdout).lines();
+    while let Ok(Some(line)) = lines.next_line().await {
+        let Ok(message) = serde_json::from_str::<Value>(&line) else {
+            tracing::warn!("unparseable claude output line: {}", &line[..line.len().min(200)]);
+            continue;
+        };
+        let (events, new_session_id) = {
+            let mut state = state.lock().unwrap();
+            let before = state.session_id.clone();
+            let events = state.translate(&message);
+            let after = state.session_id.clone();
+            (events, if before != after { after } else { None })
+        };
+        if let Some(session_id) = new_session_id {
+            write_resume_token(&session_dir, &session_id);
+        }
+        for event in events {
+            if sink.send(event).is_err() {
+                return; // session torn down
+            }
+        }
+    }
+}
+
+fn read_resume_token(session_dir: &Path) -> Option<String> {
+    let text = std::fs::read_to_string(session_dir.join(RESUME_FILE)).ok()?;
+    serde_json::from_str::<Value>(&text)
+        .ok()?
+        .get("sessionId")?
+        .as_str()
+        .map(String::from)
+}
+
+fn write_resume_token(session_dir: &Path, session_id: &str) {
+    let path = session_dir.join(RESUME_FILE);
+    if let Err(err) = std::fs::write(&path, json!({"sessionId": session_id}).to_string()) {
+        tracing::error!("couldn't persist resume token to {}: {err}", path.display());
+    }
+}
+
+/// Reads an uploaded attachment into an API image content block.
+fn attachment_block(session_dir: &Path, id: &str) -> Result<Value> {
+    // Ids are server-generated hex (see routes::upload_attachment); the
+    // check keeps a crafted "id" from naming an arbitrary file.
+    if !id.chars().all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-') {
+        anyhow::bail!("invalid attachment id");
+    }
+    let path = session_dir.join("attachments").join(id);
+    let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
+    use base64::Engine;
+    Ok(json!({
+        "type": "image",
+        "source": {
+            "type": "base64",
+            "media_type": media_type_of(id),
+            "data": base64::engine::general_purpose::STANDARD.encode(bytes),
+        }
+    }))
+}
+
+fn media_type_of(name: &str) -> &'static str {
+    match name.rsplit('.').next() {
+        Some("png") => "image/png",
+        Some("gif") => "image/gif",
+        Some("webp") => "image/webp",
+        _ => "image/jpeg",
+    }
+}
+
+/// What answering a question produced.
+enum AnswerOutcome {
+    /// Send this control_response line to the CLI.
+    Respond(Value),
+    /// Part of a multi-question request; more answers still needed.
+    Pending,
+    Unknown,
+}
+
+/// A `can_use_tool` request we've surfaced to the phone and not yet
+/// answered. For plain permissions there is one implicit question
+/// (Allow/Deny); for AskUserQuestion, one per entry in `questions`.
+struct PendingRequest {
+    request_id: String,
+    input: Value,
+    /// Question text per sub-question, in order -- the keys the answers
+    /// map uses. Empty for a plain permission request.
+    questions: Vec<String>,
+    answers: HashMap<String, String>,
+}
+
+/// Pure translation state: stream-json lines in, common events out. No
+/// I/O, so the whole dialect mapping is unit-testable from recorded lines.
+#[derive(Default)]
+struct Translator {
+    session_id: Option<String>,
+    pending: HashMap<String, PendingRequest>,
+}
+
+impl Translator {
+    fn translate(&mut self, message: &Value) -> Vec<Event> {
+        // Events from subagents (Task tool internals) carry a
+        // parent_tool_use_id; the transcript shows the Task tool's own
+        // start/end instead of every nested step.
+        if message.get("parent_tool_use_id").is_some_and(|id| !id.is_null()) {
+            return Vec::new();
+        }
+        match message.get("type").and_then(Value::as_str) {
+            Some("system") => {
+                if message.get("subtype").and_then(Value::as_str) == Some("init")
+                    && let Some(id) = message.get("session_id").and_then(Value::as_str)
+                {
+                    self.session_id = Some(id.to_string());
+                }
+                Vec::new()
+            }
+            Some("stream_event") => self.translate_stream_event(&message["event"]),
+            Some("assistant") => self.translate_assistant(&message["message"]),
+            Some("user") => translate_user(message),
+            Some("control_request") => self.translate_control_request(message),
+            Some("control_response") => {
+                let response = &message["response"];
+                if response.get("subtype").and_then(Value::as_str) == Some("error") {
+                    let error = response.get("error").and_then(Value::as_str).unwrap_or("unknown");
+                    vec![Event::Error { message: format!("claude rejected a request: {error}") }]
+                } else {
+                    Vec::new()
+                }
+            }
+            Some("result") => {
+                let usage = &message["usage"];
+                let tokens = usage.get("input_tokens").and_then(Value::as_u64).unwrap_or(0)
+                    + usage.get("output_tokens").and_then(Value::as_u64).unwrap_or(0);
+                let mut events = Vec::new();
+                if message.get("is_error").and_then(Value::as_bool).unwrap_or(false) {
+                    events.push(Event::Error {
+                        message: message
+                            .get("result")
+                            .and_then(Value::as_str)
+                            .unwrap_or("the turn ended with an error")
+                            .to_string(),
+                    });
+                }
+                if tokens > 0 {
+                    events.push(Event::UsageDelta { tokens });
+                }
+                events.push(Event::Status { state: SessionStatus::Idle });
+                events
+            }
+            _ => Vec::new(),
+        }
+    }
+
+    /// Raw API streaming: only text deltas become events. Consolidated
+    /// blocks arriving later re-carry the same text, so those are skipped
+    /// in `translate_assistant` -- one source per fact.
+    fn translate_stream_event(&mut self, event: &Value) -> Vec<Event> {
+        if event.get("type").and_then(Value::as_str) == Some("content_block_delta")
+            && let Some(delta) = event["delta"].get("text")
+            && event["delta"].get("type").and_then(Value::as_str) == Some("text_delta")
+            && let Some(text) = delta.as_str()
+        {
+            return vec![Event::AssistantText { delta: text.to_string() }];
+        }
+        Vec::new()
+    }
+
+    fn translate_assistant(&mut self, message: &Value) -> Vec<Event> {
+        let Some(content) = message.get("content").and_then(Value::as_array) else {
+            return Vec::new();
+        };
+        content
+            .iter()
+            .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use"))
+            .map(|block| Event::ToolStart {
+                id: block.get("id").and_then(Value::as_str).unwrap_or_default().to_string(),
+                tool: block.get("name").and_then(Value::as_str).unwrap_or_default().to_string(),
+                input: block.get("input").cloned().unwrap_or(Value::Null),
+            })
+            .collect()
+    }
+
+    fn translate_control_request(&mut self, message: &Value) -> Vec<Event> {
+        let request = &message["request"];
+        if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") {
+            return Vec::new();
+        }
+        let request_id =
+            message.get("request_id").and_then(Value::as_str).unwrap_or_default().to_string();
+        let tool_name = request.get("tool_name").and_then(Value::as_str).unwrap_or("a tool");
+        let input = request.get("input").cloned().unwrap_or(Value::Null);
+
+        let mut events = Vec::new();
+        let mut questions = Vec::new();
+        if tool_name == "AskUserQuestion" {
+            for (i, question) in input
+                .get("questions")
+                .and_then(Value::as_array)
+                .into_iter()
+                .flatten()
+                .enumerate()
+            {
+                let text = question
+                    .get("question")
+                    .and_then(Value::as_str)
+                    .unwrap_or("(question)")
+                    .to_string();
+                let options = question
+                    .get("options")
+                    .and_then(Value::as_array)
+                    .into_iter()
+                    .flatten()
+                    .filter_map(|option| option.get("label").and_then(Value::as_str))
+                    .map(String::from)
+                    .collect();
+                events.push(Event::Question {
+                    id: format!("{request_id}#{i}"),
+                    prompt: text.clone(),
+                    options,
+                });
+                questions.push(text);
+            }
+        } else {
+            let summary = serde_json::to_string_pretty(&input).unwrap_or_default();
+            let summary: String = summary.chars().take(600).collect();
+            events.push(Event::Question {
+                id: request_id.clone(),
+                prompt: format!("Allow {tool_name}?\n{summary}"),
+                options: vec!["Allow".to_string(), "Deny".to_string()],
+            });
+        }
+        self.pending.insert(
+            request_id.clone(),
+            PendingRequest { request_id, input, questions, answers: HashMap::new() },
+        );
+        events.push(Event::Status { state: SessionStatus::AwaitingInput });
+        events
+    }
+
+    /// Applies one answer from the phone. Question ids are the control
+    /// request id, suffixed `#i` for AskUserQuestion sub-questions.
+    fn answer(&mut self, question_id: &str, answer: &str) -> AnswerOutcome {
+        let (request_id, sub) = match question_id.split_once('#') {
+            Some((request_id, index)) => (request_id, index.parse::<usize>().ok()),
+            None => (question_id, None),
+        };
+        let Some(pending) = self.pending.get_mut(request_id) else {
+            return AnswerOutcome::Unknown;
+        };
+
+        let response = if let Some(index) = sub {
+            let Some(question) = pending.questions.get(index) else {
+                return AnswerOutcome::Unknown;
+            };
+            pending.answers.insert(question.clone(), answer.to_string());
+            if pending.answers.len() < pending.questions.len() {
+                return AnswerOutcome::Pending;
+            }
+            let mut updated = pending.input.clone();
+            updated["answers"] = serde_json::to_value(&pending.answers).expect("string map");
+            json!({"behavior": "allow", "updatedInput": updated})
+        } else if answer.eq_ignore_ascii_case("deny") {
+            json!({"behavior": "deny", "message": "The user denied this from the phone."})
+        } else {
+            json!({"behavior": "allow", "updatedInput": pending.input})
+        };
+
+        let request_id = pending.request_id.clone();
+        self.pending.remove(&request_id);
+        AnswerOutcome::Respond(json!({
+            "type": "control_response",
+            "response": {"subtype": "success", "request_id": request_id, "response": response},
+        }))
+    }
+}
+
+/// `user` messages: tool results become ToolEnd (with any images saved
+/// out-of-band by the caller -- phase 2b); replayed/synthetic user text is
+/// skipped, since the manager already recorded the user's side.
+fn translate_user(message: &Value) -> Vec<Event> {
+    let Some(content) = message["message"].get("content").and_then(Value::as_array) else {
+        return Vec::new();
+    };
+    content
+        .iter()
+        .filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_result"))
+        .map(|block| {
+            let output = match block.get("content") {
+                Some(Value::String(text)) => text.clone(),
+                Some(Value::Array(parts)) => parts
+                    .iter()
+                    .filter_map(|part| part.get("text").and_then(Value::as_str))
+                    .collect::<Vec<_>>()
+                    .join("\n"),
+                _ => String::new(),
+            };
+            Event::ToolEnd {
+                id: block
+                    .get("tool_use_id")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default()
+                    .to_string(),
+                output,
+            }
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn translate_lines(translator: &mut Translator, lines: &[&str]) -> Vec<Event> {
+        lines
+            .iter()
+            .flat_map(|line| translator.translate(&serde_json::from_str(line).expect("json")))
+            .collect()
+    }
+
+    #[test]
+    fn captures_the_resume_token_from_init() {
+        let mut translator = Translator::default();
+        let events = translate_lines(
+            &mut translator,
+            &[r#"{"type":"system","subtype":"init","cwd":"/x","session_id":"5ecf21da-d53f","tools":[],"model":"claude-haiku-4-5-20251001"}"#],
+        );
+        assert!(events.is_empty());
+        assert_eq!(translator.session_id.as_deref(), Some("5ecf21da-d53f"));
+    }
+
+    #[test]
+    fn streams_text_deltas_and_skips_the_consolidated_copy() {
+        // Real lines (trimmed) from the 2.1.237 probe.
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Done."}},"session_id":"s","parent_tool_use_id":null}"#,
+            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"Done."}]},"parent_tool_use_id":null,"session_id":"s"}"#,
+            r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hmm"}},"session_id":"s","parent_tool_use_id":null}"#,
+        ]);
+        assert_eq!(events, vec![Event::AssistantText { delta: "Done.".to_string() }]);
+    }
+
+    #[test]
+    fn tool_use_and_result_become_tool_events() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo probe-ok"}}]},"parent_tool_use_id":null}"#,
+            r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"probe-ok","is_error":false}]},"parent_tool_use_id":null}"#,
+        ]);
+        assert_eq!(events, vec![
+            Event::ToolStart {
+                id: "toolu_01".to_string(),
+                tool: "Bash".to_string(),
+                input: serde_json::json!({"command": "echo probe-ok"}),
+            },
+            Event::ToolEnd { id: "toolu_01".to_string(), output: "probe-ok".to_string() },
+        ]);
+    }
+
+    #[test]
+    fn subagent_events_are_not_duplicated_into_the_transcript() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_02","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_parent"}"#,
+        ]);
+        assert!(events.is_empty());
+    }
+
+    #[test]
+    fn a_permission_request_becomes_an_allow_deny_question() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"control_request","request_id":"req-1","request":{"subtype":"can_use_tool","tool_name":"Bash","input":{"command":"rm -rf /tmp/x"},"tool_use_id":"toolu_03"}}"#,
+        ]);
+        let Event::Question { id, prompt, options } = &events[0] else {
+            panic!("expected a question, got {events:?}");
+        };
+        assert_eq!(id, "req-1");
+        assert!(prompt.contains("Bash") && prompt.contains("rm -rf /tmp/x"));
+        assert_eq!(options, &["Allow", "Deny"]);
+        assert_eq!(events[1], Event::Status { state: SessionStatus::AwaitingInput });
+
+        // Allowing echoes the input back; the request is then gone.
+        let AnswerOutcome::Respond(response) = translator.answer("req-1", "Allow") else {
+            panic!("expected a control response");
+        };
+        assert_eq!(response["response"]["request_id"], "req-1");
+        assert_eq!(response["response"]["response"]["behavior"], "allow");
+        assert_eq!(
+            response["response"]["response"]["updatedInput"]["command"],
+            "rm -rf /tmp/x"
+        );
+        assert!(matches!(translator.answer("req-1", "Allow"), AnswerOutcome::Unknown));
+    }
+
+    #[test]
+    fn denying_a_permission_sends_deny() {
+        let mut translator = Translator::default();
+        translate_lines(&mut translator, &[
+            r#"{"type":"control_request","request_id":"req-2","request":{"subtype":"can_use_tool","tool_name":"Write","input":{"file_path":"/etc/passwd"}}}"#,
+        ]);
+        let AnswerOutcome::Respond(response) = translator.answer("req-2", "Deny") else {
+            panic!("expected a control response");
+        };
+        assert_eq!(response["response"]["response"]["behavior"], "deny");
+    }
+
+    #[test]
+    fn ask_user_question_rides_the_same_flow_with_answers_keyed_by_question() {
+        // The real 2.1.237 shape, verified live: answers go back inside
+        // updatedInput, keyed by the question text.
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"control_request","request_id":"req-3","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","input":{"questions":[{"question":"Which color?","header":"Color","options":[{"label":"Red"},{"label":"Blue"}],"multiSelect":false},{"question":"Which size?","header":"Size","options":[{"label":"S"},{"label":"L"}],"multiSelect":false}]},"tool_use_id":"toolu_04","requires_user_interaction":true}}"#,
+        ]);
+        let questions: Vec<_> = events
+            .iter()
+            .filter_map(|event| match event {
+                Event::Question { id, prompt, options } => Some((id.clone(), prompt.clone(), options.clone())),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(questions.len(), 2);
+        assert_eq!(questions[0].0, "req-3#0");
+        assert_eq!(questions[0].1, "Which color?");
+        assert_eq!(questions[0].2, vec!["Red", "Blue"]);
+
+        // First answer alone isn't enough; the response goes out when the
+        // last sub-question is answered, with all answers aboard.
+        assert!(matches!(translator.answer("req-3#0", "Blue"), AnswerOutcome::Pending));
+        let AnswerOutcome::Respond(response) = translator.answer("req-3#1", "L") else {
+            panic!("expected a control response");
+        };
+        let updated = &response["response"]["response"]["updatedInput"];
+        assert_eq!(updated["answers"]["Which color?"], "Blue");
+        assert_eq!(updated["answers"]["Which size?"], "L");
+        assert_eq!(updated["questions"][0]["question"], "Which color?");
+    }
+
+    #[test]
+    fn a_turn_result_reports_usage_and_returns_to_idle() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"result","subtype":"success","is_error":false,"num_turns":2,"stop_reason":"end_turn","session_id":"s","total_cost_usd":0.0149,"usage":{"input_tokens":18,"output_tokens":164}}"#,
+        ]);
+        assert_eq!(events, vec![
+            Event::UsageDelta { tokens: 182 },
+            Event::Status { state: SessionStatus::Idle },
+        ]);
+    }
+
+    #[test]
+    fn an_error_result_surfaces_the_message() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"something broke","usage":{}}"#,
+        ]);
+        assert_eq!(events[0], Event::Error { message: "something broke".to_string() });
+        assert_eq!(*events.last().unwrap(), Event::Status { state: SessionStatus::Idle });
+    }
+
+    #[test]
+    fn replayed_and_synthetic_user_text_is_skipped() {
+        let mut translator = Translator::default();
+        let events = translate_lines(&mut translator, &[
+            r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"isReplay":true,"parent_tool_use_id":null}"#,
+            r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"[continue]"}]},"isSynthetic":true,"parent_tool_use_id":null}"#,
+        ]);
+        assert!(events.is_empty());
+    }
+}

server/src/session/mod.rs

@@ -9,6 +9,7 @@
 //! SSE subscribers. The transcript is the source of truth -- subscribers
 //! that fall behind or reconnect catch up from the file by cursor.
 
+pub mod claude;
 pub mod driver;
 pub mod echo;
 pub mod transcript;
@@ -23,6 +24,7 @@ use serde::Serialize;
 use tokio::sync::{broadcast, mpsc};
 
 use crate::config::{Config, SessionConfig, SessionKind, TokenEntry};
+use claude::ClaudeDriver;
 use driver::{Driver, Event, ImageRef, SessionStatus};
 use echo::EchoDriver;
 use transcript::{SeqEvent, Transcript};
@@ -79,9 +81,12 @@ pub struct LiveSession {
 }
 
 /// The pump-maintained view of a session, read by the list endpoint.
+/// `model` also lives here (not in the immutable meta) because it can
+/// change mid-session via `set_model`.
 struct Shared {
     status: Mutex<SessionStatus>,
     last_activity: Mutex<f64>,
+    model: Mutex<Option<String>>,
 }
 
 impl LiveSession {
@@ -105,13 +110,6 @@ impl LiveSession {
         self.driver.interrupt();
     }
 
-    /// Hands the change to the driver. The persisted `model` field follows
-    /// when a driver that actually honors this lands (phase 2) -- echo
-    /// sessions just report the request as an error event.
-    pub fn set_model(&self, model: &str) {
-        self.driver.set_model(model);
-    }
-
     pub fn compact(&self) {
         self.driver.compact();
     }
@@ -130,7 +128,7 @@ impl LiveSession {
             kind: self.meta.kind,
             title: self.meta.title.clone(),
             host: self.meta.host.clone(),
-            model: self.meta.model.clone(),
+            model: self.shared.model.lock().unwrap().clone(),
             cwd: self.meta.cwd.clone(),
             status: *self.shared.status.lock().unwrap(),
             last_activity: *self.shared.last_activity.lock().unwrap(),
@@ -263,6 +261,28 @@ impl SessionManager {
         Ok(info)
     }
 
+    /// Changes a session's model: persisted (so a respawn keeps it and the
+    /// list shows it) and handed to the driver, which switches in place
+    /// where its dialect can. Through the manager, not the session, so the
+    /// config and the live view can't disagree.
+    pub fn set_session_model(&self, id: &str, model: &str) -> Result<()> {
+        let mut inner = self.inner.write().unwrap();
+        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
+            bail!("no session {id}");
+        }
+        let mut candidate = inner.config.clone();
+        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
+            meta.model = Some(model.to_string());
+        }
+        candidate.save(&self.config_path)?;
+        inner.config = candidate;
+        if let Some(session) = inner.live.get(id) {
+            *session.shared.model.lock().unwrap() = Some(model.to_string());
+            session.driver.set_model(model);
+        }
+        Ok(())
+    }
+
     /// Kills the process, releases everything the spawn created, and
     /// deletes the transcript and files -- the complete path out.
     pub fn delete_session(&self, id: &str) -> Result<()> {
@@ -288,6 +308,7 @@ impl SessionManager {
 fn default_title(kind: SessionKind) -> String {
     match kind {
         SessionKind::Echo => "Echo session".to_string(),
+        SessionKind::Claude => "Claude session".to_string(),
     }
 }
 
@@ -319,10 +340,12 @@ fn launch(meta: SessionConfig, data_dir: &Path) -> Result<Arc<LiveSession>> {
     let shared = Arc::new(Shared {
         status: Mutex::new(SessionStatus::Idle),
         last_activity: Mutex::new(now()),
+        model: Mutex::new(meta.model.clone()),
     });
 
     let driver: Box<dyn Driver> = match meta.kind {
         SessionKind::Echo => Box::new(EchoDriver::new(sink.clone())),
+        SessionKind::Claude => Box::new(ClaudeDriver::spawn(&meta, &dir, sink.clone())?),
     };
 
     tokio::spawn(pump(transcript, source, Arc::clone(&shared), events.clone()));