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
54 changes: 38 additions & 16 deletions src/brigade/roadmap_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -524,6 +524,23 @@ def _deferred_item_checks(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
]


def _target_owns_brigade_cli(target: Path) -> bool:
"""True when the target repo ships the brigade CLI itself.

The brigade-catalog checks (is every brigade CLI command documented, is
docs/command-inventory.md current against the full command surface) only
make sense for the repo that owns and documents the CLI. A consumer repo
that merely uses brigade as a library or work-loop tool should not be
audited against brigade's own command surface, which otherwise reports
hundreds of "undocumented" commands it never owned.
"""
try:
text = (target / "pyproject.toml").read_text()
except OSError:
return False
return re.search(r'(?m)^\s*name\s*=\s*"brigade-cli"\s*$', text) is not None


def audit_payload(target: Path) -> dict[str, Any]:
target = target.expanduser().resolve()
roadmap = _parse_roadmap(target)
Expand All @@ -534,6 +551,7 @@ def audit_payload(target: Path) -> dict[str, Any]:
checks.append({"status": OK, "name": "roadmap_exists", "detail": roadmap["path"]})
checks.extend(_section_stale_checks(roadmap["sections"]))

owns_cli = _target_owns_brigade_cli(target)
documented = _documented_brigade_commands(target)
cli_commands = _cli_command_paths()
cli_prefixes = _cli_command_prefixes(cli_commands)
Expand All @@ -545,7 +563,7 @@ def audit_payload(target: Path) -> dict[str, Any]:
for command in documented
if "..." not in command and _normalize_documented_command(command, cli_prefixes) not in cli_prefixes
)
missing_docs = sorted(command for command in cli_set if command not in documented_set)
# A bogus `brigade ...` command in the docs is worth flagging in any repo.
checks.append(
{
"status": WARN if missing_cli else OK,
Expand All @@ -554,21 +572,25 @@ def audit_payload(target: Path) -> dict[str, Any]:
"commands": missing_cli[:20],
}
)
checks.append(
{
"status": WARN if missing_docs else OK,
"name": "roadmap_cli_command_missing_docs",
"detail": f"{len(missing_docs)} CLI command(s) missing from public docs" if missing_docs else "none",
"commands": missing_docs[:20],
}
)
command_contract = command_contract_payload(target)
inventory_check = next(
(check for check in command_contract["checks"] if check.get("name") == "roadmap_command_inventory_current"),
None,
)
if inventory_check:
checks.append(dict(inventory_check))
# The reverse checks (every CLI command must be documented, the command
# inventory must be current) only apply to the repo that owns the CLI.
missing_docs = sorted(command for command in cli_set if command not in documented_set) if owns_cli else []
if owns_cli:
checks.append(
{
"status": WARN if missing_docs else OK,
"name": "roadmap_cli_command_missing_docs",
"detail": f"{len(missing_docs)} CLI command(s) missing from public docs" if missing_docs else "none",
"commands": missing_docs[:20],
}
)
command_contract = command_contract_payload(target)
inventory_check = next(
(check for check in command_contract["checks"] if check.get("name") == "roadmap_command_inventory_current"),
None,
)
if inventory_check:
checks.append(dict(inventory_check))
deferred_items = _active_queue_items()
archived_items = _archived_items()
checks.extend(_deferred_item_checks(deferred_items))
Expand Down
28 changes: 25 additions & 3 deletions src/brigade/work_cmd/scanners.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,11 @@ def _scanner_validate_import_output(
) -> tuple[Path | None, list[dict[str, Any]], list[str]]:
import_path = config_mod._scanner_import_path(target, scanner)
if import_path is None:
return None, [], [f"{scanner.get('id')}: import_path is not configured"]
# No import_path means a self-importing scanner: its command appends to
# the inbox directly (e.g. `brigade work import memory-refresh`,
# `brigade handoff sync-issues`), so there is no JSONL file for the sweep
# to ingest. Skip it silently rather than failing the whole sweep.
return None, [], []
if scanner.get("import_format", "jsonl") != "jsonl":
return import_path, [], [f"{scanner.get('id')}: import_format must be jsonl"]
if not import_path.is_file():
Expand Down Expand Up @@ -436,6 +440,23 @@ def _scanner_plan_payload(target: Path) -> dict[str, Any]:
}


