diff --git a/src/brigade/cli/harness.py b/src/brigade/cli/harness.py index e6ec938f..bc3f44b5 100644 --- a/src/brigade/cli/harness.py +++ b/src/brigade/cli/harness.py @@ -8,7 +8,7 @@ from .. import harness_profiles -SLICE1_TARGETS = harness_profiles.USER_SCOPE_SLICE1_TARGETS +USER_SCOPE_TARGETS = harness_profiles.USER_SCOPE_TARGETS def _write_mode(parser: argparse.ArgumentParser, *, default_dry: bool = True) -> None: @@ -24,9 +24,9 @@ def _write_mode(parser: argparse.ArgumentParser, *, default_dry: bool = True) -> def _common_slice1(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--target", - choices=SLICE1_TARGETS, + choices=USER_SCOPE_TARGETS, required=True, - help="Harness to configure (claude, codex, or all for both).", + help="Harness to configure (claude, codex, openclaw, kimi, grok, cursor, opencode, or all).", ) parser.add_argument("--scope", choices=["user"], required=True, help="Configuration scope.") parser.add_argument( @@ -49,7 +49,7 @@ def register(sub: argparse._SubParsersAction) -> None: commands = parser.add_subparsers(dest="harness_command", metavar="") commands.required = True - sync = commands.add_parser("sync", help="Plan or apply Claude/Codex user-scope onboarding (dry-run default).") + sync = commands.add_parser("sync", help="Plan or apply user-scope harness onboarding (dry-run default).") _common_slice1(sync) _write_mode(sync) sync.add_argument( @@ -67,8 +67,8 @@ def register(sub: argparse._SubParsersAction) -> None: uninstall_target = uninstall.add_mutually_exclusive_group(required=True) uninstall_target.add_argument( "--target", - choices=SLICE1_TARGETS, - help="Claude/Codex user-scope target to uninstall.", + choices=USER_SCOPE_TARGETS, + help="User-scope harness target to uninstall.", ) uninstall_target.add_argument( "harness", @@ -90,8 +90,8 @@ def register(sub: argparse._SubParsersAction) -> None: doctor_target = doctor.add_mutually_exclusive_group(required=True) doctor_target.add_argument( "--target", - choices=SLICE1_TARGETS, - help="Claude/Codex user-scope target to inspect.", + choices=USER_SCOPE_TARGETS, + help="User-scope harness target to inspect.", ) doctor_target.add_argument( "harness", @@ -110,10 +110,10 @@ def register(sub: argparse._SubParsersAction) -> None: doctor.add_argument( "--verify-mcp", action="store_true", - help="Also verify native MCP projections for Claude/Codex.", + help="Also verify native MCP projections for the selected targets.", ) - install = commands.add_parser("install", help="Legacy Cursor user-scope install (use sync for Claude/Codex).") + install = commands.add_parser("install", help="Legacy Cursor user-scope install (use sync for user profiles).") _common_cursor(install) _write_mode(install) install.add_argument( @@ -167,7 +167,7 @@ def dispatch(args) -> int: if args.harness_command == "install": if args.harness != "cursor": - args._brigade_parser.error("use `brigade harness sync --target --scope user`") + args._brigade_parser.error("use `brigade harness sync --target --scope user`") return 2 if args.projection_only and not args.surface: print("error: --projection-only requires --surface", file=sys.stderr) diff --git a/src/brigade/harness_profile_cmd.py b/src/brigade/harness_profile_cmd.py index 12555840..07fe2d57 100644 --- a/src/brigade/harness_profile_cmd.py +++ b/src/brigade/harness_profile_cmd.py @@ -15,8 +15,9 @@ from .toml_compat import TOMLDecodeError as _TOMLDecodeError from .toml_compat import loads as _toml_loads -_RECOVERY_COMMAND = "brigade harness sync --target --scope user --adopt --write" +_RECOVERY_COMMAND = "brigade harness sync --target --scope user --adopt --write" _SECTIONS = ("instructions", "skills", "generated", "mcp") +_HOOK_STATE_KEY = "hooks.json#sessionStart" @dataclass(frozen=True) @@ -191,6 +192,68 @@ def plan_instruction_removal(*, path: Path, state: dict[str, Any]) -> SurfacePla return SurfacePlan("instruction", path, "conflict", "preserve", detail="owned instruction block was edited") +def plan_managed_instruction(*, path: Path, desired: str, state: dict[str, Any], adopt: bool = False) -> SurfacePlan: + """Whole-file owned instruction surface (no marked block, e.g. a managed rule file).""" + instruction_state = state.get("instructions", {}) if isinstance(state.get("instructions"), dict) else {} + owned = instruction_state.get("digest") + desired_digest = digest_text(desired) + if not path.exists(): + return SurfacePlan("instruction", path, "missing", "create", desired_digest, desired) + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + return SurfacePlan("instruction", path, "conflict", "preserve", detail=str(exc)) + live_digest = digest_text(text) + if live_digest == desired_digest: + if owned == live_digest: + return SurfacePlan("instruction", path, "current", "none", desired_digest) + if adopt: + return SurfacePlan("instruction", path, "adopted", "none", desired_digest) + return SurfacePlan( + "instruction", + path, + "conflict", + "preserve", + desired_digest, + detail=f"matching managed instruction file is unowned; recover with: {_RECOVERY_COMMAND}", + ) + if owned == live_digest or adopt: + return SurfacePlan("instruction", path, "stale", "update", desired_digest, desired) + return SurfacePlan( + "instruction", + path, + "conflict", + "preserve", + detail=f"foreign managed instruction file; recover with: {_RECOVERY_COMMAND}", + ) + + +def plan_managed_instruction_removal(*, path: Path, state: dict[str, Any]) -> SurfacePlan: + if not path.exists(): + return SurfacePlan("instruction", path, "absent", "none") + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + return SurfacePlan("instruction", path, "conflict", "preserve", detail=str(exc)) + instruction_state = state.get("instructions", {}) if isinstance(state.get("instructions"), dict) else {} + if instruction_state.get("digest") == digest_text(text): + return SurfacePlan("instruction", path, "managed", "remove") + return SurfacePlan("instruction", path, "conflict", "preserve", detail="owned instruction file was edited") + + +def _instruction_plan(profile, state: dict[str, Any], *, adopt: bool) -> SurfacePlan: + desired = profile.instruction_text or harness_profiles.managed_instruction_text() + if profile.instruction_mode == "managed-file": + return plan_managed_instruction(path=profile.instruction_path, desired=desired, state=state, adopt=adopt) + return plan_instruction(path=profile.instruction_path, desired=desired, state=state, adopt=adopt) + + +def _instruction_removal_plan(profile, state: dict[str, Any]) -> SurfacePlan: + if profile.instruction_mode == "managed-file": + return plan_managed_instruction_removal(path=profile.instruction_path, state=state) + return plan_instruction_removal(path=profile.instruction_path, state=state) + + def _lstat_conflict(path: Path, *, directory: bool) -> str | None: """Reject symlinks and non-directories before native skill writes. @@ -213,17 +276,26 @@ def _lstat_conflict(path: Path, *, directory: bool) -> str | None: return None +def _mcp_config_path(profile, workspace: Path) -> Path: + if profile.mcp_path is not None: + return profile.mcp_path + return mcp_adapters.resolve_path(mcp_adapters.ADAPTERS[profile.mcp_harness], workspace) + + def _native_surface_conflicts(profile, workspace: Path) -> list[dict[str, Any]]: """Reject symlinked profile surfaces before reading or mutating them.""" - mcp_path = mcp_adapters.resolve_path(mcp_adapters.ADAPTERS[profile.mcp_harness], workspace) - surfaces = ( + mcp_path = _mcp_config_path(profile, workspace) + surfaces = [ ("profile-root", profile.user_root, True), ("instruction", profile.instruction_path, False), ("profile-directory", profile.state_path.parent, True), ("ownership-state", profile.state_path, False), ("profile-receipt", profile.receipt_path, False), ("mcp", mcp_path, False), - ) + ] + surfaces.extend(("generated", profile.user_root / generated.relative, False) for generated in profile.generated) + if profile.hook is not None: + surfaces.append(("hook", profile.hook.path, False)) conflicts: list[dict[str, Any]] = [] for surface, path, directory in surfaces: error = _lstat_conflict(path, directory=directory) @@ -434,6 +506,235 @@ def _skill_uninstall_plan(profile, state: dict[str, Any]) -> dict[str, Any]: return {"items": items, "conflicts": conflicts, "removes": removes, "prune": sorted(set(prune), reverse=True)} +def _missing_directories(root: Path, path: Path) -> list[str]: + missing: list[str] = [] + cursor = path.parent + while cursor != root and not cursor.exists(): + missing.append(str(cursor)) + cursor = cursor.parent + return list(reversed(missing)) + + +def _generated_destination(profile, relative: str) -> tuple[Path | None, str | None]: + parts = Path(relative).parts + if not relative or relative.startswith("/") or ".." in parts: + return None, "unsafe generated file path" + destination = profile.user_root / relative + candidates = [profile.user_root] + cursor = profile.user_root + for component in destination.relative_to(profile.user_root).parts[:-1]: + cursor = cursor / component + candidates.append(cursor) + for ancestor in candidates: + error = _lstat_conflict(ancestor, directory=True) + if error: + return None, error + error = _lstat_conflict(destination, directory=False) + return (None, error) if error else (destination, None) + + +def _generated_plans(profile, state: dict[str, Any], *, adopt: bool) -> dict[str, Any]: + """Plan whole-file Brigade-owned generated artifacts (plugin manifests, hook scripts).""" + items: list[dict[str, Any]] = [] + conflicts: list[dict[str, Any]] = [] + writes: list[tuple[Path, str, bool, str]] = [] + removes: list[Path] = [] + prune: list[Path] = [] + records = state["generated"] + desired = {generated.relative: generated for generated in profile.generated} + next_records: dict[str, dict[str, Any]] = {} + for generated in profile.generated: + record = records.get(generated.relative) + record = record if isinstance(record, dict) else {} + desired_digest = digest_text(generated.text) + path, error = _generated_destination(profile, generated.relative) + item = {"surface": "generated", "path": str(path or (profile.user_root / generated.relative))} + if error or path is None: + item.update(status="conflict", action="preserve", detail=error or "unsafe generated destination") + conflicts.append(item) + elif not path.exists(): + item.update(status="missing", action="create") + writes.append((path, generated.text, generated.executable, generated.relative)) + else: + try: + live_digest = digest_text(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError) as exc: + item.update(status="conflict", action="preserve", detail=str(exc)) + conflicts.append(item) + items.append(item) + next_records[generated.relative] = record + continue + if live_digest == desired_digest: + if record.get("digest") == live_digest: + item.update(status="current", action="none") + elif adopt: + item.update(status="adopted", action="none") + else: + item.update(status="conflict", action="preserve", detail="matching generated file is unowned") + conflicts.append(item) + elif record.get("digest") == live_digest or adopt: + item.update(status="stale", action="update") + writes.append((path, generated.text, generated.executable, generated.relative)) + else: + item.update(status="conflict", action="preserve", detail="owned generated file was edited") + conflicts.append(item) + items.append(item) + next_records[generated.relative] = { + "digest": desired_digest, + "created_directories": list(record.get("created_directories", [])), + } + for relative, record in sorted(records.items()): + if relative == _HOOK_STATE_KEY or relative in desired: + continue + item = {"surface": "generated", "path": str(profile.user_root / relative)} + if not isinstance(record, dict) or not isinstance(record.get("digest"), str): + item.update(status="conflict", action="preserve", detail="generated ownership record is malformed") + conflicts.append(item) + else: + path, error = _generated_destination(profile, relative) + if error or path is None: + item.update(status="conflict", action="preserve", detail=error or "unsafe generated destination") + conflicts.append(item) + elif not path.exists(): + item.update(status="absent", action="remove") + elif digest_text(path.read_text(encoding="utf-8")) == record["digest"]: + item.update(status="removed-profile", action="remove") + removes.append(path) + else: + item.update(status="conflict", action="preserve", detail="removed-profile generated file was edited") + conflicts.append(item) + if isinstance(record, dict): + prune.extend(Path(path) for path in record.get("created_directories", []) if isinstance(path, str)) + items.append(item) + return { + "items": items, + "conflicts": conflicts, + "writes": writes, + "removes": removes, + "next": next_records, + "prune": sorted(set(prune), reverse=True), + } + + +def _generated_uninstall_plan(profile, state: dict[str, Any]) -> dict[str, Any]: + items: list[dict[str, Any]] = [] + conflicts: list[dict[str, Any]] = [] + removes: list[Path] = [] + prune: list[Path] = [] + for relative, record in sorted(state["generated"].items()): + if relative == _HOOK_STATE_KEY: + continue + item = {"surface": "generated", "path": str(profile.user_root / relative)} + if not isinstance(record, dict) or not isinstance(record.get("digest"), str): + item.update(status="conflict", action="preserve", detail="generated ownership record is malformed") + conflicts.append(item) + else: + path, error = _generated_destination(profile, relative) + if error or path is None: + item.update(status="conflict", action="preserve", detail=error or "unsafe generated destination") + conflicts.append(item) + elif not path.exists(): + item.update(status="absent", action="remove") + elif digest_text(path.read_text(encoding="utf-8")) == record["digest"]: + item.update(status="managed", action="remove") + removes.append(path) + else: + item.update(status="conflict", action="preserve", detail="owned generated file was edited") + conflicts.append(item) + if isinstance(record, dict): + prune.extend(Path(path) for path in record.get("created_directories", []) if isinstance(path, str)) + items.append(item) + return {"items": items, "conflicts": conflicts, "removes": removes, "prune": sorted(set(prune), reverse=True)} + + +def _hook_plan(profile, state: dict[str, Any], *, adopt: bool) -> dict[str, Any]: + """Plan one Brigade-managed entry inside a co-owned JSON hook config.""" + from . import cursor_user_cmd + + hook = profile.hook + path = hook.path + record = state["generated"].get(_HOOK_STATE_KEY) + record = record if isinstance(record, dict) else {} + desired_fp = cursor_user_cmd._digest_value(hook.entry) + empty: dict[str, Any] = {"items": [], "conflicts": [], "doc": None, "next": record} + doc, error = cursor_user_cmd._read_json_object(path) + item: dict[str, Any] = {"surface": "hook", "path": str(path), "name": "sessionStart"} + if doc is None: + item.update(status="conflict", action="preserve", detail=error or "could not read hook configuration") + return {**empty, "items": [item], "conflicts": [item]} + hooks = doc.get("hooks") + if hooks is not None and not isinstance(hooks, dict): + item.update(status="conflict", action="preserve", detail="existing hooks field must be an object") + return {**empty, "items": [item], "conflicts": [item]} + entries = hooks.get("sessionStart") if isinstance(hooks, dict) else None + if entries is not None and not isinstance(entries, list): + item.update(status="conflict", action="preserve", detail="existing sessionStart hooks must be a list") + return {**empty, "items": [item], "conflicts": [item]} + entries = entries if isinstance(entries, list) else [] + conflicts: list[dict[str, Any]] = [] + if hook.entry in entries: + if record.get("entry_fingerprint") == desired_fp: + item.update(status="current", action="none") + elif adopt: + item.update(status="adopted", action="none") + else: + item.update( + status="conflict", + action="preserve", + detail="matching hook entry is unowned; rerun with --adopt", + ) + conflicts.append(item) + else: + prior_fp = record.get("entry_fingerprint") + prior_index = next( + ( + index + for index, entry in enumerate(entries) + if prior_fp and cursor_user_cmd._digest_value(entry) == prior_fp + ), + None, + ) + if prior_index is not None: + item.update(status="stale", action="update", prior_index=prior_index) + else: + item.update(status="missing", action="create") + return { + "items": [item], + "conflicts": conflicts, + "doc": doc, + "next": {"entry_fingerprint": desired_fp}, + } + + +def _hook_uninstall_plan(profile, state: dict[str, Any]) -> dict[str, Any]: + from . import cursor_user_cmd + + hook = profile.hook + path = hook.path + record = state["generated"].get(_HOOK_STATE_KEY) + fingerprint = record.get("entry_fingerprint") if isinstance(record, dict) else None + if not fingerprint: + return {"items": [], "conflicts": [], "doc": None, "index": None} + doc, error = cursor_user_cmd._read_json_object(path) + item: dict[str, Any] = {"surface": "hook", "path": str(path), "name": "sessionStart"} + entries: Any = None + if doc is not None and isinstance(doc.get("hooks"), dict): + entries = doc["hooks"].get("sessionStart") + index = ( + next((i for i, entry in enumerate(entries) if cursor_user_cmd._digest_value(entry) == fingerprint), None) + if isinstance(entries, list) + else None + ) + if index is not None: + item.update(status="managed", action="remove") + return {"items": [item], "conflicts": [], "doc": doc, "index": index} + if not path.exists(): + item.update(status="absent", action="none") + return {"items": [item], "conflicts": [], "doc": None, "index": None} + item.update(status="conflict", action="preserve", detail=error or "managed hook entry was edited or removed") + return {"items": [item], "conflicts": [item], "doc": None, "index": None} + + def _malformed_native_config(adapter, text: str | None) -> str | None: if text is None: return None @@ -457,7 +758,7 @@ def _mcp_plan( profile, state: dict[str, Any], workspace: Path, *, allow_global_stdio: bool, adopt: bool ) -> dict[str, Any]: adapter = mcp_adapters.ADAPTERS[profile.mcp_harness] - path = mcp_adapters.resolve_path(adapter, workspace) + path = profile.mcp_path or mcp_adapters.resolve_path(adapter, workspace) servers, errors, _warnings = mcp_cmd.load_canonical(workspace) if errors: # A workspace need not opt into project MCP at all. Existing owned @@ -608,7 +909,7 @@ def _mcp_plan( def _mcp_uninstall_plan(profile, state: dict[str, Any], workspace: Path) -> dict[str, Any]: adapter = mcp_adapters.ADAPTERS[profile.mcp_harness] - path = mcp_adapters.resolve_path(adapter, workspace) + path = profile.mcp_path or mcp_adapters.resolve_path(adapter, workspace) items: list[dict[str, Any]] = [] conflicts: list[dict[str, Any]] = [] if not state["mcp"]: @@ -645,7 +946,7 @@ def _mcp_uninstall_plan(profile, state: dict[str, Any], workspace: Path) -> dict def _verify_mcp(profile, state: dict[str, Any], workspace: Path) -> tuple[dict[str, Any], bool]: adapter = mcp_adapters.ADAPTERS[profile.mcp_harness] - path = mcp_adapters.resolve_path(adapter, workspace) + path = profile.mcp_path or mcp_adapters.resolve_path(adapter, workspace) items: list[dict[str, Any]] = [] if not state["mcp"]: return {"status": "ready", "items": items}, True @@ -726,6 +1027,11 @@ def _receipt_ownership(state: dict[str, Any]) -> dict[str, Any]: return { "instruction_fingerprint": state.get("instructions", {}).get("digest"), "skills": state.get("skills", {}), + "generated_fingerprints": { + name: record.get("digest") or record.get("entry_fingerprint") + for name, record in sorted(state.get("generated", {}).items()) + if isinstance(record, dict) + }, "mcp_fingerprints": { name: record.get("projected_fingerprint") for name, record in sorted(state["mcp"].items()) @@ -853,9 +1159,7 @@ def _sync_profile( ), False state = json.loads(json.dumps(loaded.state)) instruction_existed = profile.instruction_path.exists() - instruction = plan_instruction( - path=profile.instruction_path, desired=harness_profiles.managed_instruction_text(), state=state, adopt=adopt - ) + instruction = _instruction_plan(profile, state, adopt=adopt) items = [ { "surface": "instruction", @@ -872,6 +1176,13 @@ def _sync_profile( skill_plan = _skill_plans(profile, state, workspace) items.extend(skill_plan["items"]) conflicts.extend(skill_plan["conflicts"]) + generated_plan = _generated_plans(profile, state, adopt=adopt) + items.extend(generated_plan["items"]) + conflicts.extend(generated_plan["conflicts"]) + hook_plan = _hook_plan(profile, state, adopt=adopt) if profile.hook is not None else None + if hook_plan is not None: + items.extend(hook_plan["items"]) + conflicts.extend(hook_plan["conflicts"]) mcp_plan = _mcp_plan(profile, state, workspace, allow_global_stdio=allow_global_stdio, adopt=adopt) items.extend(mcp_plan["items"]) conflicts.extend(mcp_plan["conflicts"]) @@ -883,10 +1194,20 @@ def _sync_profile( if isinstance(old_instruction, dict) else not instruction_existed ) - proposed["instructions"] = ( - {"digest": instruction.desired_digest, "created_file": created_file} if instruction.desired_digest else {} - ) + if instruction.desired_digest: + instruction_record: dict[str, Any] = { + "digest": instruction.desired_digest, + "created_file": created_file, + } + if isinstance(old_instruction, dict) and old_instruction.get("created_directories"): + instruction_record["created_directories"] = list(old_instruction["created_directories"]) + proposed["instructions"] = instruction_record + else: + proposed["instructions"] = {} proposed["skills"] = skill_plan["next"] + proposed["generated"] = generated_plan["next"] + if hook_plan is not None: + proposed["generated"][_HOOK_STATE_KEY] = hook_plan["next"] proposed["mcp"] = mcp_plan["next"] proposed["package_version"] = BRIGADE_VERSION state_item = _state_item(profile.state_path, before=state, after=proposed) @@ -898,9 +1219,17 @@ def _sync_profile( if write and ready: changed = False if instruction.action in {"create", "update"}: + instruction_dirs = ( + _missing_directories(profile.user_root, instruction.path) + if profile.instruction_mode == "managed-file" + else [] + ) localio.write_text_atomic(instruction.path, instruction.rendered or "") files_written.append(str(instruction.path)) changed = True + if profile.instruction_mode == "managed-file": + existing_dirs = proposed["instructions"].get("created_directories", []) + proposed["instructions"]["created_directories"] = sorted(set(existing_dirs) | set(instruction_dirs)) created_dirs: dict[str, list[str]] = {} for path, data, skill_id, _relative in skill_plan["writes"]: created_dirs.setdefault(skill_id, []).extend(_missing_skill_directories(profile, path)) @@ -914,6 +1243,36 @@ def _sync_profile( path.unlink() files_removed.append(str(path)) changed = True + generated_dirs: dict[str, list[str]] = {} + for path, text, executable, relative in generated_plan["writes"]: + generated_dirs.setdefault(relative, []).extend(_missing_directories(profile.user_root, path)) + localio.write_text_atomic(path, text) + if executable: + path.chmod(path.stat().st_mode | 0o755) + files_written.append(str(path)) + changed = True + for relative, directories in generated_dirs.items(): + existing = proposed["generated"][relative].setdefault("created_directories", []) + proposed["generated"][relative]["created_directories"] = sorted(set(existing) | set(directories)) + for path in generated_plan["removes"]: + path.unlink() + files_removed.append(str(path)) + changed = True + files_removed.extend(_prune_created_directories(generated_plan["prune"], profile.user_root)) + if hook_plan is not None: + hook_item = hook_plan["items"][0] + if hook_item["action"] in {"create", "update"}: + from . import cursor_user_cmd + + hook_doc = hook_plan["doc"] if hook_plan["doc"] is not None else {} + entries = hook_doc.setdefault("hooks", {}).setdefault("sessionStart", []) + if hook_item["action"] == "create": + entries.append(profile.hook.entry) + else: + entries[hook_item["prior_index"]] = profile.hook.entry + localio.write_text_atomic(profile.hook.path, cursor_user_cmd._coowned_json_text(hook_doc)) + files_written.append(str(profile.hook.path)) + changed = True if mcp_plan["updates"] or mcp_plan["remove"]: localio.write_text_atomic( mcp_plan["path"], @@ -996,18 +1355,26 @@ def _uninstall_profile(profile, workspace: Path, *, write: bool) -> tuple[dict[s receipt_path=profile.receipt_path, receipt_state="present" if profile.receipt_path.exists() else "missing", ), False - plan = plan_instruction_removal(path=profile.instruction_path, state=state) + plan = _instruction_removal_plan(profile, state) items = [{"surface": "instruction", "path": str(plan.path), "status": plan.status, "action": plan.action}] conflicts = [] if plan.status != "conflict" else [items[0] | {"detail": plan.detail or "instruction conflict"}] skill_plan = _skill_uninstall_plan(profile, state) items.extend(skill_plan["items"]) conflicts.extend(skill_plan["conflicts"]) + generated_plan = _generated_uninstall_plan(profile, state) + items.extend(generated_plan["items"]) + conflicts.extend(generated_plan["conflicts"]) + hook_plan = _hook_uninstall_plan(profile, state) if profile.hook is not None else None + if hook_plan is not None: + items.extend(hook_plan["items"]) + conflicts.extend(hook_plan["conflicts"]) mcp_plan = _mcp_uninstall_plan(profile, state, workspace) items.extend(mcp_plan["items"]) conflicts.extend(mcp_plan["conflicts"]) proposed = json.loads(json.dumps(state)) proposed["instructions"] = {} proposed["skills"] = {} + proposed["generated"] = {} proposed["mcp"] = {} state_item = _state_item(profile.state_path, before=state, after=proposed, remove=True) receipt_item = _receipt_item(profile.receipt_path, state_action=state_item["action"], remove=True) @@ -1021,10 +1388,30 @@ def _uninstall_profile(profile, workspace: Path, *, write: bool) -> tuple[dict[s else: localio.write_text_atomic(plan.path, plan.rendered) files_removed.append(str(plan.path)) + instruction_dirs = state.get("instructions", {}) + instruction_dirs = ( + [Path(path) for path in instruction_dirs.get("created_directories", []) if isinstance(path, str)] + if isinstance(instruction_dirs, dict) + else [] + ) for path in skill_plan["removes"]: path.unlink(missing_ok=True) files_removed.append(str(path)) files_removed.extend(_prune_created_directories(skill_plan["prune"], profile.skills_root)) + for path in generated_plan["removes"]: + path.unlink(missing_ok=True) + files_removed.append(str(path)) + files_removed.extend(_prune_created_directories(instruction_dirs + generated_plan["prune"], profile.user_root)) + if hook_plan is not None and hook_plan["doc"] is not None and hook_plan["index"] is not None: + from . import cursor_user_cmd + + hook_doc = hook_plan["doc"] + entries = hook_doc["hooks"]["sessionStart"] + entries.pop(hook_plan["index"]) + if not entries: + hook_doc["hooks"].pop("sessionStart", None) + localio.write_text_atomic(profile.hook.path, cursor_user_cmd._coowned_json_text(hook_doc)) + files_removed.append(str(profile.hook.path)) if mcp_plan["remove"]: localio.write_text_atomic( mcp_plan["path"], mcp_plan["adapter"].write_file(mcp_plan["text"], {}, mcp_plan["remove"]) @@ -1077,9 +1464,7 @@ def _doctor_profile(profile, workspace: Path, *, verify_mcp: bool) -> tuple[dict receipt_state="missing", ), False state = loaded.state - instruction = plan_instruction( - path=profile.instruction_path, desired=harness_profiles.managed_instruction_text(), state=state - ) + instruction = _instruction_plan(profile, state, adopt=False) item = { "surface": "instruction", "path": str(instruction.path), @@ -1090,13 +1475,21 @@ def _doctor_profile(profile, workspace: Path, *, verify_mcp: bool) -> tuple[dict skill_plan = _skill_plans(profile, state, workspace) skill_issues = [entry for entry in skill_plan["items"] if entry["status"] != "current"] conflicts.extend(skill_issues) + generated_plan = _generated_plans(profile, state, adopt=False) + generated_issues = [entry for entry in generated_plan["items"] if entry["status"] != "current"] + conflicts.extend(generated_issues) + hook_items: list[dict[str, Any]] = [] + if profile.hook is not None: + hook_plan = _hook_plan(profile, state, adopt=False) + hook_items = hook_plan["items"] + conflicts.extend(entry for entry in hook_items if entry["status"] != "current") mcp, mcp_ok = _verify_mcp(profile, state, workspace) if verify_mcp else ({"status": "pending", "items": []}, True) ready = not conflicts and mcp_ok return _result( profile, status="current" if ready else "conflict", ready=ready, - items=[item, *skill_plan["items"]], + items=[item, *skill_plan["items"], *generated_plan["items"], *hook_items], conflicts=conflicts, files_written=[], files_removed=[], @@ -1118,7 +1511,7 @@ def _run( json_output: bool = False, home: Path | None = None, ) -> int: - profiles = harness_profiles.resolve_slice1_profiles(harness=harness, home=home or Path.home(), workspace=workspace) + profiles = harness_profiles.resolve_user_profiles(harness=harness, home=home or Path.home(), workspace=workspace) def run_profile(profile, *, profile_write: bool) -> tuple[dict[str, Any], bool]: if operation == "sync": diff --git a/src/brigade/harness_profiles.py b/src/brigade/harness_profiles.py index 61353292..857ea687 100644 --- a/src/brigade/harness_profiles.py +++ b/src/brigade/harness_profiles.py @@ -1,18 +1,39 @@ -"""Claude and Codex user-profile records for issue #438 slice 1.""" +"""User-profile records for issue #438 (Claude/Codex slice 1 + the remaining harnesses).""" from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path +from typing import Any -HARNESS_IDS = ("claude", "codex") -USER_SCOPE_SLICE1_TARGETS = (*HARNESS_IDS, "all") -SLICE1_HARNESS_IDS = HARNESS_IDS +SLICE1_HARNESS_IDS = ("claude", "codex") +HARNESS_IDS = SLICE1_HARNESS_IDS +SLICE2_HARNESS_IDS = ("openclaw", "kimi", "grok", "cursor", "opencode") +USER_SCOPE_HARNESS_IDS = (*SLICE1_HARNESS_IDS, *SLICE2_HARNESS_IDS) +USER_SCOPE_SLICE1_TARGETS = (*SLICE1_HARNESS_IDS, "all") +USER_SCOPE_TARGETS = (*USER_SCOPE_HARNESS_IDS, "all") PROFILE_STATE_VERSION = 2 INSTRUCTION_START = "" INSTRUCTION_END = "" +@dataclass(frozen=True) +class GeneratedFile: + """A whole-file Brigade-owned artifact under the profile's user root.""" + + relative: str + text: str + executable: bool = False + + +@dataclass(frozen=True) +class HookSurface: + """A co-owned JSON hook config that must carry one Brigade-managed entry.""" + + path: Path + entry: dict[str, str] + + @dataclass(frozen=True) class HarnessProfile: harness: str @@ -23,31 +44,122 @@ class HarnessProfile: receipt_path: Path mcp_harness: str reload_hint: str + instruction_text: str | None = None + instruction_mode: str = "marked-block" # or "managed-file" for whole-file ownership + generated: tuple[GeneratedFile, ...] = field(default=()) + hook: HookSurface | None = None + mcp_path: Path | None = None # overrides the MCP adapter's static path when probed + + +def _profile( + profile_id: str, + root: Path, + instruction: str, + mcp_harness: str, + reload_hint: str, + **overrides: Any, +) -> HarnessProfile: + return HarnessProfile( + harness=profile_id, + user_root=root, + instruction_path=root / instruction, + skills_root=root / "skills", + state_path=root / "brigade" / "install-state.json", + receipt_path=root / "brigade" / "profile-receipt.json", + mcp_harness=mcp_harness, + reload_hint=reload_hint, + **overrides, + ) + + +def _kimi_root(home: Path) -> Path: + """Capability probe: newer Kimi CLI installs expose ~/.kimi, older ones ~/.kimi-code. + + The newer surface wins when it exists; otherwise the documented legacy root is + used (including for a first-time sync where neither directory exists yet). + """ + newer = home / ".kimi" + return newer if newer.exists() else home / ".kimi-code" + + +def _cursor_generated() -> tuple[GeneratedFile, ...]: + """Whole-file Cursor artifacts; the rule file itself is the instruction surface.""" + from . import cursor_user_cmd + + plugin = "plugins/local/brigade-loop" + return ( + GeneratedFile(f"{plugin}/.cursor-plugin/plugin.json", cursor_user_cmd._plugin_manifest()), + GeneratedFile("hooks/brigade-session-start", cursor_user_cmd._hook_text(), executable=True), + ) def resolve_slice1_profiles(*, harness: str, home: Path, workspace: Path) -> tuple[HarnessProfile, ...]: - """Resolve the supported native user profiles without probing runtimes.""" + """Resolve the Claude/Codex native user profiles without probing runtimes.""" if harness not in USER_SCOPE_SLICE1_TARGETS: raise ValueError(f"unknown harness: {harness}") selected = SLICE1_HARNESS_IDS if harness == "all" else (harness,) specs = { - "claude": (home / ".claude", "CLAUDE.md", "claude-user", "restart Claude Code"), - "codex": (home / ".codex", "AGENTS.md", "codex-user", "restart Codex"), + "claude": _profile("claude", home / ".claude", "CLAUDE.md", "claude-user", "restart Claude Code"), + "codex": _profile("codex", home / ".codex", "AGENTS.md", "codex-user", "restart Codex"), } - return tuple( - HarnessProfile( - harness=profile_id, - user_root=root, - instruction_path=root / instruction, - skills_root=root / "skills", - state_path=root / "brigade" / "install-state.json", - receipt_path=root / "brigade" / "profile-receipt.json", - mcp_harness=mcp_harness, - reload_hint=reload_hint, - ) - for profile_id in selected - for root, instruction, mcp_harness, reload_hint in (specs[profile_id],) - ) + return tuple(specs[profile_id] for profile_id in selected) + + +def resolve_user_profiles(*, harness: str, home: Path, workspace: Path) -> tuple[HarnessProfile, ...]: + """Resolve every supported user-scope profile; slice-2 harnesses may probe surfaces.""" + if harness not in USER_SCOPE_TARGETS: + raise ValueError(f"unknown harness: {harness}") + selected = USER_SCOPE_HARNESS_IDS if harness == "all" else (harness,) + profiles: list[HarnessProfile] = [] + slice1 = { + profile.harness: profile for profile in resolve_slice1_profiles(harness="all", home=home, workspace=workspace) + } + for profile_id in selected: + if profile_id in slice1: + profiles.append(slice1[profile_id]) + elif profile_id == "openclaw": + # OpenClaw instructions target the canonical workspace AGENTS.md. + profiles.append( + _profile("openclaw", home / ".openclaw", "workspace/AGENTS.md", "openclaw", "restart OpenClaw") + ) + elif profile_id == "kimi": + root = _kimi_root(home) + profiles.append( + _profile("kimi", root, "AGENTS.md", "kimi-user", "restart Kimi Code", mcp_path=root / "mcp.json") + ) + elif profile_id == "grok": + profiles.append(_profile("grok", home / ".grok", "AGENTS.md", "grok-user", "restart Grok CLI")) + elif profile_id == "cursor": + root = home / ".cursor" + from . import cursor_user_cmd + + profiles.append( + _profile( + "cursor", + root, + "plugins/local/brigade-loop/rules/brigade-loop.mdc", + "cursor-user", + "reload Cursor windows", + instruction_text=cursor_user_cmd._rule_text(), + instruction_mode="managed-file", + generated=_cursor_generated(), + hook=HookSurface( + path=root / "hooks.json", + entry=cursor_user_cmd._hook_entry(root), + ), + ) + ) + elif profile_id == "opencode": + profiles.append( + _profile( + "opencode", + home / ".config" / "opencode", + "AGENTS.md", + "opencode-user", + "restart OpenCode", + ) + ) + return tuple(profiles) def resolve_profiles( diff --git a/src/brigade/mcp_adapters.py b/src/brigade/mcp_adapters.py index 02db5d49..f3b09826 100644 --- a/src/brigade/mcp_adapters.py +++ b/src/brigade/mcp_adapters.py @@ -1010,6 +1010,27 @@ def _make_json_mcpservers( user_scope=True, reject_invalid_existing=True, ), + # Kimi Code: mcpServers JSON. The static path is the legacy root; the user + # profile layer overrides it when the capability probe selects ~/.kimi. + "kimi-user": _make_json_mcpservers( + "kimi-user", + "~/.kimi-code/mcp.json", + user_scope=True, + reject_invalid_existing=True, + ), + "opencode-user": McpAdapter( + harness="opencode-user", + path="~/.config/opencode/opencode.json", + fmt="json", + top_key="mcp", + user_scope=True, + supports_remote=True, + env_style="expand", + to_provider=_opencode_to_provider, + from_provider=_opencode_from_provider, + read_file=lambda t: _json_read_file(t, "mcp"), + write_file=lambda t, o, r: _json_write_file(t, o, r, "mcp"), + ), "codex-user": McpAdapter( harness="codex-user", path="~/.codex/config.toml", diff --git a/src/brigade/mcp_cmd.py b/src/brigade/mcp_cmd.py index 9a2cafa7..bb4dbf24 100644 --- a/src/brigade/mcp_cmd.py +++ b/src/brigade/mcp_cmd.py @@ -153,6 +153,12 @@ def _server_targets_harness(server: CanonicalServer, harness: str) -> bool: return True if harness == "cursor-user": return "cursor" in server.targets or "cursor-user" in server.targets + if harness == "grok-user": + return "grok" in server.targets or "grok-user" in server.targets + if harness == "kimi-user": + return "kimi" in server.targets or "kimi-user" in server.targets + if harness == "opencode-user": + return "opencode" in server.targets or "opencode-user" in server.targets return harness in server.targets diff --git a/src/brigade/skills_cmd.py b/src/brigade/skills_cmd.py index 2654b683..2f8eed12 100644 --- a/src/brigade/skills_cmd.py +++ b/src/brigade/skills_cmd.py @@ -1174,7 +1174,7 @@ class UserProfileSkillPackage: _USER_PROFILE_MAX_FILES = 512 _USER_PROFILE_MAX_BYTES = 8 * 1024 * 1024 -_USER_PROFILE_HARNESSES = {"claude", "codex"} +_USER_PROFILE_HARNESSES = {"claude", "codex", "openclaw", "kimi", "grok", "cursor", "opencode"} def _user_profile_read_package_files(skill_dir: Path) -> dict[str, bytes] | None: diff --git a/tests/test_harness_user_scope.py b/tests/test_harness_user_scope.py index 55a78e07..eebf4eec 100644 --- a/tests/test_harness_user_scope.py +++ b/tests/test_harness_user_scope.py @@ -297,8 +297,9 @@ def test_target_all_preflights_both_profiles_before_any_write(tmp_path, monkeypa assert cli.main(command + ["--write", "--json"]) == 1 payload = json.loads(capsys.readouterr().out) - assert [result["harness"] for result in payload["results"]] == ["claude", "codex"] - assert payload["results"][1]["status"] == "conflict" + assert [result["harness"] for result in payload["results"]] == list(harness_profiles.USER_SCOPE_HARNESS_IDS) + codex = next(result for result in payload["results"] if result["harness"] == "codex") + assert codex["status"] == "conflict" assert _file_snapshot(home) == before diff --git a/tests/test_harness_user_scope_slice2.py b/tests/test_harness_user_scope_slice2.py new file mode 100644 index 00000000..a4302158 --- /dev/null +++ b/tests/test_harness_user_scope_slice2.py @@ -0,0 +1,463 @@ +"""Issue #438 slice 2: OpenClaw/Kimi/Grok/Cursor/OpenCode user-scope harness profiles.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from brigade import harness_profiles, mcp_cmd, skills_cmd + +# Per-harness native user surfaces (relative to the temporary home). +SURFACES = { + "openclaw": { + "instruction": ".openclaw/workspace/AGENTS.md", + "skills": ".openclaw/skills", + "state": ".openclaw/brigade/install-state.json", + "receipt": ".openclaw/brigade/profile-receipt.json", + "mcp_config": ".openclaw/openclaw.json", + "mcp_target": "openclaw", + }, + "kimi": { + "instruction": ".kimi-code/AGENTS.md", + "skills": ".kimi-code/skills", + "state": ".kimi-code/brigade/install-state.json", + "receipt": ".kimi-code/brigade/profile-receipt.json", + "mcp_config": ".kimi-code/mcp.json", + "mcp_target": "kimi", + }, + "grok": { + "instruction": ".grok/AGENTS.md", + "skills": ".grok/skills", + "state": ".grok/brigade/install-state.json", + "receipt": ".grok/brigade/profile-receipt.json", + "mcp_config": ".grok/config.toml", + "mcp_target": "grok", + }, + "cursor": { + "instruction": ".cursor/plugins/local/brigade-loop/rules/brigade-loop.mdc", + "skills": ".cursor/skills", + "state": ".cursor/brigade/install-state.json", + "receipt": ".cursor/brigade/profile-receipt.json", + "mcp_config": ".cursor/mcp.json", + "mcp_target": "cursor", + }, + "opencode": { + "instruction": ".config/opencode/AGENTS.md", + "skills": ".config/opencode/skills", + "state": ".config/opencode/brigade/install-state.json", + "receipt": ".config/opencode/brigade/profile-receipt.json", + "mcp_config": ".config/opencode/opencode.json", + "mcp_target": "opencode", + }, +} + +SLICE2_HARNESSES = tuple(SURFACES) +# Harnesses whose instruction surface is a marked block inside a user-owned AGENTS.md. +MARKED_BLOCK_HARNESSES = ("openclaw", "kimi", "grok", "opencode") + + +def _use_home(monkeypatch, tmp_path: Path) -> Path: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setattr(Path, "home", classmethod(lambda cls: home)) + monkeypatch.setenv("HOME", str(home)) + return home + + +def _workspace(tmp_path: Path) -> Path: + workspace = tmp_path / "workspace" + workspace.mkdir() + return workspace + + +def _sync_base(workspace: Path, target: str) -> list[str]: + return ["harness", "sync", "--target", target, "--scope", "user", "--workspace", str(workspace)] + + +def _uninstall_base(workspace: Path, target: str) -> list[str]: + return ["harness", "uninstall", "--target", target, "--scope", "user", "--workspace", str(workspace)] + + +def _doctor_base(workspace: Path, target: str) -> list[str]: + return ["harness", "doctor", "--target", target, "--scope", "user", "--workspace", str(workspace)] + + +def _file_snapshot(root: Path) -> dict[Path, bytes]: + return {path.relative_to(root): path.read_bytes() for path in root.rglob("*") if path.is_file()} + + +def _add_reviewed_skill(workspace: Path, name: str = "reviewed") -> None: + source = workspace / "sources" / name + source.mkdir(parents=True) + (source / "SKILL.md").write_text(f"# {name.title()}\n\nUse this skill.\n") + (source / "skill.json").write_text( + json.dumps( + { + "id": name, + "title": name.title(), + "version": "1.0.0", + "required_tools": [], + "required_mcp_servers": [], + "supported_harnesses": list(harness_profiles.USER_SCOPE_HARNESS_IDS), + "trust_level": "workspace", + "tests": [], + } + ) + ) + assert skills_cmd.import_skill(target=workspace, source=source, json_output=True) == 0 + + +def _workspace_with_stdio_server(tmp_path: Path, capsys, targets: list[str]) -> Path: + workspace = _workspace(tmp_path) + mcp_cmd.init(target=workspace, json_output=True) + capsys.readouterr() + mcp_cmd.add( + target=workspace, + name="brigade", + command="brigade", + args=["memory", "serve-mcp", "--stdio", "--target", "."], + timeout=60, + targets=targets, + json_output=True, + ) + capsys.readouterr() + return workspace + + +def test_target_all_resolves_all_seven_user_scope_harnesses(tmp_path): + home, workspace = tmp_path / "home", tmp_path / "workspace" + workspace.mkdir() + profiles = harness_profiles.resolve_user_profiles(harness="all", home=home, workspace=workspace) + assert tuple(profile.harness for profile in profiles) == ( + "claude", + "codex", + "openclaw", + "kimi", + "grok", + "cursor", + "opencode", + ) + + +def test_kimi_capability_probe_selects_surface_by_install(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + + legacy_home = tmp_path / "legacy" + profile = harness_profiles.resolve_user_profiles(harness="kimi", home=legacy_home, workspace=workspace)[0] + assert profile.user_root == legacy_home / ".kimi-code" + assert profile.mcp_path == legacy_home / ".kimi-code" / "mcp.json" + + newer_home = tmp_path / "newer" + (newer_home / ".kimi").mkdir(parents=True) + profile = harness_profiles.resolve_user_profiles(harness="kimi", home=newer_home, workspace=workspace)[0] + assert profile.user_root == newer_home / ".kimi" + assert profile.instruction_path == newer_home / ".kimi" / "AGENTS.md" + assert profile.mcp_path == newer_home / ".kimi" / "mcp.json" + + both_home = tmp_path / "both" + (both_home / ".kimi").mkdir(parents=True) + (both_home / ".kimi-code").mkdir() + profile = harness_profiles.resolve_user_profiles(harness="kimi", home=both_home, workspace=workspace)[0] + assert profile.user_root == both_home / ".kimi" + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_sync_dry_run_writes_nothing(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + assert cli.main(_sync_base(workspace, harness) + ["--json"]) == 0 + capsys.readouterr() + assert _file_snapshot(home) == {} + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_sync_write_then_resync_is_idempotent(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + base = _sync_base(workspace, harness) + + assert cli.main(base + ["--write", "--json"]) == 0 + first = json.loads(capsys.readouterr().out) + assert first["results"][0]["status"] == "updated" + instruction = home / SURFACES[harness]["instruction"] + assert instruction.is_file() + first_text = instruction.read_text() + snapshot = _file_snapshot(home) + + assert cli.main(base + ["--write", "--json"]) == 0 + second = json.loads(capsys.readouterr().out) + assert second["results"][0]["status"] == "current" + assert second["results"][0]["files_written"] == [] + assert instruction.read_text() == first_text + assert _file_snapshot(home) == snapshot + + +@pytest.mark.parametrize("harness", MARKED_BLOCK_HARNESSES) +def test_hand_authored_instruction_section_survives_sync(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + instruction = home / SURFACES[harness]["instruction"] + instruction.parent.mkdir(parents=True) + instruction.write_text("# My notes\nKeep this paragraph.\n") + + assert cli.main(_sync_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + text = instruction.read_text() + assert "# My notes" in text + assert "Keep this paragraph." in text + assert harness_profiles.INSTRUCTION_START in text + assert harness_profiles.managed_instruction_text().strip() in text + + +def test_cursor_managed_plugin_rule_hook_surface(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + hooks_json = home / ".cursor" / "hooks.json" + hooks_json.parent.mkdir(parents=True) + foreign_entry = {"command": "/usr/local/bin/foreign-hook"} + hooks_json.write_text(json.dumps({"hooks": {"sessionStart": [foreign_entry]}})) + + assert cli.main(_sync_base(workspace, "cursor") + ["--write", "--json"]) == 0 + capsys.readouterr() + + rule = home / SURFACES["cursor"]["instruction"] + assert rule.read_text().startswith("---\nalwaysApply: true\n---\n") + assert (home / ".cursor" / "plugins" / "local" / "brigade-loop" / ".cursor-plugin" / "plugin.json").is_file() + hook_script = home / ".cursor" / "hooks" / "brigade-session-start" + assert hook_script.is_file() + assert hook_script.stat().st_mode & 0o111 + entries = json.loads(hooks_json.read_text())["hooks"]["sessionStart"] + assert foreign_entry in entries + assert {"command": str(hook_script)} in entries + + +def test_cursor_edited_managed_rule_reports_conflict_and_preserves_edit(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + assert cli.main(_sync_base(workspace, "cursor") + ["--write", "--json"]) == 0 + capsys.readouterr() + rule = home / SURFACES["cursor"]["instruction"] + rule.write_text(rule.read_text() + "user edit\n") + before = rule.read_bytes() + + assert cli.main(_sync_base(workspace, "cursor") + ["--write", "--json"]) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["results"][0]["status"] == "conflict" + assert any(item["surface"] == "instruction" for item in payload["results"][0]["conflicts"]) + assert rule.read_bytes() == before + + assert cli.main(_doctor_base(workspace, "cursor") + ["--json"]) == 1 + report = json.loads(capsys.readouterr().out) + assert report["results"][0]["instruction_ready"] is False + assert rule.read_bytes() == before + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_uninstall_removes_only_brigade_owned_artifacts(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + instruction = home / SURFACES[harness]["instruction"] + if harness in MARKED_BLOCK_HARNESSES: + instruction.parent.mkdir(parents=True) + instruction.write_text("# keep\n") + else: + hooks_json = home / ".cursor" / "hooks.json" + hooks_json.parent.mkdir(parents=True) + foreign_entry = {"command": "/usr/local/bin/foreign-hook"} + hooks_json.write_text(json.dumps({"hooks": {"sessionStart": [foreign_entry]}})) + + assert cli.main(_sync_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + + # Dry-run uninstall removes nothing. + assert cli.main(_uninstall_base(workspace, harness) + ["--json"]) == 0 + capsys.readouterr() + assert (home / SURFACES[harness]["state"]).is_file() + + assert cli.main(_uninstall_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + assert not (home / SURFACES[harness]["state"]).exists() + assert not (home / SURFACES[harness]["receipt"]).exists() + if harness in MARKED_BLOCK_HARNESSES: + assert instruction.read_text() == "# keep\n" + else: + assert not instruction.exists() + assert not (home / ".cursor" / "plugins").exists() + assert not (home / ".cursor" / "hooks" / "brigade-session-start").exists() + entries = json.loads((home / ".cursor" / "hooks.json").read_text())["hooks"]["sessionStart"] + assert entries == [foreign_entry] + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_doctor_reports_drift_without_mutating(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + assert cli.main(_sync_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + state = home / SURFACES[harness]["state"] + receipt = home / SURFACES[harness]["receipt"] + before = {path: (path.read_bytes(), path.stat().st_mtime_ns) for path in (state, receipt)} + + instruction = home / SURFACES[harness]["instruction"] + instruction.write_text(instruction.read_text().replace("brigade", "drifted", 1)) + + assert cli.main(_doctor_base(workspace, harness) + ["--json"]) == 1 + report = json.loads(capsys.readouterr().out) + assert report["results"][0]["ready"] is False + assert report["results"][0]["instruction_ready"] is False + assert {path: (path.read_bytes(), path.stat().st_mtime_ns) for path in (state, receipt)} == before + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_skill_install_round_trip(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + _add_reviewed_skill(workspace) + capsys.readouterr() + + assert cli.main(_sync_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + installed = home / SURFACES[harness]["skills"] / "reviewed" / "SKILL.md" + assert installed.is_file() + + assert cli.main(_uninstall_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + assert not installed.exists() + assert not (home / SURFACES[harness]["skills"] / "reviewed").exists() + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_mcp_stdio_requires_allow_global_stdio_gate(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, [SURFACES[harness]["mcp_target"]]) + config = home / SURFACES[harness]["mcp_config"] + + assert cli.main(_sync_base(workspace, harness) + ["--write", "--json"]) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["results"][0]["status"] == "conflict" + assert not config.exists() + + assert cli.main(_sync_base(workspace, harness) + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + assert config.is_file() + assert "brigade" in config.read_text() + + +@pytest.mark.parametrize("harness", SLICE2_HARNESSES) +def test_mcp_uninstall_removes_only_owned_server(tmp_path, monkeypatch, capsys, harness): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, [SURFACES[harness]["mcp_target"]]) + config = home / SURFACES[harness]["mcp_config"] + + assert cli.main(_sync_base(workspace, harness) + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + assert "brigade" in config.read_text() + + assert cli.main(_uninstall_base(workspace, harness) + ["--write", "--json"]) == 0 + capsys.readouterr() + assert "brigade" not in config.read_text() + + +def test_kimi_mcp_config_round_trip_preserves_foreign_keys(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, ["kimi"]) + config = home / ".kimi-code" / "mcp.json" + config.parent.mkdir(parents=True) + config.write_text(json.dumps({"theme": "dark", "mcpServers": {"foreign": {"command": "keep-me"}}})) + + assert cli.main(_sync_base(workspace, "kimi") + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + doc = json.loads(config.read_text()) + assert doc["theme"] == "dark" + assert doc["mcpServers"]["foreign"] == {"command": "keep-me"} + assert "brigade" in doc["mcpServers"] + + assert cli.main(_uninstall_base(workspace, "kimi") + ["--write", "--json"]) == 0 + capsys.readouterr() + doc = json.loads(config.read_text()) + assert doc["theme"] == "dark" + assert doc["mcpServers"] == {"foreign": {"command": "keep-me"}} + + +def test_kimi_probe_routes_sync_into_newer_surface(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, ["kimi"]) + (home / ".kimi").mkdir() + + assert cli.main(_sync_base(workspace, "kimi") + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + assert (home / ".kimi" / "AGENTS.md").is_file() + assert (home / ".kimi" / "mcp.json").is_file() + assert not (home / ".kimi-code").exists() + + +def test_opencode_doctor_verify_mcp_reports_projection_drift(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, ["opencode"]) + assert cli.main(_sync_base(workspace, "opencode") + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + + assert cli.main(_doctor_base(workspace, "opencode") + ["--verify-mcp", "--json"]) == 0 + capsys.readouterr() + + config = home / ".config" / "opencode" / "opencode.json" + config.write_text(config.read_text().replace("memory", "edited", 1)) + assert cli.main(_doctor_base(workspace, "opencode") + ["--verify-mcp", "--json"]) == 1 + report = json.loads(capsys.readouterr().out) + assert report["results"][0]["mcp"]["status"] == "conflict" + + +def test_cursor_sync_preserves_foreign_mcp_server(tmp_path, monkeypatch, capsys): + from brigade import cli + + home = _use_home(monkeypatch, tmp_path) + workspace = _workspace_with_stdio_server(tmp_path, capsys, ["cursor"]) + config = home / ".cursor" / "mcp.json" + config.parent.mkdir(parents=True) + config.write_text(json.dumps({"mcpServers": {"foreign": {"command": "keep-me"}}})) + + assert cli.main(_sync_base(workspace, "cursor") + ["--allow-global-stdio", "--write", "--json"]) == 0 + capsys.readouterr() + doc = json.loads(config.read_text()) + assert doc["mcpServers"]["foreign"] == {"command": "keep-me"} + assert "brigade" in doc["mcpServers"] + + +def test_target_all_dry_run_reports_seven_results(tmp_path, monkeypatch, capsys): + from brigade import cli + + _use_home(monkeypatch, tmp_path) + workspace = _workspace(tmp_path) + assert cli.main(_sync_base(workspace, "all") + ["--json"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert [result["harness"] for result in payload["results"]] == list(harness_profiles.USER_SCOPE_HARNESS_IDS) + assert len(payload["results"]) == 7