from dataclasses import dataclass from typing import List, Optional from pathlib import Path import json @dataclass class Bead: id: str title: str description: str status: str = "active" class BeadsClient: def __init__(self, working_dir: Path): """ [C: src/mcp_client.py:_DDGParser.__init__, src/mcp_client.py:_TextExtractor.__init__] """ self.working_dir = Path(working_dir) self.repo_dir = self.working_dir / ".beads_mock" self.beads_file = self.repo_dir / "beads.json" def init_repo(self) -> None: """ Initialize the mock repository. [C: tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query, tests/test_gui_dag_beads.py:test_load_active_tickets_from_beads, tests/test_mcp_client_beads.py:test_bd_mcp_tools] """ self.repo_dir.mkdir(parents=True, exist_ok=True) if not self.beads_file.exists(): self.beads_file.write_text("[]", encoding="utf-8") def is_initialized(self) -> bool: """ Check if the repository is initialized. [C: src/mcp_client.py:dispatch, tests/test_beads_client.py:test_beads_init_and_query] """ return self.beads_file.exists() def create_bead(self, title: str, description: str) -> str: """ Create a new bead and return its ID. [C: src/mcp_client.py:dispatch, tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query, tests/test_gui_dag_beads.py:test_load_active_tickets_from_beads] """ beads = self._read_beads() bead_id = f"bead-{len(beads) + 1}" bead = {"id": bead_id, "title": title, "description": description, "status": "active"} beads.append(bead) self._write_beads(beads) return bead_id def update_bead(self, bead_id: str, status: str) -> bool: """ Update the status of an existing bead. [C: src/mcp_client.py:dispatch, tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query] """ beads = self._read_beads() for bead in beads: if bead["id"] == bead_id: bead["status"] = status self._write_beads(beads) return True return False def list_beads(self) -> List[Bead]: """ List all beads. [C: src/gui_2.py:App._render_beads_tab, src/mcp_client.py:dispatch, tests/test_beads_client.py:test_beads_init_and_query] """ return [Bead(**b) for b in self._read_beads()] def _read_beads(self) -> List[dict]: if not self.beads_file.exists(): return [] return json.loads(self.beads_file.read_text(encoding="utf-8")) def _write_beads(self, beads: List[dict]) -> None: self.beads_file.write_text(json.dumps(beads, indent=1), encoding="utf-8")