refactor(e2e): extract reusable harness into e2e/_lib

Centralize Docker / CI e2e test helpers so per-command suites can be
declarative and future commands (install, wrap, ...) can reuse the same
shim/PATH/assertion primitives without duplicating infrastructure.

The harness provides:

* Case dataclass describing one test as argv + shims + expected exit /
  stdout / stderr / files / custom callbacks
* make_shim() factory producing cross-platform executable shims (.sh on
  POSIX, .cmd on Windows) with noop / fail / record-args behaviors
* with_clean_path() context manager that isolates PATH to a minimal
  known-good value plus any extras supplied by the case
* agent_settings_path() locator mirroring headroom.cli.init so tests can
  assert the right file was written without touching private init state
* run_cases() for independent cases and run_case_sequence() for cases
  that must share scratch state (e.g. manifest-merge scenarios)

Shell / PowerShell shim-creation scripts are also shipped for CI steps
that need to drop a shim without spinning up Python first.

No behavior change in this commit - pure infrastructure. The init suite
and new subcommand suites consume the harness in follow-up commits.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

e2e/_lib/__init__.py

@@ -0,0 +1,35 @@
+"""Shared helpers for Docker / CI e2e tests.
+
+This package centralizes utilities used by the per-command e2e harnesses
+(`e2e/init/run.py`, future `e2e/install/run.py`, `e2e/wrap/run.py`, ...).
+The goal is that each command test suite is a small declarative file that
+imports from this package, so new commands can be covered with minimal
+duplication.
+"""
+
+from __future__ import annotations
+
+from .assertions import (
+    assert_exit,
+    assert_stderr_contains,
+    assert_stdout_contains,
+    read_agent_settings,
+)
+from .harness import Case, CaseContext, run_case_sequence, run_cases
+from .path_env import with_clean_path
+from .paths import agent_settings_path
+from .shims import make_shim
+
+__all__ = [
+    "Case",
+    "CaseContext",
+    "agent_settings_path",
+    "assert_exit",
+    "assert_stderr_contains",
+    "assert_stdout_contains",
+    "make_shim",
+    "read_agent_settings",
+    "run_case_sequence",
+    "run_cases",
+    "with_clean_path",
+]

e2e/_lib/assertions.py

@@ -0,0 +1,43 @@
+"""Shared assertion helpers for e2e cases.
+
+Assertions raise ``AssertionError`` with a descriptive message. The harness
+catches them and attributes the failure to the owning ``Case``.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from .paths import Agent, Scope, agent_settings_path
+
+
+def assert_exit(actual: int, expected: int, *, context: str = "") -> None:
+    if actual != expected:
+        suffix = f" ({context})" if context else ""
+        raise AssertionError(f"Expected exit code {expected}, got {actual}{suffix}")
+
+
+def assert_stdout_contains(stdout: str, needle: str) -> None:
+    if needle not in stdout:
+        raise AssertionError(f"stdout missing {needle!r}:\n---\n{stdout}\n---")
+
+
+def assert_stderr_contains(stderr: str, needle: str) -> None:
+    if needle not in stderr:
+        raise AssertionError(f"stderr missing {needle!r}:\n---\n{stderr}\n---")
+
+
+def read_agent_settings(
+    agent: Agent, *, scope: Scope, home: Path, project: Path
+) -> dict[str, Any] | str:
+    """Read an agent's settings file, returning dict for JSON and str for TOML/other."""
+
+    path = agent_settings_path(agent, scope=scope, home=home, project=project)
+    if not path.exists():
+        raise AssertionError(f"Expected settings file at {path}, not found")
+    text = path.read_text(encoding="utf-8")
+    if path.suffix == ".json":
+        return json.loads(text)
+    return text

e2e/_lib/harness.py

