Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 74 additions & 0 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,71 @@ def _check_auth(x_api_key: str | None):
raise HTTPException(status_code=401, detail="Invalid API key")


# Sentinel for "no value passed" — distinguishes _parse_path_map() (read env)
# from _parse_path_map(None) (no mapping). Closes Copilot's test-isolation
# concern on jphein/palace-daemon#1: the previous None default coupled tests
# to whatever PALACE_DAEMON_PATH_MAP happened to be in the test process env.
_PATH_MAP_USE_ENV: object = object()


def _parse_path_map(raw=_PATH_MAP_USE_ENV) -> list[tuple[str, str]]:
"""Parse PALACE_DAEMON_PATH_MAP into ordered (client_prefix, daemon_prefix) pairs.

Format: comma-separated ``client_prefix=daemon_prefix`` entries. Whitespace
around each token is stripped. Empty entries and entries missing ``=`` are
skipped silently. Order is preserved so the operator can put more-specific
prefixes first.

Args:
raw: When omitted, reads from ``PALACE_DAEMON_PATH_MAP``. Pass an
explicit string (or ``""``/``None``) to bypass env entirely —
tests use this to stay deterministic regardless of CI / dev env.

Example::

PALACE_DAEMON_PATH_MAP="/home/jp/.claude/=/mnt/raid/claude-config/,/home/jp/Projects/=/mnt/raid/projects/"
"""
if raw is _PATH_MAP_USE_ENV:
raw = os.environ.get("PALACE_DAEMON_PATH_MAP", "")
raw = (raw or "").strip()
if not raw:
return []
pairs: list[tuple[str, str]] = []
for entry in raw.split(","):
entry = entry.strip()
if not entry or "=" not in entry:
continue
client_prefix, daemon_prefix = entry.split("=", 1)
client_prefix = client_prefix.strip()
daemon_prefix = daemon_prefix.strip()
if client_prefix and daemon_prefix:
pairs.append((client_prefix, daemon_prefix))
return pairs


def _translate_client_path(path: str) -> str:
"""Translate a client-side absolute path to a daemon-side path.

Hooks running on a client machine (e.g. katana) speak in their own
filesystem namespace (``/home/jp/.claude/...``); the daemon may see the
same files at a different mount (``/mnt/raid/claude-config/...`` via
Syncthing). ``PALACE_DAEMON_PATH_MAP`` lets the operator declare those
rewrites without coupling client code to deployment specifics.

The first matching prefix wins; non-matching paths pass through
unchanged so daemon-side absolute paths still work.

Joining is normalized so mismatched trailing/leading slashes between
the two prefixes can't produce paths like ``/mnt/raid/ccprojects/...``
(Copilot finding on jphein/palace-daemon#1).
"""
for client_prefix, daemon_prefix in _parse_path_map():
if path.startswith(client_prefix):
suffix = path[len(client_prefix):]
return daemon_prefix.rstrip("/") + "/" + suffix.lstrip("/")
return path


def _sem_for(request_dict: dict) -> asyncio.Semaphore:
method = request_dict.get("method", "")
if method == "ping":
Expand Down Expand Up @@ -1013,6 +1078,15 @@ async def mine(request: Request, x_api_key: str | None = Header(default=None)):
directory = body.get("dir")
if not directory:
raise HTTPException(status_code=400, detail="'dir' is required")
if not isinstance(directory, str):
# Closes Copilot finding on jphein/palace-daemon#1: a JSON number /
# object / list would crash _translate_client_path().startswith and
# surface as 500 rather than a clean 400.
raise HTTPException(status_code=400, detail="'dir' must be a string")

# Hook clients send paths in their own filesystem namespace. Translate
# to the daemon's view via PALACE_DAEMON_PATH_MAP before validation.
directory = _translate_client_path(directory)

dir_path = Path(directory)
if not dir_path.is_absolute() or ".." in dir_path.parts:
Expand Down
Empty file added tests/__init__.py
Empty file.
161 changes: 161 additions & 0 deletions tests/test_path_translation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,161 @@
"""Unit tests for PALACE_DAEMON_PATH_MAP parsing + translation.

Run with::

cd /path/to/palace-daemon
python -m unittest tests.test_path_translation -v

These tests are pure-function and need no live daemon, palace, or
network. They exist because the path-translation logic is the only
glue between the hook's client-side path namespace and the daemon's
filesystem view; getting it wrong silently swallows transcript ingest.
"""
import os
import sys
import unittest
from unittest.mock import patch

# Ensure project root is on sys.path so ``import main`` resolves to the
# daemon's main.py regardless of where the test runner is invoked from.
_HERE = os.path.dirname(os.path.abspath(__file__))
_ROOT = os.path.dirname(_HERE)
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)

