repository-rename verify: accept descendant heads and retired dispatch after source-synced
Finishing work during the operation retires it from active dispatch, and the source repository legitimately advances after source-synced. verify now treats retired dispatch ids as present and accepts a Forge head that contains the recorded source-synced head (Forgejo compare); unrelated drift still fails. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 101414@bnt-lap001 Assistant-Session: 60b5a221-d435-42b9-b778-983c90c7eed6
This commit is contained in:
parent
6924d2410f
commit
b451bd10e5
4 changed files with 114 additions and 14 deletions
|
|
@ -83,6 +83,10 @@ class ForgeRepositoryGateway(Protocol):
|
||||||
self, *, instance: str, owner: str, old_name: str, new_name: str
|
self, *, instance: str, owner: str, old_name: str, new_name: str
|
||||||
) -> ForgeRepositorySnapshot: ...
|
) -> ForgeRepositorySnapshot: ...
|
||||||
|
|
||||||
|
async def descends_from(
|
||||||
|
self, *, instance: str, owner: str, name: str, base: str, head: str
|
||||||
|
) -> bool: ...
|
||||||
|
|
||||||
|
|
||||||
def _write_token() -> str | None:
|
def _write_token() -> str | None:
|
||||||
path = os.environ.get(WRITE_TOKEN_FILE_ENV)
|
path = os.environ.get(WRITE_TOKEN_FILE_ENV)
|
||||||
|
|
@ -207,6 +211,34 @@ class ForgejoRepositoryGateway:
|
||||||
raise ForgeRepositoryUnreadable("Renamed Forge repository is not readable")
|
raise ForgeRepositoryUnreadable("Renamed Forge repository is not readable")
|
||||||
return renamed
|
return renamed
|
||||||
|
|
||||||
|
async def descends_from(
|
||||||
|
self, *, instance: str, owner: str, name: str, base: str, head: str
|
||||||
|
) -> bool:
|
||||||
|
"""True when ``base`` is an ancestor of (or equal to) ``head``.
|
||||||
|
|
||||||
|
``compare/<head>...<base>`` lists commits reachable from ``base`` but not
|
||||||
|
from ``head``; it is empty exactly when ``base`` is contained in ``head``.
|
||||||
|
"""
|
||||||
|
if base == head:
|
||||||
|
return True
|
||||||
|
url = f"{self._api(instance, owner, name)}/compare/{quote(head, safe='')}...{quote(base, safe='')}"
|
||||||
|
try:
|
||||||
|
async with httpx.AsyncClient(
|
||||||
|
timeout=settings.repository_rename_forge_timeout_seconds
|
||||||
|
) as client:
|
||||||
|
response = await client.get(url, headers=self._headers(forge_read_token()))
|
||||||
|
if response.status_code == 404:
|
||||||
|
return False
|
||||||
|
response.raise_for_status()
|
||||||
|
data = response.json()
|
||||||
|
except (httpx.HTTPError, ValueError) as exc:
|
||||||
|
raise ForgeRepositoryUnreadable(
|
||||||
|
f"Forge ancestry check failed ({type(exc).__name__})"
|
||||||
|
) from exc
|
||||||
|
commits = data.get("commits")
|
||||||
|
total = data.get("total_commits")
|
||||||
|
return (isinstance(commits, list) and not commits) or total == 0
|
||||||
|
|
||||||
|
|
||||||
_gateway = ForgejoRepositoryGateway()
|
_gateway = ForgejoRepositoryGateway()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -47,6 +47,7 @@ from api.services.forge_repository import (
|
||||||
ForgeRepositoryGateway,
|
ForgeRepositoryGateway,
|
||||||
ForgeRepositoryRedirected,
|
ForgeRepositoryRedirected,
|
||||||
ForgeRepositorySnapshot,
|
ForgeRepositorySnapshot,
|
||||||
|
ForgeRepositoryError,
|
||||||
ForgeRepositoryUnreadable,
|
ForgeRepositoryUnreadable,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -1248,7 +1249,22 @@ async def verify_operation(
|
||||||
if forge is not None:
|
if forge is not None:
|
||||||
add("forge_coordinate", forge.name == expected_name, expected_name, forge.name)
|
add("forge_coordinate", forge.name == expected_name, expected_name, forge.name)
|
||||||
add("forge_repository_id", forge.repository_id == operation.expected_forge_repository_id, operation.expected_forge_repository_id, forge.repository_id)
|
add("forge_repository_id", forge.repository_id == operation.expected_forge_repository_id, operation.expected_forge_repository_id, forge.repository_id)
|
||||||
add("source_commit", forge.head_commit == _expected_head(operation), _expected_head(operation), forge.head_commit)
|
anchor = _expected_head(operation)
|
||||||
|
head_ok = forge.head_commit == anchor
|
||||||
|
if not head_ok and "source-synced" in (_telemetry(operation)[0].get("phases") or {}):
|
||||||
|
# After source-synced the source repository keeps moving (its own
|
||||||
|
# commits and sync writebacks); the recorded head must stay contained.
|
||||||
|
try:
|
||||||
|
head_ok = await gateway.descends_from(
|
||||||
|
instance=operation.expected_forge_instance,
|
||||||
|
owner=operation.expected_forge_owner,
|
||||||
|
name=expected_name,
|
||||||
|
base=anchor,
|
||||||
|
head=forge.head_commit,
|
||||||
|
)
|
||||||
|
except (ForgeRepositoryError, AttributeError):
|
||||||
|
head_ok = False
|
||||||
|
add("source_commit", head_ok, anchor, forge.head_commit)
|
||||||
add("default_branch", forge.default_branch == operation.expected_default_branch, operation.expected_default_branch, forge.default_branch)
|
add("default_branch", forge.default_branch == operation.expected_default_branch, operation.expected_default_branch, forge.default_branch)
|
||||||
add("projection_readable", forge.projection_readable and forge.projection_source_present, True, forge.projection_readable and forge.projection_source_present)
|
add("projection_readable", forge.projection_readable and forge.projection_source_present, True, forge.projection_readable and forge.projection_source_present)
|
||||||
expected_canonical = operation.old_slug if operation.phase in {"preflighted", "forge-renamed", "rolled-back"} else operation.new_slug
|
expected_canonical = operation.old_slug if operation.phase in {"preflighted", "forge-renamed", "rolled-back"} else operation.new_slug
|
||||||
|
|
@ -1302,6 +1318,25 @@ async def verify_operation(
|
||||||
"baseline": int(baseline_record.get("count") or 0),
|
"baseline": int(baseline_record.get("count") or 0),
|
||||||
"current": int(current_record.get("count") or 0),
|
"current": int(current_record.get("count") or 0),
|
||||||
}
|
}
|
||||||
|
# Active dispatch is a status-derived view: finishing a task during the
|
||||||
|
# operation retires it from dispatch without losing the record. Only a
|
||||||
|
# dispatch id whose task/workplan no longer exists counts as missing.
|
||||||
|
retired_dispatch: list[str] = []
|
||||||
|
live_ids = set((current.get("tasks") or {}).get("ids") or []) | set(
|
||||||
|
(current.get("workplans") or {}).get("ids") or []
|
||||||
|
)
|
||||||
|
for bucket in (missing, changed_relationships):
|
||||||
|
ids = bucket.get("active_dispatch")
|
||||||
|
if not ids:
|
||||||
|
continue
|
||||||
|
gone = [item for item in ids if item not in live_ids]
|
||||||
|
retired_dispatch = sorted(set(retired_dispatch) | (set(ids) - set(gone)))
|
||||||
|
if gone:
|
||||||
|
bucket["active_dispatch"] = gone
|
||||||
|
else:
|
||||||
|
del bucket["active_dispatch"]
|
||||||
|
if retired_dispatch:
|
||||||
|
counts.pop("active_dispatch", None)
|
||||||
# New telemetry may legitimately arrive while a phased operation is being
|
# New telemetry may legitimately arrive while a phased operation is being
|
||||||
# executed. Continuity means every baseline identity still exists; it does
|
# executed. Continuity means every baseline identity still exists; it does
|
||||||
# not freeze the repository's append-only history at the preflight count.
|
# not freeze the repository's append-only history at the preflight count.
|
||||||
|
|
@ -1309,7 +1344,7 @@ async def verify_operation(
|
||||||
"relationship_continuity",
|
"relationship_continuity",
|
||||||
not missing and not changed_relationships,
|
not missing and not changed_relationships,
|
||||||
{"missing": {}, "changed": {}},
|
{"missing": {}, "changed": {}},
|
||||||
{"missing": missing, "changed": changed_relationships},
|
{"missing": missing, "changed": changed_relationships, "retired_dispatch": retired_dispatch},
|
||||||
)
|
)
|
||||||
add(
|
add(
|
||||||
"record_counts_non_decreasing",
|
"record_counts_non_decreasing",
|
||||||
|
|
|
||||||
|
|
@ -126,6 +126,12 @@ revision pushed from the fresh clone. `source-synced` therefore requires the clo
|
||||||
`head_commit` to equal the current Forge head, records the baseline and `head_advanced`,
|
`head_commit` to equal the current Forge head, records the baseline and `head_advanced`,
|
||||||
and later phases anchor on that recorded head. Drift beyond it still fails closed.
|
and later phases anchor on that recorded head. Drift beyond it still fails closed.
|
||||||
|
|
||||||
|
After `source-synced` the source repository keeps moving. `verify` accepts a later Forge head
|
||||||
|
only when it contains the recorded `source-synced` head (Forgejo `compare`); unrelated drift
|
||||||
|
still fails. Tasks or workplans finished during the operation leave active dispatch and are
|
||||||
|
reported as `retired_dispatch` rather than missing; a dispatch id whose record no longer
|
||||||
|
exists is still a failure.
|
||||||
|
|
||||||
The private source evidence file must include the assertions enforced by State
|
The private source evidence file must include the assertions enforced by State
|
||||||
Hub. Extra non-secret references are allowed:
|
Hub. Extra non-secret references are allowed:
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -38,6 +38,7 @@ class FakeForge:
|
||||||
self.rename_calls = 0
|
self.rename_calls = 0
|
||||||
self.extra: dict[str, int] = {}
|
self.extra: dict[str, int] = {}
|
||||||
self.unreadable = False
|
self.unreadable = False
|
||||||
|
self.contains: dict[str, set[str]] = {}
|
||||||
|
|
||||||
def snapshot(self, name: str, repository_id: int | None = None) -> ForgeRepositorySnapshot:
|
def snapshot(self, name: str, repository_id: int | None = None) -> ForgeRepositorySnapshot:
|
||||||
return ForgeRepositorySnapshot(
|
return ForgeRepositorySnapshot(
|
||||||
|
|
@ -65,6 +66,9 @@ class FakeForge:
|
||||||
return self.snapshot(name, self.extra[name])
|
return self.snapshot(name, self.extra[name])
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
async def descends_from(self, *, instance: str, owner: str, name: str, base: str, head: str):
|
||||||
|
return base == head or base in self.contains.get(head, set())
|
||||||
|
|
||||||
async def rename(self, *, instance: str, owner: str, old_name: str, new_name: str):
|
async def rename(self, *, instance: str, owner: str, old_name: str, new_name: str):
|
||||||
assert old_name == self.name
|
assert old_name == self.name
|
||||||
assert new_name not in self.extra
|
assert new_name not in self.extra
|
||||||
|
|
@ -769,21 +773,44 @@ async def test_source_sync_may_advance_forge_head_past_the_baseline(client, rena
|
||||||
assert recorded["head_advanced"] is True
|
assert recorded["head_advanced"] is True
|
||||||
assert recorded["baseline_commit"] == "a" * 40
|
assert recorded["baseline_commit"] == "a" * 40
|
||||||
|
|
||||||
# Later phases anchor on the recorded head, so drift beyond it still fails.
|
# Later commits that contain the recorded head are accepted; unrelated drift is not.
|
||||||
verify = await client.get(
|
verify_url = f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify"
|
||||||
f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify"
|
|
||||||
)
|
def source_commit_ok(response):
|
||||||
assert verify.status_code == 200, verify.text
|
assert response.status_code == 200, response.text
|
||||||
assert all(
|
return [c["ok"] for c in response.json()["checks"] if c["name"] == "source_commit"][0]
|
||||||
check["ok"] for check in verify.json()["checks"] if check["name"] == "source_commit"
|
|
||||||
)
|
assert source_commit_ok(await client.get(verify_url)) is True
|
||||||
|
forge.head = "d" * 40
|
||||||
|
forge.contains[forge.head] = {advanced}
|
||||||
|
assert source_commit_ok(await client.get(verify_url)) is True
|
||||||
forge.head = "c" * 40
|
forge.head = "c" * 40
|
||||||
drifted = await client.get(
|
assert source_commit_ok(await client.get(verify_url)) is False
|
||||||
f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify"
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_task_completed_during_operation_is_retired_not_missing(client, rename_setup):
|
||||||
|
repo, _forge = rename_setup
|
||||||
|
workplan = await create_test_workplan(
|
||||||
|
client, repo_id=repo["id"], slug="FLEX-WP-0043", status="active"
|
||||||
)
|
)
|
||||||
assert any(
|
task = await client.post(
|
||||||
not check["ok"] for check in drifted.json()["checks"] if check["name"] == "source_commit"
|
"/tasks/",
|
||||||
|
json={"workplan_id": workplan["id"], "title": "Finish during rename", "status": "todo", "priority": "high"},
|
||||||
)
|
)
|
||||||
|
assert task.status_code == 201, task.text
|
||||||
|
preflight = await _preflight(client, repo["id"])
|
||||||
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
||||||
|
await _phase(client, repo["id"], operation["id"], "forge-renamed", "preflighted", confirmation)
|
||||||
|
await _phase(client, repo["id"], operation["id"], "statehub-rebound", "forge-renamed", confirmation)
|
||||||
|
done = await client.patch(f"/tasks/{task.json()['id']}", json={"status": "done"})
|
||||||
|
assert done.status_code == 200, done.text
|
||||||
|
verify = await client.get(f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify")
|
||||||
|
assert verify.status_code == 200, verify.text
|
||||||
|
checks = {c["name"]: c for c in verify.json()["checks"]}
|
||||||
|
assert checks["relationship_continuity"]["ok"] is True
|
||||||
|
assert task.json()["id"] in checks["relationship_continuity"]["actual"]["retired_dispatch"]
|
||||||
|
assert checks["record_counts_non_decreasing"]["ok"] is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue