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): wrap subcommands for cline, continue, goose, openhands

Browse source 
Adds four new `headroom wrap <agent>` subcommands so the proxy can
front-end Cline (VS Code), Continue (VS Code/JetBrains), Goose (Block CLI),
and OpenHands (CLI), extending the existing claude/codex/aider/copilot/
cursor pattern. Phase G PR-G1 of the realignment work.

Architectural decision: extended the existing `headroom/cli/wrap.py`
module in-place rather than splitting it into a `headroom/cli/wrap/`
package. The spec at REALIGNMENT/09-phase-G-rtk-observability.md
mentions per-agent files under a package, but the existing five wrap
subcommands all live in the single module and the extension is small
relative to the file. Keeping the file together preserves the simple
import surface used by tests (`from headroom.cli import wrap as wrap_mod`).

Per-agent wiring:
- cline → injects RTK block into `.clinerules` at project root
  (Cline is a VS Code extension; API base URL is configured in the UI,
  so the command prints config instructions and blocks on the proxy).
- continue → injects RTK guidance into the `systemMessage` field of
  `.continue/config.json` (idempotent; refuses malformed JSON or
  non-object roots; supports `--config` for custom paths).
- goose → injects RTK block into `.goosehints` at project root and
  launches the goose CLI with OPENAI_BASE_URL / OPENAI_API_BASE /
  ANTHROPIC_BASE_URL env vars pointed at the proxy.
- openhands → injects RTK guidance via the `OPENHANDS_INSTRUCTIONS`
  env var at launch (no on-disk artifact) and sets OPENAI_BASE_URL /
  ANTHROPIC_BASE_URL / LLM_BASE_URL. Preserves any pre-existing
  OPENHANDS_INSTRUCTIONS content.

Tests:
- tests/test_cli/test_wrap_cline.py: 4 tests covering prepare-only
  injection, idempotence, --no-context-tool, and existing content
  preservation.
- tests/test_cli/test_wrap_continue.py: 8 tests covering the new
  `_inject_continue_rtk_systemmessage` helper (new-file, existing
  keys, idempotence, malformed JSON, non-object roots) and the click
  command surface (default path, custom --config).
- tests/test_cli/test_wrap_goose.py: 5 tests covering env-var wiring,
  `.goosehints` injection, idempotence, missing-binary error, and
  --no-context-tool.
- tests/test_cli/test_wrap_openhands.py: 6 tests covering env-var
  wiring, OPENHANDS_INSTRUCTIONS injection, preservation of existing
  instructions, idempotence, missing-binary error, and
  --no-context-tool.

E2E: extended `e2e/wrap/run.py` with `--prepare-only` smoke tests for
all four new wrappers (full launches require agent CLIs not present in
the e2e image; unit tests cover the env-var wiring).
This commit is contained in:
chopratejas 2026-05-21 21:04:50 -07:00
parent 9536d6b48c
commit c375fa156d
6 changed files with 1139 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

77
e2e/wrap/run.py
View file

