"""Comprehensive tests for ``muse coord intent``. Coverage -------- Unit — core helpers create_intent roundtrip: fields survive serialise/deserialise cycle Intent.to_dict keys: all expected keys present all valid ops: create_intent accepts each without error Intent class docstring coverage: all attributes accessible load_all_intents empty dir: returns [] load_all_intents corrupt file: skips corrupt, returns rest filter_intents by run_id: exact-match filter works filter_intents by operation: exact-match filter works filter_intents by address_glob: fnmatch filter works filter_intents combined: AND semantics across filters Integration — CLI basic intent: exit 0, text output contains intent_id --detail flag: detail echoed in text output --reservation-id flag: stored and echoed in text output all valid ops via CLI: each of the 8 ops exits 0 --json flag: exit 0, valid compact JSON, required keys --json shorthand: identical schema to --json flag unknown --op rejected: exits nonzero, error on stderr missing --op rejected: exits nonzero (argparse error) multiple addresses: addresses count reflected in text output no repo exits nonzero: MUSE_REPO_ROOT pointing at non-repo exits != 0 Input validation --run-id at max length: 256 chars accepted --run-id over max length: exits USER_ERROR (1), no file written --detail at max length: 4096 chars accepted --detail over max length: exits USER_ERROR (1), no file written too many addresses: exits USER_ERROR (1), no file written --reservation-id valid content ID: accepted and stored --reservation-id invalid ID: exits USER_ERROR (1), no file written --reservation-id empty string: accepted (standalone intent) validation fires before I/O: no intent file created on bad input Security ANSI in run_id sanitized: escape codes stripped in text output ANSI in detail sanitized: escape codes stripped in text output control chars in detail: stored but sanitized in text output path traversal in address: stored safely, no FS side-effects JSON output compact: no indent=2 pretty-printing Concurrent 20 threads writing intents: all exit 0, unique intent IDs Stress 500 intents < 5 s: throughput baseline load_all_intents 500 < 1 s: read-path baseline for 500 records 1000 addresses in one intent: accepted at boundary """ from __future__ import annotations import json import pathlib import itertools import threading import time import pytest from tests.cli_test_helper import CliRunner from muse.core.coordination import ( Intent, create_intent, filter_intents, load_all_intents, ) from muse.cli.commands.intent import _MAX_ADDRESSES, _MAX_DETAIL_LEN, _MAX_RUN_ID_LEN from muse.core.errors import ExitCode from muse.core.types import content_hash, fake_id from muse.core.paths import coordination_dir, muse_dir _id_seq = itertools.count() def _new_id() -> str: return content_hash({"seq": next(_id_seq)}) cli = None runner = CliRunner() _VALID_OPS = ["rename", "move", "modify", "extract", "delete", "inline", "split", "merge"] _REQUIRED_JSON_KEYS = { "schema_version", "intent_id", "reservation_id", "run_id", "branch", "addresses", "operation", "created_at", "detail", } # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) return tmp_path # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_intent( root: pathlib.Path, *, run_id: str = "agent-1", branch: str = "main", addresses: list[str] | None = None, operation: str = "modify", detail: str = "", reservation_id: str | None = None, ) -> Intent: return create_intent( root, reservation_id=reservation_id or _new_id(), run_id=run_id, branch=branch, addresses=addresses or ["src/mod.py::foo"], operation=operation, detail=detail, ) # --------------------------------------------------------------------------- # Unit — create_intent roundtrip # --------------------------------------------------------------------------- class TestCreateIntentRoundtrip: def test_roundtrip_preserves_intent_id(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path) loaded = load_all_intents(tmp_path) assert any(i.intent_id == it.intent_id for i in loaded) def test_roundtrip_preserves_addresses(self, tmp_path: pathlib.Path) -> None: addrs = ["src/a.py::foo", "src/b.py::bar"] it = _make_intent(tmp_path, addresses=addrs) loaded = load_all_intents(tmp_path) match = next(i for i in loaded if i.intent_id == it.intent_id) assert match.addresses == addrs def test_roundtrip_preserves_operation(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path, operation="rename") loaded = load_all_intents(tmp_path) match = next(i for i in loaded if i.intent_id == it.intent_id) assert match.operation == "rename" def test_roundtrip_preserves_detail(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path, detail="rename to foo_v2") loaded = load_all_intents(tmp_path) match = next(i for i in loaded if i.intent_id == it.intent_id) assert match.detail == "rename to foo_v2" def test_roundtrip_preserves_run_id(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path, run_id="agent-99") loaded = load_all_intents(tmp_path) match = next(i for i in loaded if i.intent_id == it.intent_id) assert match.run_id == "agent-99" class TestIntentToDictKeys: def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path) d = it.to_dict() assert _REQUIRED_JSON_KEYS.issubset(d.keys()) def test_intent_id_is_content_addressed(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path) assert it.intent_id.startswith("sha256:") and len(it.intent_id) == 71 def test_addresses_is_list(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path) assert isinstance(it.to_dict()["addresses"], list) def test_created_at_is_iso_string(self, tmp_path: pathlib.Path) -> None: it = _make_intent(tmp_path) created = it.to_dict()["created_at"] assert isinstance(created, str) and "T" in created class TestAllValidOps: @pytest.mark.parametrize("op", _VALID_OPS) def test_create_intent_accepts_op(self, tmp_path: pathlib.Path, op: str) -> None: it = _make_intent(tmp_path, operation=op) assert it.operation == op # --------------------------------------------------------------------------- # Integration — CLI # --------------------------------------------------------------------------- class TestIntentCLIBasic: def test_basic_intent_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"], ) assert result.exit_code == 0 def test_basic_intent_output_contains_intent_id_label(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"], ) assert "Intent ID" in result.output def test_basic_intent_output_contains_operation(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "rename", "--run-id", "agent-1"], ) assert "rename" in result.output def test_basic_intent_output_contains_run_id(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-42"], ) assert "agent-42" in result.output def test_detail_flag_echoed_in_output(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, [ "coord", "intent", "src/billing.py::compute_total", "--op", "rename", "--detail", "rename to compute_invoice_total", "--run-id", "agent-1", ], ) assert result.exit_code == 0 assert "rename to compute_invoice_total" in result.output def test_reservation_id_flag_echoed_in_output(self, repo: pathlib.Path) -> None: res_id = fake_id("echoed-reservation") result = runner.invoke( cli, [ "coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--reservation-id", res_id, "--run-id", "agent-1", ], ) assert result.exit_code == 0 assert res_id in result.output @pytest.mark.parametrize("op", _VALID_OPS) def test_all_ops_exit_zero(self, repo: pathlib.Path, op: str) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", op, "--run-id", "agent-1"], ) assert result.exit_code == 0, f"op={op!r} failed: {result.output}" def test_multiple_addresses_count_in_output(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, [ "coord", "intent", "src/a.py::foo", "src/b.py::bar", "src/c.py::baz", "--op", "modify", "--run-id", "agent-1", ], ) assert result.exit_code == 0 assert "3" in result.output class TestIntentCLIJSON: def test_json_flag_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) assert result.exit_code == 0 def test_json_flag_is_valid_json(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) data = json.loads(result.output) assert isinstance(data, dict) def test_json_flag_has_required_keys(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) data = json.loads(result.output) assert _REQUIRED_JSON_KEYS.issubset(data.keys()) def test_json_shorthand_identical_schema(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) data = json.loads(result.output) assert _REQUIRED_JSON_KEYS.issubset(data.keys()) def test_json_operation_field_matches_op_flag(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "extract", "--json"], ) data = json.loads(result.output) assert data["operation"] == "extract" def test_json_addresses_field_contains_address(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) data = json.loads(result.output) assert "src/billing.py::compute_total" in data["addresses"] def test_json_run_id_field_reflects_flag(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "bot-7", "--json"], ) data = json.loads(result.output) assert data["run_id"] == "bot-7" def test_json_detail_field_reflects_flag(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, [ "coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--detail", "tweak logic", "--json", ], ) data = json.loads(result.output) assert data["detail"] == "tweak logic" def test_json_intent_id_is_content_addressed(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"], ) data = json.loads(result.output) assert data["intent_id"].startswith("sha256:") and len(data["intent_id"]) == 71 class TestIntentCLIErrors: def test_unknown_op_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "explode", "--run-id", "agent-1"], ) assert result.exit_code != 0 def test_unknown_op_prints_error(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "explode", "--run-id", "agent-1"], ) combined = result.output + result.stderr assert "explode" in combined or "Unknown" in combined or "error" in combined.lower() def test_missing_op_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--run-id", "agent-1"], ) assert result.exit_code != 0 def test_no_repo_exits_nonzero(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: # Point MUSE_REPO_ROOT at a directory with no .muse folder. empty = tmp_path / "notarepo" empty.mkdir() monkeypatch.setenv("MUSE_REPO_ROOT", str(empty)) result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"], ) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestIntentSecurity: def test_ansi_in_run_id_not_reflected_verbatim(self, repo: pathlib.Path) -> None: ansi_run_id = "\x1b[31magent-malicious\x1b[0m" result = runner.invoke( cli, ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", ansi_run_id], ) # The CliRunner already strips ANSI, so the raw escape byte must not appear. assert "\x1b[31m" not in result.output assert "\x1b[0m" not in result.output def test_control_chars_in_detail_does_not_crash(self, repo: pathlib.Path) -> None: # The intent command stores detail verbatim and echoes it in text output. # This test confirms the command completes without crashing when control # characters appear in --detail — no exception, exit code 0. malicious_detail = "ok\x07\x1b[2Jclear" result = runner.invoke( cli, [ "coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--detail", malicious_detail, "--run-id", "agent-1", ], ) assert result.exit_code == 0 # The intent was recorded despite control chars in the detail field. assert "Intent ID" in result.output def test_path_traversal_in_address_stored_safely(self, repo: pathlib.Path) -> None: traversal_addr = "../../etc/passwd::shadow" result = runner.invoke( cli, ["coord", "intent", traversal_addr, "--op", "modify", "--run-id", "agent-1", "--json"], ) # Command should complete without writing outside the repo tree. assert result.exit_code == 0 data = json.loads(result.output) # The address is stored as-is (advisory); no actual file access occurred. assert traversal_addr in data["addresses"] # Verify the etc/passwd file was not modified. assert not pathlib.Path("/etc/passwd_shadow").exists() def test_control_chars_in_detail_stored_in_json_field(self, repo: pathlib.Path) -> None: # Storage should preserve the raw value; display sanitizes it. malicious_detail = "ok\x07bad" result = runner.invoke( cli, [ "coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--detail", malicious_detail, "--json", ], ) assert result.exit_code == 0 data = json.loads(result.output) # The detail field stores the value; BEL is a valid JSON character. assert "ok" in data["detail"] # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestIntentStress: def test_100_intents_under_2_seconds(self, tmp_path: pathlib.Path) -> None: start = time.monotonic() for i in range(100): _make_intent( tmp_path, run_id=f"agent-{i}", addresses=[f"src/mod{i}.py::sym"], operation=_VALID_OPS[i % len(_VALID_OPS)], ) elapsed = time.monotonic() - start assert elapsed < 2.0, f"100 intents took {elapsed:.2f}s (limit 2s)" def test_load_all_intents_under_0_5_seconds(self, tmp_path: pathlib.Path) -> None: for i in range(100): _make_intent( tmp_path, run_id=f"agent-{i}", addresses=[f"src/mod{i}.py::sym"], ) start = time.monotonic() intents = load_all_intents(tmp_path) elapsed = time.monotonic() - start assert len(intents) == 100 assert elapsed < 0.5, f"load_all_intents (100 records) took {elapsed:.2f}s (limit 0.5s)" def test_500_intents_under_5_seconds(self, tmp_path: pathlib.Path) -> None: start = time.monotonic() for i in range(500): _make_intent( tmp_path, run_id=f"agent-{i}", addresses=[f"src/mod{i}.py::sym"], operation=_VALID_OPS[i % len(_VALID_OPS)], ) elapsed = time.monotonic() - start assert elapsed < 5.0, f"500 intents took {elapsed:.2f}s (limit 5s)" def test_load_all_intents_500_under_1_second(self, tmp_path: pathlib.Path) -> None: for i in range(500): _make_intent( tmp_path, run_id=f"agent-{i}", addresses=[f"src/mod{i}.py::sym"], ) start = time.monotonic() intents = load_all_intents(tmp_path) elapsed = time.monotonic() - start assert len(intents) == 500 assert elapsed < 1.0, f"load_all_intents (500 records) took {elapsed:.2f}s (limit 1s)" def test_1000_addresses_at_boundary(self, repo: pathlib.Path) -> None: addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES)] result = runner.invoke( cli, ["coord", "intent"] + addresses + ["--op", "modify", "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert len(data["addresses"]) == _MAX_ADDRESSES # --------------------------------------------------------------------------- # Input validation # --------------------------------------------------------------------------- class TestIntentInputValidation: def test_run_id_at_max_length_accepted(self, repo: pathlib.Path) -> None: run_id = "a" * _MAX_RUN_ID_LEN result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id, "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert data["run_id"] == run_id def test_run_id_over_max_length_exits_user_error(self, repo: pathlib.Path) -> None: run_id = "a" * (_MAX_RUN_ID_LEN + 1) result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id], ) assert result.exit_code == ExitCode.USER_ERROR def test_run_id_over_max_no_file_written(self, repo: pathlib.Path) -> None: run_id = "a" * (_MAX_RUN_ID_LEN + 1) runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id], ) intents_dir = coordination_dir(repo) / "intents" assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == [] def test_detail_at_max_length_accepted(self, repo: pathlib.Path) -> None: detail = "x" * _MAX_DETAIL_LEN result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail, "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert data["detail"] == detail def test_detail_over_max_length_exits_user_error(self, repo: pathlib.Path) -> None: detail = "x" * (_MAX_DETAIL_LEN + 1) result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail], ) assert result.exit_code == ExitCode.USER_ERROR def test_detail_over_max_no_file_written(self, repo: pathlib.Path) -> None: detail = "x" * (_MAX_DETAIL_LEN + 1) runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail], ) intents_dir = coordination_dir(repo) / "intents" assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == [] def test_too_many_addresses_exits_user_error(self, repo: pathlib.Path) -> None: addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES + 1)] result = runner.invoke( cli, ["coord", "intent"] + addresses + ["--op", "modify"], ) assert result.exit_code == ExitCode.USER_ERROR def test_too_many_addresses_no_file_written(self, repo: pathlib.Path) -> None: addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES + 1)] runner.invoke( cli, ["coord", "intent"] + addresses + ["--op", "modify"], ) intents_dir = coordination_dir(repo) / "intents" assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == [] def test_valid_reservation_id_accepted(self, repo: pathlib.Path) -> None: res_id = fake_id("valid-reservation") result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--reservation-id", res_id, "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert data["reservation_id"] == res_id def test_invalid_reservation_id_exits_user_error(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--reservation-id", "not-a-content-id"], ) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_reservation_id_no_file_written(self, repo: pathlib.Path) -> None: runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--reservation-id", "not-a-content-id"], ) intents_dir = coordination_dir(repo) / "intents" assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == [] def test_empty_reservation_id_creates_standalone_intent(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"], ) assert result.exit_code == 0 # no --reservation-id means standalone; field present but empty or omitted data = json.loads(result.output) assert "reservation_id" in data def test_validation_fires_before_io_on_bad_run_id(self, repo: pathlib.Path) -> None: """Validates that bad run-id rejects without touching the intents directory.""" run_id = "z" * (_MAX_RUN_ID_LEN + 1) intents_dir = coordination_dir(repo) / "intents" result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id], ) assert result.exit_code == ExitCode.USER_ERROR # The intents directory must not have been created. assert not intents_dir.exists() def test_run_id_over_max_json_mode_returns_error_field(self, repo: pathlib.Path) -> None: run_id = "a" * (_MAX_RUN_ID_LEN + 1) result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id, "--json"], ) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert "error" in data def test_invalid_reservation_id_json_mode_returns_error_field(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--reservation-id", "bad-id", "--json"], ) assert result.exit_code == ExitCode.USER_ERROR data = json.loads(result.output) assert "error" in data assert data.get("status") == "bad_reservation_id" # --------------------------------------------------------------------------- # JSON format — compact output # --------------------------------------------------------------------------- class TestIntentJsonFormat: def test_json_output_is_compact(self, repo: pathlib.Path) -> None: """No pretty-printing (no indent=2): body must be a single line.""" result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"], ) assert result.exit_code == 0 body = result.output.strip() assert "\n" not in body, "JSON output must be compact (no newlines)" def test_json_schema_version_present(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"], ) data = json.loads(result.output) assert "schema_version" in data assert isinstance(data["schema_version"], str) def test_json_branch_field_present(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"], ) data = json.loads(result.output) assert "branch" in data assert isinstance(data["branch"], str) and data["branch"] def test_json_created_at_is_iso_string(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"], ) data = json.loads(result.output) assert "T" in data["created_at"] def test_json_two_invocations_produce_unique_intent_ids(self, repo: pathlib.Path) -> None: # intent_id is content-addressed: different addresses → different IDs r1 = runner.invoke(cli, ["coord", "intent", "src/mod.py::sym_a", "--op", "modify", "--json"]) r2 = runner.invoke(cli, ["coord", "intent", "src/mod.py::sym_b", "--op", "modify", "--json"]) id1 = json.loads(r1.output)["intent_id"] id2 = json.loads(r2.output)["intent_id"] assert id1 != id2 def test_json_error_on_bad_detail_is_compact(self, repo: pathlib.Path) -> None: detail = "x" * (_MAX_DETAIL_LEN + 1) result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail, "--json"], ) body = result.output.strip() assert "\n" not in body data = json.loads(body) assert "error" in data # --------------------------------------------------------------------------- # Sanitize display — ANSI/control chars stripped in text output # --------------------------------------------------------------------------- class TestIntentSanitize: def test_ansi_in_run_id_stripped_from_text_output(self, repo: pathlib.Path) -> None: ansi_id = "\x1b[31mmalicious\x1b[0m" result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", ansi_id], ) assert result.exit_code == 0 assert "\x1b[31m" not in result.output assert "\x1b[0m" not in result.output def test_ansi_in_detail_stripped_from_text_output(self, repo: pathlib.Path) -> None: ansi_detail = "\x1b[32mgreen detail\x1b[0m" result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", ansi_detail, "--run-id", "agent-1"], ) assert result.exit_code == 0 assert "\x1b[32m" not in result.output assert "\x1b[0m" not in result.output def test_ansi_in_reservation_id_stripped_from_text_output(self, repo: pathlib.Path) -> None: # ANSI codes can't appear in a content ID (validation would reject them), # so verify the reservation_id is echoed cleanly (no raw escape sequences). res_id = fake_id("ansi-sanitize-reservation") result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--reservation-id", res_id, "--run-id", "agent-1"], ) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_bel_in_detail_does_not_appear_in_text_output(self, repo: pathlib.Path) -> None: detail = "ok\x07bad" result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail, "--run-id", "agent-1"], ) assert result.exit_code == 0 assert "\x07" not in result.output # --------------------------------------------------------------------------- # filter_intents — unit tests # --------------------------------------------------------------------------- class TestFilterIntents: def _make_n(self, root: pathlib.Path, n: int) -> list[Intent]: ops = _VALID_OPS return [ _make_intent( root, run_id=f"agent-{i % 3}", addresses=[f"src/mod{i % 5}.py::sym", "shared.py::util"], operation=ops[i % len(ops)], detail=f"detail-{i}", ) for i in range(n) ] def test_filter_by_run_id_returns_subset(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 9) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, run_id="agent-0") assert all(i.run_id == "agent-0" for i in filtered) assert len(filtered) > 0 def test_filter_by_run_id_excludes_others(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 9) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, run_id="agent-1") assert all(i.run_id == "agent-1" for i in filtered) assert not any(i.run_id == "agent-0" for i in filtered) def test_filter_by_operation_returns_subset(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 16) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, operation="rename") assert all(i.operation == "rename" for i in filtered) assert len(filtered) > 0 def test_filter_by_address_glob_returns_matches(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 10) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, address_glob="src/mod0.py::*") assert all(any("src/mod0.py" in a for a in i.addresses) for i in filtered) assert len(filtered) > 0 def test_filter_combined_and_semantics(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 24) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, run_id="agent-0", operation="rename") assert all(i.run_id == "agent-0" and i.operation == "rename" for i in filtered) def test_filter_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 5) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents, run_id="nonexistent-agent") assert filtered == [] def test_filter_empty_list_returns_empty(self, tmp_path: pathlib.Path) -> None: filtered = filter_intents([], run_id="agent-0") assert filtered == [] def test_filter_no_filters_returns_all(self, tmp_path: pathlib.Path) -> None: self._make_n(tmp_path, 6) all_intents = load_all_intents(tmp_path) filtered = filter_intents(all_intents) assert len(filtered) == len(all_intents) # --------------------------------------------------------------------------- # Concurrent writes # --------------------------------------------------------------------------- class TestIntentConcurrent: def test_20_threads_all_exit_zero(self, repo: pathlib.Path) -> None: results: list[int] = [] lock = threading.Lock() def _write() -> None: result = runner.invoke( cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", f"agent-{threading.get_ident()}", "--json"], ) with lock: results.append(result.exit_code) threads = [threading.Thread(target=_write) for _ in range(20)] for t in threads: t.start() for t in threads: t.join() assert all(c == 0 for c in results), f"some threads failed: {results}" def test_20_threads_produce_unique_intent_ids(self, tmp_path: pathlib.Path) -> None: """Uses create_intent directly to avoid CliRunner stdout-capture collision.""" intent_ids: list[str] = [] lock = threading.Lock() def _write(i: int) -> None: it = create_intent( tmp_path, reservation_id="", run_id=f"agent-concurrent-{i}", branch="main", addresses=["src/mod.py::sym"], operation="modify", detail="", ) with lock: intent_ids.append(it.intent_id) threads = [threading.Thread(target=_write, args=(i,)) for i in range(20)] for t in threads: t.start() for t in threads: t.join() assert len(intent_ids) == 20 assert len(set(intent_ids)) == 20, "intent IDs must be globally unique"