import main # noqa: E402


class TestParsePathMap(unittest.TestCase):
def test_empty_string_returns_empty(self):
self.assertEqual(main._parse_path_map(""), [])
self.assertEqual(main._parse_path_map(" "), [])

def test_explicit_none_returns_empty(self):
# None now means "no mapping" (does NOT fall back to env). The
# sentinel default keeps the env-reading behavior on the no-arg
# call below — closes Copilot finding on jphein/palace-daemon#1.
self.assertEqual(main._parse_path_map(None), [])

def test_single_pair(self):
out = main._parse_path_map("/home/u/.claude/=/mnt/raid/claude-config/")
self.assertEqual(out, [("/home/u/.claude/", "/mnt/raid/claude-config/")])

def test_multiple_pairs_preserve_order(self):
raw = "/home/u/.claude/=/mnt/raid/cc/,/home/u/Projects/=/mnt/raid/projects/"
out = main._parse_path_map(raw)
self.assertEqual(
out,
[
("/home/u/.claude/", "/mnt/raid/cc/"),
("/home/u/Projects/", "/mnt/raid/projects/"),
],
)

def test_skips_malformed_entries(self):
# Missing '=' and empty entries are skipped, valid ones survive.
raw = "noequals,,/a/=/b/, =/x/, /c/= "
out = main._parse_path_map(raw)
self.assertEqual(out, [("/a/", "/b/")])

def test_strips_whitespace_around_tokens(self):
out = main._parse_path_map(" /a/ = /b/ ")
self.assertEqual(out, [("/a/", "/b/")])

def test_reads_env_var_when_no_arg(self):
# No-arg call reads env. clear=True so a stray PALACE_DAEMON_PATH_MAP
# in the test process can't taint the assertion.
with patch.dict(
os.environ, {"PALACE_DAEMON_PATH_MAP": "/x/=/y/"}, clear=True
):
out = main._parse_path_map()
self.assertEqual(out, [("/x/", "/y/")])

def test_env_unset_no_arg_returns_empty(self):
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(main._parse_path_map(), [])


class TestTranslateClientPath(unittest.TestCase):
def test_passthrough_when_no_map(self):
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/home/u/.claude/projects/-x",
)

def test_first_matching_prefix_wins(self):
env = {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude/=/mnt/raid/cc/"}
with patch.dict(os.environ, env, clear=True):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/mnt/raid/cc/projects/-x",
)

def test_non_matching_path_passes_through(self):
env = {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude/=/mnt/raid/cc/"}
with patch.dict(os.environ, env, clear=True):
self.assertEqual(
main._translate_client_path("/var/log/syslog"),
"/var/log/syslog",
)

def test_multiple_rules_first_match_wins(self):
# Order matters: list more specific rules first if needed.
env = {
"PALACE_DAEMON_PATH_MAP": (
"/home/u/Projects/memorypalace/=/mnt/raid/mempalace/,"
"/home/u/Projects/=/mnt/raid/projects/"
)
}
with patch.dict(os.environ, env, clear=True):
self.assertEqual(
main._translate_client_path("/home/u/Projects/memorypalace/foo"),
"/mnt/raid/mempalace/foo",
)
self.assertEqual(
main._translate_client_path("/home/u/Projects/realmwatch/foo"),
"/mnt/raid/projects/realmwatch/foo",
)

def test_join_normalizes_mismatched_trailing_slash(self):
"""Closes Copilot finding on jphein/palace-daemon#1: prefixes with
mismatched trailing slashes used to produce paths like
``/mnt/raid/ccprojects/...``. Now the join normalizes to exactly
one separator regardless of operator slash style.
"""
# client trailing /, daemon no trailing
with patch.dict(
os.environ, {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude/=/mnt/raid/cc"}, clear=True
):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/mnt/raid/cc/projects/-x",
)
# client no trailing, daemon trailing /
with patch.dict(
os.environ, {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude=/mnt/raid/cc/"}, clear=True
):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/mnt/raid/cc/projects/-x",
)
# both trailing /
with patch.dict(
os.environ, {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude/=/mnt/raid/cc/"}, clear=True
):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/mnt/raid/cc/projects/-x",
)
# neither trailing /
with patch.dict(
os.environ, {"PALACE_DAEMON_PATH_MAP": "/home/u/.claude=/mnt/raid/cc"}, clear=True
):
self.assertEqual(
main._translate_client_path("/home/u/.claude/projects/-x"),
"/mnt/raid/cc/projects/-x",
)


if __name__ == "__main__":
unittest.main()