@ -26,6 +26,14 @@ CODEX_PORT = 28888
AIDER_PORT = 28889
CURSOR_PORT = 28890
OPENCLAW_PROXY_PORT = 28891
# Phase G PR-G1: new wrap subcommands. Smoke-tested via --prepare-only since
# their CLIs may not exist on the e2e image and the wrap commands without
# --prepare-only block on the proxy. The wiring is otherwise covered by the
# unit tests in tests/test_cli/test_wrap_{cline,continue,goose,openhands}.py.
CLINE_PORT = 28892
CONTINUE_PORT = 28893
GOOSE_PORT = 28894
OPENHANDS_PORT = 28895
def log(message: str) -> None:
@ -697,6 +705,71 @@ def verify_cursor_wrap(base_env: dict[str, str], project_dir: Path) -> None:
stop_process(proc)
def verify_cline_wrap(base_env: dict[str, str], project_dir: Path) -> None:
"""Smoke test: `wrap cline --prepare-only` writes RTK guidance to .clinerules."""
run(
["headroom", "wrap", "cline", "--prepare-only", "--port", str(CLINE_PORT)],
env=base_env,
cwd=project_dir,
timeout=60,
)
clinerules = project_dir / ".clinerules"
assert_true(clinerules.exists(), "Cline wrap should create .clinerules")
assert_true(
RTK_MARKER in clinerules.read_text(encoding="utf-8"),
"Cline wrap should inject RTK instructions",
)
def verify_continue_wrap(base_env: dict[str, str], project_dir: Path) -> None:
"""Smoke test: `wrap continue --prepare-only` injects RTK into .continue/config.json."""
run(
["headroom", "wrap", "continue", "--prepare-only", "--port", str(CONTINUE_PORT)],
env=base_env,
cwd=project_dir,
timeout=60,
)
config_file = project_dir / ".continue" / "config.json"
assert_true(config_file.exists(), "Continue wrap should create .continue/config.json")
data = json.loads(config_file.read_text(encoding="utf-8"))
system_message = data.get("systemMessage", "")
assert_true(
RTK_MARKER in system_message,
"Continue wrap should inject RTK instructions into systemMessage",
)
def verify_goose_wrap(base_env: dict[str, str], project_dir: Path) -> None:
"""Smoke test: `wrap goose --prepare-only` writes RTK guidance to .goosehints."""
run(
["headroom", "wrap", "goose", "--prepare-only", "--port", str(GOOSE_PORT)],
env=base_env,
cwd=project_dir,
timeout=60,
)
goosehints = project_dir / ".goosehints"
assert_true(goosehints.exists(), "Goose wrap should create .goosehints")
assert_true(
RTK_MARKER in goosehints.read_text(encoding="utf-8"),
"Goose wrap should inject RTK instructions",
)
def verify_openhands_wrap(base_env: dict[str, str], project_dir: Path) -> None:
"""Smoke test: `wrap openhands --prepare-only` exits clean and ensures rtk is present.
OpenHands wires instructions via the OPENHANDS_INSTRUCTIONS env var at launch
time (no on-disk artifact), so --prepare-only just exercises the rtk-binary
setup path. The env-var wiring is covered by the unit tests.
"""
run(
["headroom", "wrap", "openhands", "--prepare-only", "--port", str(OPENHANDS_PORT)],
env=base_env,
cwd=project_dir,
timeout=60,
)
def verify_openclaw_wrap(
base_env: dict[str, str],
project_dir: Path,
@ -827,6 +900,10 @@ def main() -> None:
verify_codex_wrap(base_env, project_dir, log_dir, mock_server)
verify_aider_wrap(base_env, project_dir, log_dir)
verify_cursor_wrap(base_env, project_dir)
verify_cline_wrap(base_env, project_dir)
verify_continue_wrap(base_env, project_dir)
verify_goose_wrap(base_env, project_dir)
verify_openhands_wrap(base_env, project_dir)
local_plugin_dir = prepare_local_openclaw_plugin(base_env, tmp_dir)
verify_openclaw_wrap(base_env, project_dir, local_plugin_dir)
finally:

540
headroom/cli/wrap.py
View file

@ -994,6 +994,64 @@ def _inject_memory_agents_md(file_path: Path) -> bool:
return True
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.
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.
"""
if config_file.exists():
try:
content = config_file.read_text()
except OSError as exc:
click.echo(f" Warning: could not read {config_file}: {exc}")
return False
if not content.strip():
data: dict[str, Any] = {}
else:
try:
parsed = json.loads(content)
except json.JSONDecodeError as exc:
click.echo(
f" Warning: {config_file} is not valid JSON ({exc.msg}); "
"not modifying — fix the file manually before re-running."
)
return False
if not isinstance(parsed, dict):
click.echo(
f" Warning: {config_file} top-level value is not an object; "
"Continue expects a JSON object — leaving file untouched."
)
return False
data = parsed
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
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
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
def _resolve_copilot_provider_type(backend: str | None, provider_type: str) -> str:
"""Resolve Copilot BYOK provider type for the current proxy backend."""
return _copilot_resolve_provider_type(backend, provider_type)
@ -1752,6 +1810,10 @@ def wrap() -> None:
headroom wrap copilot -- --model claude-sonnet-4-20250514
headroom wrap aider # Aider
headroom wrap cursor # Cursor (prints config instructions)
headroom wrap cline # Cline (VS Code; prints config instructions)
headroom wrap continue # Continue (VS Code/JetBrains; injects systemMessage)
headroom wrap goose # Goose (Block) CLI
headroom wrap openhands # OpenHands CLI
headroom wrap openclaw # OpenClaw plugin bootstrap
\b
@ -1760,9 +1822,7 @@ def wrap() -> None:
sets the right env vars, and launches the wrapped CLI.
- `headroom proxy` — just the proxy. Use this with any
OpenAI/Anthropic-compatible client by setting
ANTHROPIC_BASE_URL / OPENAI_BASE_URL yourself. Required for
tools without a dedicated `wrap` subcommand
(e.g. opencode, Cline, Continue).
ANTHROPIC_BASE_URL / OPENAI_BASE_URL yourself.
\b
Note: `headroom wrap opencode` does NOT exist. For opencode, run
@ -2626,6 +2686,480 @@ def cursor(
cleanup()
# =============================================================================
# Cline (VS Code extension)
# =============================================================================
@wrap.command(context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option(
"--no-context-tool",
"--no-rtk",
"no_rtk",
is_flag=True,
help="Skip CLI context-tool setup",
)
@click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)")
@click.option("--learn", is_flag=True, help="Enable live traffic learning")
@click.option("--memory", is_flag=True, help="Enable persistent cross-session memory")
@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
@click.option("--prepare-only", is_flag=True, hidden=True)
def cline(
port: int,
no_rtk: bool,
no_proxy: bool,
learn: bool,
memory: bool,
verbose: bool,
prepare_only: bool,
) -> None:
"""Start Headroom proxy for use with Cline (VS Code extension).
\b
Cline is a VS Code extension that reads its API configuration from the
VS Code settings UI, not from environment variables. This command starts
the proxy, sets up the selected CLI context tool (injecting RTK guidance
into .clinerules at the project root), and prints the Cline settings the
user should configure.
\b
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
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)
if prepare_only:
return
proxy_holder: list[subprocess.Popen | None] = [None]
cleanup = _make_cleanup(proxy_holder, port)
signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)
try:
click.echo()
click.echo(" ╔═══════════════════════════════════════════════╗")
click.echo(" ║ HEADROOM WRAP: CLINE ║")
click.echo(" ╚═══════════════════════════════════════════════╝")
click.echo()
proxy_holder[0] = _ensure_proxy(
port, no_proxy, learn=learn, memory=memory, agent_type="cline"
)
anthropic_base = _claude_proxy_base_url(port)
openai_base = f"http://127.0.0.1:{port}/v1"
click.echo()
click.echo(" Configure Cline in VS Code:")
click.echo(" Settings > Cline > API Provider")
click.echo(f" Anthropic Base URL: {anthropic_base}")
click.echo(f" OpenAI Compatible Base URL: {openai_base}")
if not no_rtk:
click.echo()
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" lean-ctx configured for Cline")
else:
click.echo(" rtk instructions injected into .clinerules")
click.echo(" Cline will use token-optimized commands automatically.")
click.echo()
click.echo(" Press Ctrl+C to stop the proxy.")
click.echo()
try:
while True:
time.sleep(1)
proc = proxy_holder[0]
if proc and proc.poll() is not None:
click.echo(" Proxy process exited unexpectedly.")
raise SystemExit(1)
except KeyboardInterrupt:
click.echo("\n Shutting down...")
except SystemExit:
raise
except Exception as e:
click.echo(f" Error: {e}")
raise SystemExit(1) from e
finally:
cleanup()
# =============================================================================
# Continue (VS Code / JetBrains extension)
# =============================================================================
@wrap.command("continue", context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option(
"--no-context-tool",
"--no-rtk",
"no_rtk",
is_flag=True,
help="Skip CLI context-tool setup",
)
@click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)")
@click.option("--learn", is_flag=True, help="Enable live traffic learning")
@click.option("--memory", is_flag=True, help="Enable persistent cross-session memory")
@click.option(
"--config",
"config_path",
type=click.Path(path_type=Path, file_okay=True, dir_okay=False),
default=None,
help="Path to Continue config.json (default: ./.continue/config.json)",
)
@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
@click.option("--prepare-only", is_flag=True, hidden=True)
def continue_dev(
port: int,
no_rtk: bool,
no_proxy: bool,
learn: bool,
memory: bool,
config_path: Path | None,
verbose: bool,
prepare_only: bool,
) -> None:
"""Start Headroom proxy for use with Continue (VS Code / JetBrains).
\b
Continue reads its model configuration from .continue/config.json (a JSON
document with a top-level ``systemMessage`` and a ``models`` array). This
command starts the proxy, sets up the selected CLI context tool by
extending ``systemMessage`` with RTK guidance, and prints the per-model
``apiBase`` the user should configure manually.
\b
Continue is an IDE extension — its API base URL is configured per-model
in config.json (or via the IDE UI), not via environment variables. The
config file is overridable via --config.
\b
Examples:
headroom wrap continue # Start proxy + inject systemMessage
headroom wrap continue --no-context-tool # Proxy only
headroom wrap continue --port 9999 # Custom proxy port
headroom wrap continue --config path/to/config.json
"""
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)
if prepare_only:
return
proxy_holder: list[subprocess.Popen | None] = [None]
cleanup = _make_cleanup(proxy_holder, port)
signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)
try:
click.echo()
click.echo(" ╔═══════════════════════════════════════════════╗")
click.echo(" ║ HEADROOM WRAP: CONTINUE ║")
click.echo(" ╚═══════════════════════════════════════════════╝")
click.echo()
proxy_holder[0] = _ensure_proxy(
port, no_proxy, learn=learn, memory=memory, agent_type="continue"
)
anthropic_base = _claude_proxy_base_url(port)
openai_base = f"http://127.0.0.1:{port}/v1"
click.echo()
click.echo(" Configure Continue in your IDE:")
click.echo(f" Edit {config_file} and set, per model:")
click.echo(f' "apiBase": "{openai_base}" # OpenAI-compatible models')
click.echo(f' "apiBase": "{anthropic_base}" # Anthropic models')
if not no_rtk:
click.echo()
if _selected_context_tool() == _CONTEXT_TOOL_LEAN_CTX:
click.echo(" lean-ctx configured for Continue")
else:
click.echo(f" rtk instructions injected into {config_file.name} systemMessage")
click.echo(" Continue will use token-optimized commands automatically.")
click.echo()
click.echo(" Press Ctrl+C to stop the proxy.")
click.echo()
try:
while True:
time.sleep(1)
proc = proxy_holder[0]
if proc and proc.poll() is not None:
click.echo(" Proxy process exited unexpectedly.")
raise SystemExit(1)
except KeyboardInterrupt:
click.echo("\n Shutting down...")
except SystemExit:
raise
except Exception as e:
click.echo(f" Error: {e}")
raise SystemExit(1) from e
finally:
cleanup()
# =============================================================================
# Goose (Block)
# =============================================================================
@wrap.command(context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option(
"--no-context-tool",
"--no-rtk",
"no_rtk",
is_flag=True,
help="Skip CLI context-tool setup",
)
@click.option(
"--code-graph",
is_flag=True,
help="Enable code graph indexing via codebase-memory-mcp (optional)",
)
@click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)")
@click.option("--learn", is_flag=True, help="Enable live traffic learning")
@click.option("--memory", is_flag=True, help="Enable persistent cross-session memory")
@click.option(
"--backend", default=None, help="API backend: 'anthropic', 'anyllm', 'litellm-vertex', etc."
)
@click.option("--anyllm-provider", default=None, help="Provider for any-llm backend")
@click.option("--region", default=None, help="Cloud region for Bedrock/Vertex")
@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
@click.option("--prepare-only", is_flag=True, hidden=True)
@click.argument("goose_args", nargs=-1, type=click.UNPROCESSED)
def goose(
port: int,
no_rtk: bool,
code_graph: bool,
no_proxy: bool,
learn: bool,
memory: bool,
backend: str | None,
anyllm_provider: str | None,
region: str | None,
verbose: bool,
prepare_only: bool,
goose_args: tuple,
) -> None:
"""Launch Goose (Block) CLI through Headroom proxy.
\b
Sets OPENAI_BASE_URL and ANTHROPIC_BASE_URL to route Goose's API calls
through Headroom. Sets up the selected CLI context tool by injecting RTK
guidance into .goosehints at the project root (Goose reads this file as
extra system context).
\b
Examples:
headroom wrap goose # Start proxy + context tool + goose
headroom wrap goose -- session # Start a Goose session
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)
if prepare_only:
return
goose_bin = shutil.which("goose")
if not goose_bin:
click.echo("Error: 'goose' not found in PATH.")
click.echo("Install Goose: https://block.github.io/goose/")
raise SystemExit(1)
# Goose accepts OpenAI- and Anthropic-compatible providers; route both.
env = os.environ.copy()
openai_base = f"http://127.0.0.1:{port}/v1"
anthropic_base = _claude_proxy_base_url(port)
env["OPENAI_BASE_URL"] = openai_base
env["OPENAI_API_BASE"] = openai_base
env["ANTHROPIC_BASE_URL"] = anthropic_base
env_vars_display = [
f"OPENAI_BASE_URL={openai_base}",
f"ANTHROPIC_BASE_URL={anthropic_base}",
]
_launch_tool(
binary=goose_bin,
args=goose_args,
env=env,
port=port,
no_proxy=no_proxy,
tool_label="GOOSE",
env_vars_display=env_vars_display,
learn=learn,
memory=memory,
agent_type="goose",
code_graph=code_graph,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
)
# =============================================================================
# OpenHands
# =============================================================================
@wrap.command(context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option(
"--no-context-tool",
"--no-rtk",
"no_rtk",
is_flag=True,
help="Skip CLI context-tool setup",
)
@click.option(
"--code-graph",
is_flag=True,
help="Enable code graph indexing via codebase-memory-mcp (optional)",
)
@click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)")
@click.option("--learn", is_flag=True, help="Enable live traffic learning")
@click.option("--memory", is_flag=True, help="Enable persistent cross-session memory")
@click.option(
"--backend", default=None, help="API backend: 'anthropic', 'anyllm', 'litellm-vertex', etc."
)
@click.option("--anyllm-provider", default=None, help="Provider for any-llm backend")
@click.option("--region", default=None, help="Cloud region for Bedrock/Vertex")
@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
@click.option("--prepare-only", is_flag=True, hidden=True)
@click.argument("openhands_args", nargs=-1, type=click.UNPROCESSED)
def openhands(
port: int,
no_rtk: bool,
code_graph: bool,
no_proxy: bool,
learn: bool,
memory: bool,
backend: str | None,
anyllm_provider: str | None,
region: str | None,
verbose: bool,
prepare_only: bool,
openhands_args: tuple,
) -> None:
"""Launch OpenHands CLI through Headroom proxy.
\b
Sets OPENAI_BASE_URL / ANTHROPIC_BASE_URL to route OpenHands' API calls
through Headroom. Instructions are injected via the
``OPENHANDS_INSTRUCTIONS`` environment variable at launch time so the
on-disk OpenHands config is left untouched.
\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)
if prepare_only:
return
openhands_bin = shutil.which("openhands")
if not openhands_bin:
click.echo("Error: 'openhands' not found in PATH.")
click.echo("Install OpenHands: https://docs.all-hands.dev/")
raise SystemExit(1)
env = os.environ.copy()
openai_base = f"http://127.0.0.1:{port}/v1"
anthropic_base = _claude_proxy_base_url(port)
env["OPENAI_BASE_URL"] = openai_base
env["OPENAI_API_BASE"] = openai_base
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:
# 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.
existing_instructions = env.get("OPENHANDS_INSTRUCTIONS", "")
if _RTK_MARKER in existing_instructions:
# Already injected (re-invocation in the same shell session).
pass
elif existing_instructions.strip():
env["OPENHANDS_INSTRUCTIONS"] = (
existing_instructions.rstrip() + "\n\n" + RTK_INSTRUCTIONS_BLOCK
)
else:
env["OPENHANDS_INSTRUCTIONS"] = RTK_INSTRUCTIONS_BLOCK
env_vars_display = [
f"OPENAI_BASE_URL={openai_base}",
f"ANTHROPIC_BASE_URL={anthropic_base}",
f"LLM_BASE_URL={openai_base}",
]
if not no_rtk and "OPENHANDS_INSTRUCTIONS" in env:
env_vars_display.append("OPENHANDS_INSTRUCTIONS=<rtk instructions injected>")
_launch_tool(
binary=openhands_bin,
args=openhands_args,
env=env,
port=port,
no_proxy=no_proxy,
tool_label="OPENHANDS",
env_vars_display=env_vars_display,
learn=learn,
memory=memory,
agent_type="openhands",
code_graph=code_graph,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
)
# =============================================================================
# OpenClaw
# =============================================================================

