From c1748765a07463a9315bbfdf28ebdd2b6baf36a0 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sat, 10 Oct 2026 23:24:44 -0500 Subject: [PATCH 1/7] fix: Scope native transcript admission by installed host --- tests/ingest/test_stop_hook_safety.py | 18 +- tests/test-transcript-allowlists-761.py | 409 +++++++++++++++++++++++ tests/test_issue_644_ingest_races.py | 1 + truememory/hooks/adapters/codex.py | 35 +- truememory/hooks/adapters/gemini.py | 30 +- truememory/ingest/cli.py | 35 +- truememory/ingest/hooks/_shared.py | 103 +++++- truememory/ingest/hooks/compact.py | 29 +- truememory/ingest/hooks/session_start.py | 49 ++- truememory/ingest/hooks/stop.py | 63 +++- 10 files changed, 698 insertions(+), 74 deletions(-) create mode 100644 tests/test-transcript-allowlists-761.py diff --git a/tests/ingest/test_stop_hook_safety.py b/tests/ingest/test_stop_hook_safety.py index de2ed8e5..62e148ce 100644 --- a/tests/ingest/test_stop_hook_safety.py +++ b/tests/ingest/test_stop_hook_safety.py @@ -14,6 +14,12 @@ import json import subprocess from pathlib import Path +import pytest + + +@pytest.fixture(autouse=True) +def _allow_synthetic_transcripts(monkeypatch, tmp_path): + monkeypatch.setenv("TRUEMEMORY_TRANSCRIPT_DIR", str(tmp_path)) # --------------------------------------------------------------------------- @@ -46,7 +52,7 @@ def _record_popen(*args, **kwargs): monkeypatch.setattr(subprocess, "Popen", _record_popen) stop_mod._run_background_ingestion( - transcript_path="/fake/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="sess-at-cap", user_id="alice", db_path="/tmp/fake.db", @@ -93,7 +99,7 @@ def _record_popen(*args, **kwargs): monkeypatch.setattr(subprocess, "Popen", _record_popen) stop_mod._run_background_ingestion( - transcript_path="/fake/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="sess-ok", user_id="alice", db_path="/tmp/fake.db", @@ -174,7 +180,7 @@ def _spy_inline(*args, **kwargs): # Must not raise, must not call inline ingest, must queue to backlog. stop_mod._run_background_ingestion( - transcript_path="/fake/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="sess-popen-fail", user_id="alice", db_path="/tmp/fake.db", @@ -207,7 +213,7 @@ def test_backlog_write_failure_is_swallowed(monkeypatch, tmp_path): # Should not raise stop_mod._queue_to_backlog( - transcript_path="/fake/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="sess-backlog-fail", user_id="alice", db_path="/tmp/fake.db", @@ -222,7 +228,7 @@ def test_backlog_marker_schema(monkeypatch, tmp_path): monkeypatch.setattr(stop_mod, "BACKLOG_DIR", tmp_path / "backlog") stop_mod._queue_to_backlog( - transcript_path="/conv/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="sess-abc", user_id="alice", db_path="/tmp/alice.db", @@ -250,7 +256,7 @@ def test_empty_session_id_still_queues(monkeypatch, tmp_path): monkeypatch.setattr(stop_mod, "BACKLOG_DIR", tmp_path / "backlog") stop_mod._queue_to_backlog( - transcript_path="/conv/transcript.json", + transcript_path=str(tmp_path / "transcript.json"), session_id="", user_id="", db_path="", diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py new file mode 100644 index 00000000..5a982c28 --- /dev/null +++ b/tests/test-transcript-allowlists-761.py @@ -0,0 +1,409 @@ +"""Synthetic stdlib controls for trusted transcript admission (#761). + +Run directly with Python. Production functions execute from their AST with +model, database and spawn collaborators replaced by inert boundary stubs. +""" + +from __future__ import annotations + +import argparse +import ast +from contextlib import contextmanager +import io +import json +import logging +import os +from pathlib import Path +import shlex +import subprocess +import sys +import tempfile +import types +import unittest +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] + + +def source_module(name: str, relative_path: str, names: set[str] | None = None) -> types.ModuleType: + tree = ast.parse((ROOT / relative_path).read_text(encoding="utf-8")) + if names is not None: + tree.body = [ + node for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names + ] + module = types.ModuleType(name) + module.__file__ = str(ROOT / relative_path) + module.__dict__.update({ + "Path": Path, "json": json, "os": os, "sys": sys, + "subprocess": subprocess, "argparse": argparse, + "log": logging.getLogger(name), "logging": logging, + }) + sys.modules[name] = module + exec(compile(tree, module.__file__, "exec"), module.__dict__) + return module + + +@contextmanager +def allowed_gate(): + yield True + + +class TranscriptAdmissionTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory(prefix="synthetic-transcripts-761-") + self.addCleanup(self.temp.cleanup) + self.home = Path(self.temp.name) / "home" + self.home.mkdir() + self.home_patch = patch.object(Path, "home", return_value=self.home) + self.home_patch.start() + self.addCleanup(self.home_patch.stop) + env = {key: value for key, value in os.environ.items() if key not in ( + "CODEX_HOME", "GEMINI_CLI_HOME", "SANDBOX", "TRUEMEMORY_TRANSCRIPT_DIR", + "TRUEMEMORY_EXTRACTION", "TRUEMEMORY_MAINTENANCE_CHILD", + )} + self.env_patch = patch.dict(os.environ, env, clear=True) + self.env_patch.start() + self.addCleanup(self.env_patch.stop) + packages: dict[str, types.ModuleType] = {} + for name in ( + "truememory", "truememory.ingest", "truememory.ingest.hooks", + "truememory.hooks", "truememory.hooks.adapters", + ): + module = types.ModuleType(name) + module.__path__ = [] + packages[name] = module + platform = types.ModuleType("truememory._platform") + platform._env_int = lambda name, default, **kwargs: default + platform.spawn_kwargs = lambda: {} + platform.pid_is_alive = lambda pid: False + packages["truememory"]._platform = platform + packages["truememory._platform"] = platform + core = types.ModuleType("truememory.hooks.core") + core.spawn_gate = allowed_gate + core.register_spawned_pid = lambda pid: None + core._load_cap_state = lambda: {"cap": 2} + packages["truememory.hooks.core"] = core + base = types.ModuleType("truememory.hooks.adapters.base") + base.CLIAdapter = type("CLIAdapter", (), {}) + base.atomic_write_text = lambda path, text: path.write_text(text, encoding="utf-8") + packages["truememory.hooks.adapters.base"] = base + self.module_patch = patch.dict(sys.modules, packages) + self.module_patch.start() + self.addCleanup(self.module_patch.stop) + self.shared = source_module( + "truememory.ingest.hooks._shared", "truememory/ingest/hooks/_shared.py", + ) + self.shared.check_extraction_budget = lambda: True + self.shared.refund_extraction_budget = lambda: None + self.cleanup_stale_processing = self.shared.cleanup_stale_processing + self.shared.cleanup_stale_processing = lambda path: None + self.shared.mark_session_extracted = lambda *args, **kwargs: None + self.stop = source_module("truememory.ingest.hooks.stop", "truememory/ingest/hooks/stop.py") + self.stop._writable_dirs_ok = lambda: True + self.stop.TRACE_DIR.mkdir(parents=True) + self.stop.LOG_DIR.mkdir(parents=True) + self.compact = source_module("truememory.ingest.hooks.compact", "truememory/ingest/hooks/compact.py") + self.session_start = source_module( + "truememory.ingest.hooks.session_start", "truememory/ingest/hooks/session_start.py", + {"_drain_backlog", "_run_maintenance_background"}, + ) + self.session_start.BACKLOG_DIR = self.shared.BACKLOG_DIR + self.session_start._DRAIN_CAP = 3 + self.session_start._platform = platform + self.cli = source_module("truememory.ingest.cli", "truememory/ingest/cli.py", {"_cascade_next"}) + self.parser = source_module("synthetic_transcript_parser_761", "truememory/ingest/transcript.py") + + def write(self, path: Path, text: str = "synthetic conversation") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + def native_path(self, host: str, name: str = "session.jsonl") -> Path: + roots = { + "claude": self.home / ".claude/projects/workspace", + "codex": self.home / ".codex/sessions/2026/10/10", + "gemini": self.home / ".gemini/tmp/workspace/chats", + } + return roots[host] / name + + def marker(self, host: str, name: str, path: Path, **extra: object) -> Path: + data = {"transcript_path": str(path), "transcript_host": host, + "session_id": name, "user_id": "synthetic", "db_path": ""} + data.update(extra) + return self.write(self.shared.BACKLOG_DIR / f"{name}.json", json.dumps(data)) + + def test_native_roots_are_isolated_by_installed_host(self) -> None: + for source in ("claude", "codex", "gemini"): + path = self.write(self.native_path(source)) + for host in ("claude", "codex", "gemini", "cursor", "invalid"): + with self.subTest(source=source, host=host): + self.assertEqual(self.shared.is_allowed_transcript(str(path), host), source == host) + + def test_legacy_default_remains_claude(self) -> None: + self.assertTrue(self.shared.is_allowed_transcript(str(self.native_path("claude")))) + self.assertFalse(self.shared.is_allowed_transcript(str(self.native_path("codex")))) + self.assertFalse(self.shared.is_allowed_transcript(str(self.native_path("gemini")))) + + def test_unrelated_home_and_gemini_runtime_paths_are_rejected(self) -> None: + for path in (self.home / "notes.txt", self.home / ".gemini/settings.json", + self.home / ".gemini/tmp/workspace/logs.json", + self.home / ".gemini/tmp/workspace/memory/item.json", + self.home / ".gemini/tmp/chats/item.json"): + with self.subTest(path=path): + self.assertFalse(self.shared.is_allowed_transcript(str(path), "gemini")) + + def test_gemini_supports_nested_subagent_chats(self) -> None: + self.assertTrue(self.shared.is_allowed_transcript( + str(self.native_path("gemini") / "subagent/turn.jsonl"), "gemini", + )) + + def test_codex_home_override_selects_only_sessions(self) -> None: + custom = self.home / "custom-codex" + os.environ["CODEX_HOME"] = str(custom) + self.assertTrue(self.shared.is_allowed_transcript(str(custom / "sessions/date/turn.jsonl"), "codex")) + self.assertFalse(self.shared.is_allowed_transcript(str(custom / "config.toml"), "codex")) + self.assertFalse(self.shared.is_allowed_transcript(str(self.native_path("codex")), "codex")) + self.assertFalse(self.shared.is_allowed_transcript(str(custom / "sessions/date/turn.jsonl"), "gemini")) + + def test_gemini_home_and_sandbox_select_only_documented_runtime(self) -> None: + custom = self.home / "custom-gemini" + os.environ["GEMINI_CLI_HOME"] = str(custom) + path = custom / ".gemini/tmp/workspace/chats/turn.json" + self.assertTrue(self.shared.is_allowed_transcript(str(path), "gemini")) + self.assertFalse(self.shared.is_allowed_transcript(str(self.native_path("gemini")), "gemini")) + os.environ["SANDBOX"] = "sandbox-exec" + sandbox_path = custom / ".cache/.gemini/tmp/workspace/chats/turn.jsonl" + self.assertTrue(self.shared.is_allowed_transcript(str(sandbox_path), "gemini")) + self.assertFalse(self.shared.is_allowed_transcript(str(path), "gemini")) + self.assertFalse(self.shared.is_allowed_transcript(str(sandbox_path), "codex")) + + def test_explicit_scoped_override_and_broad_home_rejection(self) -> None: + custom = self.home / "synthetic-transcripts" + os.environ["TRUEMEMORY_TRANSCRIPT_DIR"] = str(custom) + self.assertTrue(self.shared.is_allowed_transcript(str(custom / "turn.json"), "cursor")) + self.assertFalse(self.shared.is_allowed_transcript(str(custom / "turn.json"), "invalid")) + for broad in (self.home, self.home.parent, Path("/")): + os.environ["TRUEMEMORY_TRANSCRIPT_DIR"] = str(broad) + self.assertFalse(self.shared.is_allowed_transcript(str(self.home / "notes.txt"))) + + def test_path_shape_is_validated_without_coercion(self) -> None: + for value in (None, 42, True, [], {}, "", " ", "bad\0path"): + with self.subTest(value=value): + self.assertFalse(self.shared.is_allowed_transcript(value, "codex")) + for value in (None, [], {}, "CODEX", " codex", "codex;command"): + self.assertEqual(self.shared.normalize_transcript_host(value), "") + + def test_traversal_and_root_itself_are_rejected(self) -> None: + for host in ("claude", "codex", "gemini"): + with self.subTest(host=host): + self.assertFalse(self.shared.is_allowed_transcript(str(self.native_path(host).parent / "../outside.txt"), host)) + self.assertFalse(self.shared.is_allowed_transcript(str(self.home / ".codex/sessions"), "codex")) + self.assertFalse(self.shared.is_allowed_transcript(str(self.home / ".gemini/tmp/workspace/chats"), "gemini")) + + def test_symlink_escapes_and_loops_are_rejected(self) -> None: + outside = self.write(self.home / "outside.txt") + for host in ("claude", "codex", "gemini"): + path = self.native_path(host) + path.parent.mkdir(parents=True, exist_ok=True) + path.symlink_to(outside) + self.assertFalse(self.shared.is_allowed_transcript(str(path), host)) + loop = path.parent / "loop.jsonl" + loop.symlink_to(loop) + self.assertFalse(self.shared.is_allowed_transcript(str(loop), host)) + subtree = self.home / ".gemini/tmp/other/chats" + subtree.parent.mkdir(parents=True) + subtree.symlink_to(self.home) + self.assertFalse(self.shared.is_allowed_transcript(str(subtree / "outside.txt"), "gemini")) + + def test_marker_shape_and_source_are_sanitized(self) -> None: + path = str(self.native_path("codex")) + data = {"transcript_path": path, "transcript_host": "codex", "session_id": "../synthetic:id"} + self.assertEqual(self.shared.backlog_transcript_source(data, "codex"), (path, "syntheticid")) + self.assertIsNone(self.shared.backlog_transcript_source(data, "gemini")) + for key, value in (("transcript_host", []), ("transcript_host", "invalid"), + ("transcript_path", 42), ("user_id", []), ("db_path", "bad\0db"), + ("session_id", {})): + bad = dict(data) + bad[key] = value + self.assertIsNone(self.shared.backlog_transcript_source(bad, "codex")) + for bad in (None, [], 42, "text"): + self.assertIsNone(self.shared.backlog_transcript_source(bad, "codex")) + + def test_legacy_marker_is_only_admitted_by_claude(self) -> None: + data = {"transcript_path": str(self.native_path("claude")), "session_id": "synthetic"} + self.assertIsNotNone(self.shared.backlog_transcript_source(data)) + self.assertIsNone(self.shared.backlog_transcript_source(data, "codex")) + + def test_stale_marker_shape_cannot_crash_source_admission(self) -> None: + for i, data in enumerate(([], {"claimed_pid": []}, {"claimed_pid": True}, {"claimed_pid": -1})): + path = self.write(self.shared.BACKLOG_DIR / f"malformed-{i}.processing", json.dumps(data)) + os.utime(path, (1, 1)) + self.cleanup_stale_processing(self.shared.BACKLOG_DIR) + self.assertEqual(len(list(self.shared.BACKLOG_DIR.glob("*.corrupt"))), 4) + self.assertEqual(list(self.shared.BACKLOG_DIR.glob("*.json")), []) + + def test_stop_correct_host_reaches_parser_once_and_stdin_identity_cannot_override(self) -> None: + for host in ("codex", "gemini"): + text = json.dumps([{"role": "user", "content": "synthetic turn"} for _ in range(5)]) + path = self.write(self.native_path(host), text) + calls: list[tuple[str, str]] = [] + + def extract(transcript: str, session: str, user: str, db: str, transcript_host: str) -> int: + calls.append((transcript, transcript_host)) + self.assertEqual(len(self.parser.parse_transcript(Path(transcript))), 5) + return 17 + + with patch.object(self.stop, "_run_background_ingestion", side_effect=extract), \ + patch.object(sys, "argv", ["stop.py", "--host", host]), \ + patch.object(sys, "stdin", io.StringIO(json.dumps({ + "transcript_path": str(path), "session_id": "synthetic", + "host": "claude", "transcript_host": "claude", + }))): + self.stop.main() + self.assertEqual(calls, [(str(path), host)]) + calls.clear() + with patch.object(self.stop, "_run_background_ingestion", side_effect=extract), \ + patch.object(sys, "argv", ["stop.py", "--host", "claude"]), \ + patch.object(sys, "stdin", io.StringIO(json.dumps({ + "transcript_path": str(path), "session_id": "synthetic", "host": host, + }))): + self.stop.main() + self.assertEqual(calls, []) + + def test_stop_malformed_input_never_probes_source(self) -> None: + for payload in ([], 42, None, {"transcript_path": 42}, {"transcript_path": "bad\0path"}): + with patch.object(self.stop, "_has_enough_messages", side_effect=AssertionError("unexpected read")), \ + patch.object(sys, "argv", ["stop.py", "--host", "codex"]), \ + patch.object(sys, "stdin", io.StringIO(json.dumps(payload))): + self.stop.main() + + def test_compact_correct_host_snapshot_and_background_admission(self) -> None: + path = self.write(self.native_path("gemini"), "synthetic text " * 30) + with patch.object(self.compact, "save_snapshot") as snapshot, \ + patch.object(self.stop, "_run_background_ingestion", return_value=17) as spawn, \ + patch.object(sys, "argv", ["compact.py", "--host", "gemini", "--event", "PreCompress"]), \ + patch.object(sys, "stdin", io.StringIO(json.dumps({"transcript_path": str(path), "session_id": "synthetic"}))): + self.compact.main() + snapshot.assert_called_once() + self.assertEqual(spawn.call_args.kwargs["transcript_host"], "gemini") + with patch.object(self.compact, "save_snapshot") as snapshot, \ + patch.object(sys, "argv", ["compact.py", "--host", "codex"]), \ + patch.object(sys, "stdin", io.StringIO(json.dumps({"transcript_path": str(path), "session_id": "synthetic"}))): + self.compact.main() + snapshot.assert_not_called() + + def test_background_spawn_is_admitted_and_carries_trusted_identity(self) -> None: + path = self.write(self.native_path("codex")) + with patch.object(subprocess, "Popen", return_value=types.SimpleNamespace(pid=17)) as spawn: + self.assertEqual(self.stop._run_background_ingestion(str(path), "synthetic", "", "", "codex"), 17) + cmd = spawn.call_args.args[0] + self.assertEqual(cmd[cmd.index("--transcript-host") + 1], "codex") + self.assertEqual(self.stop._run_background_ingestion(str(path), "synthetic", "", "", "gemini"), 0) + self.assertEqual(spawn.call_count, 1) + + def test_queue_records_valid_source_and_refuses_wrong_host(self) -> None: + path = self.write(self.native_path("gemini")) + self.stop._queue_to_backlog(str(path), "../synthetic:id", "", "", "synthetic", "gemini") + marker = self.shared.BACKLOG_DIR / "syntheticid.json" + data = json.loads(marker.read_text()) + self.assertEqual(data["transcript_host"], "gemini") + self.assertEqual(data["session_id"], "syntheticid") + self.stop._queue_to_backlog(str(path), "wrong", "", "", "synthetic", "codex") + self.assertFalse((self.shared.BACKLOG_DIR / "wrong.json").exists()) + + def test_session_drain_preserves_other_host_and_filters_before_cap(self) -> None: + for i in range(4): + self.marker("gemini", f"a-other-{i}", self.write(self.native_path("gemini", f"{i}.json"))) + path = self.write(self.native_path("codex")) + own = self.marker("codex", "z-own", path) + with patch.object(subprocess, "Popen", return_value=types.SimpleNamespace(pid=17)) as spawn: + self.session_start._drain_backlog("codex") + self.assertEqual(spawn.call_count, 1) + cmd = spawn.call_args.args[0] + self.assertEqual(cmd[cmd.index("--transcript-host") + 1], "codex") + self.assertFalse(own.exists()) + self.assertTrue(own.with_suffix(".processing").exists()) + self.assertEqual(len(list(self.shared.BACKLOG_DIR.glob("a-other-*.json"))), 4) + + def test_session_drain_quarantines_invalid_source_without_spawn(self) -> None: + marker = self.marker("codex", "bad", self.write(self.home / "notes.txt")) + with patch.object(subprocess, "Popen") as spawn: + self.session_start._drain_backlog("codex") + spawn.assert_not_called() + self.assertTrue(marker.with_suffix(".corrupt").exists()) + + def test_cascade_uses_same_admission_and_spawns_at_most_one(self) -> None: + other = self.marker("gemini", "a-other", self.write(self.native_path("gemini"))) + bad = self.marker("codex", "b-bad", self.write(self.home / "notes.txt")) + for name in ("c-own", "d-own"): + self.marker("codex", name, self.write(self.native_path("codex", f"{name}.jsonl"))) + with patch.object(subprocess, "Popen", return_value=types.SimpleNamespace(pid=17)) as spawn: + self.cli._cascade_next("codex") + self.assertEqual(spawn.call_count, 1) + cmd = spawn.call_args.args[0] + self.assertEqual(cmd[cmd.index("--transcript-host") + 1], "codex") + self.assertTrue(other.exists()) + self.assertTrue(bad.with_suffix(".corrupt").exists()) + self.assertTrue((self.shared.BACKLOG_DIR / "d-own.json").exists()) + + def test_maintenance_child_retains_host_and_only_claude_scans_claude_archive(self) -> None: + for host in ("claude", "codex", "gemini", "cursor"): + with patch.object(subprocess, "Popen") as spawn: + self.session_start._run_maintenance_background(host) + script = spawn.call_args.args[0][2] + self.assertIn(f"_drain_backlog({host!r})", script) + self.assertEqual(script.endswith("_scan_stale_sessions()"), host == "claude") + with patch.object(subprocess, "Popen") as spawn: + self.session_start._run_maintenance_background("codex;invalid") + spawn.assert_not_called() + + def test_codex_installer_refreshes_identity_and_preserves_external_hook(self) -> None: + module = source_module("truememory.hooks.adapters.codex", "truememory/hooks/adapters/codex.py") + module._CONFIG_PATH = self.home / ".codex/config.toml" + before = ('[[hooks.Stop]]\n[[hooks.Stop.hooks]]\n' + 'type = "command"\ncommand = "external-helper"\n' + '[[hooks.SessionStart]]\n[[hooks.SessionStart.hooks]]\n' + 'type = "command"\ncommand = "python /truememory/session_start.py"\n' + 'timeout = 10\n') + self.write(module._CONFIG_PATH, before) + adapter = module.CodexAdapter() + adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") + first = module._CONFIG_PATH.read_text() + self.assertIn('command = "external-helper"', first) + hooks = adapter._parse_existing_hooks(first) + for event in module._HOOK_EVENTS: + commands = [hook["command"] for group in hooks[event] for hook in group["hooks"] + if "truememory" in hook.get("command", "")] + self.assertEqual(len(commands), 1) + parts = shlex.split(commands[0]) + self.assertEqual(parts[parts.index("--host") + 1], "codex") + self.assertEqual(parts[parts.index("--event") + 1], event) + adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") + self.assertEqual(module._CONFIG_PATH.read_text(), first) + + def test_gemini_installer_refreshes_identity_and_preserves_external_hook(self) -> None: + module = source_module("truememory.hooks.adapters.gemini", "truememory/hooks/adapters/gemini.py") + module._CONFIG_PATH = self.home / ".gemini/settings.json" + external = {"hooks": [{"type": "command", "command": "external-helper"}]} + self.write(module._CONFIG_PATH, json.dumps({"hooks": {"SessionStart": [external, + {"command": "python /truememory/session_start.py", "timeout": 10000}]}})) + adapter = module.GeminiAdapter() + adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") + first = module._CONFIG_PATH.read_text() + data = json.loads(first) + self.assertEqual(data["hooks"]["SessionStart"][0], external) + for event in module._HOOK_EVENTS: + commands = [hook["command"] for group in data["hooks"][event] for hook in group["hooks"] + if "truememory" in hook["command"]] + self.assertEqual(len(commands), 1) + parts = shlex.split(commands[0]) + self.assertEqual(parts[parts.index("--host") + 1], "gemini") + self.assertEqual(parts[parts.index("--event") + 1], event) + adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") + self.assertEqual(module._CONFIG_PATH.read_text(), first) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_644_ingest_races.py b/tests/test_issue_644_ingest_races.py index e92fe190..6366f86e 100644 --- a/tests/test_issue_644_ingest_races.py +++ b/tests/test_issue_644_ingest_races.py @@ -174,6 +174,7 @@ def test_queue_to_backlog_refuses_to_clobber_live_claim(monkeypatch, tmp_path): claim with a fresh ``.json`` (POSIX rename-overwrite race / M-15). """ from truememory.ingest.hooks import stop as stop_mod + monkeypatch.setenv("TRUEMEMORY_TRANSCRIPT_DIR", str(tmp_path)) backlog = tmp_path / "backlog" backlog.mkdir(parents=True) diff --git a/truememory/hooks/adapters/codex.py b/truememory/hooks/adapters/codex.py index b0bc603d..2d2ad939 100644 --- a/truememory/hooks/adapters/codex.py +++ b/truememory/hooks/adapters/codex.py @@ -158,6 +158,7 @@ def install_hooks( # Migrate any legacy [[hooks]] blocks before appending new ones existing_text = self._remove_legacy_hook_blocks(existing_text) + existing_text = self._refresh_hook_commands(existing_text, py, hooks_dir, user_id, db_path) existing_hooks = self._parse_existing_hooks(existing_text) @@ -167,7 +168,7 @@ def install_hooks( continue script_path = hooks_dir / info["script"] - cmd = self._build_command(py, script_path, user_id, db_path) + cmd = self._build_command(py, script_path, user_id, db_path, event=event) lines_to_append.append("") lines_to_append.append(f"[[hooks.{event}]]") @@ -223,8 +224,11 @@ def _build_command( script_path: Path, user_id: str = "", db_path: str = "", + event: str = "", ) -> str: - parts: list[str] = [python_path, str(script_path)] + parts: list[str] = [python_path, str(script_path), "--host", "codex"] + if event: + parts.extend(["--event", event]) if user_id: parts.extend(["--user", user_id]) if db_path: @@ -234,6 +238,33 @@ def _build_command( return _sp.list2cmdline(parts) return " ".join(shlex.quote(p) for p in parts) + def _refresh_hook_commands( + self, text: str, python_path: str, hooks_dir: Path, user_id: str, db_path: str, + ) -> str: + """Upgrade owned command lines while retaining other TOML fields/hooks.""" + event = "" + lines: list[str] = [] + for line in text.splitlines(keepends=True): + header = re.match(r'^\s*\[\[hooks\.(\w+)\.hooks\]\]\s*$', line) + if header: + event = header.group(1) + elif line.lstrip().startswith("["): + event = "" + if ( + event in _HOOK_EVENTS + and re.match(r'^\s*command\s*=', line) + and _TRUEMEMORY_MARKER in line.lower() + ): + cmd = self._build_command( + python_path, hooks_dir / _HOOK_EVENTS[event]["script"], + user_id, db_path, event=event, + ) + indent = line[:len(line) - len(line.lstrip())] + ending = "\n" if line.endswith("\n") else "" + line = f'{indent}command = "{_toml_escape(cmd)}"{ending}' + lines.append(line) + return "".join(lines) + @staticmethod def _parse_existing_hooks(toml_text: str) -> dict: """Parse hooks from TOML text, returning ``{event: [matcher_groups]}``.""" diff --git a/truememory/hooks/adapters/gemini.py b/truememory/hooks/adapters/gemini.py index 938b6b96..7572f469 100644 --- a/truememory/hooks/adapters/gemini.py +++ b/truememory/hooks/adapters/gemini.py @@ -147,11 +147,30 @@ def install_hooks( event_list = [] hooks[event] = event_list + script_path = hooks_dir / info["script"] + cmd = self._build_command(py, script_path, user_id, db_path, event=event) + for definition in event_list: + if not isinstance(definition, dict): + continue + inner_hooks = definition.get("hooks", []) + if isinstance(inner_hooks, list): + for hook in inner_hooks: + if ( + isinstance(hook, dict) + and isinstance(hook.get("command"), str) + and _TRUEMEMORY_MARKER in hook["command"].lower() + ): + hook["command"] = cmd + legacy_command = definition.get("command", "") + if isinstance(legacy_command, str) and _TRUEMEMORY_MARKER in legacy_command.lower(): + definition.pop("command", None) + definition.pop("timeout", None) + definition["hooks"] = [{ + "type": "command", "command": cmd, + "name": f"truememory-{event.lower()}", "timeout": info["timeout"], + }] if self._event_has_truememory(event_list): continue - - script_path = hooks_dir / info["script"] - cmd = self._build_command(py, script_path, user_id, db_path) event_list.append({ "hooks": [{ "type": "command", @@ -208,8 +227,11 @@ def _build_command( script_path: Path, user_id: str = "", db_path: str = "", + event: str = "", ) -> str: - parts: list[str] = [python_path, str(script_path)] + parts: list[str] = [python_path, str(script_path), "--host", "gemini"] + if event: + parts.extend(["--event", event]) if user_id: parts.extend(["--user", user_id]) if db_path: diff --git a/truememory/ingest/cli.py b/truememory/ingest/cli.py index 53d4ccd2..16b2d30c 100644 --- a/truememory/ingest/cli.py +++ b/truememory/ingest/cli.py @@ -44,6 +44,8 @@ def main(): p_ingest.add_argument("--provider", default="auto", help="LLM provider (auto/ollama/claude_cli/openrouter/anthropic)") p_ingest.add_argument("--model", default="", help="LLM model name") p_ingest.add_argument("--session", default="", help="Session identifier to tag this ingestion trace with") + p_ingest.add_argument("--transcript-host", choices=("claude", "codex", "gemini", "cursor"), + default="claude", help=argparse.SUPPRESS) p_ingest.add_argument("-v", "--verbose", action="store_true", help="Verbose logging") # --- install command --- @@ -301,10 +303,10 @@ def _run_ingest(args): args.session, e, ) - _cascade_next() + _cascade_next(getattr(args, "transcript_host", "claude")) -def _cascade_next() -> None: +def _cascade_next(host: str = "claude") -> None: """Spawn the next backlog ingest process if work remains and slots are free. This creates a self-sustaining chain: each ingest process spawns the @@ -328,7 +330,12 @@ def _cascade_next() -> None: record_stale_processing_pid, mark_session_extracted, _quarantine_marker, + backlog_transcript_source, + normalize_transcript_host, ) + host = normalize_transcript_host(host) + if not host: + return cleanup_stale_processing(backlog_dir) try: @@ -344,7 +351,7 @@ def _cascade_next() -> None: except ImportError: return - for marker_path in markers[:1]: + for marker_path in markers: claimed_path = marker_path.with_suffix(".processing") try: marker_path.rename(claimed_path) @@ -368,9 +375,20 @@ def _cascade_next() -> None: continue try: - transcript = data.get("transcript_path", "") - if not transcript or not Path(transcript).exists(): - claimed_path.unlink(missing_ok=True) + if ( + isinstance(data, dict) + and normalize_transcript_host(data.get("transcript_host", "claude")) + and data.get("transcript_host", "claude") != host + ): + claimed_path.rename(marker_path) + continue + source = backlog_transcript_source(data, host=host) + if source is None: + _quarantine_marker(claimed_path) + continue + transcript, session_id = source + if not Path(transcript).is_file(): + _quarantine_marker(claimed_path) continue if not check_extraction_budget(): @@ -393,8 +411,8 @@ def _cascade_next() -> None: cmd = [ sys.executable, "-m", "truememory.ingest.cli", "ingest", transcript, + "--transcript-host", host, ] - session_id = data.get("session_id", "") if session_id: cmd.extend(["--session", session_id]) if data.get("user_id"): @@ -422,8 +440,9 @@ def _cascade_next() -> None: # the session if the spawned worker exits non-zero. logging.getLogger(__name__).info( "Cascade: spawned next ingest for session %s (PID %d)", - data.get("session_id", "?"), proc.pid, + session_id, proc.pid, ) + return except Exception: try: claimed_path.rename(marker_path) diff --git a/truememory/ingest/hooks/_shared.py b/truememory/ingest/hooks/_shared.py index dc434931..81abe8fb 100644 --- a/truememory/ingest/hooks/_shared.py +++ b/truememory/ingest/hooks/_shared.py @@ -31,41 +31,106 @@ # compact hook only; it is hoisted here so stop.py and the SessionStart # backlog drain share the same check. -def _transcript_roots() -> list[Path]: - """Directories a ``transcript_path`` is allowed to live under. +def normalize_transcript_host(host: object) -> str: + """Validate the installed command identity; never infer it from stdin.""" + if isinstance(host, str) and host in ("claude", "codex", "gemini", "cursor"): + return host + return "" - Defaults to Claude Code's ``~/.claude/projects``. An explicit - ``TRUEMEMORY_TRANSCRIPT_DIR`` override (tests / non-default installs) is - honored when set. + +def transcript_input_path(value: object) -> str: + """Reject malformed source paths before any filesystem probe.""" + if not isinstance(value, str) or not value.strip() or "\0" in value: + return "" + return value + + +def _transcript_roots(host: str = "claude", transcript_path: str = "") -> list[Path]: + """Native transcript roots scoped to the trusted installed host. + + Gemini's tmp directory also contains non-transcript state. Derive only + the candidate's single-workspace ``chats`` subtree after canonical + containment, without enumerating workspace directories. """ - roots = [Path.home() / ".claude" / "projects"] + host = normalize_transcript_host(host) + if not host: + return [] + roots: list[Path] = [] + if host == "claude": + roots.append(Path.home() / ".claude" / "projects") + elif host == "codex": + codex_home = os.environ.get("CODEX_HOME", "") + roots.append((Path(codex_home) if codex_home else Path.home() / ".codex") / "sessions") + elif host == "gemini" and transcript_input_path(transcript_path): + gemini_home = os.environ.get("GEMINI_CLI_HOME", "") + home = Path(gemini_home) if gemini_home else Path.home() + runtime = home / ".gemini" + if os.environ.get("SANDBOX") == "sandbox-exec": + runtime = home / ".cache" / ".gemini" + try: + temp_root = (runtime / "tmp").resolve() + relative = Path(transcript_path).resolve().relative_to(temp_root) + if len(relative.parts) >= 3 and relative.parts[1] == "chats": + roots.append(temp_root / relative.parts[0] / "chats") + except (OSError, ValueError, RuntimeError): + pass override = os.environ.get("TRUEMEMORY_TRANSCRIPT_DIR", "") if override: - roots.insert(0, Path(override)) + try: + override_root = Path(override).resolve() + # A directory override cannot turn the allowlist into all of home. + if not Path.home().resolve().is_relative_to(override_root): + roots.insert(0, override_root) + except (OSError, ValueError, RuntimeError): + pass return roots -def is_allowed_transcript(transcript_path: str) -> bool: +def is_allowed_transcript(transcript_path: str, host: str = "claude") -> bool: """Return True only if *transcript_path* resolves inside a transcripts root. Resolves symlinks (``.resolve()``) and requires containment, so neither a ``..`` escape nor a symlink whose target is outside the root passes. """ - if not transcript_path: + transcript_path = transcript_input_path(transcript_path) + if not transcript_path or not normalize_transcript_host(host): return False try: - resolved = Path(transcript_path).resolve() - except (OSError, ValueError): + source = Path(transcript_path) + if ".." in source.parts: + return False + resolved = source.resolve() + # Non-strict resolution can leave cyclic/dangling links unresolved + # on newer Python versions. Such paths have no canonical admission. + if resolved.is_symlink() or any(parent.is_symlink() for parent in resolved.parents): + return False + except (OSError, ValueError, RuntimeError): return False - for root in _transcript_roots(): + for root in _transcript_roots(host, str(resolved)): try: - if resolved.is_relative_to(root.resolve()): + if resolved != root.resolve() and resolved.is_relative_to(root.resolve()): return True - except (OSError, ValueError): + except (OSError, ValueError, RuntimeError): continue return False +def backlog_transcript_source(data: object, host: str = "claude") -> tuple[str, str] | None: + """Validate marker shape/source without letting markers select another host.""" + active_host = normalize_transcript_host(host) + if not active_host or not isinstance(data, dict): + return None + marker_host = normalize_transcript_host(data.get("transcript_host", "claude")) + transcript = transcript_input_path(data.get("transcript_path", "")) + if marker_host != active_host or not is_allowed_transcript(transcript, host=active_host): + return None + for key in ("session_id", "user_id", "db_path"): + value = data.get(key, "") + if not isinstance(value, str) or "\0" in value: + return None + return transcript, _safe_session_id(data.get("session_id", "")) + + EXTRACTED_DIR = Path.home() / ".truememory" / "extracted" BACKLOG_DIR = Path.home() / ".truememory" / "backlog" RECALL_MARKER_DIR = Path.home() / ".truememory" / "recall_markers" @@ -118,6 +183,8 @@ def get_recall_deadline() -> float | None: def _safe_session_id(session_id: str) -> str: """Sanitize session_id to prevent path traversal.""" + if not isinstance(session_id, str): + return "" return "".join(c for c in session_id if c.isalnum() or c in "-_")[:64] @@ -344,6 +411,8 @@ def record_stale_processing_pid(processing_path: Path, pid: int) -> None: """Write the spawned PID into a .processing file for liveness checks.""" try: data = json.loads(processing_path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + return data["claimed_pid"] = pid _atomic_write_text(processing_path, json.dumps(data)) except (OSError, json.JSONDecodeError): @@ -418,7 +487,13 @@ def cleanup_stale_processing(backlog_dir: Path) -> None: continue try: data = json.loads(stale.read_text(encoding="utf-8")) + if not isinstance(data, dict): + _quarantine_marker(stale) + continue pid = data.get("claimed_pid", 0) + if not isinstance(pid, int) or isinstance(pid, bool) or pid < 0: + _quarantine_marker(stale) + continue if pid and _pid_is_alive(pid): continue except (json.JSONDecodeError, OSError): diff --git a/truememory/ingest/hooks/compact.py b/truememory/ingest/hooks/compact.py index eb26c915..2edbe494 100755 --- a/truememory/ingest/hooks/compact.py +++ b/truememory/ingest/hooks/compact.py @@ -33,16 +33,16 @@ # (hoisted so stop.py and the SessionStart backlog drain share the same guard). # These thin shims keep the original names working for callers/tests. -def _transcript_roots() -> list[Path]: +def _transcript_roots(host: str = "claude", transcript_path: str = "") -> list[Path]: """Shim — see ``_shared._transcript_roots``.""" from truememory.ingest.hooks._shared import _transcript_roots as _shared_roots - return _shared_roots() + return _shared_roots(host, transcript_path) -def _is_allowed_transcript(transcript_path: str) -> bool: +def _is_allowed_transcript(transcript_path: str, host: str = "claude") -> bool: """Shim — delegates to the shared M-90 guard (``_shared.is_allowed_transcript``).""" from truememory.ingest.hooks._shared import is_allowed_transcript - return is_allowed_transcript(transcript_path) + return is_allowed_transcript(transcript_path, host=host) def _parse_args() -> argparse.Namespace: @@ -54,11 +54,13 @@ def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(add_help=False) p.add_argument("--user", default=os.environ.get("TRUEMEMORY_USER_ID", "")) p.add_argument("--db", default=os.environ.get("TRUEMEMORY_DB_PATH", "")) + p.add_argument("--host", choices=("claude", "codex", "gemini", "cursor"), default="claude") + p.add_argument("--event", default="PreCompact") args, _ = p.parse_known_args() return args -def main(): +def main() -> None: if os.environ.get("TRUEMEMORY_EXTRACTION"): return @@ -69,10 +71,10 @@ def main(): except (json.JSONDecodeError, EOFError): input_data = {} - # M-95: stdin fields may be non-strings (e.g. an int session_id) from a - # misbehaving / hostile caller. Coerce to str before any str/Path op so the - # hook never crashes with TypeError. - transcript_path = str(input_data.get("transcript_path", "") or "") + if not isinstance(input_data, dict): + return + from truememory.ingest.hooks._shared import transcript_input_path + transcript_path = transcript_input_path(input_data.get("transcript_path", "")) session_id = str(input_data.get("session_id", "unknown") or "unknown") # Sanitize session_id to prevent injection (consistent with user_prompt_submit.py) @@ -81,13 +83,15 @@ def main(): safe_id = "unknown" session_id = safe_id - if not transcript_path or not Path(transcript_path).exists(): + if not transcript_path: return # M-90: reject transcript paths outside the expected transcripts root so an # attacker-chosen arbitrary file is never read into the memory store. - if not _is_allowed_transcript(transcript_path): - log.warning("Rejecting transcript_path outside transcripts root: %r", transcript_path) + if not _is_allowed_transcript(transcript_path, host=args.host): + log.warning("Compact transcript source rejected for host=%s", args.host) + return + if not Path(transcript_path).is_file(): return try: @@ -108,6 +112,7 @@ def main(): spawned_pid = _run_background_ingestion( transcript_path, session_id, user_id=args.user, db_path=args.db, + transcript_host=args.host, ) # Only mark extracted on a real spawn; pid==0 means the session # was queued to the backlog (spawn cap / Popen failure) and must diff --git a/truememory/ingest/hooks/session_start.py b/truememory/ingest/hooks/session_start.py index 96948de7..cfc39c6f 100755 --- a/truememory/ingest/hooks/session_start.py +++ b/truememory/ingest/hooks/session_start.py @@ -231,7 +231,7 @@ def _check_email_needed() -> str: return "" -def _drain_backlog() -> None: +def _drain_backlog(host: str = "claude") -> None: """Process queued sessions from the backlog directory. Uses the flock-based spawn gate from core to prevent the avalanche @@ -252,17 +252,25 @@ def _drain_backlog() -> None: mark_session_extracted, _quarantine_marker, is_allowed_transcript, + backlog_transcript_source, + normalize_transcript_host, ) + host = normalize_transcript_host(host) + if not host: + return cleanup_stale_processing(BACKLOG_DIR) try: - markers = sorted(BACKLOG_DIR.glob("*.json"))[:_DRAIN_CAP] + markers = sorted(BACKLOG_DIR.glob("*.json")) except Exception: return from truememory.hooks.core import spawn_gate, register_spawned_pid + drained = 0 for marker_path in markers: + if drained >= _DRAIN_CAP: + return claimed_path = marker_path.with_suffix(".processing") try: marker_path.rename(claimed_path) @@ -287,16 +295,24 @@ def _drain_backlog() -> None: continue try: - transcript = data.get("transcript_path", "") - if not transcript or not Path(transcript).exists(): - claimed_path.unlink(missing_ok=True) - continue # M-90: a backlog marker's transcript_path is attacker-influenceable; # confirm containment in an allowed transcripts root before parsing # (parse_transcript's plaintext fallback would read any readable file). - if not is_allowed_transcript(transcript): - log.warning("Drain: transcript_path %r outside allowed roots; dropping marker", transcript) - claimed_path.unlink(missing_ok=True) + if ( + isinstance(data, dict) + and normalize_transcript_host(data.get("transcript_host", "claude")) + and data.get("transcript_host", "claude") != host + ): + claimed_path.rename(marker_path) + continue + source = backlog_transcript_source(data, host=host) + if source is None: + log.warning("Drain: rejected backlog transcript source for host=%s", host) + _quarantine_marker(claimed_path) + continue + transcript, session_id = source + if not is_allowed_transcript(transcript, host=host) or not Path(transcript).is_file(): + _quarantine_marker(claimed_path) continue if not check_extraction_budget(): @@ -323,8 +339,8 @@ def _drain_backlog() -> None: cmd = [ sys.executable, "-m", "truememory.ingest.cli", "ingest", transcript, + "--transcript-host", host, ] - session_id = data.get("session_id", "") if session_id: cmd.extend(["--session", session_id]) if data.get("user_id"): @@ -334,7 +350,7 @@ def _drain_backlog() -> None: from truememory.ingest.hooks._shared import _safe_session_id _log_dir = Path.home() / ".truememory" / "logs" _log_dir.mkdir(parents=True, exist_ok=True) - _safe_sid = _safe_session_id(data.get('session_id', 'unknown')) or 'unknown' + _safe_sid = _safe_session_id(session_id) or 'unknown' _log_file = open( _log_dir / f"{_safe_sid}.log", "a", encoding="utf-8", @@ -347,6 +363,7 @@ def _drain_backlog() -> None: stdin=subprocess.DEVNULL, **_platform.spawn_kwargs(), ) + drained += 1 finally: _log_file.close() register_spawned_pid(proc.pid) @@ -624,7 +641,7 @@ def _commit_watermark(value: float) -> None: pass -def _run_maintenance_background() -> None: +def _run_maintenance_background(host: str = "claude") -> None: """Spawn backlog drain + stale-session scan in a detached subprocess. Uses ``Popen(start_new_session=True)`` so the child survives hook exit @@ -637,6 +654,10 @@ def _run_maintenance_background() -> None: # storm). The flag was previously set but never read. if os.environ.get("TRUEMEMORY_MAINTENANCE_CHILD") == "1": return + from truememory.ingest.hooks._shared import normalize_transcript_host + host = normalize_transcript_host(host) + if not host: + return import subprocess @@ -646,8 +667,8 @@ def _run_maintenance_background() -> None: script = ( "from truememory.ingest.hooks.session_start import _drain_backlog, _scan_stale_sessions; " - "_drain_backlog(); " - "_scan_stale_sessions()" + f"_drain_backlog({host!r}); " + + ("_scan_stale_sessions()" if host == "claude" else "") ) try: diff --git a/truememory/ingest/hooks/stop.py b/truememory/ingest/hooks/stop.py index 26766d1a..b85fec7d 100755 --- a/truememory/ingest/hooks/stop.py +++ b/truememory/ingest/hooks/stop.py @@ -79,6 +79,8 @@ def _safe_int_env(name: str, default: int) -> int: def _sanitize_session_id(session_id: str) -> str: """Sanitize session_id to prevent path traversal.""" + if not isinstance(session_id, str): + return "unknown" safe = "".join(c for c in session_id if c.isalnum() or c in "-_")[:64] return safe or "unknown" @@ -108,45 +110,54 @@ def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(add_help=False) p.add_argument("--user", default=os.environ.get("TRUEMEMORY_USER_ID", "")) p.add_argument("--db", default=os.environ.get("TRUEMEMORY_DB_PATH", "")) + p.add_argument("--host", choices=("claude", "codex", "gemini", "cursor"), default="claude") + p.add_argument("--event", default="Stop") args, _ = p.parse_known_args() return args -def main(): +def main() -> None: + args = _parse_args() if os.environ.get("TRUEMEMORY_EXTRACTION"): try: input_data = json.load(sys.stdin) except (json.JSONDecodeError, EOFError): input_data = {} - transcript_path = input_data.get("transcript_path", "") - session_id = input_data.get("session_id", "") - if transcript_path and session_id: - from truememory.ingest.hooks._shared import mark_session_extracted + if not isinstance(input_data, dict): + return + from truememory.ingest.hooks._shared import ( + is_allowed_transcript, mark_session_extracted, transcript_input_path, + ) + transcript_path = transcript_input_path(input_data.get("transcript_path", "")) + session_id = _sanitize_session_id(input_data.get("session_id", "")) + if is_allowed_transcript(transcript_path, host=args.host) and session_id != "unknown": try: mark_session_extracted(session_id, transcript_path) except Exception as exc: log.debug("stop hook: failed to mark extraction session %s: %s", session_id, exc) return - args = _parse_args() - try: input_data = json.load(sys.stdin) except (json.JSONDecodeError, EOFError): input_data = {} - transcript_path = input_data.get("transcript_path", "") - session_id = input_data.get("session_id", "unknown") + if not isinstance(input_data, dict): + return + from truememory.ingest.hooks._shared import is_allowed_transcript, transcript_input_path + transcript_path = transcript_input_path(input_data.get("transcript_path", "")) + session_id = _sanitize_session_id(input_data.get("session_id", "unknown")) - if not transcript_path or not Path(transcript_path).exists(): + if not transcript_path: return # M-90: transcript_path comes from attacker-influenceable hook stdin. # parse_transcript's plaintext fallback would read ANY readable file into # the memory store, so confirm containment in an allowed transcripts root # (the compact hook already did this; stop/drain previously did not). - from truememory.ingest.hooks._shared import is_allowed_transcript - if not is_allowed_transcript(transcript_path): - log.warning("stop hook: transcript_path %r outside allowed roots; skipping", transcript_path) + if not is_allowed_transcript(transcript_path, host=args.host): + log.warning("stop hook: transcript source rejected for host=%s", args.host) + return + if not Path(transcript_path).is_file(): return # Pre-flight check: ensure our write directories exist and are writable @@ -165,7 +176,9 @@ def main(): log.info("stop hook: session %s already extracted at this size, skipping", session_id) return - spawned_pid = _run_background_ingestion(transcript_path, session_id, args.user, args.db) + spawned_pid = _run_background_ingestion( + transcript_path, session_id, args.user, args.db, transcript_host=args.host, + ) # Only mark the session extracted when a real ingest process was spawned. # _run_background_ingestion returns 0 when the spawn cap denied it or Popen @@ -303,6 +316,7 @@ def _queue_to_backlog( user_id: str, db_path: str, reason: str, + transcript_host: str = "claude", ) -> None: """Drop a queue marker so a later session can re-attempt this ingestion. @@ -312,6 +326,13 @@ def _queue_to_backlog( BACKLOG_DIR and let the next session's startup hook drain it (drain path is a follow-up item). """ + from truememory.ingest.hooks._shared import is_allowed_transcript, normalize_transcript_host + transcript_host = normalize_transcript_host(transcript_host) + if not is_allowed_transcript(transcript_path, host=transcript_host): + return + if any(not isinstance(value, str) or "\0" in value for value in (user_id, db_path)): + return + session_id = _sanitize_session_id(session_id) try: from datetime import datetime, timezone BACKLOG_DIR.mkdir(parents=True, exist_ok=True) @@ -328,6 +349,7 @@ def _queue_to_backlog( marker = BACKLOG_DIR / f"{safe_sid}.json" _atomic_write_text(marker, json.dumps({ "transcript_path": transcript_path, + "transcript_host": transcript_host, "session_id": session_id, "user_id": user_id, "db_path": db_path, @@ -345,6 +367,7 @@ def _run_background_ingestion( session_id: str, user_id: str, db_path: str, + transcript_host: str = "claude", ) -> int: """Launch the ingestion pipeline as a background process. @@ -358,6 +381,14 @@ def _run_background_ingestion( for a later session to re-attempt — NEVER fall back to synchronous inline ingestion (that blocks Claude Code's shutdown). """ + from truememory.ingest.hooks._shared import is_allowed_transcript, normalize_transcript_host + transcript_host = normalize_transcript_host(transcript_host) + if not is_allowed_transcript(transcript_path, host=transcript_host): + log.warning("stop hook: background transcript source rejected") + return 0 + if any(not isinstance(value, str) or "\0" in value for value in (user_id, db_path)): + return 0 + session_id = _sanitize_session_id(session_id) # Log the effective config for debugging. Operators commonly wire up # multiple profiles and need to confirm which user/db the hook actually # saw after the arg-parse + env-var resolution. @@ -370,6 +401,7 @@ def _run_background_ingestion( cmd = [ sys.executable, "-m", "truememory.ingest.cli", "ingest", transcript_path, + "--transcript-host", transcript_host, ] if user_id: @@ -405,6 +437,7 @@ def _run_background_ingestion( _queue_to_backlog( transcript_path, session_id, user_id, db_path, reason="extraction_budget_exhausted", + transcript_host=transcript_host, ) return 0 @@ -424,6 +457,7 @@ def _run_background_ingestion( _queue_to_backlog( transcript_path, session_id, user_id, db_path, reason=f"spawn_cap_reached:SPAWN_CAP={effective_cap}", + transcript_host=transcript_host, ) return 0 @@ -449,6 +483,7 @@ def _run_background_ingestion( _queue_to_backlog( transcript_path, session_id, user_id, db_path, reason=f"popen_failed:{type(e).__name__}:{e}", + transcript_host=transcript_host, ) return 0 finally: From 06e95174b1140750f8779eb163355644704a8417 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sat, 10 Oct 2026 23:31:46 -0500 Subject: [PATCH 2/7] fix: Recognize owned hook invocations before rewriting commands --- tests/test-transcript-allowlists-761.py | 128 ++++++++++++++++++++++-- tests/test_codex_adapter.py | 16 +-- tests/test_gemini_adapter.py | 4 +- truememory/hooks/adapters/base.py | 43 ++++++++ truememory/hooks/adapters/codex.py | 34 +++++-- truememory/hooks/adapters/gemini.py | 10 +- 6 files changed, 201 insertions(+), 34 deletions(-) diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py index 5a982c28..53757994 100644 --- a/tests/test-transcript-allowlists-761.py +++ b/tests/test-transcript-allowlists-761.py @@ -35,11 +35,12 @@ def source_module(name: str, relative_path: str, names: set[str] | None = None) ] module = types.ModuleType(name) module.__file__ = str(ROOT / relative_path) - module.__dict__.update({ - "Path": Path, "json": json, "os": os, "sys": sys, - "subprocess": subprocess, "argparse": argparse, - "log": logging.getLogger(name), "logging": logging, - }) + if names is not None: + module.__dict__.update({ + "Path": Path, "json": json, "os": os, "sys": sys, + "subprocess": subprocess, "argparse": argparse, + "log": logging.getLogger(name), "logging": logging, + }) sys.modules[name] = module exec(compile(tree, module.__file__, "exec"), module.__dict__) return module @@ -85,13 +86,10 @@ def setUp(self) -> None: core.register_spawned_pid = lambda pid: None core._load_cap_state = lambda: {"cap": 2} packages["truememory.hooks.core"] = core - base = types.ModuleType("truememory.hooks.adapters.base") - base.CLIAdapter = type("CLIAdapter", (), {}) - base.atomic_write_text = lambda path, text: path.write_text(text, encoding="utf-8") - packages["truememory.hooks.adapters.base"] = base self.module_patch = patch.dict(sys.modules, packages) self.module_patch.start() self.addCleanup(self.module_patch.stop) + source_module("truememory.hooks.adapters.base", "truememory/hooks/adapters/base.py") self.shared = source_module( "truememory.ingest.hooks._shared", "truememory/ingest/hooks/_shared.py", ) @@ -365,7 +363,7 @@ def test_codex_installer_refreshes_identity_and_preserves_external_hook(self) -> before = ('[[hooks.Stop]]\n[[hooks.Stop.hooks]]\n' 'type = "command"\ncommand = "external-helper"\n' '[[hooks.SessionStart]]\n[[hooks.SessionStart.hooks]]\n' - 'type = "command"\ncommand = "python /truememory/session_start.py"\n' + 'type = "command"\ncommand = "python /truememory/ingest/hooks/session_start.py"\n' 'timeout = 10\n') self.write(module._CONFIG_PATH, before) adapter = module.CodexAdapter() @@ -388,7 +386,7 @@ def test_gemini_installer_refreshes_identity_and_preserves_external_hook(self) - module._CONFIG_PATH = self.home / ".gemini/settings.json" external = {"hooks": [{"type": "command", "command": "external-helper"}]} self.write(module._CONFIG_PATH, json.dumps({"hooks": {"SessionStart": [external, - {"command": "python /truememory/session_start.py", "timeout": 10000}]}})) + {"command": "python /truememory/ingest/hooks/session_start.py", "timeout": 10000}]}})) adapter = module.GeminiAdapter() adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") first = module._CONFIG_PATH.read_text() @@ -404,6 +402,114 @@ def test_gemini_installer_refreshes_identity_and_preserves_external_hook(self) - adapter.install_hooks(python_path="/synthetic path/python", user_id="synthetic") self.assertEqual(module._CONFIG_PATH.read_text(), first) + def test_codex_foreign_mentions_are_preserved_and_do_not_suppress_registration(self) -> None: + module = source_module("truememory.hooks.adapters.codex", "truememory/hooks/adapters/codex.py") + module._CONFIG_PATH = self.home / ".codex/config.toml" + foreign_commands = ( + "/synthetic/foreign-policy --target truememory", + "/synthetic/foreign-policy --target /pkg/truememory/ingest/hooks/session_start.py", + "python /synthetic/foreign-policy.py --target truememory", + "python -c 'print(\"truememory\")'", + "echo python /pkg/truememory/ingest/hooks/session_start.py", + "/synthetic/foreign-policy # python /pkg/truememory/ingest/hooks/session_start.py", + "python /pkg/not-truememory/ingest/hooks/session_start.py", + "python /pkg/truememory/ingest/hooks/session_start.py && /synthetic/foreign-policy", + "python /pkg/truememory/session_start.py", + "python /pkg/truememory/hooks/session_start.py", + ) + for foreign in foreign_commands: + with self.subTest(foreign=foreign): + command_line = 'command = ' + json.dumps(foreign) + '\n' + before = ('[[hooks.SessionStart]]\n[[hooks.SessionStart.hooks]]\n' + 'type = "command"\n' + command_line + 'timeout = 21\n') + self.write(module._CONFIG_PATH, before) + adapter = module.CodexAdapter() + adapter.install_hooks(python_path="/synthetic/python") + first = module._CONFIG_PATH.read_text() + self.assertIn(before, first) + commands = [hook["command"] for group in adapter._parse_existing_hooks(first)["SessionStart"] + for hook in group["hooks"]] + self.assertIn(foreign, commands) + self.assertEqual(len(commands), 2) + adapter.install_hooks(python_path="/synthetic/python") + self.assertEqual(module._CONFIG_PATH.read_text(), first) + + def test_gemini_foreign_mentions_are_preserved_and_do_not_suppress_registration(self) -> None: + module = source_module("truememory.hooks.adapters.gemini", "truememory/hooks/adapters/gemini.py") + module._CONFIG_PATH = self.home / ".gemini/settings.json" + foreign_commands = ( + "/synthetic/foreign-policy --target truememory", + "/synthetic/foreign-policy --target /pkg/truememory/ingest/hooks/session_start.py", + "python /synthetic/foreign-policy.py --target truememory", + "python -c 'print(\"truememory\")'", + "echo python /pkg/truememory/ingest/hooks/session_start.py", + "/synthetic/foreign-policy # python /pkg/truememory/ingest/hooks/session_start.py", + "python /pkg/not-truememory/ingest/hooks/session_start.py", + "python /pkg/truememory/ingest/hooks/session_start.py && /synthetic/foreign-policy", + "python /pkg/truememory/session_start.py", + "python /pkg/truememory/hooks/session_start.py", + ) + for foreign in foreign_commands: + for nested in (True, False): + with self.subTest(foreign=foreign, nested=nested): + hook = {"command": foreign, "timeout": 21} + definition = {"hooks": [hook]} if nested else hook + self.write(module._CONFIG_PATH, json.dumps({"hooks": {"SessionStart": [definition]}})) + adapter = module.GeminiAdapter() + adapter.install_hooks(python_path="/synthetic/python") + first = module._CONFIG_PATH.read_text() + definitions = json.loads(first)["hooks"]["SessionStart"] + self.assertEqual(definitions[0], definition) + self.assertEqual(len(definitions), 2) + adapter.install_hooks(python_path="/synthetic/python") + self.assertEqual(module._CONFIG_PATH.read_text(), first) + + def test_owned_hook_predicate_requires_actual_python_invocation(self) -> None: + predicate = sys.modules["truememory.hooks.adapters.base"].is_owned_hook_command + for stem in ("session_start", "user_prompt_submit", "stop", "compact"): + for command in ( + f"python /pkg/truememory/ingest/hooks/{stem}.py", + f"'/pkg with space/python3.10' '/pkg with space/truememory/ingest/hooks/{stem}.py' --user ';'", + f"python -B -m truememory.ingest.hooks.{stem}", + ): + with self.subTest(command=command): + self.assertTrue(predicate(command)) + for command in (None, [], "", "python -c 'truememory.ingest.hooks.stop'", + "python -m foreign --target truememory.ingest.hooks.stop", + "foreign /pkg/truememory/ingest/hooks/stop.py", "python-policy /pkg/truememory/ingest/hooks/stop.py", + "python /pkg/truememory/ingest/hooks/stop.py; foreign", "python /pkg/truememory/ingest/hooks/stop.py\nforeign", + "python /pkg/truememory/ingest/hooks/stop.py > output", "python '/pkg/truememory/ingest/hooks/stop.py", + "python \"$(foreign)/truememory/ingest/hooks/stop.py\"", "python /pkg/truememory/ingest/hooks/not-a-hook.py"): + with self.subTest(command=command): + self.assertFalse(predicate(command)) + with patch.object(sys, "platform", "win32"): + command = subprocess.list2cmdline([ + r"C:\Program Files\Python\python.exe", + r"C:\Synthetic Package\truememory\ingest\hooks\stop.py", + "--host", "codex", "--user", "synthetic user", + ]) + self.assertTrue(predicate(command)) + self.assertFalse(predicate(r"C:\foreign.exe --target truememory")) + + def test_owned_module_invocations_are_upgraded_without_duplication(self) -> None: + codex = source_module("truememory.hooks.adapters.codex", "truememory/hooks/adapters/codex.py") + codex._CONFIG_PATH = self.home / ".codex/config.toml" + self.write(codex._CONFIG_PATH, '[[hooks.SessionStart]]\n[[hooks.SessionStart.hooks]]\n' + 'command = "python -m truememory.ingest.hooks.session_start"\n') + adapter = codex.CodexAdapter() + adapter.install_hooks(python_path="/synthetic/python") + groups = adapter._parse_existing_hooks(codex._CONFIG_PATH.read_text())["SessionStart"] + self.assertEqual(len(groups), 1) + self.assertIn("--host codex --event SessionStart", groups[0]["hooks"][0]["command"]) + gemini = source_module("truememory.hooks.adapters.gemini", "truememory/hooks/adapters/gemini.py") + gemini._CONFIG_PATH = self.home / ".gemini/settings.json" + self.write(gemini._CONFIG_PATH, json.dumps({"hooks": {"SessionStart": [{"hooks": [ + {"command": "python -m truememory.ingest.hooks.session_start"}]}]}})) + gemini.GeminiAdapter().install_hooks(python_path="/synthetic/python") + groups = json.loads(gemini._CONFIG_PATH.read_text())["hooks"]["SessionStart"] + self.assertEqual(len(groups), 1) + self.assertIn("--host gemini --event SessionStart", groups[0]["hooks"][0]["command"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_codex_adapter.py b/tests/test_codex_adapter.py index 6a3184a4..6bd884de 100644 --- a/tests/test_codex_adapter.py +++ b/tests/test_codex_adapter.py @@ -218,7 +218,7 @@ def test_install_hooks_migrates_legacy_format(tmp_path, monkeypatch): '[general]\ntheme = "dark"\n\n' '[[hooks]]\n' 'event = "SessionStart"\n' - 'command = "/usr/bin/python3 /path/to/truememory/session_start.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py"\n' 'timeout = 10000\n', encoding="utf-8", ) @@ -338,7 +338,7 @@ def test_uninstall_cleans_legacy_format(tmp_path, monkeypatch): '[general]\ntheme = "dark"\n\n' '[[hooks]]\n' 'event = "SessionStart"\n' - 'command = "/usr/bin/python3 /path/to/truememory/session_start.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py"\n' 'timeout = 10000\n', encoding="utf-8", ) @@ -380,7 +380,7 @@ def test_remove_hook_blocks_correct_format(): '[[hooks.SessionStart]]\n\n' '[[hooks.SessionStart.hooks]]\n' 'type = "command"\n' - 'command = "/usr/bin/python3 /path/to/truememory/session_start.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py"\n' 'timeout = 10\n\n' '[[hooks.MyCustomHook]]\n\n' '[[hooks.MyCustomHook.hooks]]\n' @@ -403,7 +403,7 @@ def test_remove_legacy_hook_blocks_with_blank_lines(): '[[hooks]]\n' 'event = "SessionStart"\n' '\n' - 'command = "/usr/bin/python3 /path/to/truememory/session_start.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py"\n' 'timeout = 10000\n\n' '[[hooks]]\n' 'event = "MyCustomHook"\n' @@ -432,12 +432,12 @@ def test_parse_existing_hooks_correct_format(): '[[hooks.SessionStart]]\n\n' '[[hooks.SessionStart.hooks]]\n' 'type = "command"\n' - 'command = "/usr/bin/python3 /path/to/truememory/session_start.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py"\n' 'timeout = 10\n\n' '[[hooks.Stop]]\n\n' '[[hooks.Stop.hooks]]\n' 'type = "command"\n' - 'command = "/usr/bin/python3 /path/to/truememory/stop.py"\n' + 'command = "/usr/bin/python3 /path/to/truememory/ingest/hooks/stop.py"\n' 'timeout = 5\n' ) result = CodexAdapter._parse_existing_hooks(text) @@ -446,7 +446,7 @@ def test_parse_existing_hooks_correct_format(): assert "Stop" in result mg = result["SessionStart"][0] assert "hooks" in mg - assert mg["hooks"][0]["command"] == "/usr/bin/python3 /path/to/truememory/session_start.py" + assert mg["hooks"][0]["command"] == "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py" def test_parse_existing_hooks_empty(): @@ -465,7 +465,7 @@ def test_event_already_registered_correct_format(): "hooks": [ { "type": "command", - "command": "/usr/bin/python3 /path/to/truememory/session_start.py", + "command": "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py", "timeout": 10, } ] diff --git a/tests/test_gemini_adapter.py b/tests/test_gemini_adapter.py index b783b073..1bc1c3c0 100644 --- a/tests/test_gemini_adapter.py +++ b/tests/test_gemini_adapter.py @@ -252,7 +252,7 @@ def test_uninstall_preserves_other_entries(tmp_path, monkeypatch): "hooks": { "SessionStart": [ {"command": "my-hook", "timeout": 5000}, - {"command": "/path/to/truememory/hooks/session_start.py", "timeout": 10000}, + {"command": "/path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000}, ] }, "theme": "dark", @@ -278,7 +278,7 @@ def test_uninstall_removes_nested_format(tmp_path, monkeypatch): "hooks": { "SessionStart": [ {"hooks": [{"type": "command", "command": "my-hook", "timeout": 5000}]}, - {"hooks": [{"type": "command", "command": "/path/to/truememory/session_start.py", "timeout": 10000, "name": "truememory-sessionstart"}]}, + {"hooks": [{"type": "command", "command": "/path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000, "name": "truememory-sessionstart"}]}, ] }, }), encoding="utf-8") diff --git a/truememory/hooks/adapters/base.py b/truememory/hooks/adapters/base.py index fd8d3f71..964d1935 100644 --- a/truememory/hooks/adapters/base.py +++ b/truememory/hooks/adapters/base.py @@ -8,10 +8,53 @@ import json import os +import re +import shlex +import sys from abc import ABC, abstractmethod from pathlib import Path +def is_owned_hook_command(command: object) -> bool: + """Recognize a simple Python invocation of an installed TrueMemory hook.""" + if not isinstance(command, str) or not command.strip() or any(char in command for char in ("\0", "\n", "\r")): + return False + try: + lexer = shlex.shlex(command, posix=False, punctuation_chars=";&|<>()`") + lexer.whitespace_split = True + for token in lexer: + if token and all(char in ";&|<>()`" for char in token): + return False + if not token.startswith("'") and any(part in token for part in ("$(", "${", "`")): + return False + tokens = shlex.split(command, comments=True, posix=sys.platform != "win32") + except ValueError: + return False + if sys.platform == "win32": + tokens = [token[1:-1] if token.startswith('"') and token.endswith('"') else token for token in tokens] + if len(tokens) < 2: + return False + executable = tokens[0].replace("\\", "/").rsplit("/", 1)[-1] + if re.fullmatch(r"python(?:\d+(?:\.\d+)*)?(?:\.exe)?", executable, re.IGNORECASE) is None: + return False + index = 1 + while index < len(tokens) and tokens[index] in ("-B", "-u", "-E", "-I", "-s", "-S", "-O", "-OO"): + index += 1 + if index >= len(tokens): + return False + stems = ("session_start", "user_prompt_submit", "stop", "compact") + if tokens[index] == "-m": + return index + 1 < len(tokens) and tokens[index + 1] in ( + f"truememory.ingest.hooks.{stem}" for stem in stems + ) + script = tokens[index].replace("\\", "/").split("/") + return ( + len(script) >= 4 + and script[-4:-1] == ["truememory", "ingest", "hooks"] + and script[-1] in (f"{stem}.py" for stem in stems) + ) + + def atomic_write_text(path: Path, text: str) -> None: """Write *text* to *path* atomically (tmp-in-same-dir + os.replace). diff --git a/truememory/hooks/adapters/codex.py b/truememory/hooks/adapters/codex.py index 2d2ad939..a8813b60 100644 --- a/truememory/hooks/adapters/codex.py +++ b/truememory/hooks/adapters/codex.py @@ -10,13 +10,14 @@ from __future__ import annotations import logging +import json import re import shlex import shutil import sys from pathlib import Path -from truememory.hooks.adapters.base import CLIAdapter, atomic_write_text +from truememory.hooks.adapters.base import CLIAdapter, atomic_write_text, is_owned_hook_command log = logging.getLogger(__name__) @@ -253,7 +254,7 @@ def _refresh_hook_commands( if ( event in _HOOK_EVENTS and re.match(r'^\s*command\s*=', line) - and _TRUEMEMORY_MARKER in line.lower() + and is_owned_hook_command(self._hook_command_value(line)) ): cmd = self._build_command( python_path, hooks_dir / _HOOK_EVENTS[event]["script"], @@ -265,6 +266,25 @@ def _refresh_hook_commands( lines.append(line) return "".join(lines) + @staticmethod + def _hook_command_value(line: str) -> str | None: + """Decode the generated one-line TOML string without reading comments.""" + match = re.match(r'^\s*command\s*=\s*(.*)$', line) + if match is None: + return None + value = match.group(1).strip() + if value.startswith("'"): + literal = re.fullmatch(r"'([^']*)'\s*(?:#.*)?", value) + return literal.group(1) if literal else None + try: + command, end = json.JSONDecoder().raw_decode(value) + except (json.JSONDecodeError, ValueError): + return None + tail = value[end:].strip() + if isinstance(command, str) and (not tail or tail.startswith("#")): + return command + return None + @staticmethod def _parse_existing_hooks(toml_text: str) -> dict: """Parse hooks from TOML text, returning ``{event: [matcher_groups]}``.""" @@ -346,7 +366,7 @@ def _event_already_registered(hooks: dict, event: str) -> bool: for hook_entry in mg.get("hooks", []): handler = hook_entry if isinstance(hook_entry, dict) else {} cmd = handler.get("command", "") - if _TRUEMEMORY_MARKER in cmd.lower(): + if is_owned_hook_command(cmd): return True return False @@ -371,7 +391,7 @@ def _has_hook_entries_in_text(self, text: str) -> bool: for mg in matcher_groups: for hook_entry in mg.get("hooks", []): handler = hook_entry if isinstance(hook_entry, dict) else {} - if _TRUEMEMORY_MARKER in handler.get("command", "").lower(): + if is_owned_hook_command(handler.get("command", "")): return True return False @@ -441,8 +461,7 @@ def _remove_hook_blocks(text: str) -> str: block_lines.append(lines[i]) i += 1 - block_text = "".join(block_lines) - if _TRUEMEMORY_MARKER not in block_text.lower(): + if not any(is_owned_hook_command(CodexAdapter._hook_command_value(line)) for line in block_lines): cleaned.extend(block_lines) continue @@ -472,8 +491,7 @@ def _remove_legacy_hook_blocks(text: str) -> str: block_lines.append(lines[i]) i += 1 - block_text = "".join(block_lines) - if _TRUEMEMORY_MARKER not in block_text.lower(): + if not any(is_owned_hook_command(CodexAdapter._hook_command_value(line)) for line in block_lines): cleaned.extend(block_lines) continue diff --git a/truememory/hooks/adapters/gemini.py b/truememory/hooks/adapters/gemini.py index 7572f469..52c7af68 100644 --- a/truememory/hooks/adapters/gemini.py +++ b/truememory/hooks/adapters/gemini.py @@ -18,7 +18,7 @@ import sys from pathlib import Path -from truememory.hooks.adapters.base import CLIAdapter, atomic_write_text +from truememory.hooks.adapters.base import CLIAdapter, atomic_write_text, is_owned_hook_command _GEMINI_DIR = Path.home() / ".gemini" _CONFIG_PATH = _GEMINI_DIR / "settings.json" @@ -158,11 +158,11 @@ def install_hooks( if ( isinstance(hook, dict) and isinstance(hook.get("command"), str) - and _TRUEMEMORY_MARKER in hook["command"].lower() + and is_owned_hook_command(hook["command"]) ): hook["command"] = cmd legacy_command = definition.get("command", "") - if isinstance(legacy_command, str) and _TRUEMEMORY_MARKER in legacy_command.lower(): + if is_owned_hook_command(legacy_command): definition.pop("command", None) definition.pop("timeout", None) definition["hooks"] = [{ @@ -299,11 +299,11 @@ def _definition_has_truememory(hook_def: object) -> bool: for hc in inner_hooks: if ( isinstance(hc, dict) - and _TRUEMEMORY_MARKER in hc.get("command", "").lower() + and is_owned_hook_command(hc.get("command", "")) ): return True # Legacy flat format fallback - if _TRUEMEMORY_MARKER in hook_def.get("command", "").lower(): + if is_owned_hook_command(hook_def.get("command", "")): return True return False From 1d0b6b17c3205e9d6169936a42e47f434afee63d Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sat, 10 Oct 2026 23:34:02 -0500 Subject: [PATCH 3/7] fix: Preserve native Cursor identity across extraction hooks --- tests/test-transcript-allowlists-761.py | 34 +++++++++++++++++++++++++ truememory/ingest/hooks/_shared.py | 13 ++++++++++ truememory/ingest/hooks/compact.py | 4 +-- truememory/ingest/hooks/stop.py | 8 +++--- 4 files changed, 53 insertions(+), 6 deletions(-) diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py index 53757994..fafe6f64 100644 --- a/tests/test-transcript-allowlists-761.py +++ b/tests/test-transcript-allowlists-761.py @@ -510,6 +510,40 @@ def test_owned_module_invocations_are_upgraded_without_duplication(self) -> None self.assertEqual(len(groups), 1) self.assertIn("--host gemini --event SessionStart", groups[0]["hooks"][0]["command"]) + def test_cursor_extraction_uses_native_conversation_identity(self) -> None: + os.environ["TRUEMEMORY_TRANSCRIPT_DIR"] = str(self.home / "synthetic-transcripts") + path = self.write(self.home / "synthetic-transcripts/session.jsonl", "synthetic text " * 30) + for module, argv in ((self.stop, ["stop.py", "--host", "cursor", "--event", "stop"]), + (self.compact, ["compact.py", "--host", "cursor", "--event", "preCompact"])): + for conversation in ("synthetic-cursor-a", "synthetic-cursor-b"): + with self.subTest(module=module.__name__, conversation=conversation), \ + patch.object(self.compact, "save_snapshot") as snapshot, \ + patch.object(self.stop, "_run_background_ingestion", return_value=17) as spawn, \ + patch.object(sys, "argv", argv), \ + patch.object(sys, "stdin", io.StringIO(json.dumps({ + "transcript_path": str(path), "conversation_id": conversation, + }))): + module.main() + self.assertEqual(spawn.call_args.args[1], conversation) + if module is self.compact: + self.assertEqual(snapshot.call_args.args[1], conversation) + + def test_hook_identity_is_host_scoped_and_preserves_each_caller_default(self) -> None: + payload = {"conversation_id": "synthetic-native", "session_id": "synthetic-legacy", "host": "cursor"} + self.assertEqual(self.shared.hook_session_id(payload, "cursor"), "synthetic-native") + for host in ("claude", "codex", "gemini"): + self.assertEqual(self.shared.hook_session_id(payload, host), "synthetic-legacy") + self.assertEqual(self.shared.hook_session_id({"conversation_id": "synthetic-native"}, host), "unknown") + for malformed in (None, {}, [], 42, "", " ", "bad\0id"): + with self.subTest(malformed=malformed): + payload = {"conversation_id": malformed, "session_id": "synthetic-legacy"} + self.assertEqual(self.shared.hook_session_id(payload, "cursor"), "synthetic-legacy") + self.assertEqual(self.shared.hook_session_id({"conversation_id": malformed}, "cursor"), "unknown") + self.assertEqual(self.shared.hook_session_id({"conversation_id": malformed}, "cursor", default=""), "") + for malformed in (None, [], 42, "text"): + self.assertEqual(self.shared.hook_session_id(malformed, "cursor"), "unknown") + self.assertEqual(self.shared.hook_session_id(payload, "invalid", default=""), "") + if __name__ == "__main__": unittest.main() diff --git a/truememory/ingest/hooks/_shared.py b/truememory/ingest/hooks/_shared.py index 81abe8fb..abf4420b 100644 --- a/truememory/ingest/hooks/_shared.py +++ b/truememory/ingest/hooks/_shared.py @@ -38,6 +38,19 @@ def normalize_transcript_host(host: object) -> str: return "" +def hook_session_id(payload: object, host: str = "claude", *, default: str = "unknown") -> str: + """Read native identity using the trusted host, retaining caller fallback.""" + host = normalize_transcript_host(host) + if not host or not isinstance(payload, dict): + return default + keys = ("conversation_id", "session_id") if host == "cursor" else ("session_id",) + for key in keys: + value = payload.get(key) + if isinstance(value, str) and value.strip() and "\0" not in value: + return value.strip() + return default + + def transcript_input_path(value: object) -> str: """Reject malformed source paths before any filesystem probe.""" if not isinstance(value, str) or not value.strip() or "\0" in value: diff --git a/truememory/ingest/hooks/compact.py b/truememory/ingest/hooks/compact.py index 2edbe494..df274c65 100755 --- a/truememory/ingest/hooks/compact.py +++ b/truememory/ingest/hooks/compact.py @@ -73,9 +73,9 @@ def main() -> None: if not isinstance(input_data, dict): return - from truememory.ingest.hooks._shared import transcript_input_path + from truememory.ingest.hooks._shared import transcript_input_path, hook_session_id transcript_path = transcript_input_path(input_data.get("transcript_path", "")) - session_id = str(input_data.get("session_id", "unknown") or "unknown") + session_id = hook_session_id(input_data, host=args.host) # Sanitize session_id to prevent injection (consistent with user_prompt_submit.py) safe_id = "".join(c for c in session_id if c.isalnum() or c in "-_")[:64] diff --git a/truememory/ingest/hooks/stop.py b/truememory/ingest/hooks/stop.py index b85fec7d..3ebbad3c 100755 --- a/truememory/ingest/hooks/stop.py +++ b/truememory/ingest/hooks/stop.py @@ -126,10 +126,10 @@ def main() -> None: if not isinstance(input_data, dict): return from truememory.ingest.hooks._shared import ( - is_allowed_transcript, mark_session_extracted, transcript_input_path, + is_allowed_transcript, mark_session_extracted, transcript_input_path, hook_session_id, ) transcript_path = transcript_input_path(input_data.get("transcript_path", "")) - session_id = _sanitize_session_id(input_data.get("session_id", "")) + session_id = _sanitize_session_id(hook_session_id(input_data, host=args.host)) if is_allowed_transcript(transcript_path, host=args.host) and session_id != "unknown": try: mark_session_extracted(session_id, transcript_path) @@ -144,9 +144,9 @@ def main() -> None: if not isinstance(input_data, dict): return - from truememory.ingest.hooks._shared import is_allowed_transcript, transcript_input_path + from truememory.ingest.hooks._shared import is_allowed_transcript, transcript_input_path, hook_session_id transcript_path = transcript_input_path(input_data.get("transcript_path", "")) - session_id = _sanitize_session_id(input_data.get("session_id", "unknown")) + session_id = _sanitize_session_id(hook_session_id(input_data, host=args.host)) if not transcript_path: return From 396b4d8eadf91a3c096f512c770325709c7da922 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sat, 10 Oct 2026 23:40:12 -0500 Subject: [PATCH 4/7] fix: Retain installed host identity in recall hook foundations --- truememory/ingest/hooks/session_start.py | 4 ++- truememory/ingest/hooks/user_prompt_submit.py | 25 ++++++++++++------- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/truememory/ingest/hooks/session_start.py b/truememory/ingest/hooks/session_start.py index cfc39c6f..81a20431 100755 --- a/truememory/ingest/hooks/session_start.py +++ b/truememory/ingest/hooks/session_start.py @@ -170,6 +170,8 @@ def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(add_help=False) p.add_argument("--user", default=os.environ.get("TRUEMEMORY_USER_ID", "")) p.add_argument("--db", default=os.environ.get("TRUEMEMORY_DB_PATH", "")) + p.add_argument("--host", choices=("claude", "codex", "gemini", "cursor"), default="claude") + p.add_argument("--event", default="SessionStart") p.add_argument("--scan-stale", action="store_true", help=argparse.SUPPRESS) args, _ = p.parse_known_args() return args @@ -733,7 +735,7 @@ def main(): _scan_stale_sessions() return - _run_maintenance_background() + _run_maintenance_background(args.host) try: input_data = json.load(sys.stdin) diff --git a/truememory/ingest/hooks/user_prompt_submit.py b/truememory/ingest/hooks/user_prompt_submit.py index 93003639..9a22ee9c 100755 --- a/truememory/ingest/hooks/user_prompt_submit.py +++ b/truememory/ingest/hooks/user_prompt_submit.py @@ -83,15 +83,14 @@ def _word_overlap(a: str, b: str) -> float: def _parse_args() -> argparse.Namespace: """Parse command-line overrides the installer threads through. - UserPromptSubmit doesn't actually use ``--user`` or ``--db`` — it only - writes a per-session diagnostic buffer — but the installer passes the - same flags to every hook for consistency, so we must accept them here - without erroring out. ``parse_known_args`` ensures forward compat with - future flags. + The trusted host selects transcript admission; event retains installer context. + ``parse_known_args`` keeps the existing forward-compatible flags. """ p = argparse.ArgumentParser(add_help=False) p.add_argument("--user", default=os.environ.get("TRUEMEMORY_USER_ID", "")) p.add_argument("--db", default=os.environ.get("TRUEMEMORY_DB_PATH", "")) + p.add_argument("--host", choices=("claude", "codex", "gemini", "cursor"), default="claude") + p.add_argument("--event", default="UserPromptSubmit") args, _ = p.parse_known_args() return args @@ -737,7 +736,8 @@ def main(): return prompt = input_data.get("prompt", "").strip() - session_id = input_data.get("session_id", "unknown") + from truememory.ingest.hooks._shared import hook_session_id + session_id = hook_session_id(input_data, args.host) if not prompt or len(prompt) < 3: # A too-short first prompt is still the session's first prompt: @@ -765,10 +765,16 @@ def main(): prompt_count = _increment_prompt_count(session_id) transcript_path = input_data.get("transcript_path", "") - if transcript_path and Path(transcript_path).exists(): + if transcript_path: try: - from truememory.ingest.hooks._shared import should_extract_session, mark_session_extracted - if should_extract_session(session_id, transcript_path): + from truememory.ingest.hooks._shared import ( + is_allowed_transcript, should_extract_session, mark_session_extracted, + ) + if ( + is_allowed_transcript(transcript_path, host=args.host) + and Path(transcript_path).is_file() + and should_extract_session(session_id, transcript_path) + ): from truememory.ingest.hooks.stop import ( _has_enough_messages, _run_background_ingestion, TRACE_DIR, LOG_DIR, @@ -778,6 +784,7 @@ def main(): if _has_enough_messages(transcript_path, 10): spawned_pid = _run_background_ingestion( transcript_path, session_id, args.user, args.db, + transcript_host=args.host, ) # Only mark extracted on a real spawn; pid==0 means the # session was queued to the backlog and must stay eligible From b5c652965417898f33d7c38da8f05b4ba17a81bd Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sat, 10 Oct 2026 23:51:57 -0500 Subject: [PATCH 5/7] fix: Bind faithful native transcript and adapter fixtures --- .../ingest/test_issue_558_async_stale_scan.py | 2 +- tests/test-transcript-allowlists-761.py | 59 +++++++++++++++++-- tests/test_gemini_adapter.py | 4 +- 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/tests/ingest/test_issue_558_async_stale_scan.py b/tests/ingest/test_issue_558_async_stale_scan.py index 019dd859..dbcda497 100644 --- a/tests/ingest/test_issue_558_async_stale_scan.py +++ b/tests/ingest/test_issue_558_async_stale_scan.py @@ -70,7 +70,7 @@ def test_main_calls_maintenance_not_scan(self, monkeypatch): _scan_stale_sessions (synchronous).""" calls = {"maintenance": 0, "scan": 0} - monkeypatch.setattr(ss, "_run_maintenance_background", lambda: calls.__setitem__("maintenance", calls["maintenance"] + 1)) + monkeypatch.setattr(ss, "_run_maintenance_background", lambda host="claude": calls.__setitem__("maintenance", calls["maintenance"] + 1)) monkeypatch.setattr(ss, "_scan_stale_sessions", lambda: calls.__setitem__("scan", calls["scan"] + 1)) monkeypatch.setattr(ss, "_drain_backlog", lambda: None) monkeypatch.setattr(ss, "_is_first_run", lambda: True) diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py index fafe6f64..f1f9d7c3 100644 --- a/tests/test-transcript-allowlists-761.py +++ b/tests/test-transcript-allowlists-761.py @@ -19,6 +19,7 @@ import sys import tempfile import types +from typing import Iterator import unittest from unittest.mock import patch @@ -46,6 +47,21 @@ def source_module(name: str, relative_path: str, names: set[str] | None = None) return module +@contextmanager +def scoped_module_bindings(packages: dict[str, types.ModuleType], owned: set[str]) -> Iterator[None]: + missing = object() + saved = {name: sys.modules.get(name, missing) for name in owned} + sys.modules.update(packages) + try: + yield + finally: + for name, previous in saved.items(): + if previous is missing: + sys.modules.pop(name, None) + else: + sys.modules[name] = previous + + @contextmanager def allowed_gate(): yield True @@ -86,9 +102,16 @@ def setUp(self) -> None: core.register_spawned_pid = lambda pid: None core._load_cap_state = lambda: {"cap": 2} packages["truememory.hooks.core"] = core - self.module_patch = patch.dict(sys.modules, packages) - self.module_patch.start() - self.addCleanup(self.module_patch.stop) + owned = set(packages) | { + "truememory.hooks.adapters.base", "truememory.hooks.adapters.codex", + "truememory.hooks.adapters.gemini", "truememory.ingest.hooks._shared", + "truememory.ingest.hooks.stop", "truememory.ingest.hooks.compact", + "truememory.ingest.hooks.session_start", "truememory.ingest.cli", + "truememory.ingest.transcript", + } + self.module_bindings = scoped_module_bindings(packages, owned) + self.module_bindings.__enter__() + self.addCleanup(self.module_bindings.__exit__, None, None, None) source_module("truememory.hooks.adapters.base", "truememory/hooks/adapters/base.py") self.shared = source_module( "truememory.ingest.hooks._shared", "truememory/ingest/hooks/_shared.py", @@ -111,7 +134,7 @@ def setUp(self) -> None: self.session_start._DRAIN_CAP = 3 self.session_start._platform = platform self.cli = source_module("truememory.ingest.cli", "truememory/ingest/cli.py", {"_cascade_next"}) - self.parser = source_module("synthetic_transcript_parser_761", "truememory/ingest/transcript.py") + self.parser = source_module("truememory.ingest.transcript", "truememory/ingest/transcript.py") def write(self, path: Path, text: str = "synthetic conversation") -> Path: path.parent.mkdir(parents=True, exist_ok=True) @@ -278,7 +301,9 @@ def test_stop_malformed_input_never_probes_source(self) -> None: self.stop.main() def test_compact_correct_host_snapshot_and_background_admission(self) -> None: - path = self.write(self.native_path("gemini"), "synthetic text " * 30) + path = self.write(self.native_path("gemini"), json.dumps([ + {"role": "user", "content": "synthetic turn"} for _ in range(5) + ])) with patch.object(self.compact, "save_snapshot") as snapshot, \ patch.object(self.stop, "_run_background_ingestion", return_value=17) as spawn, \ patch.object(sys, "argv", ["compact.py", "--host", "gemini", "--event", "PreCompress"]), \ @@ -512,7 +537,9 @@ def test_owned_module_invocations_are_upgraded_without_duplication(self) -> None def test_cursor_extraction_uses_native_conversation_identity(self) -> None: os.environ["TRUEMEMORY_TRANSCRIPT_DIR"] = str(self.home / "synthetic-transcripts") - path = self.write(self.home / "synthetic-transcripts/session.jsonl", "synthetic text " * 30) + path = self.write(self.home / "synthetic-transcripts/session.jsonl", json.dumps([ + {"role": "user", "content": "synthetic turn"} for _ in range(5) + ])) for module, argv in ((self.stop, ["stop.py", "--host", "cursor", "--event", "stop"]), (self.compact, ["compact.py", "--host", "cursor", "--event", "preCompact"])): for conversation in ("synthetic-cursor-a", "synthetic-cursor-b"): @@ -544,6 +571,26 @@ def test_hook_identity_is_host_scoped_and_preserves_each_caller_default(self) -> self.assertEqual(self.shared.hook_session_id(malformed, "cursor"), "unknown") self.assertEqual(self.shared.hook_session_id(payload, "invalid", default=""), "") + def test_module_binding_scope_retains_unrelated_imports_and_restores_owned_keys(self) -> None: + owned = "synthetic_owned_module_761" + unrelated = "synthetic_later_import_761" + previous = types.ModuleType(owned) + replacement = types.ModuleType(owned) + imported = types.ModuleType(unrelated) + with scoped_module_bindings({owned: previous}, {owned, unrelated}): + with scoped_module_bindings({owned: replacement}, {owned}): + sys.modules[unrelated] = imported + self.assertIs(sys.modules[owned], replacement) + self.assertIs(sys.modules[owned], previous) + self.assertIs(sys.modules[unrelated], imported) + + def test_spawn_fixtures_require_five_real_user_turns(self) -> None: + path = self.native_path("gemini") + self.write(path, json.dumps([{"role": "user", "content": "synthetic turn"}])) + self.assertFalse(self.stop._has_enough_messages(str(path), 5)) + self.write(path, json.dumps([{"role": "user", "content": "synthetic turn"} for _ in range(5)])) + self.assertTrue(self.stop._has_enough_messages(str(path), 5)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gemini_adapter.py b/tests/test_gemini_adapter.py index 1bc1c3c0..837399c3 100644 --- a/tests/test_gemini_adapter.py +++ b/tests/test_gemini_adapter.py @@ -252,7 +252,7 @@ def test_uninstall_preserves_other_entries(tmp_path, monkeypatch): "hooks": { "SessionStart": [ {"command": "my-hook", "timeout": 5000}, - {"command": "/path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000}, + {"command": "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000}, ] }, "theme": "dark", @@ -278,7 +278,7 @@ def test_uninstall_removes_nested_format(tmp_path, monkeypatch): "hooks": { "SessionStart": [ {"hooks": [{"type": "command", "command": "my-hook", "timeout": 5000}]}, - {"hooks": [{"type": "command", "command": "/path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000, "name": "truememory-sessionstart"}]}, + {"hooks": [{"type": "command", "command": "/usr/bin/python3 /path/to/truememory/ingest/hooks/session_start.py", "timeout": 10000, "name": "truememory-sessionstart"}]}, ] }, }), encoding="utf-8") From b53c35d6b8bcdd705c5893d672976a65eb35cd0b Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sun, 11 Oct 2026 01:45:54 -0500 Subject: [PATCH 6/7] fix: Use platform-aware quoted hook fixtures --- tests/test-transcript-allowlists-761.py | 42 +++++++++++++++---------- 1 file changed, 26 insertions(+), 16 deletions(-) diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py index f1f9d7c3..1b2ba193 100644 --- a/tests/test-transcript-allowlists-761.py +++ b/tests/test-transcript-allowlists-761.py @@ -491,22 +491,32 @@ def test_gemini_foreign_mentions_are_preserved_and_do_not_suppress_registration( def test_owned_hook_predicate_requires_actual_python_invocation(self) -> None: predicate = sys.modules["truememory.hooks.adapters.base"].is_owned_hook_command - for stem in ("session_start", "user_prompt_submit", "stop", "compact"): - for command in ( - f"python /pkg/truememory/ingest/hooks/{stem}.py", - f"'/pkg with space/python3.10' '/pkg with space/truememory/ingest/hooks/{stem}.py' --user ';'", - f"python -B -m truememory.ingest.hooks.{stem}", - ): - with self.subTest(command=command): - self.assertTrue(predicate(command)) - for command in (None, [], "", "python -c 'truememory.ingest.hooks.stop'", - "python -m foreign --target truememory.ingest.hooks.stop", - "foreign /pkg/truememory/ingest/hooks/stop.py", "python-policy /pkg/truememory/ingest/hooks/stop.py", - "python /pkg/truememory/ingest/hooks/stop.py; foreign", "python /pkg/truememory/ingest/hooks/stop.py\nforeign", - "python /pkg/truememory/ingest/hooks/stop.py > output", "python '/pkg/truememory/ingest/hooks/stop.py", - "python \"$(foreign)/truememory/ingest/hooks/stop.py\"", "python /pkg/truememory/ingest/hooks/not-a-hook.py"): - with self.subTest(command=command): - self.assertFalse(predicate(command)) + for platform in ("linux", "win32"): + with patch.object(sys, "platform", platform): + for stem in ("session_start", "user_prompt_submit", "stop", "compact"): + quoted_command = ( + subprocess.list2cmdline([ + r"C:\Program Files\Python\python.exe", + rf"C:\Synthetic Package\truememory\ingest\hooks\{stem}.py", + "--user", "synthetic user", + ]) if platform == "win32" else + f"'/pkg with space/python3.10' '/pkg with space/truememory/ingest/hooks/{stem}.py' --user ';'" + ) + for command in ( + f"python /pkg/truememory/ingest/hooks/{stem}.py", + quoted_command, + f"python -B -m truememory.ingest.hooks.{stem}", + ): + with self.subTest(platform=platform, command=command): + self.assertTrue(predicate(command)) + for command in (None, [], "", "python -c 'truememory.ingest.hooks.stop'", + "python -m foreign --target truememory.ingest.hooks.stop", + "foreign /pkg/truememory/ingest/hooks/stop.py", "python-policy /pkg/truememory/ingest/hooks/stop.py", + "python /pkg/truememory/ingest/hooks/stop.py; foreign", "python /pkg/truememory/ingest/hooks/stop.py\nforeign", + "python /pkg/truememory/ingest/hooks/stop.py > output", "python '/pkg/truememory/ingest/hooks/stop.py", + "python \"$(foreign)/truememory/ingest/hooks/stop.py\"", "python /pkg/truememory/ingest/hooks/not-a-hook.py"): + with self.subTest(platform=platform, command=command): + self.assertFalse(predicate(command)) with patch.object(sys, "platform", "win32"): command = subprocess.list2cmdline([ r"C:\Program Files\Python\python.exe", From 003705e9e3f4cd38f7a3abf5c79607af2ba45e7e Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Sun, 11 Oct 2026 01:49:25 -0500 Subject: [PATCH 7/7] fix: Cover quoted Windows literal semicolons --- tests/test-transcript-allowlists-761.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/test-transcript-allowlists-761.py b/tests/test-transcript-allowlists-761.py index 1b2ba193..c3221888 100644 --- a/tests/test-transcript-allowlists-761.py +++ b/tests/test-transcript-allowlists-761.py @@ -498,8 +498,7 @@ def test_owned_hook_predicate_requires_actual_python_invocation(self) -> None: subprocess.list2cmdline([ r"C:\Program Files\Python\python.exe", rf"C:\Synthetic Package\truememory\ingest\hooks\{stem}.py", - "--user", "synthetic user", - ]) if platform == "win32" else + ]) + ' --user ";"' if platform == "win32" else f"'/pkg with space/python3.10' '/pkg with space/truememory/ingest/hooks/{stem}.py' --user ';'" ) for command in (