diff --git a/src/portfolio_repository_state.py b/src/portfolio_repository_state.py index 3e10808..e0a11cf 100644 --- a/src/portfolio_repository_state.py +++ b/src/portfolio_repository_state.py @@ -1,10 +1,13 @@ from __future__ import annotations +import re import subprocess from datetime import UTC, datetime from pathlib import Path from typing import Any +_GIT_OID_RE = re.compile(r"^[0-9a-f]{40,64}$") + def observe_repository_state( path: Path, @@ -13,67 +16,397 @@ def observe_repository_state( remote_default_branch: dict[str, Any] | None = None, ) -> dict[str, Any]: """Read local Git/worktree state without changing refs or exposing file names.""" - remote = dict(remote_default_branch) if remote_default_branch else { - "state": "unknown", - "reason_code": "not_requested", - "reason": "no independent live remote read was performed by portfolio generation", - } - if not (path / ".git").exists(): + remote = ( + dict(remote_default_branch) + if remote_default_branch + else { + "state": "unknown", + "reason_code": "not_requested", + "reason": ( + "no independent live remote read was performed by portfolio generation" + ), + } + ) + repository_kind = _repository_kind(path) + if repository_kind is None: return { "state": "not_a_repository", "observed_at": observed_at.astimezone(UTC).isoformat(), "remote_default_branch": remote, } + try: - head = _git(path, "rev-parse", "HEAD") - branch = _git(path, "branch", "--show-current") or None - dirty = _git(path, "status", "--porcelain", "--untracked-files=all") - upstream = _git_optional(path, "rev-parse", "--abbrev-ref", "@{upstream}") - ahead = behind = None - if upstream: - counts = _git(path, "rev-list", "--left-right", "--count", f"{upstream}...HEAD") - behind_text, ahead_text = counts.split() - behind, ahead = int(behind_text), int(ahead_text) - worktrees = [] - for item in _worktrees(path): - worktree_path = Path(item["path"]) - worktree_dirty = _git( - worktree_path, "status", "--porcelain", "--untracked-files=all" + worktrees = _observe_worktrees(path) + selection = _select_remote_default_worktree(worktrees, remote) + topology = { + "kind": repository_kind, + "configured_path": str(path), + "worktree_count": len(worktrees), + "linked_worktree_count": sum( + item.get("state") != "coordinator" for item in worktrees + ) + - (0 if repository_kind == "bare_coordinator" else 1), + "selection": selection, + } + + if repository_kind == "bare_coordinator": + topology["coordinator"] = _observe_bare_coordinator(path) + return _bare_repository_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + ) + + configured_local = _observe_working_tree(path) + if remote.get("state") == "observed": + if selection["state"] != "selected": + return _unknown_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=configured_local, + reason_code=str(selection["reason_code"]), + reason=str(selection["reason"]), + ) + selected_local = _local_from_worktree( + _selected_worktree(worktrees, selection) + ) + return _observed_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=selected_local, ) - worktrees.append( + + if _tracks_nonmatching_branch(configured_local): + return _unknown_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=configured_local, + reason_code="nonstandard_upstream_requires_remote_default_evidence", + reason=( + "the configured worktree tracks a differently named branch, " + "and independent remote-default evidence is unavailable" + ), + ) + + return _observed_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=configured_local, + ) + except (OSError, subprocess.CalledProcessError, ValueError) as exc: + return { + "state": "unknown", + "observed_at": observed_at.astimezone(UTC).isoformat(), + "reason_code": "repository_observation_failed", + "reason": str(exc), + "remote_default_branch": remote, + } + + +def _repository_kind(path: Path) -> str | None: + try: + is_bare = _git(path, "rev-parse", "--is-bare-repository") + except (OSError, subprocess.CalledProcessError): + return None + if is_bare == "true": + return "bare_coordinator" + if is_bare == "false": + return "working_repository" + raise ValueError("git returned an invalid repository kind") + + +def _observe_working_tree(path: Path) -> dict[str, Any]: + head = _git(path, "rev-parse", "HEAD") + branch = _git(path, "branch", "--show-current") or None + dirty = _git(path, "status", "--porcelain", "--untracked-files=all") + upstream = _git_optional(path, "rev-parse", "--abbrev-ref", "@{upstream}") + ahead = behind = None + if upstream: + counts = _git( + path, + "rev-list", + "--left-right", + "--count", + f"{upstream}...HEAD", + ) + behind_text, ahead_text = counts.split() + behind, ahead = int(behind_text), int(ahead_text) + return { + "path": str(path), + "head": head, + "branch": branch, + "dirty": bool(dirty), + "dirty_path_count": len(dirty.splitlines()) if dirty else 0, + "upstream": upstream, + "upstream_observation_source": ( + "local_tracking_ref" if upstream else "unavailable" + ), + "ahead": ahead, + "behind": behind, + } + + +def _observe_bare_coordinator(path: Path) -> dict[str, Any]: + head = _git_optional(path, "rev-parse", "--verify", "HEAD^{commit}") + branch = _git_optional(path, "symbolic-ref", "--short", "HEAD") + return { + "path": str(path), + "head": head, + "head_state": "observed" if head else "dangling", + "branch": branch, + } + + +def _observe_worktrees(path: Path) -> list[dict[str, Any]]: + observed: list[dict[str, Any]] = [] + for item in _worktrees(path): + worktree_path = Path(item["path"]) + if item.get("bare"): + observed.append( { + "state": "coordinator", + "path": str(worktree_path), + "head": item.get("head"), + "branch": item.get("branch"), + "detached": False, + "bare": True, + "dirty": None, + "dirty_path_count": None, + } + ) + continue + try: + local = _observe_working_tree(worktree_path) + except (OSError, subprocess.CalledProcessError, ValueError): + observed.append( + { + "state": "unknown", + "reason_code": "worktree_observation_failed", + "reason": "git could not observe the linked worktree", "path": str(worktree_path), "head": item.get("head"), "branch": item.get("branch"), "detached": item.get("detached", False), - "dirty": bool(worktree_dirty), - "dirty_path_count": len(worktree_dirty.splitlines()) if worktree_dirty else 0, + "bare": False, + "dirty": None, + "dirty_path_count": None, } ) + continue + observed.append( + { + "state": "observed", + **local, + "detached": item.get("detached", False), + "bare": False, + } + ) + return observed + + +def _select_remote_default_worktree( + worktrees: list[dict[str, Any]], remote: dict[str, Any] +) -> dict[str, Any]: + remote_state = str(remote.get("state") or "unknown") + source = str(remote.get("source") or "remote_default_branch") + if remote_state != "observed": return { - "state": "observed", - "observed_at": observed_at.astimezone(UTC).isoformat(), - "local": { - "path": str(path), - "head": head, - "branch": branch, - "dirty": bool(dirty), - "dirty_path_count": len(dirty.splitlines()) if dirty else 0, - "upstream": upstream, - "upstream_observation_source": "local_tracking_ref" if upstream else "unavailable", - "ahead": ahead, - "behind": behind, - }, - "remote_default_branch": remote, - "worktrees": worktrees, + "source": source, + "state": "unknown", + "reason_code": "remote_default_branch_unavailable", + "reason": ( + "independent remote-default evidence is not observed " + f"(state={remote_state})" + ), + "candidate_count": 0, } - except (OSError, subprocess.CalledProcessError, ValueError) as exc: + + default_branch = str(remote.get("default_branch") or "") + head_sha = str(remote.get("head_sha") or "") + if not default_branch or not _GIT_OID_RE.fullmatch(head_sha): return { + "source": source, "state": "unknown", - "observed_at": observed_at.astimezone(UTC).isoformat(), - "reason": str(exc), - "remote_default_branch": remote, + "reason_code": "remote_default_branch_malformed", + "reason": "observed remote-default evidence lacks a valid branch and head", + "candidate_count": 0, + } + + head_matches = [ + item + for item in worktrees + if item.get("state") == "observed" and item.get("head") == head_sha + ] + branch_matches = [ + item for item in head_matches if item.get("branch") == default_branch + ] + if len(head_matches) == 1: + return _selected_worktree_result( + head_matches[0], + source=source, + reason_code="unique_remote_head_match", + candidate_count=1, + ) + if len(branch_matches) == 1: + return _selected_worktree_result( + branch_matches[0], + source=source, + reason_code="default_branch_tiebreak", + candidate_count=len(head_matches), + ) + if not head_matches: + return { + "source": source, + "state": "unknown", + "reason_code": "remote_default_worktree_not_found", + "reason": "no observed worktree matches the remote default head", + "candidate_count": 0, } + return { + "source": source, + "state": "unknown", + "reason_code": "ambiguous_remote_default_worktrees", + "reason": ( + "multiple observed worktrees match the remote default head and no " + "single default-branch worktree resolves the ambiguity" + ), + "candidate_count": len(head_matches), + } + + +def _selected_worktree_result( + worktree: dict[str, Any], + *, + source: str, + reason_code: str, + candidate_count: int, +) -> dict[str, Any]: + return { + "source": source, + "state": "selected", + "reason_code": reason_code, + "reason": None, + "candidate_count": candidate_count, + "path": worktree["path"], + "head": worktree["head"], + "branch": worktree.get("branch"), + } + + +def _selected_worktree( + worktrees: list[dict[str, Any]], selection: dict[str, Any] +) -> dict[str, Any]: + selected_path = selection.get("path") + matches = [ + item + for item in worktrees + if item.get("state") == "observed" and item.get("path") == selected_path + ] + if len(matches) != 1: + raise ValueError("selected worktree is no longer uniquely observable") + return matches[0] + + +def _local_from_worktree(worktree: dict[str, Any]) -> dict[str, Any]: + keys = ( + "path", + "head", + "branch", + "dirty", + "dirty_path_count", + "upstream", + "upstream_observation_source", + "ahead", + "behind", + ) + return {key: worktree.get(key) for key in keys} + + +def _tracks_nonmatching_branch(local: dict[str, Any]) -> bool: + branch = str(local.get("branch") or "") + upstream = str(local.get("upstream") or "") + if not branch or "/" not in upstream: + return False + _remote_name, upstream_branch = upstream.split("/", 1) + return upstream_branch != branch + + +def _bare_repository_result( + *, + observed_at: datetime, + remote: dict[str, Any], + worktrees: list[dict[str, Any]], + topology: dict[str, Any], +) -> dict[str, Any]: + selection = topology["selection"] + if selection["state"] != "selected": + return _unknown_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=None, + reason_code=str(selection["reason_code"]), + reason=str(selection["reason"]), + ) + return _observed_result( + observed_at=observed_at, + remote=remote, + worktrees=worktrees, + topology=topology, + local=_local_from_worktree(_selected_worktree(worktrees, selection)), + ) + + +def _observed_result( + *, + observed_at: datetime, + remote: dict[str, Any], + worktrees: list[dict[str, Any]], + topology: dict[str, Any], + local: dict[str, Any], +) -> dict[str, Any]: + return { + "state": "observed", + "observed_at": observed_at.astimezone(UTC).isoformat(), + "local": local, + "remote_default_branch": remote, + "topology": topology, + "worktrees": worktrees, + } + + +def _unknown_result( + *, + observed_at: datetime, + remote: dict[str, Any], + worktrees: list[dict[str, Any]], + topology: dict[str, Any], + local: dict[str, Any] | None, + reason_code: str, + reason: str, +) -> dict[str, Any]: + result = { + "state": "unknown", + "observed_at": observed_at.astimezone(UTC).isoformat(), + "reason_code": reason_code, + "reason": reason, + "remote_default_branch": remote, + "topology": topology, + "worktrees": worktrees, + } + if local is not None: + result["local"] = local + return result def _git(path: Path, *args: str) -> str: @@ -85,7 +418,7 @@ def _git(path: Path, *args: str) -> str: def _git_optional(path: Path, *args: str) -> str | None: try: return _git(path, *args) or None - except subprocess.CalledProcessError: + except (OSError, subprocess.CalledProcessError): return None @@ -108,4 +441,10 @@ def _worktrees(path: Path) -> list[dict[str, Any]]: current["branch"] = value.removeprefix("refs/heads/") elif key == "detached": current["detached"] = True + elif key == "bare": + current["bare"] = True + elif key == "locked": + current["locked"] = True + elif key == "prunable": + current["prunable"] = True return items diff --git a/src/portfolio_truth_sources.py b/src/portfolio_truth_sources.py index fbec2b2..684424e 100644 --- a/src/portfolio_truth_sources.py +++ b/src/portfolio_truth_sources.py @@ -581,7 +581,7 @@ def _read_small_json(path: Path) -> dict[str, Any]: def _gather_git_facts(project_path: Path) -> dict[str, Any]: git_dir = project_path / ".git" - if not git_dir.exists(): + if not git_dir.exists() and not _is_bare_repository_root(project_path): return { "has_git": False, "last_commit_at": None, @@ -621,6 +621,34 @@ def _gather_git_facts(project_path: Path) -> dict[str, Any]: return base +def _is_bare_repository_root(project_path: Path) -> bool: + """Recognize a conventional bare repository without climbing into parents.""" + try: + result = subprocess.run( + [ + "git", + "-C", + str(project_path), + "rev-parse", + "--is-bare-repository", + "--absolute-git-dir", + ], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return False + lines = result.stdout.splitlines() + if result.returncode != 0 or len(lines) != 2 or lines[0].strip() != "true": + return False + try: + return Path(lines[1].strip()).resolve() == project_path.resolve() + except OSError: + return False + + def _git_default_branch(project_path: Path) -> str: """The repo's default branch from the local ``origin/HEAD`` ref, if set. diff --git a/tests/test_portfolio_repository_state.py b/tests/test_portfolio_repository_state.py index e92ee25..d841e06 100644 --- a/tests/test_portfolio_repository_state.py +++ b/tests/test_portfolio_repository_state.py @@ -3,6 +3,7 @@ import subprocess from datetime import UTC, datetime from pathlib import Path +from typing import Any from src.portfolio_repository_state import observe_repository_state @@ -25,7 +26,35 @@ def _repo(tmp_path: Path) -> Path: return repo -def test_observation_reports_dirty_no_upstream_and_unknown_remote(tmp_path: Path) -> None: +def _bare_repo(tmp_path: Path) -> tuple[Path, str]: + source = _repo(tmp_path) + head = _git(source, "rev-parse", "HEAD") + bare = tmp_path / "coordinator.git" + subprocess.run( + ["git", "clone", "--bare", str(source), str(bare)], + check=True, + capture_output=True, + text=True, + ) + return bare, head + + +def _remote_default(head: str, branch: str = "main") -> dict[str, Any]: + return { + "source": "fixture-live-remote-default", + "state": "observed", + "reason_code": "observed", + "reason": None, + "observed_at": "2026-07-12T00:00:00+00:00", + "default_branch": branch, + "head_sha": head, + "archived": False, + } + + +def test_observation_reports_dirty_no_upstream_and_unknown_remote( + tmp_path: Path, +) -> None: repo = _repo(tmp_path) (repo / "dirty.txt").write_text("dirty\n") @@ -49,7 +78,150 @@ def test_observation_reports_linked_worktree_without_file_names(tmp_path: Path) state = observe_repository_state(repo, observed_at=datetime.now(UTC)) assert len(state["worktrees"]) == 2 - linked_state = next(item for item in state["worktrees"] if item["path"] == str(linked)) + linked_state = next( + item for item in state["worktrees"] if item["path"] == str(linked) + ) assert linked_state["dirty"] is True assert linked_state["dirty_path_count"] == 1 assert "untracked.txt" not in str(state) + + +def test_dangling_bare_head_uses_clean_matching_linked_worktree( + tmp_path: Path, +) -> None: + bare, head = _bare_repo(tmp_path) + linked = tmp_path / "linked" + _git(bare, "worktree", "add", str(linked), "main") + _git(bare, "symbolic-ref", "HEAD", "refs/heads/missing-default") + + state = observe_repository_state( + bare, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + remote_default_branch=_remote_default(head), + ) + + assert state["state"] == "observed" + assert state["topology"]["kind"] == "bare_coordinator" + assert state["topology"]["coordinator"]["head"] is None + assert state["topology"]["coordinator"]["head_state"] == "dangling" + assert state["topology"]["selection"]["state"] == "selected" + assert state["local"]["path"] == str(linked) + assert state["local"]["head"] == head + assert state["local"]["dirty"] is False + + +def test_bare_coordinator_selects_unique_clean_remote_default_worktree( + tmp_path: Path, +) -> None: + bare, head = _bare_repo(tmp_path) + linked = tmp_path / "linked" + _git(bare, "worktree", "add", str(linked), "main") + + state = observe_repository_state( + bare, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + remote_default_branch=_remote_default(head), + ) + + assert state["state"] == "observed" + assert state["local"]["path"] == str(linked) + assert state["topology"]["selection"]["reason_code"] == "unique_remote_head_match" + assert len(state["worktrees"]) == 2 + coordinator = next(item for item in state["worktrees"] if item["bare"]) + assert coordinator["state"] == "coordinator" + assert coordinator["dirty"] is None + + +def test_multiple_remote_head_candidates_use_default_branch_tiebreak( + tmp_path: Path, +) -> None: + repo = _repo(tmp_path) + head = _git(repo, "rev-parse", "HEAD") + detached = tmp_path / "detached" + _git(repo, "worktree", "add", "--detach", str(detached), head) + + state = observe_repository_state( + repo, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + remote_default_branch=_remote_default(head), + ) + + selection = state["topology"]["selection"] + assert state["state"] == "observed" + assert selection["state"] == "selected" + assert selection["reason_code"] == "default_branch_tiebreak" + assert selection["candidate_count"] == 2 + assert state["local"]["path"] == str(repo) + + +def test_recovery_tracking_does_not_impersonate_remote_default( + tmp_path: Path, +) -> None: + repo = _repo(tmp_path) + recovery_head = _git(repo, "rev-parse", "HEAD") + _git(repo, "checkout", "-b", "remote-main") + (repo / "remote.txt").write_text("remote\n") + _git(repo, "add", "remote.txt") + _git(repo, "commit", "-m", "remote default") + remote_head = _git(repo, "rev-parse", "HEAD") + _git(repo, "checkout", "main") + _git(repo, "remote", "add", "origin", str(repo)) + _git( + repo, + "update-ref", + "refs/remotes/origin/recovery/repo-main", + recovery_head, + ) + _git(repo, "branch", "--set-upstream-to=origin/recovery/repo-main", "main") + + state = observe_repository_state( + repo, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + remote_default_branch=_remote_default(remote_head), + ) + + assert state["state"] == "unknown" + assert state["reason_code"] == "remote_default_worktree_not_found" + assert state["local"]["head"] == recovery_head + assert state["local"]["upstream"] == "origin/recovery/repo-main" + assert state["remote_default_branch"]["head_sha"] == remote_head + + +def test_bare_coordinator_missing_remote_evidence_is_precise_unknown( + tmp_path: Path, +) -> None: + bare, _head = _bare_repo(tmp_path) + linked = tmp_path / "linked" + _git(bare, "worktree", "add", str(linked), "main") + + state = observe_repository_state( + bare, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + ) + + assert state["state"] == "unknown" + assert state["reason_code"] == "remote_default_branch_unavailable" + assert state["topology"]["kind"] == "bare_coordinator" + assert state["topology"]["selection"]["candidate_count"] == 0 + assert "local" not in state + + +def test_ambiguous_remote_default_worktrees_fail_closed(tmp_path: Path) -> None: + bare, head = _bare_repo(tmp_path) + first = tmp_path / "first" + second = tmp_path / "second" + _git(bare, "worktree", "add", "--detach", str(first), head) + _git(bare, "worktree", "add", "--detach", str(second), head) + + state = observe_repository_state( + bare, + observed_at=datetime(2026, 7, 12, tzinfo=UTC), + remote_default_branch=_remote_default(head), + ) + + selection = state["topology"]["selection"] + assert state["state"] == "unknown" + assert state["reason_code"] == "ambiguous_remote_default_worktrees" + assert selection["state"] == "unknown" + assert selection["candidate_count"] == 2 + assert "local" not in state diff --git a/tests/test_portfolio_truth_sources.py b/tests/test_portfolio_truth_sources.py index f18ca9e..50765ed 100644 --- a/tests/test_portfolio_truth_sources.py +++ b/tests/test_portfolio_truth_sources.py @@ -8,6 +8,7 @@ from __future__ import annotations +import subprocess from datetime import datetime, timezone from src.portfolio_truth_sources import ( @@ -174,3 +175,23 @@ def _project(*parts: str) -> None: "scratch-container": 1, "temporary-checkout": 1, } + + +def test_discovery_recognizes_conventional_bare_coordinator(tmp_path) -> None: + coordinator = tmp_path / "bare-coordinator" + subprocess.run( + ["git", "init", "--bare", str(coordinator)], + check=True, + capture_output=True, + text=True, + ) + + result = discover_workspace_projects( + tmp_path, + catalog_data={}, + now=datetime(2026, 7, 30, tzinfo=timezone.utc), + ) + + project = next(item for item in result if item["name"] == coordinator.name) + assert project["has_git"] is True + assert project["project_path"] == coordinator