Don't let one namespace delete another's server records

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.
 

declib/api/server_registry.py

@@ -56,6 +56,7 @@ def register_server(info: Dict) -> Path:
  payload = dict(info)
  payload.setdefault("started_at", time.time())
  payload.setdefault("pid", os.getpid())
+ payload.setdefault("namespace", _namespace_token())
  tmp_path = path.with_suffix(".json.tmp")
  with open(tmp_path, "w") as f:
  json.dump(payload, f, indent=2, default=str)
@@ -72,6 +73,52 @@ def unregister_server(server_id: str) -> bool:
  return False
 
 
+# A registry directory can be shared by processes that cannot see each other's
+# servers -- most easily when several containers bind-mount the same home
+# directory, which is exactly how agent harnesses tend to run. Sockets live in
+# each container's own /tmp and PIDs are per-namespace, so a sibling's record
+# looks dead by both tests, and pruning deletes a perfectly healthy server's
+# record. The owner then finds itself missing from the registry.
+#
+# Stamping each record with the namespace that created it makes "not mine"
+# distinguishable from "dead". Foreign records are ignored -- never returned,
+# never deleted -- so the owner stays the only writer of its own entries.
+def _namespace_token() -> str:
+ """Identify the PID/mount namespace this process can observe servers in."""
+ import socket as _socket
+
+ parts = [_socket.gethostname()]
+ for ns in ("pid", "mnt"):
+ try:
+ parts.append(os.readlink(f"/proc/self/ns/{ns}"))
+ except OSError:
+ parts.append("?")
+ return "|".join(parts)
+
+
+def _probe_socket(socket_path: str, timeout: float = 0.5) -> bool:
+ """Whether something is actually accepting connections on `socket_path`.
+
+ A socket file on disk proves only that a server once bound there; the file
+ outlives an unclean death. Connecting is the authoritative test, and it is
+ the one ida-pro-mcp uses (`probe_instance`) for the same reason.
+ """
+ import socket as _socket
+
+ sock = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ sock.settimeout(timeout)
+ sock.connect(socket_path)
+ return True
+ except OSError:
+ return False
+ finally:
+ try:
+ sock.close()
+ except OSError:
+ pass
+
+
 # A pruned record is the only evidence that a server ever existed. Deleting it
 # outright means the *next* command can explain what happened but every command
 # after that cannot -- and the command agents run most is `list`, which prunes.
@@ -160,7 +207,9 @@ def _is_record_live(record: Dict) -> bool:
  return False
  except Exception:
  return False
- return True
+ # Both checks above are proxies: a socket file survives an unclean death,
+ # and a PID can be reused by an unrelated process. Ask the server directly.
+ return _probe_socket(socket_path)
 
 
 def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None) -> List[Dict]:
@@ -188,6 +237,12 @@ def list_servers(prune_stale: bool = True, pruned: Optional[List[Dict]] = None)
  _l.debug("Failed to read server registry file %s: %s", entry, exc)
  continue
 
+ # A record from another namespace is not ours to judge or delete: its
+ # socket and pid are meaningless here. Skip it entirely.
+ record_ns = record.get("namespace")
+ if record_ns is not None and record_ns != _namespace_token():
+ continue
+
  if prune_stale and not _is_record_live(record):
  if pruned is not None:
  pruned.append(record)

tests/test_decompiler_cli.py

@@ -12,6 +12,7 @@
 """
 import json
 import os
+import pathlib
 import re
 import shutil
 import subprocess
@@ -2193,6 +2194,123 @@ def test_read_memory(self):
  unittest.main()
 
 
+class TestRegistryLivenessProbe(unittest.TestCase):
+ """A socket file is not proof of life; connecting to it is.
+
+ The file outlives an unclean death and a PID can be recycled, so both of
+ the cheap checks can say "alive" about a server that is gone. ida-pro-mcp
+ settles this by probing the endpoint; so do we.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_orphaned_socket_file_is_not_alive(self):
+ """The exact residue an unclean death leaves behind."""
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock = os.path.join(sockdir, "decompiler.sock")
+ open(sock, "wb").close() # a file, nothing listening
+ self.assertFalse(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock, "pid": os.getpid()}))
+
+ def test_a_listening_socket_is_alive(self):
+ import socket as _socket
+ import tempfile as _tf
+ from declib.api import server_registry
+
+ with _tf.TemporaryDirectory() as sockdir:
+ sock_path = os.path.join(sockdir, "decompiler.sock")
+ srv = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
+ try:
+ srv.bind(sock_path)
+ srv.listen(1)
+ self.assertTrue(server_registry._is_record_live(
+ {"id": "x", "socket_path": sock_path, "pid": os.getpid()}))
+ finally:
+ srv.close()
+
+ def test_probe_reports_false_for_a_missing_path(self):
+ from declib.api import server_registry
+
+ self.assertFalse(server_registry._probe_socket("/nonexistent/nope.sock"))
+
+
+class TestRegistryNamespaceSafety(unittest.TestCase):
+ """A shared registry directory must not let one namespace delete another's.
+
+ Agent harnesses commonly bind-mount one home directory into several
+ containers at once. Sockets live in each container's own /tmp and PIDs are
+ per-namespace, so a sibling's record fails both liveness tests and gets
+ pruned -- deleting the record of a server that is alive and well. The owner
+ then cannot find its own server.
+ """
+
+ def setUp(self):
+ import tempfile
+ self._tmp = tempfile.TemporaryDirectory()
+ os.environ["DECLIB_SERVER_REGISTRY"] = self._tmp.name
+
+ def tearDown(self):
+ os.environ.pop("DECLIB_SERVER_REGISTRY", None)
+ self._tmp.cleanup()
+
+ def test_foreign_records_are_never_deleted(self):
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "sibling01.json"
+ entry.write_text(_json.dumps({
+ "id": "sibling01",
+ "socket_path": "/tmp/declib_server_sibling01/decompiler.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 4242,
+ "namespace": "some-other-container|pid:[4026531999]|mnt:[4026532999]",
+ }))
+ self.assertEqual(server_registry.list_servers(), [],
+ "a foreign server must not be reported as ours")
+ self.assertTrue(entry.exists(),
+ "a foreign record must survive our pruning")
+
+ def test_own_records_still_pruned_when_dead(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine01", "socket_path": "/nonexistent/x.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ })
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse((pathlib.Path(self._tmp.name) / "mine01.json").exists())
+
+ def test_new_records_carry_a_namespace(self):
+ from declib.api import server_registry
+
+ server_registry.register_server({
+ "id": "mine02", "socket_path": "/nonexistent/y.sock",
+ })
+ import json as _json
+ rec = _json.loads((pathlib.Path(self._tmp.name) / "mine02.json").read_text())
+ self.assertTrue(rec.get("namespace"))
+
+ def test_legacy_records_without_a_namespace_still_work(self):
+ """Records written before this change must not become undeletable."""
+ import json as _json
+ from declib.api import server_registry
+
+ entry = pathlib.Path(self._tmp.name) / "legacy01.json"
+ entry.write_text(_json.dumps({
+ "id": "legacy01", "socket_path": "/nonexistent/z.sock",
+ "binary_path": "/bin/true", "backend": "ida", "pid": 999999,
+ }))
+ self.assertEqual(server_registry.list_servers(), [])
+ self.assertFalse(entry.exists(), "legacy dead records are still pruned")
 class TestGhidraSafeNames(unittest.TestCase):
  """Ghidra rejects characters that filenames are perfectly free to contain.