fix(hooks): deliver detached Claude failure and permission events

architecture/notes/nebula_app/ai_hook/2026-10-25-claude-detached-hook-delivery.md

@@ -0,0 +1,61 @@
+# Deliver detached Claude hooks through its terminal sequence output
+
+## Status
+
+Implemented for the notification-delivery defect in issue #540.
+
+## Context
+
+Interactive Claude can start a hook without a controlling terminal. Opening
+`/dev/tty` then fails with ENXIO and the fail-open bridge discards the event.
+A sequence file advancing does not prove delivery to the owning pane.
+
+## Evidence
+
+The real detached-process regression starts the production bridge with a new
+session and receives empty stdout on the previous implementation. Claude's
+[official changelog](https://github.com/anthropics/claude-code/blob/main/CHANGELOG.md),
+version 2.1.141, documents `terminalSequence` for hooks without a controlling
+terminal. PR [#542](https://github.com/Kuddev/pebrel/pull/542) supplied the fallback
+design and reported an interactive Claude/WSL observation.
+
+## Decision
+
+Keep authenticated OSC framing, process ownership, bounded payloads, ordering and
+terminal writes in the shared remote bridge. Only Claude, only ENXIO from opening
+the terminal, and only before any terminal bytes are written may return the same
+OSC inside one `terminalSequence` JSON object. Other providers and other errors
+retain their existing behavior. A partial terminal write is cancelled, never
+duplicated through stdout.
+
+## Rejected alternatives
+
+- Consider sequence-file updates successful delivery: the file precedes output.
+- Fall back after partial writes: this can duplicate or interleave an event.
+- Return the field for every provider: their stdout contracts differ.
+- Bundle automatic `continue`: that is the independent feature in issue #541;
+  notification repair does not require injecting another command.
+
+## Consequences
+
+No dependency, setting, local credential, hook installation or retry policy
+changes. Interactive clients supporting this output field can restore delivery.
+This does not promise delivery in non-interactive CLI/SDK modes or old clients
+that ignore the field, and it does not solve other bridge transport failures.
+
+## Validation
+
+The real detached-process regression covers permission/failure events, identity
+envelopes and invalid-token rejection. Tests preserve provider/error isolation,
+normal TTY single-delivery and partial-frame cancellation. Existing real PTY and
+transaction tests remain. A physical WSL/Claude acceptance run is not claimed by
+these Linux process tests.
+
+## Supersedes
+
+None; isolates the delivery portion proposed in #542 from automatic continuation.
+
+## Revisit when
+
+The provider offers an authenticated direct hook transport, or retires the
+documented interactive terminal sequence contract.

nebula_app/res/hooks/remote_bridge.py

@@ -1,5 +1,6 @@
 """Bounded, fail-open provider hook delivery to the owning SSH terminal."""
 import base64
+import errno
 from contextlib import contextmanager
 import fcntl
 import hashlib
@@ -157,11 +158,19 @@ def send(source, native, mode, payload, token):
         if len(envelope) > MAX_ENVELOPE:
             return
         osc = b"\x1b]777;nebula-hook;" + token.encode() + b";" + base64.b64encode(envelope) + b"\x07"
-        write_terminal(osc)
+        write_terminal(osc, source)
 
 
-def write_terminal(osc):
-    fd = os.open("/dev/tty", os.O_WRONLY | os.O_NONBLOCK)
+def write_terminal(osc, source=None):
+    try:
+        fd = os.open("/dev/tty", os.O_WRONLY | os.O_NONBLOCK)
+    except OSError as error:
+        if source != "claude" or error.errno != errno.ENXIO:
+            raise
+        # Interactive Claude consumes hook JSON even when its hook has no TTY.
+        # This fallback is only possible before any part of the OSC was written.
+        print(json.dumps({"terminalSequence": osc.decode("ascii")}), flush=True)
+        return
     offset = 0
     deadline = time.monotonic() + 0.5
     try:

scripts/tests/test_remote_hooks.py

@@ -1,6 +1,8 @@
 """Real PTY delivery and transactional SSH file-adapter regressions."""
 import base64
+import errno
 import importlib.util
+import io
 import json
 import os
 from pathlib import Path
@@ -40,6 +42,75 @@ def setUp(self):
         self.environment.start()
         self.addCleanup(self.environment.stop)
 
+    def test_detached_claude_hook_returns_one_authenticated_terminal_sequence(self):
+        for event in ["PermissionRequest", "StopFailure"]:
+            payload = {"hook_event_name": event, "session_id": "fixture", "error": "unknown"}
+            result = subprocess.run(
+                [sys.executable, str(ASSETS / "remote_bridge.py"), "claude"],
+                input=json.dumps(payload), capture_output=True, text=True,
+                start_new_session=True, timeout=8, check=True,
+            )
+            self.assertEqual(result.stderr, "")
+            response = json.loads(result.stdout)
+            self.assertEqual(list(response), ["terminalSequence"])
+            frame = response["terminalSequence"]
+            self.assertTrue(frame.startswith("\x1b]777;nebula-hook;" + TOKEN + ";"))
+            self.assertTrue(frame.endswith("\x07"))
+            header, body = base64.b64decode(frame.split(";", 3)[3][:-1]).split(b"\n", 1)
+            self.assertTrue(header.startswith(b"nebula-hook/1 source=claude"))
+            body = json.loads(body)
+            self.assertEqual(body["hook_event_name"], event)
+            self.assertEqual(body["session_id"], "fixture")
+            self.assertEqual(body["error"], "unknown")
+            self.assertGreater(body["bridge_sequence"], 0)
+        with patch.dict(os.environ, {"PEBREL_REMOTE_HOOK_TOKEN": "invalid"}):
+            rejected = subprocess.run(
+                [sys.executable, str(ASSETS / "remote_bridge.py"), "claude"],
+                input=json.dumps(payload), capture_output=True, text=True,
+                start_new_session=True, timeout=8, check=True,
+            )
+        self.assertEqual(rejected.stdout, "")
+
+    def test_stdout_fallback_requires_claude_and_no_controlling_terminal(self):
+        bridge = module("remote_bridge")
+        for source, error in [("codex", errno.ENXIO), ("pi", errno.ENXIO),
+                              ("opencode", errno.ENXIO), ("claude", errno.EACCES)]:
+            with self.subTest(source=source, error=error), patch.object(
+                bridge.os, "open", side_effect=OSError(error, "unavailable")
+            ), patch.object(bridge.sys, "stdout", io.StringIO()) as output:
+                with self.assertRaises(OSError):
+                    bridge.write_terminal(b"\x07", source)
+                self.assertEqual(output.getvalue(), "")
+
+    def test_claude_with_controlling_terminal_does_not_duplicate_stdout(self):
+        bridge = module("remote_bridge")
+        with patch.object(bridge.os, "open", return_value=42), patch.object(
+            bridge.os, "close"
+        ), patch.object(bridge.select, "select", return_value=([], [42], [])), patch.object(
+            bridge.os, "write", return_value=1
+        ) as write, patch.object(bridge.sys, "stdout", io.StringIO()) as output:
+            bridge.write_terminal(b"\x07", "claude")
+            write.assert_called_once_with(42, b"\x07")
+            self.assertEqual(output.getvalue(), "")
+
+    def test_claude_partial_terminal_frame_is_cancelled_without_stdout_retry(self):
+        bridge = module("remote_bridge")
+        writes = []
+
+        def write(fd, value):
+            writes.append(value)
+            return 2 if len(writes) == 1 else len(value)
+
+        with patch.object(bridge.os, "open", return_value=5), patch.object(
+            bridge.os, "close"
+        ), patch.object(bridge.os, "write", write), patch.object(
+            bridge.select, "select", side_effect=[([], [5], []), ([], [], []), ([], [5], [])]
+        ), patch.object(bridge.sys, "stdout", io.StringIO()) as output:
+            with self.assertRaises(TimeoutError):
+                bridge.write_terminal(b"\x1b]777;partial\x07", "claude")
+            self.assertEqual(writes[-1], b"\x18")
+            self.assertEqual(output.getvalue(), "")
+
     def test_discovery_finds_nvm_cli_and_its_interpreter_without_login_path(self):
         files = module("remote_files")
         bindir = self.root / ".nvm/versions/node/v24.18.0/bin"