akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(cli): G1 remediation — non-string clobber, per-model systemMessage, openhands gate

Browse source 
Addresses 1 High + 4 Medium findings from the PR-G1 code review.

H1: `_inject_continue_rtk_systemmessage` previously fell through to an
unconditional `data["systemMessage"] = RTK_INSTRUCTIONS_BLOCK` when the
existing value was non-string (dict / list / number), silently clobbering
user data despite a docstring promising otherwise. Extracted a small helper
`_apply_rtk_to_systemmessage_field` that returns `(changed, ok)` and refuses
loudly on non-string user data with guidance to clear the field before
re-running. The injecting helper reports `ok=False` on any refusal so the
caller surfaces it as a warning instead of pretending the injection
succeeded. Tests cover dict, list, and int values for both top-level and
per-model sites.

M2: Continue overrides top-level `systemMessage` with per-model
`systemMessage` when set, so users with per-model configs were silently
getting no RTK guidance. The helper now visits every `models[i]` dict in
addition to the top-level field, applying the same idempotency and non-
string-clobber rules at each site. Non-dict entries in `models[]` are
skipped.

M3: The openhands subcommand previously called `_ensure_rtk_binary()` and
ignored the result, then proceeded to inject `OPENHANDS_INSTRUCTIONS` even
when rtk install had failed. Mirrored the cline/continue/goose pattern —
if rtk install fails (and `--no-context-tool` was not passed), exit 1 with
a clear error explaining how to install rtk manually or skip rtk. No
silent fallback to env-only injection.

M4: Wrapped the marker-injection + rtk-setup prelude of all four new
subcommands (cline, continue, goose, openhands) in a try/except for
KeyboardInterrupt. On Ctrl-C between marker injection and proxy startup,
we emit a clear "wrap was interrupted; marker file at <path> is on disk;
rerun to retry — it's idempotent" message and exit 130. Pre-compute the
marker path so the message can name it even if the interrupt fires before
`_inject_rtk_instructions` returns. Introduces a small `_emit_wrap_
interrupted` helper.

M1 + M5: Documented the uninstall procedure (hand-remove the
`<!-- headroom:rtk-instructions -->` block) and the lean-ctx agent-name
caveat in each of the four new subcommand docstrings. We chose docstring
guidance over `unwrap cline|continue|goose|openhands` subcommands to keep
the PR scoped. Also documented Continue's modern YAML-first config in the
`continue` docstring so users on the YAML schema know this command only
handles the JSON variant.

Tests: +9 new tests across the 4 wrap test files exercising H1 refusal
(dict/list/int parametrized × top-level + per-model), M2 per-model
injection + idempotency + non-dict-entry skip, M3 rtk install failure
abort + `--no-context-tool` bypass, and M4 KeyboardInterrupt-during-
prelude flows for all four agents.

Cosmetic: Removed the misleading "re-invocation in the same shell session"
comment from openhands; the marker guard is for pre-existing env vars.
This commit is contained in:
chopratejas 2026-05-25 11:54:06 -07:00
parent c375fa156d
commit ea1976e37a
5 changed files with 577 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

314
headroom/cli/wrap.py
View file

