fix(install): consolidate Windows fallback and cleanup safety (#2980)

## Description

Consolidates two fully reviewed installation-safety fixes whose original
PRs can no longer merge under current branch protection: Windows
persistent-service deployments need a supported Task Scheduler fallback,
and legacy context-tool cleanup must never delete user-owned
RTK/lean-ctx artifacts.

Closes #2552
Closes #2817

Supersedes #2600 and #2828 while preserving their authors' commits and
review-driven corrections.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [x] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Convert Windows `persistent-service` plans to the supported
`persistent-task` supervisor and make the fallback explicit in CLI
output.
- Restrict context-tool cleanup to artifacts proven to live under
Headroom's managed directory.
- Recognize wrapped, relative, and platform-specific managed commands
without accepting prefixed/path-boundary lookalikes.
- Scope cleanup completion state correctly across projects and alternate
agent homes.
- Stamp cleanup complete only after all managed remnants are settled.
- Preserve the original focused regression suites and behavior-proof
artifact.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ uv run pytest -q tests/test_install/test_planner.py tests/test_install/test_supervisors.py tests/test_cli/test_install_cli.py tests/test_context_tool_cleanup.py tests/test_cli/test_unwrap_claude.py
135 passed in 0.45s

$ uv run ruff check <changed Python and test files>
All checks passed!

$ uv run ruff format --check <changed Python and test files>
8 files already formatted
```

## Real Behavior Proof

- Environment: macOS arm64 for consolidated current-main validation; the
Windows fallback source PR was independently validated on Windows and
includes its captured verification artifact.
- Exact command / steps: run the planner, supervisor, install CLI,
cleanup provenance, and unwrap suites on the rebased combined branch.
- Observed result: 135/135 focused tests pass. Windows service requests
resolve to `persistent-task`; cleanup rejects user-owned and path-prefix
lookalikes while removing managed artifacts.
- Not tested: a fresh privileged Windows host deployment in this local
pass; #2600's accepted review contains the Windows-specific proof.

## Runtime Rollout Safety

- Rollout-managed feature(s): Install supervisor selection and one-time
legacy cleanup.
- Minimum rollout channel: Stable/default; both prevent currently
destructive or nonfunctional install paths.
- Stable/default behavior changed: Windows service requests use Task
Scheduler; cleanup requires managed provenance.
- Kill switch / disable path: Select `persistent-task` explicitly;
cleanup remains bounded by its completion stamp and provenance checks.
- Unsafe override required: No.
- Qualification impact: Windows native install and wrap/unwrap cleanup
suites.
- Rollback path: Revert this PR, restoring the two pre-fix behaviors.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Screenshots (if applicable)

The Windows verification artifact from #2600 is retained at
`.github/pr-images/issue-2552-windows-fallback-verification.png`.

## Additional Notes

This is intentionally an installation-safety batch rather than two
replacement PRs. Original commit authorship is preserved, and the
combined diff was applied cleanly to current `main` after #2832 and
#1628 landed.

---------

Co-authored-by: Inference1 <68734681+Inference1@users.noreply.github.com>
Co-authored-by: Dennis Alexis Valin Dittrich <dd+github@dr-dittrich.de>
This commit is contained in:
JD Davis
2026-08-13 15:05:45 -05:00
committed by GitHub
parent a3fe5cb65b
commit ddd2a259ec
9 changed files with 1080 additions and 55 deletions
Binary file not shown.

After

Width:  |  Height:  |  Size: 180 KiB

+16
View File
@@ -139,6 +139,12 @@ def _require_manifest(profile: str) -> DeploymentManifest:
raise _missing_profile_error(profile, installed)
def _is_windows() -> bool:
"""Return whether this command is running on Windows."""
return sys.platform.startswith("win")
def _start_deployment(manifest: DeploymentManifest, *, assume_start_lock: bool = False) -> None:
if not assume_start_lock:
with acquire_runtime_start_lock(manifest.profile) as acquired:
@@ -658,6 +664,16 @@ def install_apply(
bedrock_profile=bedrock_profile,
extra_env=combined_env,
)
if (
preset == InstallPreset.PERSISTENT_SERVICE.value
and manifest.preset == InstallPreset.PERSISTENT_TASK.value
and _is_windows()
):
click.echo(
"Warning: persistent-service is not supported on Windows because the "
"Python runner cannot act as a Windows service. Falling back to "
"persistent-task with Task Scheduler."
)
_apply_manifest(manifest)
_echo_installed(manifest)
+5 -3
View File
@@ -795,7 +795,7 @@ _RETIRED_CONTEXT_TOOL_MESSAGE = (
"rewrote shell commands through a third-party binary Headroom no longer "
"manages. Drop --context-tool / --no-context-tool and unset "
f"{_RETIRED_CONTEXT_TOOL_ENV}; `headroom wrap` uninstalls what they left "
"behind on first run."
"behind automatically."
)
@@ -855,8 +855,10 @@ def _report_context_tool_purge() -> None:
default: the Claude ``PreToolUse`` hook, the vendored binaries and the
injected hint-file guidance are all durable on disk. Running this once per
``wrap`` / ``unwrap`` invocation is what actually makes the tools go away.
Silent when there is nothing to do, which is the steady state after the first
run, and never fatal a cleanup failure must not block launching the tool.
Silent when there is nothing to do the common case once the machine-global
half is stamped done, though the project- and config-directory-scoped half
still runs every launch and never fatal: a cleanup failure must not block
launching the tool.
Reports on **stderr**: some subcommands (``wrap/unwrap openclaw
--prepare-only``) emit machine-readable JSON on stdout as their entire
+377 -37
View File
@@ -10,26 +10,87 @@ Deleting the code is not enough: everything above is *durable state on the
user's disk*. Left alone, the Claude hooks keep rewriting every Bash command
through binaries Headroom no longer manages, and the injected guidance keeps
telling agents to use tools that may not resolve. So ``headroom wrap`` /
``headroom unwrap`` call :func:`purge_context_tool_artifacts` once per run to
remove what earlier versions installed.
``headroom unwrap`` call :func:`purge_context_tool_artifacts` on every run to
remove what earlier versions installed — machine-global artifacts (hooks,
binaries, Claude Code's MCP registration) are removed once per workspace and
then stamped done (see the stamp below), while project- and config-directory-
scoped guidance (``CODEX_HOME`` / ``OPENCODE_HOME`` hint files, Continue's
config) is inspected on every launch, since a later launch can sit in a
different project or point at a different ``CODEX_HOME`` / ``OPENCODE_HOME``.
Everything here is idempotent, best-effort and deliberately conservative:
* only files Headroom installed (or caused a context tool to install) are
deleted;
deleted. An MCP entry's ``command`` or a hook script's body counts as
Headroom's only when it names a path inside :func:`paths.bin_dir`
(:func:`_references_managed_bin` — which is where the matching rules and
the reasons behind them live);
* a hook entry is Headroom's when it names such a path directly, or — the
common case, since a hook command names a script rather than the binary —
when it names one of the ``~/.claude/hooks`` scripts already classified as
Headroom's, whose verdict it inherits
(:func:`_references_context_tool`, :func:`_names_a_managed_script`). A
Cursor ``hooks.json`` entry naming a script under ``~/.cursor`` cannot
inherit a verdict this way, since the map covers ``~/.claude/hooks`` only;
it is still caught when its ``command`` names the managed directory;
* ``.rtk-hook.sha256`` is never read for its own provenance (it holds a hex
digest, not a path) and instead inherits ``rtk-rewrite.sh``'s
classification; a ``<name>.lean-ctx.bak`` backup inherits ``<name>``'s
(:func:`_classify_hook_scripts`);
* a hook script that exists but cannot be read is classified unknown —
deleted by nothing, and named in the report so the user can remove it by
hand;
* ``~/.local/bin/{rtk,lean-ctx}`` is unlinked only when it is a symlink into
Headroom's own bin directory — a user's own build is never touched;
* a JSON config that does not parse is reported and **skipped**, never
overwritten (a hand-edited typo must not cost the user their settings);
* the tools' own backups of *config* files (``~/.claude.json.lean-ctx.bak`` and
friends) are left in place — they hold the user's real settings history. Only
backups of the hook scripts being deleted are cleaned up.
backups of the hook scripts proven to be Headroom's are cleaned up;
* two cases cannot be decided at all, and are accepted as limits rather than
fixed:
* ``get_lean_ctx_path`` used to check ``PATH`` before Headroom's own bin
directory, so on a machine that already had ``lean-ctx`` on ``PATH``,
the tool that ran was the user's own, and the config it wrote looks
exactly like config the user wrote by hand. That leftover survives the
purge — it still points at a binary that exists, so nothing dangles;
* an rtk hook written *after* #1698 execs a bare ``rtk`` and never mentions
:func:`paths.bin_dir`, so it reads exactly like a hook a user wrote by
hand, and ``rtk-rewrite.sh``, its ``.rtk-hook.sha256`` and its
``settings.json`` entry all survive while step 3 removes the managed
binary — leaving a hook that silently no-ops (#487, #1698). Earlier
hooks are decidable: Headroom patched the absolute managed path into
them (``_patch_rtk_hook_absolute_path``, removed by #1698), so the
window this misses is rtk setups run between #1698 and the tools'
removal in #2677;
* the marker-fenced guidance block is the one step with no provenance check
to make — ``<!-- headroom:rtk-instructions -->`` is Headroom's own fence,
and no third party writes it.
Removing the retired integration's machine-global footprint — hook
registrations, hook scripts, PATH symlinks, managed binaries and Claude
Code's own MCP registration — is a one-time migration: the first completed
run of that half stamps ``.context-tools-purged`` beside the managed bin
directory, and every later run skips that half outright. Without the stamp
this would keep rewriting the same machine-wide files on every ``wrap``
invocation forever, and a user who installs one of these tools *after* the
migration would have Headroom auditing files at each launch for a leftover
that cannot exist there.
Project- and config-directory-scoped state is not covered by that stamp: a
later invocation can sit in a different project, or point ``CODEX_HOME`` /
``OPENCODE_HOME`` somewhere the stamped run never inspected, and whatever
guidance an earlier Headroom left behind there is still worth removing — so
those steps run on every invocation instead (:func:`_purge_invocation_scoped`).
"""
from __future__ import annotations
import json
import os
import posixpath
import re
from pathlib import Path
from typing import Any
@@ -69,52 +130,138 @@ _HOOK_SCRIPTS = (
"lean-ctx-redirect-native",
)
# rtk's integrity digest never names a path (see ``_classify_hook_scripts``)
# and rtk-rewrite.sh is the script it authenticates.
_RTK_DIGEST_NAME = ".rtk-hook.sha256"
_RTK_SCRIPT_NAME = "rtk-rewrite.sh"
# MCP server entries the tools registered, and the config files holding them.
# lean-ctx registers itself as an MCP server during ``lean-ctx init``; rtk never
# did, but it is matched too so a stale hand-added entry is cleaned up as well.
_MCP_SERVER_NAMES = ("lean-ctx", "lean_ctx", "rtk")
# Report-line prefixes meaning "this one is not settled" — a config that would
# not parse, a script that would not read, a file that would not unlink. Such a
# run leaves a leftover behind, so it must not be stamped as the completed
# migration. Emitted by _purge_hook_config, _purge_mcp_entries,
# _purge_fenced_block, _purge_continue_system_messages and _remove_files.
_DEFERRED_PREFIXES = ("skipped ", "could not remove ")
def purge_context_tool_artifacts() -> list[str]:
"""Remove every rtk / lean-ctx artifact an earlier Headroom version installed.
Returns human-readable descriptions of what was removed — plus a line for
any config that had to be skipped because the user must fix it by hand. An
empty list means there was nothing to do, which is the steady state after
the first run.
any config that had to be skipped because the user must fix it by hand.
Machine-global cleanup (hooks, binaries, Claude Code's MCP registration)
runs once and is then skipped via the stamp below; project- and config-
directory-scoped cleanup (hint files, ``CODEX_HOME`` / ``OPENCODE_HOME``,
Continue's config) runs on every call, so a later call in a different
project or a repointed ``CODEX_HOME`` / ``OPENCODE_HOME`` can still report
something even after the global half is long since stamped done.
"""
marker = _purge_marker()
home = Path.home()
project = Path.cwd()
report: list[str] = []
# 1. Hook registrations (Claude Code's settings.json, Cursor's hooks.json).
for config in (home / ".claude" / "settings.json", home / ".cursor" / "hooks.json"):
report += _purge_hook_config(config)
if not marker.exists():
# Classify every hook script's provenance once, up front: both step 1
# (is a settings.json entry pointing at *our* script?) and step 2 (is
# the script itself ours?) need the same answer, and each file is
# read once.
hooks_dir = home / ".claude" / "hooks"
hook_classification = _classify_hook_scripts(hooks_dir)
# 2. The generated hook scripts, their integrity digests and stale backups.
hooks_dir = home / ".claude" / "hooks"
report += _remove_files(
*(hooks_dir / name for name in _HOOK_SCRIPTS),
*(hooks_dir / f"{name}.lean-ctx.bak" for name in _HOOK_SCRIPTS),
)
# 1. Hook registrations (Claude Code's settings.json, Cursor's hooks.json).
for config in (home / ".claude" / "settings.json", home / ".cursor" / "hooks.json"):
report += _purge_hook_config(config, hooks_dir, hook_classification)
# 3. The PATH symlinks, then the managed binaries they pointed at.
for name in ("rtk", "lean-ctx"):
report += _remove_managed_path_link(home / ".local" / "bin" / name)
report += _remove_files(*(paths.bin_dir() / name for name in _BINARY_NAMES))
# 2. The generated hook scripts, their integrity digests and stale
# backups — only the ones proven to reference Headroom's managed bin
# directory.
managed_names = [name for name in _HOOK_SCRIPTS if hook_classification.get(name)]
report += _remove_files(
*(hooks_dir / name for name in managed_names),
*(hooks_dir / f"{name}.lean-ctx.bak" for name in managed_names),
)
for name in _HOOK_SCRIPTS:
if hook_classification.get(name, False) is not None:
continue
if name == _RTK_DIGEST_NAME:
report.append(
f"skipped {hooks_dir / name} (inherits {_RTK_SCRIPT_NAME}'s unreadable verdict)"
" — remove any stale hook script by hand"
)
else:
report.append(
f"skipped {hooks_dir / name} (could not read to verify it was Headroom's)"
" — remove any stale hook script by hand"
)
# 4. MCP server registrations (lean-ctx registers itself during init).
report += _purge_mcp_entries(home / ".claude.json", "mcpServers")
# 3. The PATH symlinks, then the managed binaries they pointed at.
for name in ("rtk", "lean-ctx"):
report += _remove_managed_path_link(home / ".local" / "bin" / name)
report += _remove_files(*(paths.bin_dir() / name for name in _BINARY_NAMES))
# 4. Claude Code's own MCP server registration (lean-ctx registers
# itself during init). OpenCode's is invocation-scoped — see below.
report += _purge_mcp_entries(home / ".claude.json", "mcpServers")
# Only a global half that settled everything is the completed
# migration. One that could not read a script or parse a config left
# a leftover behind, and the user needs both the reminder on the next
# launch and the cleanup once the permissions or the typo are fixed.
# A deferral in the invocation-scoped half below must not withhold
# this stamp — that half re-runs every time regardless, so nothing is
# lost by stamping the global half done now.
if not any(line.startswith(_DEFERRED_PREFIXES) for line in report):
try:
# Cleanup must not become the first mutation on a pristine
# machine. In particular, ``wrap <missing-tool>`` validates
# the binary after the wrap-group migration hook; creating
# ``~/.headroom`` merely to stamp an empty scan violates that
# command's no-side-effects-on-failure contract. Established
# Headroom installs already have the state directory and get
# the one-time fast path; clean machines cheaply rescan until
# some real Headroom state exists.
if marker.parent.is_dir():
marker.touch()
except OSError:
pass # Unwritable workspace: the purge simply runs again next time.
report += _purge_invocation_scoped(home, project)
return report
def _purge_invocation_scoped(home: Path, project: Path) -> list[str]:
"""Steps the one-time stamp must never withhold.
``OPENCODE_HOME``'s config, the hint files in ``project`` /
``CODEX_HOME`` / ``OPENCODE_HOME``, and Continue's config are all a
function of *this* invocation's cwd and environment, not of the machine —
a later run can sit in a different project or point ``CODEX_HOME`` /
``OPENCODE_HOME`` somewhere the global-half stamp never inspected. Each
step is cheap and side-effect-free when nothing matches, so re-running
them on every invocation costs a handful of reads in the steady state.
"""
report: list[str] = []
report += _purge_mcp_entries(_opencode_home(home) / "opencode.json", "mcp")
# 5. Marker-fenced guidance in every hint file the wrap harnesses wrote to.
for hint_file in _instruction_files(home, project):
report += _purge_fenced_block(hint_file)
report += _purge_continue_system_messages(project / ".continue" / "config.json")
return report
def _purge_marker() -> Path:
"""Path of the "already migrated" stamp.
Derived from :func:`paths.bin_dir` rather than ``workspace_dir`` so it
cannot escape a temporary tree through ``HEADROOM_WORKSPACE_DIR``.
"""
return paths.bin_dir().parent / ".context-tools-purged"
def _instruction_files(home: Path, project: Path) -> list[Path]:
"""Hint files the wrap subcommands injected the context-tool block into.
@@ -151,15 +298,194 @@ def _opencode_home(home: Path) -> Path:
# --- hook registrations -------------------------------------------------------
def _references_context_tool(entry: Any) -> bool:
"""Whether a hook entry's command is one a retired context tool registered."""
# Characters that can never be part of a path: a match ending right before
# one of these (or at end-of-string) sits at a real word boundary. Quotes,
# `=`/`:` (`export PATH="<dir>:$PATH"`, `BIN=<dir>/x`), `;`/`,`/`|` (command
# joiners) and `()` (subshells) all end a path the same way whitespace does.
_PATH_BOUNDARY_CHARS = frozenset("\"'=:;,()|")
# Splits a command into path-shaped tokens on whitespace plus the same
# boundary punctuation above — used by _names_a_managed_script, which (unlike
# _references_managed_bin's needle-anchored scan) tokenizes the whole command.
_PATH_TOKEN_SPLIT = re.compile(r"[\s" + re.escape("".join(_PATH_BOUNDARY_CHARS)) + r"]+")
def _norm_path_text(value: str) -> str:
"""Case-fold ``value`` and give it one separator, so paths compare as text."""
return os.path.normcase(value).replace("\\", "/")
def _references_managed_bin(text: str) -> bool:
"""Whether ``text`` names a path inside Headroom's managed bin directory.
A hook command or script body is free text we don't control, so this
scans ``text`` for raw occurrences of the managed directory — the
unresolved and resolved bin directory, and its ``~``-relative form
(home-relative, since a script may reference it unexpanded) — matched
against the text as given and as ``expanduser``'d, case-folded, with both
path separators. Deliberately *not* tokenized on whitespace first: a
quoted or ``$HOME``-derived path can itself contain a space, and slicing
the text into words before searching would sever it.
A hit is only a real reference at a path boundary on *both* ends. The
character immediately before the match, if any, must be whitespace or a
:data:`_PATH_BOUNDARY_CHARS` character, or the match is just the tail of
some longer, unrelated path segment (e.g. ``/prefix<bin_dir>/lean-ctx``)
and is rejected. The match must then be immediately followed by
end-of-string or a :data:`_PATH_BOUNDARY_CHARS` character (an exact
reference, e.g. a bare ``PATH=<dir>`` export), or by ``/`` — in which
case the run of characters up to the next boundary is lexically
normalized (``.``/``..`` collapsed) and re-compared, so neither a sibling
directory like ``bin-backup``/``binfoo`` nor a ``bin/../evil`` traversal
can borrow the managed prefix.
# ponytail: boundary-aware substring scan, not a shell parse — upgrade to
# shlex if a command ever embeds a managed path it does not execute.
"""
try:
bin_dir = paths.bin_dir()
resolved_bin_dir = bin_dir.resolve()
except OSError:
return False
needles = {_norm_path_text(str(bin_dir)), _norm_path_text(str(resolved_bin_dir))}
home = Path.home()
for base in (bin_dir, resolved_bin_dir):
try:
needles.add(_norm_path_text(f"~/{base.relative_to(home).as_posix()}"))
except ValueError:
pass
for haystack in (text, os.path.expanduser(text)):
normalized_haystack = _norm_path_text(haystack)
if any(_names_managed_dir(normalized_haystack, needle) for needle in needles):
return True
return False
def _names_managed_dir(haystack: str, needle: str) -> bool:
"""Whether a normalized ``haystack`` names ``needle`` at a path boundary
on both ends: the character right before the match, if any, must be
whitespace or a :data:`_PATH_BOUNDARY_CHARS` character too, or a
user-owned path that merely has the managed directory as a substring
(e.g. ``/prefix<bin_dir>/lean-ctx``) would be misread as naming it.
"""
search_from = 0
while True:
index = haystack.find(needle, search_from)
if index < 0:
return False
end = index + len(needle)
search_from = index + 1 # keep scanning; occurrences may overlap
if index > 0 and not (
haystack[index - 1].isspace() or haystack[index - 1] in _PATH_BOUNDARY_CHARS
):
continue
following = haystack[end : end + 1]
if not following or following.isspace() or following in _PATH_BOUNDARY_CHARS:
return True
if following != "/":
continue
tail_end = end
while tail_end < len(haystack) and not (
haystack[tail_end].isspace() or haystack[tail_end] in _PATH_BOUNDARY_CHARS
):
tail_end += 1
candidate = posixpath.normpath(haystack[index:tail_end])
if candidate == needle or candidate.startswith(needle + "/"):
return True
def _classify_hook_scripts(hooks_dir: Path) -> dict[str, bool | None]:
"""Classify each existing ``_HOOK_SCRIPTS`` file by whether it is Headroom's.
``True`` — the file exists and its body references the managed bin
directory. ``False`` — it exists and does not. ``None`` — it exists but
could not be read, so its provenance is unprovable. A basename with no
file on disk is simply absent from the map. Each file is read at most once.
``.rtk-hook.sha256`` holds a hex digest that can never reference a path,
so it is never read; it inherits ``rtk-rewrite.sh``'s classification.
"""
classification: dict[str, bool | None] = {}
for name in _HOOK_SCRIPTS:
if name == _RTK_DIGEST_NAME:
continue
path = hooks_dir / name
if not path.is_file():
continue
try:
body = fsutil.read_text(path)
except OSError:
classification[name] = None
continue
classification[name] = _references_managed_bin(body)
if (hooks_dir / _RTK_DIGEST_NAME).is_file():
classification[_RTK_DIGEST_NAME] = classification.get(_RTK_SCRIPT_NAME, False)
return classification
def _references_context_tool(
entry: Any, hooks_dir: Path, hook_classification: dict[str, bool | None]
) -> bool:
"""Whether a hook entry is one a retired context tool registered.
A command-marker hit alone is not enough — a user could author a script
with a matching name. It must also either name the managed bin directory
directly, or name — as a path resolving to that exact file, not merely
sharing its basename, so a same-named script of the user's own in a
different directory is never caught — a hook script in ``hooks_dir`` the
classification map marks ``True`` (see :func:`_names_a_managed_script`).
"""
if not isinstance(entry, dict):
return False
command = str(entry.get("command", "")).lower()
return any(marker in command for marker in _HOOK_COMMAND_MARKERS)
command = str(entry.get("command", ""))
if not any(marker in command.lower() for marker in _HOOK_COMMAND_MARKERS):
return False
if _references_managed_bin(command):
return True
return _names_a_managed_script(command, hooks_dir, hook_classification)
def _prune_hooks(hooks: Any) -> tuple[Any, bool]:
def _names_a_managed_script(
command: str, hooks_dir: Path, hook_classification: dict[str, bool | None]
) -> bool:
"""Whether ``command`` names, by absolute path, a hook script the map marks ``True``.
A real command is rarely the bare script path: ``bash <script>`` wraps
it, a shell often quotes it, and it may carry a redundant ``./`` segment.
``command`` is split into path-shaped tokens on the same boundary
punctuation :data:`_PATH_BOUNDARY_CHARS` (and whitespace) mark as *not*
part of a path, each token is lexically normalized, and compared against
the script's absolute path only.
Deliberately not resolved against ``~``/:func:`Path.home` or against the
process's working directory: a *relative* hook command in
``~/.claude/settings.json`` is resolved by the harness against the
project's cwd, not against home — this module never writes or inspects a
project-relative path, so treating one as if it named a home-relative
script would delete a project-local hook this purge has no business
touching. A relative token, and a ``$VAR``-style unexpanded reference
(e.g. ``$HOME/...``), are both simply not recognised — unprovable, so
kept, per the guard's own rule.
"""
tokens = {
token
for haystack in (command, os.path.expanduser(command))
for token in _PATH_TOKEN_SPLIT.split(haystack)
if token
}
normalized_tokens = {posixpath.normpath(_norm_path_text(token)) for token in tokens}
return any(
verdict is True and _norm_path_text(str(hooks_dir / name)) in normalized_tokens
for name, verdict in hook_classification.items()
)
def _prune_hooks(
hooks: Any, hooks_dir: Path, hook_classification: dict[str, bool | None]
) -> tuple[Any, bool]:
"""Drop retired-tool entries from a ``hooks`` mapping; return ``(pruned, changed)``.
Handles both shapes Headroom's installers produced: Claude Code nests
@@ -180,13 +506,17 @@ def _prune_hooks(hooks: Any) -> tuple[Any, bool]:
retained: list[Any] = []
for entry in entries:
# Cursor shape: the command sits on the entry itself.
if _references_context_tool(entry):
if _references_context_tool(entry, hooks_dir, hook_classification):
changed = True
continue
# Claude shape: a matcher entry holding a list of hooks.
inner = entry.get("hooks") if isinstance(entry, dict) else None
if isinstance(inner, list):
kept_inner = [item for item in inner if not _references_context_tool(item)]
kept_inner = [
item
for item in inner
if not _references_context_tool(item, hooks_dir, hook_classification)
]
if len(kept_inner) != len(inner):
changed = True
if not kept_inner:
@@ -206,7 +536,9 @@ def _prune_hooks(hooks: Any) -> tuple[Any, bool]:
return pruned, changed
def _purge_hook_config(path: Path) -> list[str]:
def _purge_hook_config(
path: Path, hooks_dir: Path, hook_classification: dict[str, bool | None]
) -> list[str]:
"""Remove retired-tool hook registrations from a JSON hook config."""
if not path.is_file():
return []
@@ -217,7 +549,7 @@ def _purge_hook_config(path: Path) -> list[str]:
if not isinstance(payload, dict):
return [f"skipped {path} (not a JSON object) — remove any stale hook by hand"]
hooks, changed = _prune_hooks(payload.get("hooks"))
hooks, changed = _prune_hooks(payload.get("hooks"), hooks_dir, hook_classification)
if not changed:
return []
if hooks:
@@ -236,8 +568,10 @@ def _purge_mcp_entries(path: Path, container_key: str) -> list[str]:
``lean-ctx init`` registers lean-ctx as an MCP server in the harness's own
config — Claude Code keeps them under ``mcpServers``, OpenCode under ``mcp``.
Only the exactly-named entries are removed; every other server, and every
unrelated top-level key, is preserved byte-for-byte.
A name match alone is not enough — a user can register their own server
under the same name — so an entry is removed only when its ``command``
also names Headroom's managed bin directory. Every other server, and
every unrelated top-level key, is preserved byte-for-byte.
"""
if not path.is_file():
return []
@@ -251,7 +585,13 @@ def _purge_mcp_entries(path: Path, container_key: str) -> list[str]:
servers = payload.get(container_key)
if not isinstance(servers, dict):
return []
removed = [name for name in _MCP_SERVER_NAMES if name in servers]
removed = [
name
for name in _MCP_SERVER_NAMES
if isinstance(servers.get(name), dict)
and isinstance(servers[name].get("command"), str)
and _references_managed_bin(servers[name]["command"])
]
if not removed:
return []
for name in removed:
+14 -3
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import shutil
import sys
from collections.abc import Iterable
import click
@@ -142,9 +143,19 @@ def build_manifest(
normalized_profile = validate_profile_name(profile)
if preset == InstallPreset.PERSISTENT_SERVICE.value:
# A Windows service must implement the Service Control Manager protocol.
# The Python runner is an ordinary console process, so registering it with
# ``sc.exe create`` always fails at start with SCM error 1053. Task
# Scheduler can run the same runner safely and already provides startup
# plus periodic health recovery, so make it the effective preset on
# Windows instead of creating a service that can never start (#2552).
effective_preset = preset
if sys.platform.startswith("win") and preset == InstallPreset.PERSISTENT_SERVICE.value:
effective_preset = InstallPreset.PERSISTENT_TASK.value
if effective_preset == InstallPreset.PERSISTENT_SERVICE.value:
supervisor_kind = SupervisorKind.SERVICE.value
elif preset == InstallPreset.PERSISTENT_TASK.value:
elif effective_preset == InstallPreset.PERSISTENT_TASK.value:
supervisor_kind = SupervisorKind.TASK.value
else:
supervisor_kind = SupervisorKind.NONE.value
@@ -234,7 +245,7 @@ def build_manifest(
container_name = f"headroom-{normalized_profile}"
return DeploymentManifest(
profile=normalized_profile,
preset=preset,
preset=effective_preset,
runtime_kind=runtime_kind,
supervisor_kind=supervisor_kind,
scope=scope,
+38
View File
@@ -172,6 +172,44 @@ def test_install_apply_starts_service_supervisor(monkeypatch) -> None:
assert calls == ["save", "start_service", "apply", "save"]
def test_install_apply_announces_windows_service_fallback(monkeypatch) -> None:
runner = CliRunner()
calls: list[str] = []
class Manifest:
profile = "default"
preset = "persistent-task"
runtime_kind = "python"
supervisor_kind = "task"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
mutations: list[object] = []
targets: list[str] = []
artifacts: list[object] = []
monkeypatch.setattr("headroom.cli.install._is_windows", lambda: True)
monkeypatch.setattr("headroom.cli.install.build_manifest", lambda **_: Manifest())
monkeypatch.setattr("headroom.cli.install.load_manifest", lambda profile: None)
monkeypatch.setattr("headroom.cli.install.install_supervisor", lambda deployment: [])
monkeypatch.setattr("headroom.cli.install.save_manifest", lambda deployment: None)
monkeypatch.setattr("headroom.cli.install.apply_mutations", lambda deployment: [])
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.install.runtime_status", lambda manifest: "stopped")
monkeypatch.setattr(
"headroom.cli.install.start_detached_agent", lambda profile: calls.append("start_agent")
)
monkeypatch.setattr(
"headroom.cli.install.wait_ready", lambda deployment, timeout_seconds=45: True
)
result = runner.invoke(main, ["install", "apply", "--preset", "persistent-service"])
assert result.exit_code == 0, result.output
assert "Falling back to persistent-task with Task Scheduler" in result.output
assert "sc.exe" not in result.output
assert calls == ["start_agent"]
def test_install_apply_forwards_no_http2_to_build_manifest(monkeypatch) -> None:
runner = CliRunner()
captured: dict[str, object] = {}
+8 -3
View File
@@ -7,6 +7,7 @@ from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom import paths
from headroom.cli import wrap as wrap_cli
from headroom.cli.main import main
@@ -68,8 +69,14 @@ def test_unwrap_claude_removes_mcp_purges_retired_hook_and_stops_proxy(
home = str(tmp_path)
monkeypatch.setenv("HOME", home)
monkeypatch.setenv("USERPROFILE", home)
monkeypatch.delenv("HEADROOM_WORKSPACE_DIR", raising=False)
bin_dir = paths.bin_dir()
claude_dir = tmp_path / ".claude"
claude_dir.mkdir()
hooks_dir = claude_dir / "hooks"
hooks_dir.mkdir()
hook_script = hooks_dir / "rtk-rewrite.sh"
hook_script.write_text(f'#!/bin/sh\nexec {bin_dir / "rtk"} "$@"\n', encoding="utf-8")
settings = claude_dir / "settings.json"
settings.write_text(
json.dumps(
@@ -78,9 +85,7 @@ def test_unwrap_claude_removes_mcp_purges_retired_hook_and_stops_proxy(
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [
{"type": "command", "command": str(claude_dir / "rtk-rewrite.sh")}
],
"hooks": [{"type": "command", "command": str(hook_script)}],
}
]
}
+598 -9
View File
@@ -11,6 +11,8 @@ else.
from __future__ import annotations
import json
import os
import sys
import pytest
@@ -37,6 +39,19 @@ def _write(path, content):
def test_removes_hooks_for_both_tools_but_keeps_user_hooks(home):
bin_dir = paths.bin_dir()
hooks_dir = home / ".claude" / "hooks"
# Managed: a script whose body execs the Headroom-installed binary.
managed_script = _write(
hooks_dir / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
# User-owned: same marker-matching filename, but the body execs the
# user's own install — no path inside bin_dir anywhere.
user_script = _write(
hooks_dir / "lean-ctx-redirect.sh",
'#!/bin/sh\nexec /usr/bin/lean-ctx "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps(
@@ -44,12 +59,16 @@ def test_removes_hooks_for_both_tools_but_keeps_user_hooks(home):
"permissions": {"allow": ["Bash"]},
"hooks": {
"PreToolUse": [
{"hooks": [{"type": "command", "command": str(managed_script)}]},
{
"hooks": [
{"type": "command", "command": "~/.claude/hooks/rtk-rewrite.sh"}
{
"type": "command",
"command": f"{bin_dir / 'lean-ctx'} hook rewrite",
}
]
},
{"hooks": [{"type": "command", "command": "lean-ctx hook rewrite"}]},
{"hooks": [{"type": "command", "command": str(user_script)}]},
{"hooks": [{"type": "command", "command": "my-own-linter --check"}]},
],
"SessionStart": [{"hooks": [{"type": "command", "command": "echo hi"}]}],
@@ -64,7 +83,7 @@ def test_removes_hooks_for_both_tools_but_keeps_user_hooks(home):
commands = [
item["command"] for entry in payload["hooks"]["PreToolUse"] for item in entry["hooks"]
]
assert commands == ["my-own-linter --check"]
assert commands == [str(user_script), "my-own-linter --check"]
# Unrelated events and unrelated top-level keys survive untouched.
assert payload["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "echo hi"
assert payload["permissions"] == {"allow": ["Bash"]}
@@ -72,11 +91,21 @@ def test_removes_hooks_for_both_tools_but_keeps_user_hooks(home):
def test_removes_binaries_hook_scripts_and_backups(home):
bin_dir = home / ".headroom" / "bin"
bin_dir = paths.bin_dir()
hooks_dir = home / ".claude" / "hooks"
rtk = _write(bin_dir / "rtk", "binary")
lean = _write(bin_dir / "lean-ctx", "binary")
script = _write(home / ".claude" / "hooks" / "lean-ctx-rewrite.sh", "#!/bin/sh\n")
backup = _write(home / ".claude" / "hooks" / "lean-ctx-rewrite.sh.lean-ctx.bak", "#!/bin/sh\n")
script = _write(
hooks_dir / "lean-ctx-rewrite.sh", f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n'
)
backup = _write(
hooks_dir / "lean-ctx-rewrite.sh.lean-ctx.bak",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
managed_rtk_script = _write(
hooks_dir / "rtk-rewrite.sh", f'#!/bin/sh\nexec {bin_dir / "rtk"} "$@"\n'
)
managed_rtk_digest = _write(hooks_dir / ".rtk-hook.sha256", "deadbeef\n")
context_tool_cleanup.purge_context_tool_artifacts()
@@ -84,6 +113,22 @@ def test_removes_binaries_hook_scripts_and_backups(home):
assert not lean.exists()
assert not script.exists()
assert not backup.exists()
# .rtk-hook.sha256 follows rtk-rewrite.sh's classification: both managed,
# both removed.
assert not managed_rtk_script.exists()
assert not managed_rtk_digest.exists()
def test_leaves_a_users_own_rtk_digest_alone(home):
"""The digest is a hex hash, so it can only follow the script it authenticates."""
hooks_dir = home / ".claude" / "hooks"
script = _write(hooks_dir / "rtk-rewrite.sh", '#!/bin/sh\nexec /usr/bin/rtk "$@"\n')
digest = _write(hooks_dir / ".rtk-hook.sha256", "cafef00d\n")
context_tool_cleanup.purge_context_tool_artifacts()
assert script.exists()
assert digest.exists()
def test_leaves_a_users_own_binary_on_path_alone(home):
@@ -100,13 +145,14 @@ def test_leaves_a_users_own_binary_on_path_alone(home):
def test_removes_mcp_entry_and_preserves_siblings(home):
bin_dir = paths.bin_dir()
config = _write(
home / ".claude.json",
json.dumps(
{
"projects": {"/some/path": {"history": []}},
"mcpServers": {
"lean-ctx": {"command": "lean-ctx", "args": ["mcp"]},
"lean-ctx": {"command": str(bin_dir / "lean-ctx"), "args": ["mcp"]},
"headroom": {"command": "headroom", "args": ["mcp"]},
},
}
@@ -146,13 +192,25 @@ def test_skips_malformed_json_instead_of_clobbering_it(home):
def test_is_idempotent(home):
_write(home / ".headroom" / "bin" / "rtk", "binary")
"""Re-running the purge body reports nothing new.
Normally the completion stamp stops a second run, but a workspace the
stamp cannot be written to falls back to running every time — so the body
itself has to stay idempotent. Removing the stamp between runs is what
that machine does.
"""
bin_dir = paths.bin_dir()
_write(bin_dir / "rtk", "binary")
script = _write(
home / ".claude" / "hooks" / "rtk-rewrite.sh", f'#!/bin/sh\nexec {bin_dir / "rtk"} "$@"\n'
)
_write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": "rtk rewrite"}]}]}}),
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
assert context_tool_cleanup.purge_context_tool_artifacts()
(bin_dir.parent / ".context-tools-purged").unlink()
# Steady state after the first run: nothing left to report.
assert context_tool_cleanup.purge_context_tool_artifacts() == []
@@ -161,6 +219,98 @@ def test_no_op_on_a_clean_machine(home):
assert context_tool_cleanup.purge_context_tool_artifacts() == []
def test_a_completed_purge_never_runs_again(home):
"""Machine-global artifacts stay stamped-done: no per-launch re-audit.
A tool installed under the *global* half (hooks, binaries) after the
migration is not a leftover, so a later run must leave it alone without
even looking — this is what stops `headroom wrap` from re-litigating
machine-global state on every launch, forever. Project- and config-
directory-scoped guidance is a different story: see
`test_a_completed_purge_still_cleans_a_different_project`.
"""
bin_dir = paths.bin_dir()
bin_dir.parent.mkdir(parents=True)
assert context_tool_cleanup.purge_context_tool_artifacts() == []
assert (bin_dir.parent / ".context-tools-purged").exists()
# Artifacts that would otherwise be removed, installed after the migration.
binary = _write(bin_dir / "lean-ctx", "binary")
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
assert context_tool_cleanup.purge_context_tool_artifacts() == []
assert binary.exists()
assert script.exists()
def test_a_completed_purge_still_cleans_a_different_project(home, monkeypatch):
"""The one-time stamp is machine-global; project guidance is not.
A completed run in project A must not leave project B's fenced guidance in
place forever — the stamp only ever covered a snapshot of ``Path.cwd()``.
"""
paths.bin_dir().parent.mkdir(parents=True)
assert context_tool_cleanup.purge_context_tool_artifacts() == []
assert (paths.bin_dir().parent / ".context-tools-purged").exists()
project_b = home / "project-b"
project_b.mkdir()
agents = _write(
project_b / "AGENTS.md",
"# Project B\n\n<!-- headroom:rtk-instructions -->\nAlways prefix with rtk.\n"
"<!-- /headroom:rtk-instructions -->\n",
)
monkeypatch.chdir(project_b)
report = context_tool_cleanup.purge_context_tool_artifacts()
assert "rtk" not in agents.read_text()
assert any(str(agents) in line for line in report)
def test_a_completed_purge_still_inspects_a_repointed_codex_home(home, monkeypatch):
"""``CODEX_HOME`` can point somewhere new after the stamp; that target is not exempt."""
paths.bin_dir().parent.mkdir(parents=True)
assert context_tool_cleanup.purge_context_tool_artifacts() == []
assert (paths.bin_dir().parent / ".context-tools-purged").exists()
new_codex_home = home / "elsewhere-codex"
new_codex_home.mkdir()
agents = _write(
new_codex_home / "AGENTS.md",
"<!-- headroom:rtk-instructions -->\nAlways prefix with rtk.\n"
"<!-- /headroom:rtk-instructions -->\n",
)
monkeypatch.setenv("CODEX_HOME", str(new_codex_home))
report = context_tool_cleanup.purge_context_tool_artifacts()
# Nothing but the fence was in the file, so it is removed outright.
assert not agents.exists()
assert any(str(agents) in line for line in report)
def test_a_scoped_deferral_does_not_withhold_the_global_stamp(home):
"""A scoped step's leftover must not re-run the whole global half forever.
Only the invocation-scoped half is unprovable here (malformed Continue
config); the machine-global half has nothing to defer, so its stamp must
still be written — otherwise a permanently-broken ``.continue/config.json``
would force every hook/binary/MCP step to be re-walked on every launch.
"""
paths.bin_dir().parent.mkdir(parents=True)
_write(home / "project" / ".continue" / "config.json", "{not json")
report = context_tool_cleanup.purge_context_tool_artifacts()
assert any(line.startswith("skipped ") for line in report)
assert (paths.bin_dir().parent / ".context-tools-purged").exists()
def test_purge_reports_on_stderr_so_json_stdout_stays_parseable(home):
"""`wrap openclaw --prepare-only` emits machine-readable JSON as its whole contract.
@@ -216,3 +366,442 @@ def test_selfheal_does_not_purge(home, monkeypatch):
CliRunner().invoke(main, ["wrap", "selfheal", "--marker", "headroom-wrap-selfheal"])
assert binary.exists(), "selfheal performed filesystem cleanup"
def test_leaves_a_users_own_mcp_entry_alone(home):
config = _write(
home / ".claude.json",
json.dumps({"mcpServers": {"lean-ctx": {"command": "lean-ctx", "args": ["mcp"]}}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(config.read_text())
assert payload["mcpServers"] == {"lean-ctx": {"command": "lean-ctx", "args": ["mcp"]}}
def test_leaves_a_users_own_hook_script_and_hook_entry_alone(home):
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
'#!/bin/sh\nexec /usr/bin/lean-ctx "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert script.exists()
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == str(script)
def test_removes_the_managed_hook_script_and_its_hook_entry(home):
bin_dir = paths.bin_dir()
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert not script.exists()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_removes_a_hook_entry_pointing_at_the_managed_binary_directly(home):
bin_dir = paths.bin_dir()
settings = _write(
home / ".claude" / "settings.json",
json.dumps(
{
"hooks": {
"PreToolUse": [
{
"hooks": [
{
"type": "command",
"command": f"{bin_dir / 'lean-ctx'} hook rewrite",
}
]
}
]
}
}
),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_matches_a_managed_path_written_in_tilde_form(home):
"""The dangling-hook regression test: bin_dir is <tmp>/.headroom/bin, and the
script references it in unexpanded tilde form — the guard must normalize
both sides before comparing, or it wrongly treats this as unprovable and
leaves a hook pointing at a script Headroom itself no longer manages.
"""
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
'#!/bin/sh\nexec ~/.headroom/bin/lean-ctx "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert not script.exists()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_leaves_a_hook_script_in_a_sibling_bin_named_directory_alone(home):
"""A directory that merely starts with the bin dir's name is not the bin dir.
``<workspace>/.headroom/binaries`` shares a prefix with
``<workspace>/.headroom/bin`` but is a different, user-owned directory —
a naive substring match (no directory-boundary check) would treat the
shared prefix as a reference to the managed bin dir and wrongly delete
this script.
"""
bin_dir = paths.bin_dir()
sibling = bin_dir.parent / "binaries"
own_binary = _write(sibling / "lean-ctx", "my own build")
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {own_binary} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert script.exists()
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == str(script)
def test_leaves_an_mcp_entry_in_a_sibling_bin_named_directory_alone(home):
"""Same sibling-directory hazard as above, for the MCP-entry command check."""
bin_dir = paths.bin_dir()
sibling_command = str(bin_dir.parent / "bin-backup" / "lean-ctx")
config = _write(
home / ".claude.json",
json.dumps({"mcpServers": {"lean-ctx": {"command": sibling_command, "args": ["mcp"]}}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(config.read_text())
assert payload["mcpServers"] == {"lean-ctx": {"command": sibling_command, "args": ["mcp"]}}
def test_leaves_a_parent_traversal_path_through_the_bin_dir_alone(home):
"""``bin/../evil`` contains the managed prefix as literal text but does not
resolve inside it — the guard must collapse ``..`` before comparing, or a
crafted (or coincidental) traversal path would be treated as Headroom's.
"""
bin_dir = paths.bin_dir()
evil_binary = _write(bin_dir.parent / "evil" / "lean-ctx", "not ours")
traversal_command = f"{bin_dir}/../evil/lean-ctx"
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {traversal_command} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert script.exists()
assert evil_binary.exists()
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == str(script)
def test_leaves_a_hook_script_whose_body_has_the_bin_dir_as_a_path_segment_alone(home):
"""``/prefix<bin_dir>/lean-ctx`` contains the managed prefix as literal text
but at a position with no boundary before it — the run of characters
immediately preceding the match is a filename character (``x``), not
whitespace or a :data:`_PATH_BOUNDARY_CHARS` character, so this names a
different, user-owned directory that only happens to end in the managed
path's tail. Only checking the trailing boundary (the pre-fix behavior)
would misclassify this as Headroom's and delete the user's script.
"""
bin_dir = paths.bin_dir()
lookalike = f"/prefix{bin_dir}/lean-ctx"
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {lookalike} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert script.exists()
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == str(script)
def test_leaves_an_mcp_entry_whose_command_has_the_bin_dir_as_a_path_segment_alone(home):
"""Same leading-boundary hazard as above, for the MCP-entry command check."""
bin_dir = paths.bin_dir()
lookalike_command = f"/prefix{bin_dir}/lean-ctx"
config = _write(
home / ".claude.json",
json.dumps({"mcpServers": {"lean-ctx": {"command": lookalike_command, "args": ["mcp"]}}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(config.read_text())
assert payload["mcpServers"] == {"lean-ctx": {"command": lookalike_command, "args": ["mcp"]}}
def test_leaves_a_hook_entry_whose_command_has_the_bin_dir_as_a_path_segment_alone(home):
"""Same hazard for a hook entry whose ``command`` names the managed
directory directly (no script indirection). The command still carries a
``_HOOK_COMMAND_MARKERS`` token (``lean-ctx hook``) so it reaches the
provenance guard rather than being filtered out earlier.
"""
bin_dir = paths.bin_dir()
lookalike_command = f"/prefix{bin_dir}/lean-ctx hook rewrite"
settings = _write(
home / ".claude" / "settings.json",
json.dumps(
{
"hooks": {
"PreToolUse": [{"hooks": [{"type": "command", "command": lookalike_command}]}]
}
}
),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == lookalike_command
def test_removes_a_hook_script_whose_body_has_a_lookalike_prefix_before_a_genuine_managed_path(
home,
):
"""A body can contain both a rejected lookalike occurrence and a later
genuine, boundary-correct occurrence of the managed path. Rejecting the
first must not short-circuit the scan (``continue``, not
``return False``) or the genuine occurrence right after it would never
be seen.
"""
bin_dir = paths.bin_dir()
lookalike = f"/prefix{bin_dir}/lean-ctx-fake"
genuine = str(bin_dir / "lean-ctx")
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {lookalike} --check\nexec {genuine} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert not script.exists()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_removes_a_hook_script_whose_body_names_the_managed_dir_only_via_quoting_or_delimiters(
home,
):
"""Real shell scripts quote paths and join them with `=`/`:`/`()`, not bare
whitespace. The guard must not miss the managed directory just because it
sits inside a quoted string, a `VAR=` assignment, a `PATH=...:` join, or a
subshell — a naive whitespace-token split would sever every one of these
(and the quote/paren characters would still be glued onto the token).
"""
bin_dir = paths.bin_dir()
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
"#!/bin/sh\n"
f'exec "{bin_dir}/lean-ctx" "$@"\n'
f"# or: exec '{bin_dir}/lean-ctx'\n"
f'export PATH="{bin_dir}:$PATH"\n'
f"BIN={bin_dir}/lean-ctx\n"
f"({bin_dir}/lean-ctx)\n",
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert not script.exists()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_matches_a_managed_path_when_home_contains_a_space(home, monkeypatch):
"""A ``$HOME`` with a space is real, and yields a bin dir with a literal
space inside it. Splitting a script's body on whitespace before searching
would sever the path at that space and miss it entirely.
"""
bin_dir = home / "space here" / ".headroom" / "bin"
monkeypatch.setattr(paths, "bin_dir", lambda: bin_dir)
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir}/lean-ctx "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert not script.exists()
payload = json.loads(settings.read_text())
assert "hooks" not in payload
def test_leaves_a_same_named_hook_script_in_a_different_directory_alone(home):
"""A hook entry's command must name the exact managed script in
``~/.claude/hooks`` — not merely share a basename with one. A user's own
``~/mytools/lean-ctx-rewrite.sh`` must never inherit the classification of
Headroom's ``~/.claude/hooks/lean-ctx-rewrite.sh`` just because the
filename matches — but a wrapper invocation (``bash <script>``), a quoted
command, or a redundant ``./`` segment naming the *managed* script by
absolute path must still be recognised, or the entry survives while step
2 deletes the very script it names (the same class of stale, silently
no-op hook the rtk case documents as an accepted limitation, not one to
introduce here).
A *relative* command (``.claude/hooks/lean-ctx-rewrite.sh``) must survive
even though it shares wording with the managed script's home-relative
form: a relative hook command is resolved by the harness against the
project's cwd, never against home, so it names a project-local script
this purge never inspects — treating it as a home-relative reference
would delete a different file than the one the guard just proved nothing
about.
"""
bin_dir = paths.bin_dir()
hooks_dir = home / ".claude" / "hooks"
# The managed script that gives "lean-ctx-rewrite.sh" a True verdict.
managed_script = _write(
hooks_dir / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
# The user's own script: same basename, different directory, own binary.
user_script = _write(
home / "mytools" / "lean-ctx-rewrite.sh",
'#!/bin/sh\nexec /usr/bin/lean-ctx "$@"\n',
)
relative_command = ".claude/hooks/lean-ctx-rewrite.sh"
settings = _write(
home / ".claude" / "settings.json",
json.dumps(
{
"hooks": {
"PreToolUse": [
{"hooks": [{"command": f"bash {managed_script}"}]},
{"hooks": [{"command": f'"{managed_script}"'}]},
{"hooks": [{"command": f"{hooks_dir}/./lean-ctx-rewrite.sh"}]},
{"hooks": [{"command": relative_command}]},
{"hooks": [{"command": str(user_script)}]},
]
}
}
),
)
context_tool_cleanup.purge_context_tool_artifacts()
assert user_script.exists()
payload = json.loads(settings.read_text())
commands = [
item["command"] for entry in payload["hooks"]["PreToolUse"] for item in entry["hooks"]
]
assert commands == [relative_command, str(user_script)]
def test_reports_an_unreadable_hook_script_instead_of_guessing(home):
if sys.platform.startswith("win") or os.geteuid() == 0:
pytest.skip("chmod 0o000 does not deny access on Windows or when running as root")
bin_dir = paths.bin_dir()
script = _write(
home / ".claude" / "hooks" / "lean-ctx-rewrite.sh",
f'#!/bin/sh\nexec {bin_dir / "lean-ctx"} "$@"\n',
)
settings = _write(
home / ".claude" / "settings.json",
json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"command": str(script)}]}]}}),
)
script.chmod(0o000)
try:
report = context_tool_cleanup.purge_context_tool_artifacts()
finally:
if script.exists():
script.chmod(0o644)
# Unprovable, so kept — not deleted on a guess — but named in the report.
assert script.exists()
assert any(str(script) in line for line in report)
payload = json.loads(settings.read_text())
assert payload["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == str(script)
# A run that could not decide is not the completed migration: leaving the
# stamp off is what gets this script looked at again once it is readable.
assert not (paths.bin_dir().parent / ".context-tools-purged").exists()
def test_an_unparseable_config_defers_the_migration_stamp(home):
"""A config the user must fix by hand still holds a leftover.
Stamping the migration complete here would retire the only reminder they
get, and the entry would never be cleaned once the typo is fixed.
"""
_write(home / ".claude" / "settings.json", "{not json")
report = context_tool_cleanup.purge_context_tool_artifacts()
assert any(line.startswith("skipped ") for line in report)
assert not (paths.bin_dir().parent / ".context-tools-purged").exists()
def test_leaves_an_mcp_entry_without_a_command_alone(home):
config = _write(
home / ".claude.json",
json.dumps(
{
"mcpServers": {
"lean-ctx": {"args": ["mcp"]},
"rtk": "not-a-dict",
"lean_ctx": {"command": ["lean-ctx", "mcp"]},
}
}
),
)
context_tool_cleanup.purge_context_tool_artifacts()
payload = json.loads(config.read_text())
assert set(payload["mcpServers"]) == {"lean-ctx", "rtk", "lean_ctx"}
+24
View File
@@ -75,6 +75,30 @@ def test_build_manifest_python_runtime_keeps_explicit_memory_db_path() -> None:
assert "--memory-db-path" in manifest.proxy_args
def test_build_manifest_falls_back_from_windows_service_to_task(monkeypatch) -> None:
monkeypatch.setattr("headroom.install.planner.sys.platform", "win32")
manifest = build_manifest(
profile="default",
preset=InstallPreset.PERSISTENT_SERVICE.value,
runtime_kind="python",
scope="user",
provider_mode="manual",
targets=["claude"],
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
proxy_mode="token",
memory_enabled=False,
telemetry_enabled=False,
image="ghcr.io/headroomlabs-ai/headroom:latest",
)
assert manifest.preset == InstallPreset.PERSISTENT_TASK.value
assert manifest.supervisor_kind == "task"
def test_build_manifest_uses_provider_slice_env_builders_for_all_supported_targets() -> None:
manifest = build_manifest(
profile="default",