def _required_scanner_ids(target: Path) -> tuple[str, ...]:
"""Required local-producer scanner ids for this target.

chat-memory-sweep only matters when the repo actually has an enabled chat
surface. A code repo with every surface disabled (or no chat-surfaces.toml
at all) has nothing to sweep, so it should not be nagged to enable it.
"""
from .. import chat_cmd

surfaces = chat_cmd.health(target).get("surfaces")
surfaces = surfaces if isinstance(surfaces, list) else []
chat_active = any(isinstance(surface, dict) and surface.get("enabled") for surface in surfaces)
return tuple(
scanner_id for scanner_id in constants.SCANNER_REQUIRED_IDS if scanner_id != "chat-memory-sweep" or chat_active
)


def _scanner_health(target: Path) -> dict[str, Any]:
target = target.expanduser().resolve()
plan = _scanner_plan_payload(target)
Expand All @@ -455,10 +476,11 @@ def _scanner_health(target: Path) -> dict[str, Any]:
checks.append({"status": constants.FAIL, "name": "scanner_config", "detail": "; ".join(plan.get("errors", []))})

by_id = {scanner.get("id"): scanner for scanner in scanners if isinstance(scanner, dict)}
missing_required = [scanner_id for scanner_id in constants.SCANNER_REQUIRED_IDS if scanner_id not in by_id]
required_ids = _required_scanner_ids(target)
missing_required = [scanner_id for scanner_id in required_ids if scanner_id not in by_id]
disabled_required = [
scanner_id
for scanner_id in constants.SCANNER_REQUIRED_IDS
for scanner_id in required_ids
if isinstance(by_id.get(scanner_id), dict) and not by_id[scanner_id].get("enabled", True)
]
if missing_required or disabled_required:
Expand Down
19 changes: 19 additions & 0 deletions tests/test_roadmap_inventory_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ def test_roadmap_commands_check_reports_stale_inventory(tmp_path, capsys):


def test_roadmap_audit_includes_command_inventory_drift(tmp_path):
# The command-inventory drift check only applies to the repo that owns the
# brigade CLI, so mark this target as brigade-cli.
(tmp_path / "pyproject.toml").write_text('[project]\nname = "brigade-cli"\n')
(tmp_path / "README.md").write_text("Use `brigade roadmap commands`.\n")
(tmp_path / "ROADMAP.md").write_text("# Roadmap\n")

Expand All @@ -48,6 +51,22 @@ def test_roadmap_audit_includes_command_inventory_drift(tmp_path):
assert any(issue["name"] == "roadmap_command_inventory_current" for issue in payload["issues"])


def test_roadmap_audit_skips_command_catalog_checks_for_consumer_repo(tmp_path):
# A repo that merely consumes brigade (no brigade-cli pyproject) must not be
# audited against brigade's own command surface.
(tmp_path / "README.md").write_text("Use `brigade work verify run`.\n")
(tmp_path / "ROADMAP.md").write_text("# Roadmap\n")

payload = roadmap_cmd.audit_payload(tmp_path)

names = {check["name"] for check in payload["checks"]}
assert "roadmap_cli_command_missing_docs" not in names
assert "roadmap_command_inventory_current" not in names
# The universal "documented a command that does not exist" check still runs.
assert "roadmap_documented_command_missing_cli" in names
assert payload["missing_documented_commands"] == []


def test_roadmap_commands_cli_dispatch(tmp_path, monkeypatch):
seen = []

Expand Down
122 changes: 122 additions & 0 deletions tests/test_work_cmd_scanners.py
Original file line number Diff line number Diff line change
Expand Up @@ -717,3 +717,125 @@ def fake_scanners_run_show(**kwargs):
("runs", {"target": tmp_path, "json_output": True, "limit": 5}),
("run-show", {"target": tmp_path, "run_id": "run-1", "json_output": True}),
]