@ -890,6 +890,26 @@ def _restore_codex_provider_config() -> tuple[str, Path]:
return "noop", config_file
def _emit_wrap_interrupted(agent: str, marker_path: Path | None) -> None:
"""Log a clear interruption message after a partial wrap setup.
Called when a wrap subcommand catches ``KeyboardInterrupt`` between marker
injection and proxy startup. The marker file (if any) is left on disk —
re-running the same ``headroom wrap <agent>`` command is idempotent and
safe.
"""
if marker_path is not None:
click.echo(
f"\n Wrap was interrupted; marker file at {marker_path} is on "
f"disk. Rerun `headroom wrap {agent}` to retry — it's idempotent."
)
else:
click.echo(
f"\n Wrap was interrupted before any on-disk changes. Rerun "
f"`headroom wrap {agent}` to retry — it's idempotent."
)
def _inject_rtk_instructions(file_path: Path, verbose: bool = False) -> bool:
"""Inject rtk instructions into a file (AGENTS.md, .cursorrules, etc.).
@ -994,19 +1014,79 @@ def _inject_memory_agents_md(file_path: Path) -> bool:
return True
def _apply_rtk_to_systemmessage_field(
container: dict[str, Any],
location_label: str,
verbose: bool = False,
) -> tuple[bool, bool]:
"""Apply the RTK block to ``container["systemMessage"]`` in place.
Returns ``(changed, ok)``:
* ``changed`` is ``True`` if the field was written (or rewritten) on this
call. ``False`` for idempotent skips and for refusals.
* ``ok`` is ``True`` for "RTK guidance is now present (or refused-safely)",
``False`` only for refusals where the user must intervene. Callers
surface ``not ok`` as a warning to the user.
Refusal cases (loud, no silent overwrite):
* ``systemMessage`` exists and is **not a string** (dict / list / number).
We never clobber user data of an unknown shape. The user must remove or
clear the field before re-running.
"""
existing_msg = container.get("systemMessage")
if isinstance(existing_msg, str) and _RTK_MARKER in existing_msg:
if verbose:
click.echo(f" rtk instructions already in {location_label}")
return False, True
if existing_msg is None or (isinstance(existing_msg, str) and not existing_msg.strip()):
container["systemMessage"] = RTK_INSTRUCTIONS_BLOCK
return True, True
if isinstance(existing_msg, str):
container["systemMessage"] = existing_msg.rstrip() + "\n\n" + RTK_INSTRUCTIONS_BLOCK
return True, True
# Non-string, non-null value present — refuse loudly. We will not clobber
# user data of unknown shape.
click.echo(
f" Warning: {location_label} systemMessage is not a string "
f"(type={type(existing_msg).__name__}); refusing to overwrite. "
"To opt in, remove or clear the existing systemMessage value and re-run."
)
return False, False
def _inject_continue_rtk_systemmessage(config_file: Path, verbose: bool = False) -> bool:
"""Inject the rtk instructions block into Continue's ``.continue/config.json``.
Continue's schema supports a top-level ``systemMessage`` string applied to
every model. We treat the RTK marker as the idempotency token: if a prior
``systemMessage`` already contains the ``<!-- headroom:rtk-instructions -->``
marker we leave it alone. Otherwise we either set the field (if absent) or
append the rtk block to the existing string with a separator.
Continue's schema supports both a top-level ``systemMessage`` string and a
per-model ``systemMessage`` on each entry in the ``models`` array. The
per-model value, when set, overrides the top-level one — so users with
per-model configs would otherwise silently get no RTK guidance. This
helper writes the RTK block into **every** ``systemMessage`` site:
* top-level ``systemMessage``
* each ``models[i].systemMessage`` where ``models[i]`` is a dict
The RTK marker (``<!-- headroom:rtk-instructions -->``) is the idempotency
token: if a prior ``systemMessage`` already contains the marker we leave
that site alone. If the existing value is a non-empty string we append
with a separator. If the existing value is **non-string** (dict / list /
number) we refuse loudly and leave it untouched — we do not clobber user
data of unknown shape. To opt in to overwrite, the user must clear the
existing value first.
The config file is read/written as JSON. Malformed JSON is left untouched
and the helper returns ``False`` — we do not silently overwrite user data.
Returns ``True`` if the instructions were successfully written or already
present.
and the helper returns ``False``. Note: Continue's modern config is
YAML-first; users on the YAML schema should configure systemMessage
through that file instead — this helper only handles the JSON variant.
Returns ``True`` if injection succeeded (or was already idempotent at
every site); ``False`` if any site refused or the file was malformed.
"""
if config_file.exists():
try:
@ -1035,21 +1115,41 @@ def _inject_continue_rtk_systemmessage(config_file: Path, verbose: bool = False)
else:
data = {}
existing_msg = data.get("systemMessage")
if isinstance(existing_msg, str) and _RTK_MARKER in existing_msg:
if verbose:
click.echo(f" rtk instructions already in {config_file.name}")
return True
any_changed = False
all_ok = True
if isinstance(existing_msg, str) and existing_msg.strip():
data["systemMessage"] = existing_msg.rstrip() + "\n\n" + RTK_INSTRUCTIONS_BLOCK
else:
data["systemMessage"] = RTK_INSTRUCTIONS_BLOCK
# 1. Top-level systemMessage.
changed, ok = _apply_rtk_to_systemmessage_field(
data, location_label=f"{config_file.name} (top-level)", verbose=verbose
)
any_changed = any_changed or changed
all_ok = all_ok and ok
config_file.parent.mkdir(parents=True, exist_ok=True)
config_file.write_text(json.dumps(data, indent=2) + "\n")
click.echo(f" rtk instructions injected into {config_file}")
return True
# 2. Per-model systemMessage. Continue's models[] entry overrides the
# top-level value when set, so we must visit each one.
models = data.get("models")
if isinstance(models, list):
for idx, model in enumerate(models):
if not isinstance(model, dict):
continue
label = f"{config_file.name} models[{idx}]"
if isinstance(model.get("title"), str):
label = f"{config_file.name} models[{idx}] ({model['title']})"
changed_i, ok_i = _apply_rtk_to_systemmessage_field(
model, location_label=label, verbose=verbose
)
any_changed = any_changed or changed_i
all_ok = all_ok and ok_i
if any_changed:
config_file.parent.mkdir(parents=True, exist_ok=True)
config_file.write_text(json.dumps(data, indent=2) + "\n")
click.echo(f" rtk instructions injected into {config_file}")
elif all_ok and verbose:
# Idempotent re-run with no refusals — nothing to do.
click.echo(f" rtk instructions already present in {config_file.name}")
return all_ok
def _resolve_copilot_provider_type(backend: str | None, provider_type: str) -> str:
@ -2727,22 +2827,40 @@ def cline(
After running this command, open Cline's settings in VS Code and configure
the API Base URL to point at the local Headroom proxy.
\b
Uninstall: there is no ``headroom unwrap cline`` subcommand. To remove the
injected guidance, hand-edit ``.clinerules`` at the project root and
delete everything between ``<!-- headroom:rtk-instructions -->`` and
``<!-- /headroom:rtk-instructions -->`` (inclusive). If ``lean-ctx`` mode
is selected, the lean-ctx agent name ``cline`` may not be recognized by
the local lean-ctx binary; a warning is printed in that case and setup
is skipped silently.
\b
Examples:
headroom wrap cline # Start proxy + .clinerules instructions
headroom wrap cline --no-context-tool # Proxy only, no CLI context tool
headroom wrap cline --port 9999 # Custom proxy port
"""
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Cline...")
_setup_lean_ctx_agent("cline", verbose=verbose)
else:
click.echo(" Setting up rtk for Cline...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path:
clinerules = Path.cwd() / ".clinerules"
_inject_rtk_instructions(clinerules, verbose=verbose)
# Pre-compute the marker path so the KeyboardInterrupt handler can report
# its location even if the interrupt fires before _inject_rtk_instructions
# returns (e.g., during the inner _ensure_rtk_binary download).
clinerules: Path | None = Path.cwd() / ".clinerules" if not no_rtk else None
try:
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Cline...")
_setup_lean_ctx_agent("cline", verbose=verbose)
else:
click.echo(" Setting up rtk for Cline...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path and clinerules is not None:
_inject_rtk_instructions(clinerules, verbose=verbose)
except KeyboardInterrupt:
_emit_wrap_interrupted(
"cline", clinerules if (clinerules and clinerules.exists()) else None
)
raise SystemExit(130) from None
if prepare_only:
return
@ -2850,6 +2968,29 @@ def continue_dev(
in config.json (or via the IDE UI), not via environment variables. The
config file is overridable via --config.
\b
Note: Continue's modern config is YAML-first (``.continue/config.yaml``).
This helper only writes the JSON variant. Users on the YAML schema should
configure ``systemMessage`` through that file by hand.
\b
Per-model handling: Continue overrides top-level ``systemMessage`` with
per-model ``systemMessage`` when set, so this command also injects into
each ``models[i].systemMessage`` if the ``models`` array is present.
Existing non-string ``systemMessage`` values are NEVER overwritten — the
command warns loudly and leaves them in place. To opt in, clear the
existing value first.
\b
Uninstall: there is no ``headroom unwrap continue`` subcommand. To remove
the injected guidance, hand-edit ``.continue/config.json`` and delete
everything between ``<!-- headroom:rtk-instructions -->`` and
``<!-- /headroom:rtk-instructions -->`` (inclusive) from every
``systemMessage`` field — both top-level and inside ``models[*]``. If
``lean-ctx`` mode is selected, the lean-ctx agent name ``continue`` may
not be recognized by the local lean-ctx binary; a warning is printed in
that case and setup is skipped silently.
\b
Examples:
headroom wrap continue # Start proxy + inject systemMessage
@ -2859,15 +3000,19 @@ def continue_dev(
"""
config_file = config_path or (Path.cwd() / ".continue" / "config.json")
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Continue...")
_setup_lean_ctx_agent("continue", verbose=verbose)
else:
click.echo(" Setting up rtk for Continue...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path:
_inject_continue_rtk_systemmessage(config_file, verbose=verbose)
try:
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Continue...")
_setup_lean_ctx_agent("continue", verbose=verbose)
else:
click.echo(" Setting up rtk for Continue...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path:
_inject_continue_rtk_systemmessage(config_file, verbose=verbose)
except KeyboardInterrupt:
_emit_wrap_interrupted("continue", config_file if config_file.exists() else None)
raise SystemExit(130) from None
if prepare_only:
return
@ -2977,6 +3122,15 @@ def goose(
guidance into .goosehints at the project root (Goose reads this file as
extra system context).
\b
Uninstall: there is no ``headroom unwrap goose`` subcommand. To remove the
injected guidance, hand-edit ``.goosehints`` at the project root and
delete everything between ``<!-- headroom:rtk-instructions -->`` and
``<!-- /headroom:rtk-instructions -->`` (inclusive). If ``lean-ctx`` mode
is selected, the lean-ctx agent name ``goose`` may not be recognized by
the local lean-ctx binary; a warning is printed in that case and setup
is skipped silently.
\b
Examples:
headroom wrap goose # Start proxy + context tool + goose
@ -2984,17 +3138,27 @@ def goose(
headroom wrap goose -- --provider anthropic # Pass args to goose
headroom wrap goose --no-context-tool # Skip CLI context-tool setup
"""
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Goose...")
_setup_lean_ctx_agent("goose", verbose=verbose)
else:
click.echo(" Setting up rtk for Goose...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path:
# Goose reads .goosehints from the project root as extra context.
goosehints = Path.cwd() / ".goosehints"
_inject_rtk_instructions(goosehints, verbose=verbose)
# Pre-compute the marker path so the KeyboardInterrupt handler can report
# its location even if the interrupt fires before _inject_rtk_instructions
# returns (e.g., during the inner _ensure_rtk_binary download).
goosehints: Path | None = Path.cwd() / ".goosehints" if not no_rtk else None
try:
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for Goose...")
_setup_lean_ctx_agent("goose", verbose=verbose)
else:
click.echo(" Setting up rtk for Goose...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if rtk_path and goosehints is not None:
# Goose reads .goosehints from the project root as extra
# context.
_inject_rtk_instructions(goosehints, verbose=verbose)
except KeyboardInterrupt:
_emit_wrap_interrupted(
"goose", goosehints if (goosehints and goosehints.exists()) else None
)
raise SystemExit(130) from None
if prepare_only:
return
@ -3087,19 +3251,42 @@ def openhands(
``OPENHANDS_INSTRUCTIONS`` environment variable at launch time so the
on-disk OpenHands config is left untouched.
\b
The ``OPENHANDS_INSTRUCTIONS`` value injected by this command contains the
``<!-- headroom:rtk-instructions -->`` marker. To uninstall, simply do not
set ``OPENHANDS_INSTRUCTIONS`` in the parent shell — this command never
writes to disk, so nothing to clean up. If ``lean-ctx`` mode is selected,
the lean-ctx agent name ``openhands`` may not be recognized by the local
lean-ctx binary; a warning is printed in that case and rtk-style guidance
falls through.
\b
Examples:
headroom wrap openhands # Start proxy + context tool + openhands
headroom wrap openhands -- --task ... # Pass args to openhands
headroom wrap openhands --no-context-tool
"""
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for OpenHands...")
_setup_lean_ctx_agent("openhands", verbose=verbose)
else:
click.echo(" Setting up rtk for OpenHands...")
_ensure_rtk_binary(verbose=verbose)
rtk_path: Path | None = None
try:
if not no_rtk:
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" Setting up lean-ctx for OpenHands...")
_setup_lean_ctx_agent("openhands", verbose=verbose)
else:
click.echo(" Setting up rtk for OpenHands...")
rtk_path = _ensure_rtk_binary(verbose=verbose)
if not rtk_path:
click.echo(
" Error: rtk install failed; refusing to inject "
"OPENHANDS_INSTRUCTIONS without rtk. Install rtk "
"manually and re-run, or pass --no-context-tool to "
"skip rtk."
)
raise SystemExit(1)
except KeyboardInterrupt:
# openhands never writes to disk — no marker file to flag.
_emit_wrap_interrupted("openhands", None)
raise SystemExit(130) from None
if prepare_only:
return
@ -3118,14 +3305,15 @@ def openhands(
env["ANTHROPIC_BASE_URL"] = anthropic_base
# Also set LLM_BASE_URL for OpenHands' generic LLM provider config.
env["LLM_BASE_URL"] = openai_base
if not no_rtk:
if not no_rtk and rtk_path:
# Inject rtk guidance via env var so OpenHands picks it up as the
# session's instruction prefix. Appending instead of overwriting any
# pre-existing OPENHANDS_INSTRUCTIONS so user-supplied instructions are
# preserved.
# session's instruction prefix. Appending instead of overwriting
# any pre-existing OPENHANDS_INSTRUCTIONS so user-supplied content
# is preserved. The marker check guards against double-injection
# when the user inherits an env var that already has the rtk block.
existing_instructions = env.get("OPENHANDS_INSTRUCTIONS", "")
if _RTK_MARKER in existing_instructions:
# Already injected (re-invocation in the same shell session).
# Already injected — pre-existing env var contains marker.
pass
elif existing_instructions.strip():
env["OPENHANDS_INSTRUCTIONS"] = (

34
tests/test_cli/test_wrap_cline.py
View file

@ -91,3 +91,37 @@ def test_wrap_cline_preserves_existing_clinerules_content(
content = clinerules.read_text()
assert "Always use Python 3.12." in content
assert wrap_mod._RTK_MARKER in content
# ---------------------------------------------------------------------------
# M4: Ctrl-C during prelude emits a clear "interrupted, marker may be on disk"
# message and exits non-zero.
# ---------------------------------------------------------------------------
def test_wrap_cline_keyboardinterrupt_during_prelude_emits_clear_message(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ctrl-C between marker injection and proxy startup must signal clearly."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
def raise_kbd_interrupt(*args, **kwargs): # noqa: ANN002, ANN003
# Simulate the user hitting Ctrl-C right after the prelude wrote the
# .clinerules marker but before _ensure_proxy returns. We trigger via
# _ensure_rtk_binary side-effect so the marker file exists on disk.
marker_path = tmp_path / ".clinerules"
marker_path.write_text(wrap_mod.RTK_INSTRUCTIONS_BLOCK)
raise KeyboardInterrupt
with patch.object(wrap_mod, "_ensure_rtk_binary", side_effect=raise_kbd_interrupt):
result = runner.invoke(main, ["wrap", "cline", "--prepare-only"])
assert result.exit_code == 130
assert "interrupted" in result.output.lower()
assert "idempotent" in result.output.lower()
# marker file is on disk
assert (tmp_path / ".clinerules").exists()
assert str(tmp_path / ".clinerules") in result.output

180
tests/test_cli/test_wrap_continue.py
View file

@ -30,7 +30,7 @@ def test_inject_continue_rtk_systemmessage_new_file(tmp_path: Path) -> None:
def test_inject_continue_rtk_systemmessage_preserves_existing_keys(tmp_path: Path) -> None:
"""Pre-existing keys like ``models`` are not touched."""
"""Pre-existing keys are not touched; per-model entries get systemMessage."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text(json.dumps({"models": [{"title": "GPT-4o", "provider": "openai"}]}))
@ -38,8 +38,14 @@ def test_inject_continue_rtk_systemmessage_preserves_existing_keys(tmp_path: Pat
wrap_mod._inject_continue_rtk_systemmessage(config_file)
data = json.loads(config_file.read_text())
assert data["models"] == [{"title": "GPT-4o", "provider": "openai"}]
# Pre-existing fields on the model entry are preserved verbatim.
assert data["models"][0]["title"] == "GPT-4o"
assert data["models"][0]["provider"] == "openai"
# Top-level systemMessage is set.
assert wrap_mod._RTK_MARKER in data["systemMessage"]
# Per-model systemMessage is also populated (Continue overrides top-level
# with per-model when set, so we must visit each model).
assert wrap_mod._RTK_MARKER in data["models"][0]["systemMessage"]
def test_inject_continue_rtk_systemmessage_appends_to_existing_message(
@ -138,3 +144,173 @@ def test_wrap_continue_respects_custom_config_path(
assert not (tmp_path / ".continue" / "config.json").exists()
data = json.loads(custom_config.read_text())
assert wrap_mod._RTK_MARKER in data["systemMessage"]
# ---------------------------------------------------------------------------
# H1: non-string systemMessage must NOT be silently clobbered.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"non_string_value",
[
{"role": "system", "content": "You are helpful."}, # dict
["You are helpful.", "Respond in JSON."], # list
42, # int
],
ids=["dict", "list", "int"],
)
def test_inject_continue_rtk_systemmessage_refuses_non_string_top_level(
tmp_path: Path,
non_string_value: object,
) -> None:
"""A non-string top-level systemMessage must NOT be overwritten."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
original_payload = {"systemMessage": non_string_value, "other": "untouched"}
config_file.write_text(json.dumps(original_payload))
original_bytes = config_file.read_bytes()
result = wrap_mod._inject_continue_rtk_systemmessage(config_file)
assert result is False, "must report refusal when user data would be clobbered"
# File must be byte-identical to before.
assert config_file.read_bytes() == original_bytes
data = json.loads(config_file.read_text())
assert data["systemMessage"] == non_string_value
assert data["other"] == "untouched"
@pytest.mark.parametrize(
"non_string_value",
[
{"role": "system", "content": "Per-model system."},
["List", "of", "strings"],
7,
],
ids=["dict", "list", "int"],
)
def test_inject_continue_rtk_systemmessage_refuses_non_string_per_model(
tmp_path: Path,
non_string_value: object,
) -> None:
"""A non-string per-model systemMessage must NOT be overwritten."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
original_payload = {
"models": [
{"title": "GPT-4o", "provider": "openai", "systemMessage": non_string_value},
],
}
config_file.write_text(json.dumps(original_payload))
result = wrap_mod._inject_continue_rtk_systemmessage(config_file)
assert result is False, "must report refusal when per-model user data would be clobbered"
data = json.loads(config_file.read_text())
# The non-string per-model value must be preserved.
assert data["models"][0]["systemMessage"] == non_string_value
# ---------------------------------------------------------------------------
# M2: per-model systemMessage handling.
# ---------------------------------------------------------------------------
def test_inject_continue_rtk_systemmessage_visits_each_model(
tmp_path: Path,
) -> None:
"""Each models[i].systemMessage gets the RTK block."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text(
json.dumps(
{
"models": [
{"title": "A", "systemMessage": "user value"},
{"title": "B"}, # no systemMessage yet
],
}
)
)
assert wrap_mod._inject_continue_rtk_systemmessage(config_file) is True
data = json.loads(config_file.read_text())
# Pre-existing per-model systemMessage is preserved + RTK block appended.
assert "user value" in data["models"][0]["systemMessage"]
assert wrap_mod._RTK_MARKER in data["models"][0]["systemMessage"]
# Model with no systemMessage gets the RTK block fresh.
assert wrap_mod._RTK_MARKER in data["models"][1]["systemMessage"]
# Top-level also populated.
assert wrap_mod._RTK_MARKER in data["systemMessage"]
def test_inject_continue_rtk_systemmessage_per_model_idempotent(
tmp_path: Path,
) -> None:
"""Re-running must not duplicate per-model RTK blocks."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text(
json.dumps({"models": [{"title": "A", "systemMessage": "user"}]}),
)
wrap_mod._inject_continue_rtk_systemmessage(config_file)
wrap_mod._inject_continue_rtk_systemmessage(config_file)
data = json.loads(config_file.read_text())
assert data["models"][0]["systemMessage"].count(wrap_mod._RTK_MARKER) == 1
assert data["systemMessage"].count(wrap_mod._RTK_MARKER) == 1
def test_inject_continue_rtk_systemmessage_skips_non_dict_model_entries(
tmp_path: Path,
) -> None:
"""A models[] entry that isn't a dict must be left untouched."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text(
json.dumps({"models": ["just a string entry", {"title": "B"}]}),
)
wrap_mod._inject_continue_rtk_systemmessage(config_file)
data = json.loads(config_file.read_text())
# Non-dict entry preserved verbatim.
assert data["models"][0] == "just a string entry"
# Dict entry got the RTK block.
assert wrap_mod._RTK_MARKER in data["models"][1]["systemMessage"]
# ---------------------------------------------------------------------------
# M4: Ctrl-C during prelude emits a clear message.
# ---------------------------------------------------------------------------
def test_wrap_continue_keyboardinterrupt_during_prelude_emits_clear_message(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ctrl-C between marker injection and proxy start must signal clearly."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
config_file = tmp_path / ".continue" / "config.json"
def raise_kbd_after_inject(*args, **kwargs): # noqa: ANN002, ANN003
# Simulate the user hitting Ctrl-C right after we wrote config.json
# but before the proxy started.
config_file.parent.mkdir(parents=True, exist_ok=True)
config_file.write_text('{"systemMessage": "marker block"}')
raise KeyboardInterrupt
with patch.object(wrap_mod, "_ensure_rtk_binary", side_effect=raise_kbd_after_inject):
result = runner.invoke(main, ["wrap", "continue", "--prepare-only"])
assert result.exit_code == 130
assert "interrupted" in result.output.lower()
assert "idempotent" in result.output.lower()
assert config_file.exists()
assert str(config_file) in result.output

31
tests/test_cli/test_wrap_goose.py
View file

@ -115,3 +115,34 @@ def test_wrap_goose_no_context_tool_skips_goosehints(
assert result.exit_code == 0, result.output
assert not (tmp_path / ".goosehints").exists()
ensure.assert_not_called()
# ---------------------------------------------------------------------------
# M4: Ctrl-C during prelude.
# ---------------------------------------------------------------------------
def test_wrap_goose_keyboardinterrupt_during_prelude_emits_clear_message(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ctrl-C between marker injection and proxy start must signal clearly."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
def raise_kbd(*args, **kwargs): # noqa: ANN002, ANN003
# Simulate Ctrl-C after the prelude wrote .goosehints but before the
# tool was launched. The marker file exists on disk.
marker_path = tmp_path / ".goosehints"
marker_path.write_text(wrap_mod.RTK_INSTRUCTIONS_BLOCK)
raise KeyboardInterrupt
with patch.object(wrap_mod, "_ensure_rtk_binary", side_effect=raise_kbd):
result = runner.invoke(main, ["wrap", "goose", "--prepare-only"])
assert result.exit_code == 130
assert "interrupted" in result.output.lower()
assert "idempotent" in result.output.lower()
assert (tmp_path / ".goosehints").exists()
assert ".goosehints" in result.output

83
tests/test_cli/test_wrap_openhands.py
View file

@ -173,3 +173,86 @@ def test_wrap_openhands_no_context_tool_does_not_inject(
env = captured["env"]
assert isinstance(env, dict)
assert "OPENHANDS_INSTRUCTIONS" not in env or env["OPENHANDS_INSTRUCTIONS"] == ""
# ---------------------------------------------------------------------------
# M3: rtk install failure must fail loudly — no silent fallback to env
# injection without rtk on disk.
# ---------------------------------------------------------------------------
def test_wrap_openhands_rtk_install_failure_aborts_loudly(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If rtk install fails, command must exit non-zero with a clear error."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
monkeypatch.delenv("OPENHANDS_INSTRUCTIONS", raising=False)
launch_called: list[bool] = []
def fake_launch_tool(**kwargs): # noqa: ANN003
launch_called.append(True)
with patch.object(wrap_mod.shutil, "which", return_value="openhands"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=None):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 1
assert "rtk install failed" in result.output
assert "--no-context-tool" in result.output
# _launch_tool must NOT have been invoked when rtk install fails.
assert launch_called == []
def test_wrap_openhands_rtk_install_failure_with_no_context_tool_still_launches(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--no-context-tool bypasses rtk entirely — should still launch."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("OPENHANDS_INSTRUCTIONS", raising=False)
captured: dict[str, object] = {}
def fake_launch_tool(**kwargs): # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="openhands"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=None) as ensure:
result = runner.invoke(main, ["wrap", "openhands", "--no-context-tool"])
assert result.exit_code == 0, result.output
# rtk should never have been queried.
ensure.assert_not_called()
env = captured["env"]
assert "OPENHANDS_INSTRUCTIONS" not in env or env["OPENHANDS_INSTRUCTIONS"] == ""
# ---------------------------------------------------------------------------
# M4: Ctrl-C during prelude emits a clear "no on-disk changes" message.
# openhands never writes to disk (env-var injection only).
# ---------------------------------------------------------------------------
def test_wrap_openhands_keyboardinterrupt_during_prelude_emits_clear_message(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ctrl-C during the prelude must signal cleanly with no on-disk artifact."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
monkeypatch.delenv("OPENHANDS_INSTRUCTIONS", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", side_effect=KeyboardInterrupt):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 130
assert "interrupted" in result.output.lower()
assert "idempotent" in result.output.lower()