93
tests/test_cli/test_wrap_cline.py Normal file
View file

@ -0,0 +1,93 @@
"""Tests for `headroom wrap cline` command (PR-G1, Phase G)."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli import wrap as wrap_mod
from headroom.cli.main import main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def test_wrap_cline_prepare_only_injects_rtk_into_clinerules(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""`wrap cline --prepare-only` writes RTK guidance to .clinerules at cwd."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "cline", "--prepare-only"])
assert result.exit_code == 0, result.output
clinerules = tmp_path / ".clinerules"
assert clinerules.exists(), ".clinerules should be created"
content = clinerules.read_text()
assert wrap_mod._RTK_MARKER in content
assert "RTK (Rust Token Killer)" in content
def test_wrap_cline_prepare_only_idempotent_no_duplicate_block(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Running wrap cline twice must not duplicate the RTK block."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
runner.invoke(main, ["wrap", "cline", "--prepare-only"])
runner.invoke(main, ["wrap", "cline", "--prepare-only"])
clinerules = tmp_path / ".clinerules"
content = clinerules.read_text()
assert content.count(wrap_mod._RTK_MARKER) == 1
def test_wrap_cline_no_context_tool_does_not_create_clinerules(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--no-context-tool must not create .clinerules."""
monkeypatch.chdir(tmp_path)
# Patch RTK to fail if accidentally called.
with patch.object(wrap_mod, "_ensure_rtk_binary") as ensure:
result = runner.invoke(main, ["wrap", "cline", "--prepare-only", "--no-context-tool"])
assert result.exit_code == 0, result.output
assert not (tmp_path / ".clinerules").exists()
ensure.assert_not_called()
def test_wrap_cline_preserves_existing_clinerules_content(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Pre-existing .clinerules content must be preserved when RTK is appended."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
clinerules = tmp_path / ".clinerules"
original = "# Project conventions\n\nAlways use Python 3.12.\n"
clinerules.write_text(original)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "cline", "--prepare-only"])
assert result.exit_code == 0, result.output
content = clinerules.read_text()
assert "Always use Python 3.12." in content
assert wrap_mod._RTK_MARKER in content

