From e5a843306f0aa9590bf28c04493672c09be44874 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Thu, 23 Jul 2026 13:15:51 -0400 Subject: [PATCH] feat(harness): add surface-aware harness selection Co-Authored-By: Codex --- src/brigade/cli/harness.py | 25 ++- src/brigade/install.py | 37 ++++- src/brigade/selection.py | 126 ++++++++++++++- tests/test_surface_aware_install.py | 243 ++++++++++++++++++++++++++++ 4 files changed, 428 insertions(+), 3 deletions(-) create mode 100644 tests/test_surface_aware_install.py diff --git a/src/brigade/cli/harness.py b/src/brigade/cli/harness.py index f6b7d63d..af4ad1ee 100644 --- a/src/brigade/cli/harness.py +++ b/src/brigade/cli/harness.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import sys def _write_mode(parser: argparse.ArgumentParser) -> None: @@ -25,6 +26,16 @@ def register(sub: argparse._SubParsersAction) -> None: install = commands.add_parser("install", help="Plan or apply a harness onboarding profile.") _common(install) _write_mode(install) + install.add_argument( + "--surface", + choices=["cursor-cli", "cursor-gui"], + help="Install Brigade's cursor projection for this explicitly selected vendor surface.", + ) + install.add_argument( + "--projection-only", + action="store_true", + help="Allow a runtime-absent surface to receive Brigade projections without claiming a native runtime.", + ) uninstall = commands.add_parser("uninstall", help="Remove only Brigade-owned harness configuration.") _common(uninstall) @@ -38,9 +49,21 @@ def register(sub: argparse._SubParsersAction) -> None: def dispatch(args) -> int: from .. import cursor_user_cmd + from ..install import ensure_surface_installable + from ..selection import SurfaceInstallRefusal, SurfaceRecord if args.harness_command == "install": - return cursor_user_cmd.install(write=args.write, json_output=args.json) + if args.projection_only and not args.surface: + print("error: --projection-only requires --surface", file=sys.stderr) + return 2 + try: + if args.surface: + surface = SurfaceRecord.resolve_known(args.surface) + ensure_surface_installable(surface, projection_only=args.projection_only) + return cursor_user_cmd.install(write=args.write, json_output=args.json) + except SurfaceInstallRefusal as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 if args.harness_command == "uninstall": return cursor_user_cmd.uninstall(write=args.write, json_output=args.json) if args.harness_command == "doctor": diff --git a/src/brigade/install.py b/src/brigade/install.py index 3a57c2e8..cb15e02e 100644 --- a/src/brigade/install.py +++ b/src/brigade/install.py @@ -7,6 +7,7 @@ from __future__ import annotations +import json import os import shutil import sys @@ -14,7 +15,7 @@ from typing import List, Tuple from .config import Config, write_config -from .selection import Selection, WRITER_INBOXES +from .selection import Selection, SurfaceInstallRefusal, SurfaceRecord, WRITER_INBOXES from .templates import ( harness_memory_owner, is_text, @@ -30,6 +31,7 @@ LEGACY_GITIGNORE_BEGIN = "# >>> solo-mise gitignore block >>>" LEGACY_GITIGNORE_END = "# <<< solo-mise gitignore block <<<" DEFAULT_WIRED_SKILLS = ("brigade-work", "ultra-work-scout") +SURFACE_EVIDENCE_REL_PATH = ".brigade/surface-evidence.json" def build_gitignore_block(selection: Selection) -> str: @@ -275,6 +277,36 @@ def resolve_manifests(selection: Selection) -> Tuple[List[dict], List[str], List return deduped_files, deduped_dirs, notes +def ensure_surface_installable(surface: SurfaceRecord, *, projection_only: bool) -> None: + """Refuse runtime-absent surfaces unless projection-only mode is explicit.""" + state = surface.availability.get("state") + reason = surface.availability.get("reason") + if state == "externally_blocked" and reason == "binary_not_found" and not projection_only: + raise SurfaceInstallRefusal(surface.surface_id, "availability is externally_blocked: binary_not_found") + if state == "external_only" and not projection_only: + raise SurfaceInstallRefusal(surface.surface_id, "external-only surfaces require --projection-only") + + +def _preflight_surfaces(selection: Selection, *, projection_only: bool) -> None: + for surface in selection.surfaces: + ensure_surface_installable(surface, projection_only=projection_only) + + +def _write_surface_evidence(target: Path, selection: Selection, *, projection_only: bool) -> None: + if not selection.surfaces: + return + path = target / SURFACE_EVIDENCE_REL_PATH + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "version": 1, + "surfaces": { + surface.surface_id: surface.persisted_evidence(projection_only=projection_only) + for surface in selection.surfaces + }, + } + path.write_text(json.dumps(payload, indent=2) + "\n") + + def install_selection( target: Path, selection: Selection, @@ -284,10 +316,12 @@ def install_selection( use_git_exclude: bool = False, update_gitignore: bool = True, wire_skills: bool = True, + projection_only: bool = False, ) -> int: """Install a Selection into `target`. Returns process exit code.""" selection.validate() target = target.expanduser().resolve() + _preflight_surfaces(selection, projection_only=projection_only) if target == Path.home() and not allow_home: print( @@ -360,6 +394,7 @@ def install_selection( # Persist config.json. write_config(target, Config(version=1, selection=selection)) + _write_surface_evidence(target, selection, projection_only=projection_only) # Wire Brigade's built-in skills into each harness's skills directory so # agents actually USE Brigade and can scout large work before editing. diff --git a/src/brigade/selection.py b/src/brigade/selection.py index 2ab521b0..e2d90e3e 100644 --- a/src/brigade/selection.py +++ b/src/brigade/selection.py @@ -2,8 +2,10 @@ from __future__ import annotations +import shutil +from copy import deepcopy from dataclasses import dataclass, field -from typing import List, Optional +from typing import Any, Callable, List, Optional KNOWN_DEPTHS = ("repo", "workspace") @@ -56,6 +58,112 @@ # also appears in the selection becomes the canonical memory owner unless # the user passes --owner. HARNESS_PRIORITY = ["openclaw", "hermes", "claude", "codex", "this-repo"] +SURFACE_PROJECTIONS = { + "cursor-cli": "cursor", + "cursor-gui": "cursor", +} +SURFACE_BINARIES = { + "cursor-cli": ("cursor-agent",), +} +EXTERNAL_ONLY_SURFACES = {"cursor-gui"} + + +class SurfaceInstallRefusal(ValueError): + """A selected vendor surface cannot be installed with the requested mode.""" + + def __init__(self, surface_id: str, detail: str) -> None: + self.surface_id = surface_id + self.detail = detail + super().__init__(f"surface {surface_id!r}: {detail}") + + +@dataclass +class SurfaceRecord: + """Probe evidence for one vendor surface projected through a base harness.""" + + surface_id: str + projection_harness: str + capabilities: list[dict[str, Any]] = field(default_factory=list) + availability: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + self.capabilities = deepcopy(self.capabilities) + self.availability = deepcopy(self.availability) + + @classmethod + def from_fixture( + cls, + fixture: dict[str, Any], + *, + projection_harness: str, + availability: dict[str, Any], + ) -> "SurfaceRecord": + harness = fixture.get("harness", {}) + surface_id = harness.get("id") + if not isinstance(surface_id, str) or not surface_id: + raise ValueError("surface fixture must contain a non-empty harness.id") + capabilities = fixture.get("capabilities", []) + if not isinstance(capabilities, list) or not isinstance(availability, dict): + raise ValueError("surface fixture capabilities and resolved availability must be collections") + return cls( + surface_id=surface_id, + projection_harness=projection_harness, + capabilities=capabilities, + availability=availability, + ) + + @classmethod + def resolve_known( + cls, + surface_id: str, + *, + which: Callable[[str], str | None] | None = None, + ) -> "SurfaceRecord": + """Resolve the availability state for a built-in surface without executing it.""" + projection_harness = SURFACE_PROJECTIONS.get(surface_id) + if projection_harness is None: + raise ValueError(f"unknown harness surface: {surface_id!r} (valid: {tuple(SURFACE_PROJECTIONS)})") + availability: dict[str, Any] + if surface_id in EXTERNAL_ONLY_SURFACES: + availability = { + "state": "external_only", + "reason": "desktop_or_gui_surface", + } + else: + resolver = which or shutil.which + commands = list(SURFACE_BINARIES[surface_id]) + command_available = {command: resolver(command) is not None for command in commands} + available_commands = [command for command, available in command_available.items() if available] + availability = { + "state": "available" if available_commands else "externally_blocked", + "commands": commands, + "command_available": command_available, + } + if available_commands: + availability["available_commands"] = available_commands + else: + availability["reason"] = "binary_not_found" + return cls( + surface_id=surface_id, + projection_harness=projection_harness, + availability=availability, + ) + + def persisted_evidence(self, *, projection_only: bool) -> dict[str, Any]: + state = self.availability.get("state") + if state == "available": + install_mode = "native" + elif projection_only: + install_mode = "projection_only" + else: + install_mode = "unverified" + return { + "projection_harness": self.projection_harness, + "availability": deepcopy(self.availability), + "capabilities": deepcopy(self.capabilities), + "runtime_present": state == "available", + "install_mode": install_mode, + } @dataclass @@ -64,6 +172,7 @@ class Selection: harnesses: List[str] = field(default_factory=list) owner: str = "this-repo" includes: List[str] = field(default_factory=list) + surfaces: List[SurfaceRecord] = field(default_factory=list) def validate(self) -> None: if self.depth not in KNOWN_DEPTHS: @@ -76,6 +185,21 @@ def validate(self) -> None: raise ValueError(f"unknown include: {inc!r} (valid: {KNOWN_INCLUDES})") if self.owner != "this-repo" and self.owner not in self.harnesses: raise ValueError(f"owner {self.owner!r} not in selected harnesses {self.harnesses}") + surface_ids: set[str] = set() + for surface in self.surfaces: + if surface.surface_id in surface_ids: + raise ValueError(f"duplicate selected surface: {surface.surface_id!r}") + surface_ids.add(surface.surface_id) + expected_projection = SURFACE_PROJECTIONS.get(surface.surface_id) + if expected_projection is not None and surface.projection_harness != expected_projection: + raise ValueError( + f"surface {surface.surface_id!r} must project through harness {expected_projection!r}, " + f"not {surface.projection_harness!r}" + ) + if surface.projection_harness not in self.harnesses: + raise ValueError( + f"surface {surface.surface_id!r} projection harness {surface.projection_harness!r} is not selected" + ) def resolve_owner(harnesses: List[str], override: Optional[str] = None) -> str: diff --git a/tests/test_surface_aware_install.py b/tests/test_surface_aware_install.py new file mode 100644 index 00000000..a5d64dbd --- /dev/null +++ b/tests/test_surface_aware_install.py @@ -0,0 +1,243 @@ +import json +from pathlib import Path + +import pytest + +from brigade import cli +from brigade.install import install_selection +from brigade.selection import Selection, SurfaceInstallRefusal, SurfaceRecord + + +FIXTURES = Path(__file__).parents[1] / "docs" / "research" / "fixtures" / "harness-contract.v1" + + +def _fixture(name: str) -> dict: + return json.loads((FIXTURES / f"{name}.json").read_text()) + + +def _surface(name: str, availability: dict) -> SurfaceRecord: + fixture = _fixture(name) + return SurfaceRecord.from_fixture( + fixture, + projection_harness="cursor", + availability=availability, + ) + + +def _selection(*surfaces: SurfaceRecord) -> Selection: + return Selection(depth="repo", harnesses=["cursor"], owner="cursor", surfaces=list(surfaces)) + + +def test_native_available_surface_installs_unchanged(tmp_path: Path) -> None: + surface = _surface("cursor-cli", {"state": "available"}) + + assert install_selection(tmp_path, _selection(surface), wire_skills=False) == 0 + + assert (tmp_path / ".cursor" / "memory-handoffs" / "TEMPLATE.md").is_file() + persisted = json.loads((tmp_path / ".brigade" / "surface-evidence.json").read_text()) + evidence = persisted["surfaces"]["cursor-cli"] + assert evidence["projection_harness"] == "cursor" + assert evidence["runtime_present"] is True + assert evidence["install_mode"] == "native" + + +def test_binary_not_found_refusal_happens_before_any_filesystem_write(tmp_path: Path) -> None: + surface = _surface("cursor-cli", {"state": "externally_blocked", "reason": "binary_not_found"}) + + with pytest.raises(SurfaceInstallRefusal, match="cursor-cli.*binary_not_found"): + install_selection(tmp_path, _selection(surface), wire_skills=False) + + assert list(tmp_path.iterdir()) == [] + + +def test_binary_not_found_projection_only_opt_in_writes_without_runtime_claim(tmp_path: Path) -> None: + surface = _surface("cursor-cli", {"state": "externally_blocked", "reason": "binary_not_found"}) + + assert install_selection(tmp_path, _selection(surface), projection_only=True, wire_skills=False) == 0 + + evidence = json.loads((tmp_path / ".brigade" / "surface-evidence.json").read_text())["surfaces"]["cursor-cli"] + assert evidence["runtime_present"] is False + assert evidence["install_mode"] == "projection_only" + + +def test_external_only_surface_projection_only_opt_in_writes_brigade_projections(tmp_path: Path) -> None: + surface = _surface("cursor-gui", {"state": "external_only"}) + + assert install_selection(tmp_path, _selection(surface), projection_only=True, wire_skills=False) == 0 + + assert (tmp_path / ".cursor" / "memory-handoffs" / "TEMPLATE.md").is_file() + persisted = json.loads((tmp_path / ".brigade" / "surface-evidence.json").read_text()) + evidence = persisted["surfaces"]["cursor-gui"] + assert evidence["runtime_present"] is False + assert evidence["install_mode"] == "projection_only" + + +def test_external_only_surface_without_projection_only_raises(tmp_path: Path) -> None: + surface = _surface("cursor-gui", {"state": "external_only"}) + + with pytest.raises(SurfaceInstallRefusal, match="cursor-gui.*projection-only"): + install_selection(tmp_path, _selection(surface), wire_skills=False) + + assert list(tmp_path.iterdir()) == [] + + +def test_other_externally_blocked_reason_is_not_refused(tmp_path: Path) -> None: + surface = _surface("cursor-cli", {"state": "externally_blocked", "reason": "output_overflow"}) + + assert install_selection(tmp_path, _selection(surface), wire_skills=False) == 0 + persisted = json.loads((tmp_path / ".brigade" / "surface-evidence.json").read_text()) + assert persisted["surfaces"]["cursor-cli"]["runtime_present"] is False + + +def test_cursor_cli_and_cursor_gui_preserve_separate_fixture_capability_evidence(tmp_path: Path) -> None: + cli_fixture = _fixture("cursor-cli") + gui_fixture = _fixture("cursor-gui") + cli_surface = SurfaceRecord.from_fixture( + cli_fixture, + projection_harness="cursor", + availability={"state": "available"}, + ) + gui_surface = SurfaceRecord.from_fixture( + gui_fixture, + projection_harness="cursor", + availability={"state": "external_only"}, + ) + cli_fixture["capabilities"][0]["claim"] = "mutated after selection" + gui_fixture["capabilities"][0]["scope"] = "mutated after selection" + + assert ( + install_selection( + tmp_path, + _selection(cli_surface, gui_surface), + projection_only=True, + wire_skills=False, + ) + == 0 + ) + + persisted = json.loads((tmp_path / ".brigade" / "surface-evidence.json").read_text())["surfaces"] + assert set(persisted) == {"cursor-cli", "cursor-gui"} + assert persisted["cursor-cli"]["capabilities"][0]["scope"] == "project" + assert persisted["cursor-gui"]["capabilities"][0]["scope"] == "desktop" + assert persisted["cursor-cli"]["capabilities"][0]["claim"] != "mutated after selection" + assert persisted["cursor-gui"]["capabilities"][0]["scope"] != "mutated after selection" + assert persisted["cursor-cli"]["runtime_present"] is True + assert persisted["cursor-gui"]["runtime_present"] is False + assert persisted["cursor-cli"]["install_mode"] == "native" + assert persisted["cursor-gui"]["install_mode"] == "projection_only" + + +def test_harness_install_cursor_user_scope_requires_projection_only_for_gui_projection(capsys) -> None: + rc = cli.main(["harness", "install", "cursor", "--scope", "user", "--surface", "cursor-gui"]) + captured = capsys.readouterr() + + assert rc == 2 + assert "projection-only" in captured.err + assert "Traceback" not in captured.err + + +def test_harness_native_cursor_cli_surface_keeps_existing_install_path(monkeypatch) -> None: + calls: list[dict] = [] + + def native_install(*, write: bool, json_output: bool) -> int: + calls.append({"write": write, "json_output": json_output}) + return 31 + + monkeypatch.setattr("brigade.selection.shutil.which", lambda command: f"/fixture-bin/{command}") + monkeypatch.setattr("brigade.cursor_user_cmd.install", native_install) + + rc = cli.main(["harness", "install", "cursor", "--scope", "user", "--surface", "cursor-cli", "--write"]) + + assert rc == 31 + assert calls == [{"write": True, "json_output": False}] + + +def test_harness_blocked_cursor_cli_surface_refuses_without_projection_only(monkeypatch, capsys) -> None: + monkeypatch.setattr("brigade.selection.shutil.which", lambda command: None) + + rc = cli.main(["harness", "install", "cursor", "--scope", "user", "--surface", "cursor-cli"]) + captured = capsys.readouterr() + + assert rc == 2 + assert "externally_blocked: binary_not_found" in captured.err + assert "Traceback" not in captured.err + + +def test_harness_install_cursor_user_scope_keeps_legacy_behavior_without_surface(monkeypatch) -> None: + calls: list[dict] = [] + + def legacy_install(*, write: bool, json_output: bool) -> int: + calls.append({"write": write, "json_output": json_output}) + return 23 + + monkeypatch.setattr("brigade.cursor_user_cmd.install", legacy_install) + + assert cli.main(["harness", "install", "cursor", "--scope", "user", "--write", "--json"]) == 23 + assert calls == [{"write": True, "json_output": True}] + + +def test_harness_explicit_surface_projection_only_delegates_to_user_installer_without_repo_write( + tmp_path: Path, monkeypatch +) -> None: + calls: list[dict] = [] + + def user_install(*, write: bool, json_output: bool) -> int: + calls.append({"write": write, "json_output": json_output}) + return 29 + + monkeypatch.chdir(tmp_path) + monkeypatch.setattr("brigade.selection.shutil.which", lambda command: None) + monkeypatch.setattr("brigade.cursor_user_cmd.install", user_install) + + rc = cli.main( + [ + "harness", + "install", + "cursor", + "--scope", + "user", + "--surface", + "cursor-gui", + "--projection-only", + "--write", + "--json", + ] + ) + + assert rc == 29 + assert calls == [{"write": True, "json_output": True}] + assert list(tmp_path.iterdir()) == [] + + +def test_harness_projection_only_requires_an_explicit_surface(capsys) -> None: + rc = cli.main(["harness", "install", "cursor", "--scope", "user", "--projection-only"]) + captured = capsys.readouterr() + + assert rc == 2 + assert "requires --surface" in captured.err + assert "Traceback" not in captured.err + + +def test_harness_surface_refusal_is_normal_cli_error(monkeypatch, capsys) -> None: + def refuse(*args, **kwargs) -> int: + raise SurfaceInstallRefusal("cursor-cli", "availability is externally_blocked: binary_not_found") + + monkeypatch.setattr("brigade.cursor_user_cmd.install", refuse) + + rc = cli.main( + [ + "harness", + "install", + "cursor", + "--scope", + "user", + "--surface", + "cursor-cli", + "--projection-only", + ] + ) + captured = capsys.readouterr() + + assert rc == 2 + assert "binary_not_found" in captured.err + assert "Traceback" not in captured.err