fix: harden learn path handling across platforms

This commit is contained in:
Tejas Chopra 2026-05-09 15:45:26 -07:00
parent 4a88c698ad
commit 5ceca13c65
15 changed files with 203 additions and 32 deletions

View file

@ -226,7 +226,13 @@ def learn(
total_recommendations += len(recommendations)
click.echo(f" Recommendations: {len(recommendations)}")
result = writer.write(recommendations, proj, dry_run=not apply)
try:
result = writer.write(recommendations, proj, dry_run=not apply)
except OSError as e:
click.echo(
f" Warning: failed to write recommendations for {proj.project_path}: {e}"
)
continue
for file_path, content in result.content_by_file.items():
click.echo(f"\n {'[WOULD WRITE]' if result.dry_run else '[WROTE]'} {file_path}")

View file

@ -8,7 +8,7 @@ from __future__ import annotations
import json
import logging
import re
from pathlib import Path
from pathlib import Path, PureWindowsPath
from .._shared import classify_error, is_error_content
from ..base import ConversationScanner, LearnPlugin
@ -77,7 +77,7 @@ class ClaudeCodePlugin(LearnPlugin, ConversationScanner):
else:
project_path = Path("/" + entry.name[1:].replace("-", "/"))
name = project_path.name if project_path != Path("/") else entry.name
name = _project_display_name(project_path, entry.name)
context_file = None
if project_path.exists():
@ -323,6 +323,8 @@ def _decode_project_path(escaped_name: str) -> Path | None:
result = _greedy_path_decode(win_base, parts[2:])
if result:
return result
if len(parts) > 1 and parts[1].lower() == "users":
return win_path
simple = Path("/" + escaped_name[1:].replace("-", "/"))
if simple.exists():
@ -344,6 +346,16 @@ def _decode_project_path(escaped_name: str) -> Path | None:
return None
def _project_display_name(project_path: Path, fallback: str) -> str:
"""Return a human project name for POSIX and Windows-style decoded paths."""
rendered = str(project_path)
if re.match(r"^[A-Za-z]:[\\/]", rendered):
return PureWindowsPath(rendered).name or fallback
if project_path == Path("/"):
return fallback
return project_path.name or fallback
def _greedy_path_decode(base: Path, parts: list[str]) -> Path | None:
"""Greedily decode remaining path parts using real child directories."""
if not parts:

View file

@ -153,7 +153,7 @@ def _merge_recommendations(
"""
if not file_path.exists():
return new_recommendations
prior = _parse_prior_recommendations(file_path.read_text())
prior = _parse_prior_recommendations(file_path.read_text(encoding="utf-8"))
if not prior:
return new_recommendations
new_sections = {r.section for r in new_recommendations}
@ -166,9 +166,9 @@ def _merge_into_file(file_path: Path, new_recommendations: list[Recommendation])
merged = _merge_recommendations(file_path, new_recommendations)
section = _build_section(merged)
if file_path.exists():
existing = file_path.read_text()
existing = file_path.read_text(encoding="utf-8")
if _MARKER_START in existing:
return _MARKER_PATTERN.sub(section, existing)
return _MARKER_PATTERN.sub(lambda _match: section, existing)
return existing.rstrip() + "\n\n" + section + "\n"
return section + "\n"
@ -199,7 +199,7 @@ class ClaudeCodeWriter(ContextWriter):
result.add(claude_md_path, full_content)
if not dry_run:
claude_md_path.parent.mkdir(parents=True, exist_ok=True)
claude_md_path.write_text(full_content)
claude_md_path.write_text(full_content, encoding="utf-8")
if memory_recs:
memory_path = self._resolve_memory_path(project)
@ -207,7 +207,7 @@ class ClaudeCodeWriter(ContextWriter):
result.add(memory_path, full_content)
if not dry_run:
memory_path.parent.mkdir(parents=True, exist_ok=True)
memory_path.write_text(full_content)
memory_path.write_text(full_content, encoding="utf-8")
return result
@ -252,7 +252,7 @@ class CodexWriter(ContextWriter):
result.add(agents_md, full_content)
if not dry_run:
agents_md.parent.mkdir(parents=True, exist_ok=True)
agents_md.write_text(full_content)
agents_md.write_text(full_content, encoding="utf-8")
if memory_recs:
instructions_md = project.memory_file or (project.data_path.parent / "instructions.md")
@ -260,7 +260,7 @@ class CodexWriter(ContextWriter):
result.add(instructions_md, full_content)
if not dry_run:
instructions_md.parent.mkdir(parents=True, exist_ok=True)
instructions_md.write_text(full_content)
instructions_md.write_text(full_content, encoding="utf-8")
return result
@ -290,6 +290,6 @@ class GeminiWriter(ContextWriter):
result.add(gemini_md, full_content)
if not dry_run:
gemini_md.parent.mkdir(parents=True, exist_ok=True)
gemini_md.write_text(full_content)
gemini_md.write_text(full_content, encoding="utf-8")
return result

View file

@ -35,6 +35,21 @@ def _sanitize_for_filename(text: str) -> str:
return slug or "memory"
def encode_claude_project_path(project_path: Path | str) -> str:
"""Encode a project path the way Claude Code names project directories.
POSIX absolute paths naturally become ``-Users-me-repo``. Windows drive
paths should become ``-C-Users-me-repo`` rather than ``C:-Users-me-repo``.
"""
rendered = str(project_path)
drive_match = re.match(r"^([A-Za-z]):[\\/](.*)$", rendered)
if drive_match:
drive, rest = drive_match.groups()
rest = rest.replace("\\", "-").replace("/", "-")
return f"-{drive.upper()}-{rest}" if rest else f"-{drive.upper()}"
return rendered.replace("/", "-").replace("\\", "-")
def _parse_frontmatter(content: str) -> tuple[dict[str, str], str]:
"""Parse YAML frontmatter from a markdown file.
@ -232,6 +247,5 @@ def get_claude_memory_dir(project_path: Path | None = None) -> Path:
~/.claude/projects/-<sanitized-path>/memory/
"""
project = project_path or Path.cwd()
# Replace both Unix and Windows path separators (Claude Code does the same)
sanitized = str(project).replace("/", "-").replace("\\", "-")
sanitized = encode_claude_project_path(project)
return Path.home() / ".claude" / "projects" / sanitized / "memory"

View file

@ -85,7 +85,7 @@ class CodexAdapter(AgentMemoryAdapter):
if self._path.exists():
content = self._path.read_text(encoding="utf-8")
if _MARKER_START in content:
content = _MARKER_PATTERN.sub(section, content)
content = _MARKER_PATTERN.sub(lambda _match: section, content)
else:
content = content.rstrip() + "\n\n" + section + "\n"
else:

View file

@ -1256,7 +1256,7 @@ def _project_for_pattern(pattern: ExtractedPattern, roots: list[ProjectInfo]) ->
for cand in candidates:
for root in roots_sorted:
root_str = str(root.project_path).rstrip("/")
root_str = str(root.project_path).rstrip("/\\")
if not root_str:
continue
if (

View file

@ -160,7 +160,7 @@ class AgentWriter(ABC):
if not dry_run:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(full_content)
target.write_text(full_content, encoding="utf-8")
return result
@ -185,8 +185,8 @@ class AgentWriter(ABC):
def _merge_section(file_path: Path, section: str) -> str:
"""Merge a marker-delimited section into an existing file."""
if file_path.exists():
existing = file_path.read_text()
existing = file_path.read_text(encoding="utf-8")
if MARKER_START in existing:
return MARKER_PATTERN.sub(section, existing)
return MARKER_PATTERN.sub(lambda _match: section, existing)
return existing.rstrip() + "\n\n" + section + "\n"
return section + "\n"

View file

@ -12,6 +12,7 @@ from __future__ import annotations
from collections import defaultdict
from pathlib import Path
from headroom.memory.sync_adapters.claude_code import encode_claude_project_path
from headroom.memory.writers.base import AgentWriter, MemoryEntry
@ -64,8 +65,7 @@ class ClaudeCodeMemoryWriter(AgentWriter):
project_path = self._project_path
# Claude Code stores per-project memory at:
# ~/.claude/projects/-<sanitized-path>/memory/MEMORY.md
# Replace both Unix and Windows path separators
sanitized = str(project_path).replace("/", "-").replace("\\", "-")
sanitized = encode_claude_project_path(project_path)
claude_memory_dir = Path.home() / ".claude" / "projects" / sanitized / "memory"
return claude_memory_dir / "MEMORY.md"
@ -116,6 +116,6 @@ class ClaudeCodeMemoryWriter(AgentWriter):
if not dry_run:
memory_dir.mkdir(parents=True, exist_ok=True)
(memory_dir / filename).write_text(content)
(memory_dir / filename).write_text(content, encoding="utf-8")
return topic_files

View file

@ -100,10 +100,10 @@ class CursorMemoryWriter(AgentWriter):
# If file exists and has our markers, only replace marker section
if target.exists():
existing = target.read_text()
existing = target.read_text(encoding="utf-8")
if MARKER_START in existing:
section = f"{MARKER_START}\n{body}\n{MARKER_END}"
full_content = MARKER_PATTERN.sub(section, existing)
full_content = MARKER_PATTERN.sub(lambda _match: section, existing)
else:
# Append our section
section = f"{MARKER_START}\n{body}\n{MARKER_END}"
@ -127,6 +127,6 @@ class CursorMemoryWriter(AgentWriter):
if not dry_run:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(full_content)
target.write_text(full_content, encoding="utf-8")
return result

View file

@ -20,9 +20,12 @@ def runner() -> CliRunner:
class FakeWriter:
def __init__(self) -> None:
self.calls: list[tuple[list[object], object, bool]] = []
self.fail_for: object | None = None
def write(self, recommendations, project, dry_run: bool): # noqa: ANN001, ANN201
self.calls.append((recommendations, project, dry_run))
if project is self.fail_for:
raise PermissionError(f"cannot write {project.project_path}")
return SimpleNamespace(
dry_run=dry_run,
content_by_file={
@ -218,6 +221,33 @@ def test_learn_analyze_all_uses_default_workers_and_prints_summary(
assert plugin_b.scan_calls == [(projects_b[0], 8)]
def test_learn_analyze_all_continues_when_one_project_write_fails(
monkeypatch: pytest.MonkeyPatch, runner: CliRunner, tmp_path: Path
) -> None:
blocked = SimpleNamespace(name="blocked", project_path=tmp_path / "blocked")
ok = SimpleNamespace(name="ok", project_path=tmp_path / "ok")
plugin = FakePlugin("claude", "Claude Code", [blocked, ok])
plugin.writer.fail_for = blocked
analyzer = FakeAnalyzer()
monkeypatch.setattr("headroom.learn.analyzer._detect_default_model", lambda: "gpt-4o")
monkeypatch.setattr("headroom.learn.registry.get_plugin", lambda name: plugin)
monkeypatch.setattr("headroom.learn.analyzer.SessionAnalyzer", lambda model=None: analyzer)
result = runner.invoke(
main,
["learn", "--agent", "claude", "--all", "--apply"],
catch_exceptions=False,
)
assert result.exit_code == 0, result.output
assert "Warning: failed to write recommendations" in result.output
assert str(blocked.project_path) in result.output
assert "[WROTE]" in result.output
assert str(ok.project_path / "AGENTS.md") in result.output
assert plugin.scan_calls == [(blocked, 8), (ok, 8)]
def test_learn_handles_empty_sessions_and_no_pattern_outputs(
monkeypatch: pytest.MonkeyPatch, runner: CliRunner, tmp_path: Path
) -> None:

View file

@ -8,12 +8,13 @@ making it impossible to reconstruct names formed from three or more tokens.
from __future__ import annotations
from collections.abc import Generator
from pathlib import Path
from uuid import uuid4
import pytest
from headroom.learn.scanner import _decode_project_path, _greedy_path_decode
from headroom.learn.scanner import ClaudeCodeScanner, _decode_project_path, _greedy_path_decode
# ---------------------------------------------------------------------------
# Helpers
@ -191,7 +192,7 @@ class TestDecodeProjectPath:
# ------------------------------------------------------------------
@pytest.fixture()
def users_tmp(self, tmp_path: Path) -> Path:
def users_tmp(self, tmp_path: Path) -> Generator[Path, None, None]:
"""Return a temporary directory whose path starts with /Users/…
On macOS the system temp dir is under /private/var, so we create a
@ -315,10 +316,31 @@ class TestDecodeProjectPath:
def test_windows_users_path(self) -> None:
"""Encoded name -C-Users-foo-project detects drive letter."""
import sys
result = _decode_project_path("-C-Users-foo-project")
if sys.platform == "win32":
assert result is None or "Users" in str(result)
else:
assert result is None
assert result is not None
assert str(result).startswith("C:")
assert "Users" in str(result)
def test_windows_username_with_dot_stays_single_component(self) -> None:
"""Windows profile names like john.doe must not decode as john/doe."""
result = _decode_project_path("-C-Users-john.doe-work")
assert result is not None
rendered = str(result)
assert rendered.startswith("C:")
assert "john.doe" in rendered
assert "john\\doe" not in rendered
assert "john/doe" not in rendered
def test_discover_windows_project_uses_leaf_name(self, tmp_path: Path) -> None:
"""A syntactic Windows path decoded on Unix should still display the project leaf."""
claude_dir = tmp_path / ".claude"
project_dir = claude_dir / "projects" / "-C-Users-john.doe-work"
project_dir.mkdir(parents=True)
(project_dir / "session.jsonl").write_text("{}\n")
projects = ClaudeCodeScanner(claude_dir=claude_dir).discover_projects()
assert len(projects) == 1
assert projects[0].name == "work"
assert str(projects[0].project_path).startswith("C:")

View file

@ -7,6 +7,7 @@ from headroom.learn.writer import (
_MARKER_END,
_MARKER_START,
ClaudeCodeWriter,
_merge_into_file,
_parse_prior_recommendations,
extract_marker_block,
)
@ -146,6 +147,33 @@ class TestClaudeCodeWriter:
# Only one Environment section in the final block
assert content.count("### Environment") == 1
def test_replacing_existing_block_handles_literal_backslash_escapes(self, tmp_path):
"""LLM text with backslash escapes must not be interpreted as a regex replacement."""
proj = _project(tmp_path)
claude_md = proj.project_path / "CLAUDE.md"
claude_md.write_text(
f"# Existing\n\n{_MARKER_START}\n"
"## Headroom Learned Patterns\n\n"
"### Windows Paths\n"
"- stale\n\n"
f"{_MARKER_END}\n"
)
full_content = _merge_into_file(
claude_md,
[
_rec(
RecommendationTarget.CONTEXT_FILE,
"Windows Paths",
r"- Keep the literal \u sequence and C:\Users\john.doe\repo path",
)
],
)
assert r"\u sequence" in full_content
assert r"C:\Users\john.doe\repo" in full_content
assert "stale" not in full_content
def test_memory_md_carry_forward(self, tmp_path):
"""Carry-forward also works for MEMORY.md."""
proj = _project(tmp_path)

View file

@ -521,6 +521,16 @@ class TestProjectForPattern:
)
assert _project_for_pattern(pattern, [proj]) is proj
def test_windows_root_with_trailing_backslash_matches_child_path(self):
proj = self._project(r"C:\Users\john.doe\repo\\")
pattern = ExtractedPattern(
category=PatternCategory.ERROR_RECOVERY,
content=r"File `C:\Users\john.doe\repo\src\main.py` does not exist.",
importance=0.5,
)
assert _project_for_pattern(pattern, [proj]) is proj
def test_no_false_match_on_prefix_boundary(self):
# /x/ab should not match a project rooted at /x/a
proj_a = self._project("/x/a")

View file

@ -99,6 +99,17 @@ class TestMergeSection:
assert "# Header" in result
assert "# Footer" in result
def test_replace_existing_markers_handles_literal_backslashes(self, tmp_path: Path):
existing = tmp_path / "marked.md"
existing.write_text(f"# Header\n\n{MARKER_START}\nold content\n{MARKER_END}\n")
section = f"{MARKER_START}\n- Keep C:\\Users\\john.doe\\repo and literal \\u\n{MARKER_END}"
result = _merge_section(existing, section)
assert r"C:\Users\john.doe\repo" in result
assert r"literal \u" in result
assert "old content" not in result
# =============================================================================
# Claude Code Writer Tests
@ -183,6 +194,14 @@ class TestClaudeCodeWriter:
assert filename.startswith("headroom_")
assert "---" in content # YAML frontmatter
def test_default_path_encodes_windows_user_with_dot(self):
writer = ClaudeCodeMemoryWriter(project_path=Path(r"C:\Users\john.doe\work"))
rendered = str(writer.default_path())
assert "-C-Users-john.doe-work" in rendered
assert "john-doe" not in rendered
assert rendered.endswith("MEMORY.md")
# =============================================================================
# Cursor Writer Tests

View file

@ -30,6 +30,8 @@ from headroom.memory.sync import (
from headroom.memory.sync_adapters.claude_code import (
ClaudeCodeAdapter,
_parse_frontmatter,
encode_claude_project_path,
get_claude_memory_dir,
)
from headroom.memory.sync_adapters.codex_agent import CodexAdapter
@ -408,6 +410,17 @@ class TestClaudeCodeAdapter:
assert fm == {}
assert body == "Just plain content."
def test_encode_claude_project_path_windows_user_with_dot(self):
assert encode_claude_project_path(r"C:\Users\john.doe\work") == "-C-Users-john.doe-work"
def test_get_claude_memory_dir_uses_windows_safe_project_encoding(self):
memory_dir = get_claude_memory_dir(Path(r"C:\Users\john.doe\work"))
rendered = str(memory_dir)
assert "-C-Users-john.doe-work" in rendered
assert "john-doe" not in rendered
assert rendered.endswith("memory")
@pytest.mark.asyncio
async def test_read_memories_skips_memory_md(self, memory_dir):
(memory_dir / "MEMORY.md").write_text("# Index\n- entry")
@ -537,6 +550,23 @@ class TestCodexAdapter:
assert "new fact" in content
assert "old fact" not in content
@pytest.mark.asyncio
async def test_write_replaces_existing_section_with_literal_backslashes(self, agents_md):
agents_md.write_text(
"# Instructions\n\n"
"<!-- headroom:memory:start -->\n"
"## Old\n- old fact\n"
"<!-- headroom:memory:end -->\n"
)
adapter = CodexAdapter(agents_md)
await adapter.write_memories([{"content": r"Use C:\Users\john.doe\repo and literal \u"}])
content = agents_md.read_text()
assert r"C:\Users\john.doe\repo" in content
assert r"literal \u" in content
assert "old fact" not in content
@pytest.mark.asyncio
async def test_read_empty_agents_md(self, agents_md):
agents_md.write_text("# No memory section\n")