def _required_scanner_config(tmp_path: Path, *, chat_enabled: bool) -> None:
config = tmp_path / ".brigade" / "scanners.toml"
config.parent.mkdir(parents=True, exist_ok=True)
chat = "true" if chat_enabled else "false"
config.write_text(
f"""
[[scanner]]
id = "chat-memory-sweep"
source = "chat-memory-sweep"
command = "brigade work import chat-sweep --json"
cadence = "daily@02:15"
enabled = {chat}
timeout = 300
output_path = ".brigade/chat-memory-sweeps/latest.json"
conflict_window = "02:00-02:30"

[[scanner]]
id = "memory-refresh"
source = "memory-refresh"
command = "brigade work import memory-refresh --json"
cadence = "daily@02:45"
enabled = true
timeout = 300
output_path = "memory/cards/decay/refresh-queue.json"
conflict_window = "02:30-03:00"

[[scanner]]
id = "handoff-ingest"
source = "handoff-ingest"
command = "brigade handoff sync-issues --json"
cadence = "hourly@15"
enabled = true
timeout = 180
output_path = ".brigade/handoff-sources.json"
conflict_window = "00:10-00:25"
"""
)


def _scanner_required_check(tmp_path: Path, capsys) -> dict:
assert work_cmd.scanners_doctor(target=tmp_path, json_output=True) in (0, 1)
payload = json.loads(capsys.readouterr().out)
return next(check for check in payload["checks"] if check["name"] == "scanner_required")


def test_scanner_required_ignores_chat_sweep_without_enabled_surface(tmp_path, capsys):
_init_git_repo(tmp_path)
_required_scanner_config(tmp_path, chat_enabled=False)

# No chat-surfaces.toml at all: chat-memory-sweep has nothing to sweep, so a
# disabled chat-memory-sweep must not raise scanner_required.
check = _scanner_required_check(tmp_path, capsys)
assert check["status"] == "ok"


def test_scanner_required_flags_chat_sweep_when_surface_enabled(tmp_path, capsys):
_init_git_repo(tmp_path)
_required_scanner_config(tmp_path, chat_enabled=False)
surfaces = tmp_path / ".brigade" / "chat-surfaces.toml"
surfaces.write_text(
"""
[[surface]]
id = "discord-export"
provider = "discord-export"
enabled = true
"""
)

# With a live chat surface, a disabled chat-memory-sweep is a real gap.
check = _scanner_required_check(tmp_path, capsys)
assert check["status"] == "warn"
assert "disabled=chat-memory-sweep" in check["detail"]


def test_scanners_run_ingest_skips_self_importing_scanner(tmp_path, capsys):
_init_git_repo(tmp_path)
script = tmp_path / "self_import.py"
script.write_text(
"""
import json
from pathlib import Path

inbox = Path.cwd() / ".brigade" / "work" / "imports" / "inbox.jsonl"
inbox.parent.mkdir(parents=True, exist_ok=True)
record = {
"id": "self-import-1",
"kind": "task",
"source": "self-import",
"text": "self-imported work",
"status": "pending",
"created_at": "2026-05-28T12:00:00+00:00",
"updated_at": "2026-05-28T12:00:00+00:00",
}
inbox.write_text(json.dumps(record) + "\\n")
print("self import complete")
"""
)
config = tmp_path / ".brigade" / "scanners.toml"
config.parent.mkdir(parents=True, exist_ok=True)
# No import_path: this scanner appends to the inbox itself. The ingest step
# must skip it, not fail the whole sweep.
config.write_text(
f"""
[[scanner]]
id = "self-import"
source = "self-import"
command = "{sys.executable} {script}"
cadence = "daily@02:00"
enabled = true
timeout = 30
output_path = ".brigade/self-import-output.json"
conflict_window = "02:00-02:10"
"""
)

assert work_cmd.scanners_run(target=tmp_path, scanner_id="self-import", ingest_output=True, json_output=True) == 0
payload = json.loads(capsys.readouterr().out)
assert payload["completed"] == 1
assert payload["failed"] == 0
assert payload["ingest_errors"] == []