@@ -0,0 +1,309 @@
+"""Declarative test-case harness for Docker e2e runners.
+
+Each command gets its own ``run.py`` file that builds a list of ``Case``
+objects and calls ``run_cases(cases)``. The harness handles:
+
+* creating a scratch HOME and project directory per case
+* dropping the requested shims into a dedicated shim dir
+* building a clean PATH that only exposes the shim dir + minimal system dirs
+* invoking the ``headroom`` subprocess with the case's argv
+* running the case's assertions against stdout / stderr / exit code / files
+* reporting pass/fail per case and a final summary
+
+``run_cases`` returns a non-zero exit code if any case fails, so Docker
+containers driving it can fail fast.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from .assertions import assert_exit, assert_stderr_contains, assert_stdout_contains
+from .path_env import with_clean_path
+from .shims import ShimBehavior, make_shim
+
+CaseCallback = Callable[["CaseContext"], None]
+
+
+@dataclass
+class CaseContext:
+    """Runtime context passed to assertion callbacks."""
+
+    name: str
+    home: Path
+    project: Path
+    shim_dir: Path
+    shim_log: Path
+    stdout: str
+    stderr: str
+    exit_code: int
+
+
+@dataclass
+class Case:
+    """Declarative specification of a single e2e test case.
+
+    Attributes:
+        name: Human-readable identifier, printed on success/failure.
+        argv: Arguments passed to ``headroom`` (e.g. ``["init", "-g", "claude"]``).
+        shims: Mapping of shim name -> behavior to drop into the shim dir.
+        env_extra: Extra env vars layered on top of the clean env.
+        expected_exit: Required exit code (default 0).
+        expected_stdout_contains: Substrings that must appear on stdout.
+        expected_stderr_contains: Substrings that must appear on stderr.
+        expected_files: Paths (relative to home or project) that must exist.
+                        Use ``{home}/...`` or ``{project}/...`` placeholders.
+        extra_assertions: Optional list of callbacks invoked after exit-code /
+                          stdout / stderr / file checks pass. Receives a
+                          ``CaseContext``. Use for JSON-content assertions,
+                          shim-log inspection, etc.
+    """
+
+    name: str
+    argv: list[str]
+    shims: dict[str, ShimBehavior] = field(default_factory=dict)
+    env_extra: dict[str, str] = field(default_factory=dict)
+    expected_exit: int = 0
+    expected_stdout_contains: list[str] = field(default_factory=list)
+    expected_stderr_contains: list[str] = field(default_factory=list)
+    expected_files: list[str] = field(default_factory=list)
+    extra_assertions: list[CaseCallback] = field(default_factory=list)
+
+
+def _log(message: str) -> None:
+    print(f"[e2e] {message}", flush=True)
+
+
+def _resolve_placeholder(spec: str, *, home: Path, project: Path) -> Path:
+    return Path(spec.format(home=str(home), project=str(project)))
+
+
+def _run_single(case: Case, headroom_bin: str = "headroom") -> bool:
+    """Execute one case. Return True on pass, False on fail."""
+
+    with tempfile.TemporaryDirectory(prefix=f"headroom-e2e-{case.name}-") as temp_raw:
+        temp_root = Path(temp_raw)
+        home = temp_root / "home"
+        project = temp_root / "project"
+        shim_dir = temp_root / "bin"
+        shim_log = temp_root / "shim-log.jsonl"
+        home.mkdir(parents=True)
+        project.mkdir(parents=True)
+
+        for shim_name, behavior in case.shims.items():
+            make_shim(shim_name, shim_dir, behavior=behavior)
+
+        with with_clean_path([shim_dir]) as env:
+            env["HOME"] = str(home)
+            env["USERPROFILE"] = str(home)
+            env["HEADROOM_E2E_SHIM_LOG"] = str(shim_log)
+            env.update(case.env_extra)
+
+            proc = subprocess.run(
+                [headroom_bin, *case.argv],
+                env=env,
+                cwd=str(project),
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=180,
+            )
+
+        ctx = CaseContext(
+            name=case.name,
+            home=home,
+            project=project,
+            shim_dir=shim_dir,
+            shim_log=shim_log,
+            stdout=proc.stdout,
+            stderr=proc.stderr,
+            exit_code=proc.returncode,
+        )
+
+        try:
+            assert_exit(proc.returncode, case.expected_exit, context=f"case {case.name}")
+            for needle in case.expected_stdout_contains:
+                assert_stdout_contains(proc.stdout, needle)
+            for needle in case.expected_stderr_contains:
+                assert_stderr_contains(proc.stderr, needle)
+            for spec in case.expected_files:
+                path = _resolve_placeholder(spec, home=home, project=project)
+                if not path.exists():
+                    raise AssertionError(f"Expected file {path} not found")
+            for callback in case.extra_assertions:
+                callback(ctx)
+        except AssertionError as exc:
+            _log(f"FAIL {case.name}: {exc}")
+            if proc.stdout.strip():
+                _log(f"  stdout: {proc.stdout.rstrip()}")
+            if proc.stderr.strip():
+                _log(f"  stderr: {proc.stderr.rstrip()}")
+            return False
+
+        _log(f"PASS {case.name}")
+        return True
+
+
+def _run_in_scratch(
+    case: Case,
+    *,
+    home: Path,
+    project: Path,
+    shim_dir: Path,
+    shim_log: Path,
+    headroom_bin: str,
+) -> bool:
+    """Execute one case inside a pre-existing scratch layout.
+
+    Shims are *added* to ``shim_dir`` (existing shims from prior sequence
+    steps are preserved). This enables sequence cases to build up shim state.
+    """
+
+    for shim_name, behavior in case.shims.items():
+        make_shim(shim_name, shim_dir, behavior=behavior)
+
+    with with_clean_path([shim_dir]) as env:
+        env["HOME"] = str(home)
+        env["USERPROFILE"] = str(home)
+        env["HEADROOM_E2E_SHIM_LOG"] = str(shim_log)
+        env.update(case.env_extra)
+
+        proc = subprocess.run(
+            [headroom_bin, *case.argv],
+            env=env,
+            cwd=str(project),
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=180,
+        )
+
+    ctx = CaseContext(
+        name=case.name,
+        home=home,
+        project=project,
+        shim_dir=shim_dir,
+        shim_log=shim_log,
+        stdout=proc.stdout,
+        stderr=proc.stderr,
+        exit_code=proc.returncode,
+    )
+
+    try:
+        assert_exit(proc.returncode, case.expected_exit, context=f"case {case.name}")
+        for needle in case.expected_stdout_contains:
+            assert_stdout_contains(proc.stdout, needle)
+        for needle in case.expected_stderr_contains:
+            assert_stderr_contains(proc.stderr, needle)
+        for spec in case.expected_files:
+            path = _resolve_placeholder(spec, home=home, project=project)
+            if not path.exists():
+                raise AssertionError(f"Expected file {path} not found")
+        for callback in case.extra_assertions:
+            callback(ctx)
+    except AssertionError as exc:
+        _log(f"FAIL {case.name}: {exc}")
+        if proc.stdout.strip():
+            _log(f"  stdout: {proc.stdout.rstrip()}")
+        if proc.stderr.strip():
+            _log(f"  stderr: {proc.stderr.rstrip()}")
+        return False
+
+    _log(f"PASS {case.name}")
+    return True
+
+
+def run_cases(
+    cases: list[Case],
+    *,
+    headroom_bin: str = "headroom",
+    fail_fast: bool = False,
+) -> int:
+    """Run each case in its own scratch dir. Return exit code (0 = all pass)."""
+
+    passed = 0
+    failed = 0
+    for case in cases:
+        ok = _run_single(case, headroom_bin=headroom_bin)
+        if ok:
+            passed += 1
+        else:
+            failed += 1
+            if fail_fast:
+                break
+
+    _log(f"Summary: {passed} passed, {failed} failed, {len(cases)} total")
+    return 0 if failed == 0 else 1
+
+
+def run_case_sequence(
+    cases: list[Case],
+    *,
+    headroom_bin: str = "headroom",
+    label: str = "sequence",
+    fail_fast: bool = True,
+) -> int:
+    """Run cases sequentially inside a single shared scratch dir.
+
+    Useful when later cases must observe state left by earlier ones (e.g.
+    ``headroom init`` accumulating targets in a shared manifest across
+    successive calls).
+    """
+
+    passed = 0
+    failed = 0
+    with tempfile.TemporaryDirectory(prefix=f"headroom-e2e-{label}-") as temp_raw:
+        temp_root = Path(temp_raw)
+        home = temp_root / "home"
+        project = temp_root / "project"
+        shim_dir = temp_root / "bin"
+        shim_log = temp_root / "shim-log.jsonl"
+        home.mkdir(parents=True)
+        project.mkdir(parents=True)
+
+        for case in cases:
+            ok = _run_in_scratch(
+                case,
+                home=home,
+                project=project,
+                shim_dir=shim_dir,
+                shim_log=shim_log,
+                headroom_bin=headroom_bin,
+            )
+            if ok:
+                passed += 1
+            else:
+                failed += 1
+                if fail_fast:
+                    break
+
+    _log(f"Summary ({label}): {passed} passed, {failed} failed, {len(cases)} total")
+    return 0 if failed == 0 else 1
+
+
+# Allow callers to adopt a different exit strategy (e.g. raising) easily.
+def main_from_cases(cases: list[Case]) -> None:
+    """Convenience entry point for ``run.py`` scripts."""
+
+    code = run_cases(cases)
+    sys.exit(code)
+
+
+__all__ = [
+    "Case",
+    "CaseContext",
+    "main_from_cases",
+    "run_case_sequence",
+    "run_cases",
+]
+
+# Silence unused-import lint for re-exports used by callers.
+_ = os

e2e/_lib/make_shim.ps1

@@ -0,0 +1,24 @@
+# Create a noop executable shim at $Dir\$Name.cmd for use in PATH during
+# native (non-Docker) e2e tests on Windows. Mirrors e2e/_lib/shims.py
+# make_shim(noop).
+#
+# Usage: make_shim.ps1 -Name <name> -Dir <dir>
+
+param(
+    [Parameter(Mandatory = $true)][string]$Name,
+    [Parameter(Mandatory = $true)][string]$Dir
+)
+
+$ErrorActionPreference = "Stop"
+
+if (-not (Test-Path $Dir)) {
+    New-Item -ItemType Directory -Path $Dir -Force | Out-Null
+}
+
+$path = Join-Path $Dir "$Name.cmd"
+$content = @"
+@echo off
+exit /b 0
+"@
+Set-Content -Path $path -Value $content -Encoding ASCII -NoNewline
+Write-Output $path

e2e/_lib/make_shim.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+# Create a noop executable shim at $2/$1 suitable for use in PATH during
+# native (non-Docker) e2e tests. Mirrors e2e/_lib/shims.py make_shim(noop).
+#
+# Usage: make_shim.sh <name> <dir>
+#
+# Exit codes:
+#   0 on success
+#   2 on usage error
+
+set -euo pipefail
+
+if [ $# -ne 2 ]; then
+    echo "usage: $0 <name> <dir>" >&2
+    exit 2
+fi
+
+name="$1"
+dir="$2"
+
+mkdir -p "$dir"
+path="$dir/$name"
+cat >"$path" <<'EOS'
+#!/usr/bin/env bash
+exit 0
+EOS
+chmod +x "$path"
+echo "$path"

e2e/_lib/path_env.py

@@ -0,0 +1,54 @@
+"""PATH environment helpers for e2e test isolation."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+
+
+def _minimal_path_dirs() -> list[str]:
+    """Directories always needed so Python / basic shell utilities work."""
+
+    if os.name == "nt":
+        system_root = os.environ.get("SystemRoot", r"C:\Windows")
+        return [
+            rf"{system_root}\System32",
+            system_root,
+            rf"{system_root}\System32\Wbem",
+            rf"{system_root}\System32\WindowsPowerShell\v1.0",
+        ]
+    # POSIX: keep enough for bash, python3, mkdir, chmod, etc.
+    return ["/usr/local/bin", "/usr/bin", "/bin", "/usr/sbin", "/sbin"]
+
+
+@contextmanager
+def with_clean_path(extra_dirs: list[Path] | None = None) -> Iterator[dict[str, str]]:
+    """Set PATH to a minimal known-good value plus ``extra_dirs``.
+
+    Yields the (already-mutated) environment dict so callers can pass it
+    directly to ``subprocess.run(env=...)``. On exit, the previous PATH is
+    restored.
+    """
+
+    extras = [str(Path(p)) for p in (extra_dirs or [])]
+    new_path = os.pathsep.join(extras + _minimal_path_dirs())
+    env = os.environ.copy()
+    previous = env.get("PATH")
+    env["PATH"] = new_path
+    # Also mutate the real environment so ``shutil.which`` inside this process
+    # sees the clean PATH. Restore on exit.
+    real_previous = os.environ.get("PATH")
+    os.environ["PATH"] = new_path
+    try:
+        yield env
+    finally:
+        if real_previous is None:
+            os.environ.pop("PATH", None)
+        else:
+            os.environ["PATH"] = real_previous
+        if previous is None:
+            env.pop("PATH", None)
+        else:
+            env["PATH"] = previous

