Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion src/brigade/cli/harness.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import argparse
import sys


def _write_mode(parser: argparse.ArgumentParser) -> None:
Expand All @@ -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)
Expand All @@ -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":
Expand Down
37 changes: 36 additions & 1 deletion src/brigade/install.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,15 @@

from __future__ import annotations

import json
import os
import shutil
import sys
from pathlib import Path
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,
Expand All @@ -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:
Expand Down Expand Up @@ -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,
Expand All @@ -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(
Expand Down Expand Up @@ -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.
Expand Down
126 changes: 125 additions & 1 deletion src/brigade/selection.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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:
Expand Down
Loading
Loading