Skip to content
23 changes: 23 additions & 0 deletions PLAN.md
Original file line number Diff line number Diff line change
Expand Up @@ -109,3 +109,26 @@ new mutating capabilities.
|------|------------|-------|--------|
| `mocks/code_locator.py` | `RealCodeLocatorAdapter` in `adapters/code_locator.py` | Phase 1 | **Deleted** |
| `mocks/decision_ledger.py` | `SurrealDBLedgerAdapter` in `ledger/adapter.py` | Phase 2 | **Deleted** |

---

## Ledger Locator (#368)

Plan: [`thoughts/shared/plans/2026-05-16-ledger-locator-and-migration.md`](thoughts/shared/plans/2026-05-16-ledger-locator-and-migration.md).
R4 audit: PASS (locator + delegation + wizard split + git-native onboarding + migrate-state + gc).

- [x] **Phase 1** — `ledger_locator/` module: `_project_id.py`, `_origin_guard.py`,
`__init__.py` with `resolve_*` paths (ledger, code_graph, bm25,
watermark, pending/processed transcripts, operator_config); explicit
VCS contract; origin-collision guard
- [x] **Phase 2A** — derived-state resolvers (bm25, watermark, transcripts)
- [x] **Phase 2B-i** — ledger + code-graph delegation in adapter / runtime / config
- [x] **Phase 2B-ii** — events/materializer + events/transcript_queue delegate to locator
- [x] **Phase 2C** — setup_wizard env-var strip + R4 config split
(`context._CONFIG_KEY_ROUTING`, `_write_collaboration_config` atomic
two-file write, git-native onboarding detection + divergence guard,
`run_config_wizard` two-pane editor)
- [x] **Phase 3** — `cli/migrate_state.py` + `migrate-state` / `migrate-ledger`
server subparsers + R4 `config.yaml` key partition + bicameral-update
skill post-upgrade hook
- [x] **Phase 4** — `cli/gc.py` orphan project-dir reclaim (list + `--delete`)
6 changes: 4 additions & 2 deletions adapters/ledger.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,11 +110,13 @@ def get_ledger():
data_path = os.getenv("BICAMERAL_DATA_PATH", repo_path)
bicameral_dir = Path(data_path) / ".bicameral"
events_dir = bicameral_dir / "events"
local_dir = bicameral_dir / "local"

author = _get_git_email(repo_path)
writer = EventFileWriter(events_dir, author)
materializer = EventMaterializer(events_dir, local_dir)
# #368 Phase 2B-ii: watermark lives at the locator-resolved
# project dir, not bicameral_dir/"local"/"watermark". The
# materializer derives the watermark path from `repo_path`.
materializer = EventMaterializer(events_dir, Path(repo_path))

cfg.setdefault("team", {})["author"] = author
try:
Expand Down
129 changes: 129 additions & 0 deletions cli/gc.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
"""`bicameral-mcp gc` — reclaim project dirs whose origin no longer resolves.

Each `~/.bicameral/projects/<id>/origin.txt` names the absolute git
common-dir path the locator hashed at first resolve. When the named
path no longer exists (project deleted, relocated, or rebased), the
project dir is orphaned. `gc` lists orphans by default and deletes
them under `--delete` after a per-item prompt (or `--yes` to skip).
"""

from __future__ import annotations

import argparse
import shutil
from pathlib import Path
from typing import Literal

Status = Literal["live", "orphan", "unreadable"]


def _scan(state_root: Path) -> list[tuple[str, Path, Status, str | None]]:
"""Yield (project_id, project_dir, status, origin_text) for every
immediate subdirectory of ``state_root``.

- ``live`` — origin.txt points at an existing directory.
- ``orphan`` — origin.txt parses but the named path is gone.
- ``unreadable`` — origin.txt missing, empty, or not a regular file.
"""
out: list[tuple[str, Path, Status, str | None]] = []
if not state_root.is_dir():
return out
for child in sorted(state_root.iterdir()):
if not child.is_dir():
continue
origin = child / "origin.txt"
if not origin.is_file():
out.append((child.name, child, "unreadable", None))
continue
try:
text = origin.read_text(encoding="utf-8").strip()
except OSError:
out.append((child.name, child, "unreadable", None))
continue
if not text:
out.append((child.name, child, "unreadable", None))
continue
if Path(text).is_dir():
out.append((child.name, child, "live", text))
else:
out.append((child.name, child, "orphan", text))
return out


def _print_table(rows: list[tuple[str, Path, Status, str | None]]) -> None:
if not rows:
print(" (no projects under ~/.bicameral/projects/)")
return
print(f" {'STATUS':<10} {'PROJECT-ID':<18} ORIGIN")
for _project_id, _path, status, origin_text in rows:
# Pad the project id column with the first 16 chars of the dir name.
print(f" {status:<10} {_path.name:<18} {origin_text or '(unreadable)'}")


def _confirm_delete(project_dir: Path, status: Status, origin: str | None) -> bool:
"""Per-item prompt. Empty / `y` / `yes` confirms; anything else declines."""
try:
response = (
input(
f" Delete {project_dir.name} ({status} — origin={origin or 'unreadable'})? [y/N]: "
)
.strip()
.lower()
)
except (EOFError, OSError):
return False
return response in ("y", "yes")


def _build_argparser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="bicameral-mcp gc",
description="List or delete orphan project dirs under ~/.bicameral/projects/ (#368).",
)
p.add_argument(
"--delete",
action="store_true",
help="prompt to delete each orphan (and unreadable) project dir",
)
p.add_argument(
"--yes",
action="store_true",
help="under --delete: skip per-item prompts and delete every orphan/unreadable dir",
)
p.add_argument(
"--state-root",
default=None,
metavar="PATH",
help="override ~/.bicameral/projects/ (test fixture knob)",
)
return p


def main(argv: list[str] | None = None) -> int:
args = _build_argparser().parse_args(argv)

if args.state_root is not None:
state_root = Path(args.state_root).expanduser().resolve()
else:
try:
from ledger_locator import STATE_ROOT

state_root = STATE_ROOT
except ImportError as exc: # pragma: no cover
print(f" ERROR: ledger_locator unavailable: {exc}")
return 2

rows = _scan(state_root)
if not args.delete:
_print_table(rows)
return 0

to_delete = [r for r in rows if r[2] in ("orphan", "unreadable")]
if not to_delete:
print(" No orphan project dirs.")
return 0
for _project_id, path, status, origin in to_delete:
if args.yes or _confirm_delete(path, status, origin):
shutil.rmtree(path, ignore_errors=False)
print(f" Removed {path}")
return 0
Loading
Loading