Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion src/portfolio_repository_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,12 @@ def observe_repository_state(
topology=topology,
)

configured_local = _observe_working_tree(path)
# `_observe_worktrees` already captured the configured worktree. Re-reading
# it here creates two different snapshots when another owner is actively
# editing that checkout: `worktrees` can contain the first dirty count while
# `local` contains the second. Derive both views from the same observation so
# the envelope remains internally consistent without hiding the dirty state.
configured_local = _configured_local_from_worktrees(worktrees, path)
if remote.get("state") == "observed":
if selection["state"] != "selected":
return _unknown_result(
Expand Down Expand Up @@ -409,6 +414,21 @@ def _local_from_worktree(worktree: dict[str, Any]) -> dict[str, Any]:
return {key: worktree.get(key) for key in keys}


def _configured_local_from_worktrees(
worktrees: list[dict[str, Any]], path: Path
) -> dict[str, Any]:
configured_path = path.resolve()
matches = [
item
for item in worktrees
if item.get("state") == "observed"
and Path(str(item.get("path") or "")).resolve() == configured_path
]
if len(matches) != 1:
raise ValueError("configured worktree is not uniquely observable")
return _local_from_worktree(matches[0])


def _tracks_nonmatching_branch(local: dict[str, Any]) -> bool:
branch = str(local.get("branch") or "")
upstream_branch = str(local.get("upstream_branch") or "")
Expand Down
64 changes: 64 additions & 0 deletions tests/test_portfolio_repository_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,70 @@ def test_observation_reports_dirty_no_upstream_and_unknown_remote(
assert state["remote_default_branch"]["state"] == "unknown"


def test_configured_local_reuses_one_worktree_observation(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = _repo(tmp_path)
head = _git(repo, "rev-parse", "HEAD")
observations = [
{
"path": str(repo),
"head": head,
"branch": "main",
"dirty": False,
"dirty_path_count": 0,
"upstream": None,
"upstream_branch": None,
"upstream_remote": None,
"upstream_observation_source": "unavailable",
"ahead": None,
"behind": None,
},
{
"path": str(repo),
"head": head,
"branch": "main",
"dirty": True,
"dirty_path_count": 1,
"upstream": None,
"upstream_branch": None,
"upstream_remote": None,
"upstream_observation_source": "unavailable",
"ahead": None,
"behind": None,
},
]
call_count = 0

def changing_observation(_path: Path) -> dict[str, Any]:
nonlocal call_count
observation = observations[min(call_count, len(observations) - 1)]
call_count += 1
return observation

monkeypatch.setattr(
repository_state, "_observe_working_tree", changing_observation
)
observed_at = datetime(2026, 7, 12, tzinfo=UTC)
state = observe_repository_state(
repo,
observed_at=observed_at,
remote_default_branch=_remote_default("f" * 40),
)

assert call_count == 1
assert state["state"] == "unknown"
assert state["reason_code"] == "remote_default_worktree_not_found"
assert state["local"] == _local_from_worktree(state["worktrees"][0])
assert state["local"]["dirty"] is False
_validate_repository_state_shape(
state,
expected_remote=state["remote_default_branch"],
project_key="fixture/changing-configured-worktree",
generated_at=observed_at,
)


def test_local_branch_upstream_is_observed_and_divergence_fails_closed(
tmp_path: Path,
) -> None:
Expand Down