kuddev
97985b73d5
Native Codex hooks may identify a running session/group instead of the rollout thread accepted by resume. Retain transcript paths, normalize the thread identity, and share the parser with session discovery and probes. Keep recovery paths when a matching provider acknowledgement omits them, retain native identity metadata while reducing large SSH/WSL events, and show the existing failure message when a saved target has no resume command. Add regressions for Windows/guest paths and multiple CLI startup flows.
215 lines
11 KiB
Python
215 lines
11 KiB
Python
"""Real PTY delivery and transactional SSH file-adapter regressions."""
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ASSETS = Path(__file__).resolve().parents[2] / "nebula_app/res/hooks"
|
|
TOKEN = "0123456789abcdef0123456789abcdef"
|
|
|
|
|
|
def module(name):
|
|
spec = importlib.util.spec_from_file_location(name, ASSETS / (name + ".py"))
|
|
result = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(result)
|
|
return result
|
|
|
|
|
|
@unittest.skipUnless(os.name == "posix", "remote adapter targets POSIX SSH hosts")
|
|
class RemoteHooksTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="pebrel-remote-hooks-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
# macOS exposes its temporary directory through a system symlink.
|
|
# Ordinary fixture paths must exercise transactions, while redirection
|
|
# is supplied explicitly by the negative tests below.
|
|
self.root = Path(self.tmp.name).resolve()
|
|
env = {"HOME": str(self.root), "PATH": os.environ["PATH"], "SHELL": "/bin/bash",
|
|
"XDG_DATA_HOME": str(self.root / "data"), "XDG_CACHE_HOME": str(self.root / "cache"),
|
|
"PEBREL_REMOTE_HOOK_TOKEN": TOKEN}
|
|
self.environment = patch.dict(os.environ, env, clear=True)
|
|
self.environment.start()
|
|
self.addCleanup(self.environment.stop)
|
|
|
|
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"
|
|
bindir.mkdir(parents=True)
|
|
node = bindir / "node"
|
|
node.write_text("#!/bin/sh\nprintf 'hooks stable true\\n'\n")
|
|
node.chmod(0o700)
|
|
cli = bindir / "codex"
|
|
cli.write_text("#!/usr/bin/env node\n")
|
|
cli.chmod(0o700)
|
|
with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}):
|
|
self.assertEqual(files.find_program("codex"), str(cli))
|
|
self.assertEqual(files.capture(str(cli), "features", "list"), "hooks stable true\n")
|
|
configured = self.root / "bin"
|
|
configured.mkdir()
|
|
preferred = configured / "codex"
|
|
preferred.write_text("#!/bin/sh\n")
|
|
preferred.chmod(0o700)
|
|
with patch.dict(os.environ, {"PATH": str(configured)}):
|
|
self.assertEqual(files.find_program("codex"), str(preferred))
|
|
|
|
def test_compare_and_swap_does_not_overwrite_concurrent_user_edit(self):
|
|
files = module("remote_files")
|
|
snapshot = files.snapshot()
|
|
root, paths = files.locations()
|
|
paths["claude"].parent.mkdir(parents=True)
|
|
paths["claude"].write_text("user change")
|
|
plan = [{"name": "claude", "expected": snapshot["files"]["claude"]["sha256"], "content": "new"}]
|
|
with self.assertRaisesRegex(ValueError, "changed during setup"):
|
|
files.apply(plan)
|
|
self.assertEqual(paths["claude"].read_text(), "user change")
|
|
|
|
def test_transaction_rolls_back_only_its_writes(self):
|
|
files = module("remote_files")
|
|
real_replace = files.replace
|
|
root, paths = files.locations()
|
|
def replace(path, content, executable=False):
|
|
if path == paths["bridge.py"]:
|
|
raise OSError("fixture write failure")
|
|
real_replace(path, content, executable)
|
|
plan = [{"name": name, "expected": None, "content": "new"} for name in ["pebrel-hook", "bridge.py"]]
|
|
with patch.object(files, "replace", replace), self.assertRaises(OSError):
|
|
files.apply(plan)
|
|
self.assertFalse(paths["pebrel-hook"].exists())
|
|
self.assertFalse(paths["bridge.py"].exists())
|
|
|
|
def test_unknown_name_and_symlink_cannot_redirect_writes(self):
|
|
files = module("remote_files")
|
|
with self.assertRaises(ValueError):
|
|
files.apply([{"name": "../outside", "expected": None, "content": "bad"}])
|
|
root, paths = files.locations()
|
|
outside = self.root / "outside"
|
|
outside.write_text("preserve")
|
|
paths["bridge.py"].symlink_to(outside)
|
|
with self.assertRaises(ValueError):
|
|
files.apply([{"name": "bridge.py", "expected": None, "content": "bad"}])
|
|
self.assertEqual(outside.read_text(), "preserve")
|
|
|
|
def test_symlinked_integration_directory_cannot_redirect_writes(self):
|
|
files = module("remote_files")
|
|
root, paths = files.locations()
|
|
outside = self.root / "outside"
|
|
outside.mkdir()
|
|
root.parent.mkdir(parents=True)
|
|
root.symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaisesRegex(ValueError, "symbolic link"):
|
|
files.apply([{"name": "bridge.py", "expected": None, "content": "bad"}])
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
|
|
def test_bsd_ancestry_recognizes_interpreted_provider_without_trusting_other_arguments(self):
|
|
bridge = module("remote_bridge")
|
|
prefix = "400 Sat Sep 19 12:00:00 2026 "
|
|
for source, command in [
|
|
("codex", "/usr/bin/python3 /private/var/fixture/codex"),
|
|
("claude", "/usr/bin/node /opt/modules/@anthropic-ai/claude-code/cli.js"),
|
|
("pi", "/usr/bin/node /opt/modules/pi-coding-agent/dist/cli.js"),
|
|
]:
|
|
def ps_output(args, **kwargs):
|
|
observed = command if "args=" in args else command.split()[0]
|
|
return (prefix + observed).encode()
|
|
with self.subTest(source=source), patch.object(bridge.os, "getppid", return_value=500), patch.object(bridge.Path, "read_text", side_effect=FileNotFoundError), patch.object(bridge.subprocess, "check_output", side_effect=ps_output):
|
|
self.assertRegex(bridge.process_identity(source), r"^500:[a-f0-9]{16}$")
|
|
row = b"1 Sat Sep 19 12:00:00 2026 /usr/bin/python3 worker.py unrelated codex"
|
|
with patch.object(bridge.os, "getppid", return_value=500), patch.object(bridge.Path, "read_text", side_effect=FileNotFoundError), patch.object(bridge.subprocess, "check_output", return_value=row):
|
|
self.assertIsNone(bridge.process_identity("codex"))
|
|
|
|
def test_installed_launcher_delivers_native_stdin_over_real_controlling_tty(self):
|
|
import pty
|
|
provider = self.root / "codex"
|
|
bridge = ASSETS / "remote_bridge.py"
|
|
# Parent ancestry supplies process identity, never a PID from JSON.
|
|
provider.write_text(
|
|
"import json, subprocess, sys\n"
|
|
f"bridge={str(bridge)!r}\n"
|
|
"for event in ['SessionStart','UserPromptSubmit','PermissionRequest','PreToolUse','Stop']:\n"
|
|
" p={'hook_event_name':event,'session_id':'main','turn_id':'turn','pid':1, 'transcript_path':'/home/user/.codex/sessions/rollout-date-0199a213-c2a4-7cf5-8f6b-d746fbb6e86c.jsonl'}\n"
|
|
" if event == 'PreToolUse': p.update(tool_name='request_user_input',tool_input={'questions':[{'id':'scope','question':'Which scope?'}]})\n"
|
|
" if event == 'Stop': p['last_assistant_message']='x'*70000\n"
|
|
" subprocess.run([sys.executable,bridge,'codex','--hooks=full'],input=json.dumps(p).encode(),check=True)\n"
|
|
)
|
|
pid, master = pty.fork()
|
|
if pid == 0:
|
|
os.execv(sys.executable, [sys.executable, str(provider)])
|
|
output = b""
|
|
try:
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
if select.select([master], [], [], 0.1)[0]:
|
|
try:
|
|
part = os.read(master, 65536)
|
|
except OSError:
|
|
break
|
|
if not part:
|
|
break
|
|
output += part
|
|
waited, status = os.waitpid(pid, 0)
|
|
self.assertEqual(status, 0, output.decode(errors="replace"))
|
|
finally:
|
|
os.close(master)
|
|
envelopes = re.findall(rb"\x1b\]777;nebula-hook;" + TOKEN.encode() + rb";([^\x07]+)\x07", output)
|
|
self.assertEqual(len(envelopes), 5, output[:2000])
|
|
decoded = [base64.b64decode(value).split(b"\n", 1) for value in envelopes]
|
|
for header, payload in decoded:
|
|
self.assertIn(f"process={pid}:".encode(), header)
|
|
self.assertIn(b"codex_hooks=full", header)
|
|
self.assertLessEqual(len(header) + len(payload) + 1, 65536)
|
|
payloads = [json.loads(payload) for _, payload in decoded]
|
|
self.assertEqual([p["bridge_sequence"] for p in payloads], [1, 2, 3, 4, 5])
|
|
self.assertEqual(payloads[3]["tool_input"]["questions"][0]["question"], "Which scope?")
|
|
self.assertEqual(payloads[-1]["hook_event_name"], "Stop")
|
|
self.assertEqual(payloads[-1]["transcript_path"], payloads[0]["transcript_path"])
|
|
self.assertNotIn("last_assistant_message", payloads[-1])
|
|
|
|
def test_large_pi_events_keep_the_native_file_and_bridge_lifetime(self):
|
|
bridge = module("remote_bridge")
|
|
payload = {"kind": "done", "session_id": "pi-native", "session_file": "/sessions/pi.jsonl",
|
|
"bridge_instance": "process-start", "stop_reason": "stop",
|
|
"last_assistant_message": "x" * 70000}
|
|
with patch.object(bridge, "write_terminal") as write:
|
|
bridge.send("pi", False, "", payload, TOKEN)
|
|
frame = write.call_args.args[0]
|
|
encoded = frame.split(b";", 3)[3].removesuffix(b"\x07")
|
|
body = json.loads(base64.b64decode(encoded).split(b"\n", 1)[1])
|
|
for key in ["session_id", "session_file", "bridge_instance", "stop_reason"]:
|
|
self.assertEqual(body[key], payload[key])
|
|
self.assertNotIn("last_assistant_message", body)
|
|
|
|
def test_invalid_token_never_writes_and_failed_delivery_keeps_notify_chain(self):
|
|
bridge = module("remote_bridge")
|
|
with patch.dict(os.environ, {"PEBREL_REMOTE_HOOK_TOKEN": "bad"}), patch.object(bridge, "send") as send:
|
|
bridge.run(["codex", '{"type":"agent-turn-complete"}'])
|
|
send.assert_not_called()
|
|
for payload in ['{"type":"agent-turn-complete"}', "malformed"]:
|
|
with patch.object(bridge, "send", side_effect=OSError("no tty")), patch.object(bridge.subprocess, "Popen") as popen:
|
|
with self.assertRaises((OSError, ValueError)):
|
|
bridge.run(["codex", "--chain", "user-notifier", "original-arg", payload])
|
|
self.assertEqual(popen.call_args.args[0], ["user-notifier", "original-arg", payload])
|
|
|
|
def test_partial_terminal_frame_is_cancelled_on_timeout(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], [])]):
|
|
with self.assertRaises(TimeoutError):
|
|
bridge.write_terminal(b"\x1b]777;partial\x07")
|
|
self.assertEqual(writes[-1], b"\x18")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|