from __future__ import annotations import json from pathlib import Path from recoil.pipeline.tools import linear_drain_headless as drain_headless from recoil.pipeline.tools import linear_queue as lq def _event(index: int = 1, **overrides: object) -> dict: values = { "source_type": "manual_file", "category": "bug", "title": f"Drain finding {index}", "evidence": f"Evidence {index}", "recommendation": f"Recommendation {index}", "file": f"recoil/file_{index}.py", "event_ts": f"2026-06-03T00:00:{index:02d}Z", } values.update(overrides) return lq.build_event(**values) def _write_event(root: Path, event: dict) -> Path: return lq.atomic_write_inbox(event, inbox=root / "inbox") def _ledger_rows(root: Path) -> list[dict]: return lq.load_ledger(ledger=root / "ledger.jsonl") def _inbox_names(root: Path) -> list[str]: return sorted(path.name for path in (root / "inbox").glob("*.json")) class FakeLinear: def __init__( self, *, open_by_key: dict[str, dict] | None = None, fail_create: bool = False, ): self.open_by_key = open_by_key or {} self.fail_create = fail_create self.calls: list[dict] = [] self.created_payloads: list[dict] = [] def __call__(self, _url, _headers, payload): self.calls.append(payload) query = payload["query"] variables = payload["variables"] if "LinearFindingKey" in query: key = variables["q"].removeprefix("finding_key:") issue = self.open_by_key.get(key) return { "data": { "issues": { "nodes": [ { "id": issue["id"], "identifier": issue["identifier"], "url": issue.get("url", ""), "state": {"type": "started"}, } ] if issue else [] } } } if "AutonomyComment" in query: return {"data": {"commentCreate": {"success": True}}} if "LinearTeamByName" in query: return {"data": {"teams": {"nodes": [{"id": "team-1", "name": "Recoil"}]}}} if "LinearIssueLabels" in query: return { "data": { "issueLabels": { "nodes": [{"id": "label-auto", "name": "auto-filed"}] } } } if "LinearCreateIssue" in query: if self.fail_create: raise RuntimeError("boom") self.created_payloads.append(payload) identifier = f"REC-{100 + len(self.created_payloads)}" return { "data": { "issueCreate": { "success": True, "issue": { "id": f"issue-{len(self.created_payloads)}", "identifier": identifier, "url": f"https://linear.app/recoil/issue/{identifier}", }, } } } raise AssertionError(query) def count(self, needle: str) -> int: return sum(1 for call in self.calls if needle in call["query"]) def test_dedup_path_terminal_ledger_moves_to_done_without_create(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() _write_event(tmp_path, event) lq.append_ledger( { "schema_version": 1, "ledger_ts": "2026-06-03T00:00:00Z", "finding_key": event["finding_key"], "event_id": "prior", "state": "filed", "linear_issue": "REC-1", "reason": None, }, ledger=tmp_path / "ledger.jsonl", ) fake = FakeLinear() summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) assert summary["exit_code"] == 0 assert summary["deduped"] == 1 assert fake.calls == [] assert _inbox_names(tmp_path) == [] assert (tmp_path / "done" / f"{event['event_id']}.json").exists() assert _ledger_rows(tmp_path)[-1]["state"] == "deduped" assert _ledger_rows(tmp_path)[-1]["reason"] == "ledger already handled" def test_create_path_writes_pending_then_filed_and_applies_label(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() event["file"] = None event["pr_url"] = None _write_event(tmp_path, event) fake = FakeLinear() summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) assert summary["exit_code"] == 0 assert summary["created"] == 1 rows = _ledger_rows(tmp_path) assert [row["state"] for row in rows] == ["pending", "filed"] assert rows[1]["linear_issue"] == "REC-101" create_vars = fake.created_payloads[0]["variables"] assert create_vars["labelIds"] == ["label-auto"] assert create_vars["title"] == event["title"] assert f"finding_key:{event['finding_key']}" in create_vars["description"] assert "- file: " in create_vars["description"] assert "- file: None" not in create_vars["description"] assert "- pr_url: " in create_vars["description"] assert "- pr_url: None" not in create_vars["description"] assert _inbox_names(tmp_path) == [] assert (tmp_path / "done" / f"{event['event_id']}.json").exists() def test_cap_creates_only_max_and_leaves_later_events(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") events = [_event(index) for index in range(1, 5)] for event in events: _write_event(tmp_path, event) fake = FakeLinear() summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=2, transport=fake, ) assert summary["exit_code"] == 0 assert summary["created"] == 2 assert len(fake.created_payloads) == 2 assert len(_inbox_names(tmp_path)) == 2 assert len(list((tmp_path / "done").glob("*.json"))) == 2 assert len(summary["remaining"]) == 2 def test_backstop_open_issue_dedupes_without_create(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() _write_event(tmp_path, event) lq.append_ledger( { "schema_version": 1, "ledger_ts": "2026-06-03T00:00:00Z", "finding_key": event["finding_key"], "event_id": event["event_id"], "state": "pending", "linear_issue": None, "reason": None, }, ledger=tmp_path / "ledger.jsonl", ) fake = FakeLinear( open_by_key={ event["finding_key"]: {"id": "issue-open", "identifier": "REC-8"} } ) summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) assert summary["exit_code"] == 0 assert summary["deduped"] == 1 assert fake.count("LinearCreateIssue") == 0 assert fake.count("AutonomyComment") == 1 assert _ledger_rows(tmp_path)[-1]["linear_issue"] == "REC-8" assert _ledger_rows(tmp_path)[-1]["reason"] == "linear backstop finding_key match" assert _inbox_names(tmp_path) == [] def test_malformed_event_quarantined_and_valid_still_processed(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() _write_event(tmp_path, event) malformed = tmp_path / "inbox" / "malformed.json" malformed.write_text("{not-json", encoding="utf-8") fake = FakeLinear() summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) assert summary["exit_code"] == 0 assert summary["created"] == 1 assert len(summary["quarantined"]) == 1 assert (tmp_path / "quarantine" / "malformed.json").exists() assert _inbox_names(tmp_path) == [] def test_dry_run_prints_summary_without_transport_calls_or_fs_mutation( tmp_path, monkeypatch, capsys, ): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() _write_event(tmp_path, event) before = { path.relative_to(tmp_path): path.read_text(encoding="utf-8") for path in tmp_path.rglob("*") if path.is_file() } fake = FakeLinear() rc = drain_headless.main( ["--dry-run", "--queue-root", str(tmp_path)], transport=fake, ) after = { path.relative_to(tmp_path): path.read_text(encoding="utf-8") for path in tmp_path.rglob("*") if path.is_file() } assert rc == 0 assert fake.calls == [] assert before == after assert "dry_run_summary" in capsys.readouterr().out def test_live_without_linear_api_key_exits_nonzero_without_writes(tmp_path, monkeypatch): monkeypatch.delenv("LINEAR_API_KEY", raising=False) event = _event() event_path = _write_event(tmp_path, event) rc = drain_headless.main(["--live", "--queue-root", str(tmp_path)]) assert rc != 0 assert event_path.exists() assert not (tmp_path / "ledger.jsonl").exists() assert not (tmp_path / "done").exists() def test_create_failure_leaves_pending_and_stops_with_later_events(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") first = _event(1) second = _event(2) _write_event(tmp_path, first) _write_event(tmp_path, second) fake = FakeLinear(fail_create=True) summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) rows = _ledger_rows(tmp_path) assert summary["exit_code"] != 0 assert [row["state"] for row in rows] == ["pending"] assert not any(row["state"] == "filed" for row in rows) assert (tmp_path / "inbox" / f"{first['event_id']}.json").exists() assert (tmp_path / "inbox" / f"{second['event_id']}.json").exists() assert not (tmp_path / "done").exists() def test_non_positive_max_exits_nonzero_without_writes(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() event_path = _write_event(tmp_path, event) assert ( drain_headless.main(["--live", "--max", "0", "--queue-root", str(tmp_path)]) != 0 ) assert ( drain_headless.main(["--live", "--max", "-1", "--queue-root", str(tmp_path)]) != 0 ) assert event_path.exists() assert not (tmp_path / "ledger.jsonl").exists() assert not (tmp_path / "done").exists() def test_nullable_fields_render_blank_not_none(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() event["subject_kind"] = None event["subject_id"] = None event["file"] = None event["pr_url"] = None event["branch"] = None event["head_sha"] = None _write_event(tmp_path, event) fake = FakeLinear() summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=fake, ) assert summary["exit_code"] == 0 body = fake.created_payloads[0]["variables"]["description"] for field in ("subject_kind", "subject_id", "file", "pr_url", "branch", "head_sha"): assert f"- {field}: " in body assert f"- {field}: None" not in body def test_queue_root_isolation_uses_only_explicit_paths(tmp_path, monkeypatch): monkeypatch.setenv("LINEAR_API_KEY", "lin_test") event = _event() _write_event(tmp_path, event) touched: list[Path] = [] original_read = drain_headless.lq.read_inbox original_load = drain_headless.lq.load_ledger original_append = drain_headless.lq.append_ledger original_move = drain_headless.lq.move_to_done def read_spy(*, inbox): touched.append(Path(inbox)) return original_read(inbox=inbox) def load_spy(*, ledger): touched.append(Path(ledger)) return original_load(ledger=ledger) def append_spy(row, *, ledger): touched.append(Path(ledger)) return original_append(row, ledger=ledger) def move_spy(event_id, *, inbox, done): touched.extend([Path(inbox), Path(done)]) return original_move(event_id, inbox=inbox, done=done) monkeypatch.setattr(drain_headless.lq, "read_inbox", read_spy) monkeypatch.setattr(drain_headless.lq, "load_ledger", load_spy) monkeypatch.setattr(drain_headless.lq, "append_ledger", append_spy) monkeypatch.setattr(drain_headless.lq, "move_to_done", move_spy) summary = drain_headless.drain( queue_root=tmp_path, live=True, max_creates=3, transport=FakeLinear(), ) assert summary["exit_code"] == 0 assert touched assert all(path == tmp_path or tmp_path in path.parents for path in touched) assert all(path != lq.DEFAULT_QUEUE_ROOT for path in touched) assert json.loads((tmp_path / "done" / f"{event['event_id']}.json").read_text())