fix(node): preserve Claude SDK inbox delivery with mobile enabled

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased - Major]
 
+### Fixed
+
+- `agent-relay-node` keeps messages to registered Claude Code app chats, Anthropic Agent SDK sessions, `claude --print` and structured JSON runs on their ordinary inbox route when mobile continuation is enabled.
+- Updated Claude mobile plugins and cores explain recognized stopped-session and host-limit refusals; older counterparts retain generic failure notices. Refused and uncertain input is never replayed.
+- A rejected first Claude mobile submission no longer leaves provisional mobile control active.
+
 ### Changed
 
 - The core is now `agent-relay-node`: `relay init` installs it from `relay-node-v<version>` releases, and `AGENT_RELAY_NODE_PATH` points at a binary of your own.

crates/relay-node/README.md

@@ -56,6 +56,16 @@ contract), `call`, the harness hooks and `connect`. The account methods
 answer `method_not_found`, and the core uses the workspace the `agent-relay`
 CLI selects (`src/relay/workspace/cli.rs`), with no sign-in needed.
 
+## Claude mobile continuation
+
+On macOS and Linux, Claude Code terminal sessions can receive the verified owner's
+mobile messages as native user turns through Relay's plugin. Agent SDK,
+`claude --print` and pipe-driven sessions, including app-hosted chats, keep their
+ordinary peer inbox instead: their host owns the input and interrupt lifecycle. Turning mobile continuation on does not
+replace that inbox. A native submission refused after Stop or by a host limit
+is reported without automatic replay; unknown outcomes are never resubmitted.
+Native mobile continuation is unavailable on Windows.
+
 ## Build and test
 
 ```sh

crates/relay-node/scripts/test-claude-mobile-plugin.mjs

@@ -80,7 +80,9 @@ async function fixture(options = {}) {
     prompt: {
       submit: async (value) => {
         submitted.push(value);
-        return options.drop ? { drop: 'Policy refused' } : { text: value.text };
+        return options.drop !== undefined
+          ? { drop: options.drop === true ? 'Policy refused' : options.drop }
+          : { text: value.text };
       },
     },
   };
@@ -854,3 +856,50 @@ test(
     assert.deepEqual(f.submitted, [{ text: turn.text, asUser: true }]);
   }
 );
+
+test('recognized refusals are bounded, negotiated and never retried', async () => {
+  for (const [drop, refusal] of [
+    ['the session was stopped and nobody has sent input since', 'stopped'],
+    ['the prompt was removed from the queue before it ran', 'stopped'],
+    ['the host has ended its input', 'host_ended'],
+    ['the run reached its --max-turns limit', 'turn_limit'],
+    ['the run is past its --max-budget-usd limit', 'budget_limit'],
+    ['secret prompt /private/path refused by a user hook', undefined],
+  ]) {
+    const f = await fixture({ action: { ...turn, refusal_codes: true }, drop });
+    await f.tick();
+    await f.tick();
+    const expected = { lease_id: turn.lease_id, outcome: 'failed' };
+    if (refusal) expected.refusal = refusal;
+    assert.deepEqual(
+      f.calls.filter((call) => call.route === 'ack').map((call) => call.body),
+      [expected]
+    );
+    assert.equal(f.submitted.length, 1);
+    assert.equal(
+      f.calls.some((call) => JSON.stringify(call.body).includes(drop)),
+      false
+    );
+  }
+});
+
+test('a new plugin keeps failure acknowledgements compatible with old cores', async () => {
+  const f = await fixture({ action: turn, drop: 'the session was stopped and nobody has sent input since' });
+  await f.tick();
+  assert.deepEqual(
+    f.calls.filter((call) => call.route === 'ack').map((call) => call.body),
+    [{ lease_id: turn.lease_id, outcome: 'failed' }]
+  );
+});
+
+test('every defined drop value refuses input even when it is falsy', async () => {
+  for (const drop of ['', false, 0, null]) {
+    const f = await fixture({ action: { ...turn, refusal_codes: true }, drop });
+    await f.tick();
+    assert.equal(f.submitted.length, 1);
+    assert.deepEqual(
+      f.calls.filter((call) => call.route === 'ack').map((call) => call.body),
+      [{ lease_id: turn.lease_id, outcome: 'failed' }]
+    );
+  }
+});

crates/relay-node/src/relay/claude_mobile_plugin.rs

@@ -164,22 +164,22 @@ impl Core {
 pub(crate) fn session_roots(
     config: &crate::relay::claude::ClaudeConfig,
     sessions: &[crate::relay::claude::TerminalSession],
+    table: &dyn crate::relay::process::ProcessTable,
 ) -> std::collections::BTreeMap<crate::relay::store::Target, PathBuf> {
     let mut roots = std::collections::BTreeMap::new();
     for session in sessions {
         let target = crate::relay::store::Target::new(
             crate::relay::store::SessionKind::ClaudeTerminal,
             &session.session_id,
         );
-        let root =
-            matches!(session.kind.as_deref(), Some("interactive" | "background")).then(|| {
-                session
-                    .registry
-                    .as_deref()
-                    .and_then(Path::parent)
-                    .unwrap_or(&config.home)
-                    .to_path_buf()
-            });
+        let root = native_input_supported(session, table).then(|| {
+            session
+                .registry
+                .as_deref()
+                .and_then(Path::parent)
+                .unwrap_or(&config.home)
+                .to_path_buf()
+        });
         roots
             .entry(target)
             .and_modify(|known| *known = None)
@@ -191,6 +191,28 @@ pub(crate) fn session_roots(
         .collect()
 }
 
+/// An app can publish an `interactive` session while owning Claude's input
+/// through the SDK. Its stop/input lifecycle does not belong to the mod:
+/// `prompt.submit` may reject even while the peer inbox remains available.
+/// Keep those sessions on their existing inbox route, independently of the
+/// mobile setting. Check the running process too: entrypoint labels alone
+/// do not distinguish every host. Require positive terminal input, not only
+/// the absence of SDK flags.
+pub(crate) fn native_input_supported(
+    session: &crate::relay::claude::TerminalSession,
+    table: &dyn crate::relay::process::ProcessTable,
+) -> bool {
+    if !matches!(session.kind.as_deref(), Some("interactive" | "background"))
+        || session
+            .entrypoint
+            .as_deref()
+            .is_some_and(|entry| entry == "sdk" || entry.starts_with("sdk-"))
+    {
+        return false;
+    }
+    crate::relay::claude::is_interactive_terminal(table, session)
+}
+
 pub(crate) fn install(config: &Path, socket: &Path) -> Result<(), String> {
     if !config.is_absolute() || !socket.is_absolute() {
         return Err("Claude's configuration and Agent Relay bridge paths must be absolute.".into());

crates/relay-node/src/relay/claude_mobile_plugin/hooks/register.mjs

@@ -24,6 +24,19 @@ const PERMISSION_WAIT_MS = 300000;
 // comfortably below the core's 96 KiB limit, including the visible notice.
 const MAX_ANSWER_UNITS = 24000;
 
+// Do not forward arbitrary hook text to other devices. These are provider
+// refusals, not transport failures, and none makes a submission retryable.
+function refusalCode(reason) {
+  switch (reason) {
+    case "the session was stopped and nobody has sent input since":
+    case "the prompt was removed from the queue before it ran": return "stopped";
+    case "the host has ended its input": return "host_ended";
+    case "the run reached its --max-turns limit": return "turn_limit";
+    case "the run is past its --max-budget-usd limit": return "budget_limit";
+    default: return undefined;
+  }
+}
+
 function mobileAnswer(text) {
   if (text.length <= MAX_ANSWER_UNITS) return text;
   let end = MAX_ANSWER_UNITS;
@@ -205,10 +218,15 @@ async function tick($) {
       } else if (forward) {
         controlToken = action.control_token;
       }
-      await request($, "ack", {
+      const acknowledgement = {
         lease_id: action.lease_id,
         outcome: admitted?.drop === undefined ? "started" : "failed",
-      });
+      };
+      const refusal = refusalCode(admitted?.drop);
+      // Older cores accept exactly two fields. Negotiate diagnostics on
+      // each action so a plugin upgrade or rollback cannot lose its ack.
+      if (action.refusal_codes === true && refusal) acknowledgement.refusal = refusal;
+      await request($, "ack", acknowledgement);
     } finally {
       confirming = false;
     }

crates/relay-node/src/relay/claude_mobile_plugin/tests.rs

@@ -324,25 +324,36 @@ fn native_config_root_resolution_rejects_duplicate_session_ids() {
     let temp = tempfile::tempdir().unwrap();
     let config = crate::relay::claude::ClaudeConfig::resolve(&|_| None, temp.path(), &[]);
     let mut session = crate::relay::claude::TerminalSession::decode(br#"{"sessionId":"00000000-0000-4000-8000-000000000001","pid":123,"messagingSocketPath":"/tmp/claude.sock","kind":"interactive"}"#).unwrap();
+    let table = crate::relay::process::FakeProcesses::default();
+    for pid in [123, 124] {
+        table.insert(
+            pid,
+            crate::relay::process::FakeProcess {
+                arguments: vec!["claude".into()],
+                has_terminal: true,
+                ..Default::default()
+            },
+        );
+    }
     let target = crate::relay::store::Target::new(
         crate::relay::store::SessionKind::ClaudeTerminal,
         &session.session_id,
     );
     assert_eq!(
-        session_roots(&config, &[session.clone()])[&target],
+        session_roots(&config, &[session.clone()], &table)[&target],
         config.home
     );
     session.registry = Some(temp.path().join("work/sessions"));
     assert_eq!(
-        session_roots(&config, &[session.clone()])[&target],
+        session_roots(&config, &[session.clone()], &table)[&target],
         temp.path().join("work")
     );
     let mut duplicate = session.clone();
     duplicate.pid = 124;
     duplicate.registry = None;
-    assert!(session_roots(&config, &[session.clone(), duplicate]).is_empty());
+    assert!(session_roots(&config, &[session.clone(), duplicate], &table).is_empty());
     session.kind = Some("unknown".into());
-    assert!(session_roots(&config, &[session]).is_empty());
+    assert!(session_roots(&config, &[session], &table).is_empty());
 }
 
 #[test]

crates/relay-node/src/relay/local.rs

@@ -6445,6 +6445,22 @@ printf '{{"jsonrpc":"2.0","id":"%s","result":{{"stopReason":"end_turn","outputTo
     #[test]
     fn mobile_native_runtime_requires_exact_native_process_and_rejects_tool_descendants() {
         let world = world();
+        world
+            .table
+            .processes
+            .lock()
+            .unwrap()
+            .get_mut(&300)
+            .unwrap()
+            .arguments = vec!["claude".into()];
+        world
+            .table
+            .processes
+            .lock()
+            .unwrap()
+            .get_mut(&300)
+            .unwrap()
+            .has_terminal = true;
         let path = world.sessions.config.claude.registry().join("300.json");
         let mut record: serde_json::Value =
             serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
@@ -6462,6 +6478,30 @@ printf '{{"jsonrpc":"2.0","id":"%s","result":{{"stopReason":"end_turn","outputTo
             "ordinary Relay descendant identification stays available"
         );
         assert_eq!((runtime.pid, runtime.start, runtime.idle), (300, 100, true));
+        world
+            .table
+            .processes
+            .lock()
+            .unwrap()
+            .get_mut(&300)
+            .unwrap()
+            .arguments = vec!["claude".into(), "--input-format=stream-json".into()];
+        assert!(
+            world.sessions.mobile_claude_runtime(&target, 300).is_none(),
+            "a running SDK host cannot submit mobile input even with an interactive registry record",
+        );
+        assert!(
+            world.sessions.caller(320).is_ok(),
+            "the ordinary session route stays available"
+        );
+        world
+            .table
+            .processes
+            .lock()
+            .unwrap()
+            .get_mut(&300)
+            .unwrap()
+            .arguments = vec!["claude".into()];
         assert_eq!(
             world.sessions.mobile_claude_runtime(&target, 300),
             Some(runtime.clone())

crates/relay-node/src/relay/local/directory_trait.rs

@@ -36,7 +36,7 @@ impl SessionDirectory for LocalSessions {
         if matches.next().is_some() {
             return None;
         }
-        if !matches!(session.kind.as_deref()?, "background" | "interactive") {
+        if !crate::relay::claude_mobile_plugin::native_input_supported(session, &*self.table) {
             return None;
         }
         // The native Mods HTTP client runs inside Claude itself. A tool,

crates/relay-node/src/relay/local/directory_trait/mobile_claude_roots.rs

@@ -50,7 +50,7 @@ pub(super) fn read(config: &ClaudeConfig, table: &dyn ProcessTable) -> BTreeMap<
             }
         }
     }
-    crate::relay::claude_mobile_plugin::session_roots(config, &sessions)
+    crate::relay::claude_mobile_plugin::session_roots(config, &sessions, table)
 }
 
 #[cfg(test)]

crates/relay-node/src/relay/local/directory_trait/mobile_claude_roots/tests.rs

@@ -37,6 +37,8 @@ fn processes(pids: &[i32]) -> FakeProcesses {
             *pid,
             FakeProcess {
                 start: 123,
+                arguments: vec!["claude".into()],
+                has_terminal: true,
                 ..Default::default()
             },
         );
@@ -102,3 +104,80 @@ fn unreadable_live_record_suspends_attribution_but_dead_record_does_not() {
         [(Target::new(SessionKind::ClaudeTerminal, SESSION), a)].into()
     );
 }
+
+#[test]
+fn sdk_host_keeps_its_inbox_but_never_advertises_native_mobile_input() {
+    let dir = tempfile::tempdir().unwrap();
+    let a = dir.path().join("login");
+    let b = dir.path().join("unused");
+    record(&a, 42, 123);
+    let config = config(&a, &b);
+    let table = processes(&[42]);
+    assert_eq!(read(&config, &table).len(), 1);
+    for arguments in [
+        vec!["claude", "--input-format", "stream-json"],
+        vec!["claude", "--input-format=stream-json"],
+        vec!["claude", "--output-format", "stream-json"],
+        vec!["claude", "--output-format=stream-json"],
+        vec!["claude", "-p", "hello"],
+        vec!["claude", "--print"],
+        vec![],
+    ] {
+        table
+            .processes
+            .lock()
+            .unwrap()
+            .get_mut(&42)
+            .unwrap()
+            .arguments = arguments.iter().map(|arg| (*arg).into()).collect();
+        assert!(read(&config, &table).is_empty(), "{arguments:?}");
+        assert_eq!(
+            claude::registry(&config, &table).sessions.len(),
+            1,
+            "excluding native capability must not remove the peer inbox"
+        );
+    }
+    table
+        .processes
+        .lock()
+        .unwrap()
+        .get_mut(&42)
+        .unwrap()
+        .arguments = vec!["claude".into()];
+    table
+        .processes
+        .lock()
+        .unwrap()
+        .get_mut(&42)
+        .unwrap()
+        .has_terminal = false;
+    assert!(
+        read(&config, &table).is_empty(),
+        "a pipe without SDK flags is not native terminal input"
+    );
+    assert_eq!(claude::registry(&config, &table).sessions.len(), 1);
+    table
+        .processes
+        .lock()
+        .unwrap()
+        .get_mut(&42)
+        .unwrap()
+        .has_terminal = true;
+    let path = a.join("sessions/42.json");
+    let mut value: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+    for entrypoint in ["sdk-cli", "sdk-ts", "sdk"] {
+        value["entrypoint"] = json!(entrypoint);
+        std::fs::write(&path, value.to_string()).unwrap();
+        assert!(read(&config, &table).is_empty(), "{entrypoint}");
+    }
+    value["entrypoint"] = json!("cli");
+    std::fs::write(&path, value.to_string()).unwrap();
+    assert_eq!(read(&config, &table).len(), 1);
+    for name in ["sdk-task", "SDK-task", "Sdk-task"] {
+        value["name"] = json!(name);
+        std::fs::write(&path, value.to_string()).unwrap();
+        assert!(read(&config, &table).is_empty(), "{name}");
+        assert_eq!(claude::registry(&config, &table).sessions.len(), 1);
+    }
+}