140
tests/test_cli/test_wrap_continue.py Normal file
View file

@ -0,0 +1,140 @@
"""Tests for `headroom wrap continue` command (PR-G1, Phase G)."""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli import wrap as wrap_mod
from headroom.cli.main import main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def test_inject_continue_rtk_systemmessage_new_file(tmp_path: Path) -> None:
"""Writing into a non-existent config.json creates parents + sets systemMessage."""
config_file = tmp_path / ".continue" / "config.json"
assert not config_file.exists()
assert wrap_mod._inject_continue_rtk_systemmessage(config_file) is True
data = json.loads(config_file.read_text())
assert wrap_mod._RTK_MARKER in data["systemMessage"]
def test_inject_continue_rtk_systemmessage_preserves_existing_keys(tmp_path: Path) -> None:
"""Pre-existing keys like ``models`` are not touched."""
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"}]}))
wrap_mod._inject_continue_rtk_systemmessage(config_file)
data = json.loads(config_file.read_text())
assert data["models"] == [{"title": "GPT-4o", "provider": "openai"}]
assert wrap_mod._RTK_MARKER in data["systemMessage"]
def test_inject_continue_rtk_systemmessage_appends_to_existing_message(
tmp_path: Path,
) -> None:
"""Pre-existing systemMessage content is preserved; rtk block is appended."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
existing_msg = "You are a helpful assistant."
config_file.write_text(json.dumps({"systemMessage": existing_msg}))
wrap_mod._inject_continue_rtk_systemmessage(config_file)
data = json.loads(config_file.read_text())
assert data["systemMessage"].startswith(existing_msg)
assert wrap_mod._RTK_MARKER in data["systemMessage"]
def test_inject_continue_rtk_systemmessage_idempotent(tmp_path: Path) -> None:
"""Re-injection must not duplicate the marker."""
config_file = tmp_path / ".continue" / "config.json"
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["systemMessage"].count(wrap_mod._RTK_MARKER) == 1
def test_inject_continue_rtk_systemmessage_refuses_invalid_json(
tmp_path: Path,
) -> None:
"""Malformed JSON must be left untouched and the helper must return False."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
malformed = '{ "models": [ this is not valid json'
config_file.write_text(malformed)
result = wrap_mod._inject_continue_rtk_systemmessage(config_file)
assert result is False
assert config_file.read_text() == malformed
def test_inject_continue_rtk_systemmessage_refuses_non_object_root(
tmp_path: Path,
) -> None:
"""A JSON array at the root is not a valid Continue config; leave untouched."""
config_file = tmp_path / ".continue" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("[]")
result = wrap_mod._inject_continue_rtk_systemmessage(config_file)
assert result is False
assert config_file.read_text() == "[]"
def test_wrap_continue_prepare_only_injects_systemmessage(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""`wrap continue --prepare-only` injects into ./.continue/config.json by default."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "continue", "--prepare-only"])
assert result.exit_code == 0, result.output
config_file = tmp_path / ".continue" / "config.json"
assert config_file.exists()
data = json.loads(config_file.read_text())
assert wrap_mod._RTK_MARKER in data["systemMessage"]
def test_wrap_continue_respects_custom_config_path(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--config writes to the user-specified path, not the cwd default."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
custom_config = tmp_path / "custom" / "my-continue.json"
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(
main,
["wrap", "continue", "--prepare-only", "--config", str(custom_config)],
)
assert result.exit_code == 0, result.output
assert custom_config.exists()
assert not (tmp_path / ".continue" / "config.json").exists()
data = json.loads(custom_config.read_text())
assert wrap_mod._RTK_MARKER in data["systemMessage"]

117
tests/test_cli/test_wrap_goose.py Normal file
View file

@ -0,0 +1,117 @@
"""Tests for `headroom wrap goose` command (PR-G1, Phase G)."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli import wrap as wrap_mod
from headroom.cli.main import main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def test_wrap_goose_sets_provider_envs(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""OPENAI_BASE_URL, OPENAI_API_BASE, ANTHROPIC_BASE_URL are set on launch."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", 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="goose"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "goose", "--port", "9000", "--", "session"])
assert result.exit_code == 0, result.output
env = captured["env"]
assert isinstance(env, dict)
assert env["OPENAI_BASE_URL"] == "http://127.0.0.1:9000/v1"
assert env["OPENAI_API_BASE"] == "http://127.0.0.1:9000/v1"
assert env["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:9000"
assert captured["tool_label"] == "GOOSE"
assert captured["agent_type"] == "goose"
assert captured["args"] == ("session",)
def test_wrap_goose_injects_rtk_into_goosehints(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""RTK block must be written to .goosehints at the project root."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "goose", "--prepare-only"])
assert result.exit_code == 0, result.output
goosehints = tmp_path / ".goosehints"
assert goosehints.exists()
content = goosehints.read_text()
assert wrap_mod._RTK_MARKER in content
assert "RTK (Rust Token Killer)" in content
def test_wrap_goose_idempotent_no_duplicate_block(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Re-running prepare-only must not duplicate the .goosehints RTK block."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
runner.invoke(main, ["wrap", "goose", "--prepare-only"])
runner.invoke(main, ["wrap", "goose", "--prepare-only"])
content = (tmp_path / ".goosehints").read_text()
assert content.count(wrap_mod._RTK_MARKER) == 1
def test_wrap_goose_missing_binary_errors_clearly(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If the goose binary is missing the command must fail with a clear error."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod.shutil, "which", return_value=None):
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "goose"])
assert result.exit_code == 1
assert "'goose' not found in PATH" in result.output
def test_wrap_goose_no_context_tool_skips_goosehints(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--no-context-tool must not create .goosehints."""
monkeypatch.chdir(tmp_path)
with patch.object(wrap_mod, "_ensure_rtk_binary") as ensure:
result = runner.invoke(main, ["wrap", "goose", "--prepare-only", "--no-context-tool"])
assert result.exit_code == 0, result.output
assert not (tmp_path / ".goosehints").exists()
ensure.assert_not_called()

175
tests/test_cli/test_wrap_openhands.py Normal file
View file

@ -0,0 +1,175 @@
"""Tests for `headroom wrap openhands` command (PR-G1, Phase G)."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli import wrap as wrap_mod
from headroom.cli.main import main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def test_wrap_openhands_sets_provider_envs(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""OPENAI_BASE_URL, ANTHROPIC_BASE_URL, LLM_BASE_URL are set on launch."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", 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=Path("/tmp/rtk")):
result = runner.invoke(
main, ["wrap", "openhands", "--port", "9000", "--", "--task", "demo"]
)
assert result.exit_code == 0, result.output
env = captured["env"]
assert isinstance(env, dict)
assert env["OPENAI_BASE_URL"] == "http://127.0.0.1:9000/v1"
assert env["OPENAI_API_BASE"] == "http://127.0.0.1:9000/v1"
assert env["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:9000"
assert env["LLM_BASE_URL"] == "http://127.0.0.1:9000/v1"
assert captured["tool_label"] == "OPENHANDS"
assert captured["agent_type"] == "openhands"
assert captured["args"] == ("--task", "demo")
def test_wrap_openhands_injects_rtk_via_env_var(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""OPENHANDS_INSTRUCTIONS env var must contain the RTK block at launch."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
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=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 0, result.output
env = captured["env"]
assert isinstance(env, dict)
instructions = env.get("OPENHANDS_INSTRUCTIONS", "")
assert wrap_mod._RTK_MARKER in instructions
assert "RTK (Rust Token Killer)" in instructions
def test_wrap_openhands_preserves_existing_openhands_instructions(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Pre-existing OPENHANDS_INSTRUCTIONS env content is preserved, rtk is appended."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
monkeypatch.setenv("OPENHANDS_INSTRUCTIONS", "Prefer typed Python.")
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=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 0, result.output
env = captured["env"]
instructions = env.get("OPENHANDS_INSTRUCTIONS", "")
assert "Prefer typed Python." in instructions
assert wrap_mod._RTK_MARKER in instructions
def test_wrap_openhands_idempotent_already_injected(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If OPENHANDS_INSTRUCTIONS already contains the marker, do not re-append."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
pre_existing = "Prefer typed Python.\n\n" + wrap_mod.RTK_INSTRUCTIONS_BLOCK
monkeypatch.setenv("OPENHANDS_INSTRUCTIONS", pre_existing)
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=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 0, result.output
env = captured["env"]
instructions = env.get("OPENHANDS_INSTRUCTIONS", "")
assert instructions.count(wrap_mod._RTK_MARKER) == 1
def test_wrap_openhands_missing_binary_errors_clearly(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If the openhands binary is missing the command must fail with a clear error."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod.shutil, "which", return_value=None):
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", "openhands"])
assert result.exit_code == 1
assert "'openhands' not found in PATH" in result.output
def test_wrap_openhands_no_context_tool_does_not_inject(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--no-context-tool must skip OPENHANDS_INSTRUCTIONS injection."""
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") as ensure:
result = runner.invoke(main, ["wrap", "openhands", "--no-context-tool"])
assert result.exit_code == 0, result.output
ensure.assert_not_called()
env = captured["env"]
assert isinstance(env, dict)
assert "OPENHANDS_INSTRUCTIONS" not in env or env["OPENHANDS_INSTRUCTIONS"] == ""