e2e/_lib/paths.py

@@ -0,0 +1,47 @@
+"""Per-agent settings-file locators for e2e assertions.
+
+These paths mirror the logic in ``headroom.cli.init`` so e2e tests can
+verify that the right file was written without importing private init
+helpers.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Literal
+
+Agent = Literal["claude", "codex", "copilot", "openclaw"]
+Scope = Literal["user", "local"]
+
+
+def agent_settings_path(agent: Agent, *, scope: Scope, home: Path, project: Path) -> Path:
+    """Return the file that ``headroom init`` should have written for ``agent``.
+
+    ``home`` is the test's simulated HOME directory and ``project`` is the cwd
+    used when invoking ``headroom init``. For global (``-g``) invocations only
+    ``home`` matters; for local invocations only ``project`` matters.
+    """
+
+    home = Path(home)
+    project = Path(project)
+
+    if agent == "claude":
+        if scope == "user":
+            return home / ".claude" / "settings.json"
+        return project / ".claude" / "settings.local.json"
+
+    if agent == "codex":
+        if scope == "user":
+            return home / ".codex" / "config.toml"
+        return project / ".codex" / "config.toml"
+
+    if agent == "copilot":
+        # Copilot init requires -g; no local scope.
+        return home / ".copilot" / "config.json"
+
+    if agent == "openclaw":
+        # OpenClaw init is delegated to `headroom wrap openclaw`; it writes
+        # the openclaw json under $HOME.
+        return home / ".openclaw" / "openclaw.json"
+
+    raise ValueError(f"Unknown agent: {agent!r}")

e2e/_lib/shims.py

@@ -0,0 +1,96 @@
+"""Cross-platform agent binary shim factory for e2e tests.
+
+A "shim" is a tiny executable with a given name (e.g. `claude`, `codex`) that
+the harness drops into a temporary directory and prepends to PATH. It lets
+tests drive `headroom init` without requiring a real Claude/Codex install.
+
+Three behaviors are supported:
+
+* ``noop``         — exits 0 with no output. Default.
+* ``fail``         — exits 1 with a short stderr message.
+* ``record-args``  — appends a JSON record of (tool, argv, cwd) to the file at
+                     ``$HEADROOM_E2E_SHIM_LOG``, then exits 0. Useful for
+                     asserting that `init claude` invoked
+                     `claude plugin install` with the right arguments.
+"""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from pathlib import Path
+from typing import Literal
+
+ShimBehavior = Literal["noop", "fail", "record-args"]
+
+_NOOP_SH = """#!/usr/bin/env bash
+exit 0
+"""
+
+_FAIL_SH = """#!/usr/bin/env bash
+echo "${0##*/}: simulated failure" >&2
+exit 1
+"""
+
+_RECORD_SH = """#!/usr/bin/env bash
+tool="${0##*/}"
+log="${HEADROOM_E2E_SHIM_LOG:-/dev/null}"
+mkdir -p "$(dirname "$log")" 2>/dev/null || true
+python3 - "$tool" "$log" "$@" <<'PY'
+import json, os, sys
+tool, log, *argv = sys.argv[1:]
+record = {"tool": tool, "argv": argv, "cwd": os.getcwd()}
+if log != "/dev/null":
+    with open(log, "a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record) + "\\n")
+print(f"{tool} shim executed")
+PY
+exit 0
+"""
+
+# Windows equivalents. Use `.cmd` so `shutil.which` and PATHEXT find them.
+_NOOP_CMD = "@echo off\r\nexit /b 0\r\n"
+
+_FAIL_CMD = "@echo off\r\necho %~n0: simulated failure 1>&2\r\nexit /b 1\r\n"
+
+_RECORD_CMD = (
+    "@echo off\r\n"
+    "setlocal\r\n"
+    'if "%HEADROOM_E2E_SHIM_LOG%"=="" set HEADROOM_E2E_SHIM_LOG=NUL\r\n'
+    "python -c \"import json,os,sys; name=r'%~n0'; log=os.environ['HEADROOM_E2E_SHIM_LOG']; "
+    "rec={'tool':name,'argv':sys.argv[1:],'cwd':os.getcwd()};\r\n"
+    "open(log,'a',encoding='utf-8').write(json.dumps(rec)+chr(10)) if log!='NUL' else None;\r\n"
+    "print(f'{name} shim executed')\" %*\r\n"
+    "exit /b 0\r\n"
+)
+
+
+def _is_windows() -> bool:
+    return os.name == "nt" or sys.platform == "win32"
+
+
+def make_shim(name: str, dir: Path, behavior: ShimBehavior = "noop") -> Path:
+    """Create an executable shim named ``name`` inside ``dir``.
+
+    Returns the absolute path to the created shim. On POSIX this is a ``.sh``
+    file made executable and named without extension (so ``shutil.which(name)``
+    finds it). On Windows this is a ``.cmd`` file — again, ``shutil.which``
+    honours ``PATHEXT`` and will find it.
+    """
+
+    dir = Path(dir)
+    dir.mkdir(parents=True, exist_ok=True)
+
+    if _is_windows():
+        body = {"noop": _NOOP_CMD, "fail": _FAIL_CMD, "record-args": _RECORD_CMD}[behavior]
+        path = dir / f"{name}.cmd"
+        path.write_text(body, encoding="utf-8")
+        return path
+
+    body = {"noop": _NOOP_SH, "fail": _FAIL_SH, "record-args": _RECORD_SH}[behavior]
+    path = dir / name
+    path.write_text(body, encoding="utf-8")
+    mode = path.stat().st_mode
+    path.chmod(mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+    return path