diff --git a/.custodian-brief.md b/.custodian-brief.md index a513467..14040eb 100644 --- a/.custodian-brief.md +++ b/.custodian-brief.md @@ -2,7 +2,7 @@ # Custodian Brief — state-hub **Domain:** infotech -**Last synced:** 2026-08-29 11:07 UTC +**Last synced:** 2026-08-29 13:03 UTC **State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)* ## Active Workstreams @@ -30,11 +30,10 @@ Progress: 1/4 done | workplan_id: `9c4b8c79-edba-5d2d-ad37-a45707e89304` - ► Teach the derivation to use it `b22b24d9` ### Lineage-preserving repository rename workflow and adoption-plan generator -Progress: 6/9 done | workplan_id: `4f5661de-5922-55d6-bd02-5dcb00b13f73` +Progress: 7/9 done | workplan_id: `4f5661de-5922-55d6-bd02-5dcb00b13f73` **Open tasks:** - ! Generate and review the flex-auth adoption plan `df34f6ec` -- · Prove failure recovery and telemetry continuity `5160ad75` - · Document operations and repository-boundary handoffs `06b6cde4` --- diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 0045f3b..1b25d10 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -327,6 +327,6 @@ | task | STATE-WP-0085-T04 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | | task | STATE-WP-0085-T05 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | | task | STATE-WP-0085-T06 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | -| task | STATE-WP-0085-T07 | todo | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | +| task | STATE-WP-0085-T07 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | | task | STATE-WP-0085-T08 | todo | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | | task | STATE-WP-0085-T09 | wait | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md | diff --git a/api/routers/repository_renames.py b/api/routers/repository_renames.py index e6febe2..cfdef31 100644 --- a/api/routers/repository_renames.py +++ b/api/routers/repository_renames.py @@ -1,12 +1,15 @@ from __future__ import annotations +import asyncio import uuid from typing import Any from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.encoders import jsonable_encoder from sqlalchemy.ext.asyncio import AsyncSession from api.database import get_session +from api.events import EventEnvelope, publish_event from api.models.repository_rename import RepositoryRenameOperation from api.schemas.repository_rename import ( ForgeIdentityVerifyRequest, @@ -44,10 +47,77 @@ operation_router = APIRouter( ) +def repository_rename_events( + operation: RepositoryRenameOperation, + *, + outcome: str, + no_op: bool = False, + verification: dict[str, Any] | None = None, +) -> list[tuple[str, EventEnvelope]]: + """Build credential-free lifecycle events from the durable journal.""" + telemetry = (operation.evidence or {}).get("telemetry") or {} + phase_durations = telemetry.get("phase_durations_ms") or {} + attributes = { + "operation_id": str(operation.id), + "repo_id": str(operation.repo_id), + "forge_repository_id": operation.expected_forge_repository_id, + "old_slug": operation.old_slug, + "new_slug": operation.new_slug, + "phase": operation.phase, + "actor": operation.actor, + "expected_source_commit": operation.expected_source_commit, + "outcome": outcome, + "no_op": no_op, + "phase_duration_ms": int(phase_durations.get(operation.phase) or 0), + "retries": int(telemetry.get("retries") or 0), + "failures": int(telemetry.get("failures") or 0), + "rollback_attempts": int(telemetry.get("rollback_attempts") or 0), + "rollback_outcome": telemetry.get("rollback_outcome"), + "verification_outcome": ( + "passed" if verification and verification.get("ok") else + "failed" if verification else + telemetry.get("verification_outcome", "pending") + ), + "error_code": operation.error_code if outcome == "failed" else None, + "evidence_ref": f"repository-renames/operations/{operation.id}", + } + subject = ( + "org.statehub.repo.rename.failed" + if outcome == "failed" + else "org.statehub.repo.rename.verified" + if verification is not None + else "org.statehub.repo.rename.rolled_back" + if operation.phase == "rolled-back" + else "org.statehub.repo.rename.phase" + ) + events = [(subject, EventEnvelope.new(subject, attributes=attributes))] + if operation.phase == "completed" and outcome == "succeeded": + renamed_subject = "org.statehub.repo.renamed" + events.append( + (renamed_subject, EventEnvelope.new(renamed_subject, attributes=attributes)) + ) + return events + + +def _publish_rename_events( + operation: RepositoryRenameOperation, + *, + outcome: str, + no_op: bool = False, + verification: dict[str, Any] | None = None, +) -> None: + for subject, envelope in repository_rename_events( + operation, outcome=outcome, no_op=no_op, verification=verification + ): + asyncio.create_task(publish_event(subject, envelope)) + + def _raise(exc: RenameLifecycleError) -> None: raise HTTPException( status_code=exc.status_code, - detail={"code": exc.code, "message": str(exc), "details": exc.details}, + detail=jsonable_encoder( + {"code": exc.code, "message": str(exc), "details": exc.details} + ), ) from exc @@ -129,6 +199,7 @@ async def create_repository_rename_operation( operation, no_op = await create_operation(session, gateway, repo_id, body) except RenameLifecycleError as exc: _raise(exc) + _publish_rename_events(operation, outcome="succeeded", no_op=no_op) return _operation_read(operation, no_op=no_op) @@ -194,7 +265,11 @@ async def apply_repository_rename_phase( session, gateway, repo_id, operation_id, phase, body ) except RenameLifecycleError as exc: + failed = await session.get(RepositoryRenameOperation, operation_id) + if failed is not None: + _publish_rename_events(failed, outcome="failed") _raise(exc) + _publish_rename_events(operation, outcome="succeeded", no_op=no_op) return _operation_read(operation, no_op=no_op) @@ -210,9 +285,15 @@ async def verify_repository_rename_operation( ) -> dict[str, Any]: try: operation = await load_operation(session, repo_id, operation_id) - return await verify_operation(session, gateway, operation) + verification = await verify_operation(session, gateway, operation) except RenameLifecycleError as exc: _raise(exc) + _publish_rename_events( + operation, + outcome="succeeded" if verification["ok"] else "failed", + verification=verification, + ) + return verification @router.post( @@ -236,7 +317,11 @@ async def preflight_repository_rename_rollback( confirmation=body.confirmation, ) except RenameLifecycleError as exc: + failed = await session.get(RepositoryRenameOperation, operation_id) + if failed is not None: + _publish_rename_events(failed, outcome="failed") _raise(exc) + _publish_rename_events(operation, outcome="succeeded") return { "operation_id": operation.id, "repo_id": operation.repo_id, @@ -266,5 +351,9 @@ async def rollback_repository_rename_operation( confirmation=body.confirmation, ) except RenameLifecycleError as exc: + failed = await session.get(RepositoryRenameOperation, operation_id) + if failed is not None: + _publish_rename_events(failed, outcome="failed") _raise(exc) + _publish_rename_events(operation, outcome="succeeded", no_op=no_op) return _operation_read(operation, no_op=no_op) diff --git a/api/schemas/repository_rename.py b/api/schemas/repository_rename.py index 797de45..aed4083 100644 --- a/api/schemas/repository_rename.py +++ b/api/schemas/repository_rename.py @@ -101,6 +101,7 @@ class RepositoryRenameVerificationRead(BaseModel): checks: list[dict[str, Any]] baseline_checksum: str current_checksum: str + relationship_checksums: dict[str, dict[str, str]] class RepositoryRenameRollbackPreflightRead(BaseModel): diff --git a/api/services/repository_rename.py b/api/services/repository_rename.py index ec9d832..4b09095 100644 --- a/api/services/repository_rename.py +++ b/api/services/repository_rename.py @@ -176,23 +176,78 @@ def _record_summary(ids: list[str]) -> dict[str, Any]: return {"count": len(ids), "ids": ids, "checksum": checksum(ids)} +async def _relationship_rows( + session: AsyncSession, query, fields: tuple[str, ...] +) -> list[dict[str, Any]]: + def normalize(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, datetime): + return value.astimezone(timezone.utc).isoformat() + if isinstance(value, uuid.UUID): + return str(value) + if hasattr(value, "value"): + return value.value + return str(value) + + result = await session.execute(query) + rows = [ + {field: normalize(getattr(row, field)) for field in fields} + for row in result.all() + ] + return sorted(rows, key=lambda row: canonical_json(row)) + + +def _relationship_summary( + rows: list[dict[str, Any]], *, totals: dict[str, int] | None = None +) -> dict[str, Any]: + ids = sorted(row["id"] for row in rows) + result = { + "count": len(rows), + "ids": ids, + "checksum": checksum(ids), + "relationships": rows, + "relationship_checksum": checksum(rows), + } + if totals is not None: + result["totals"] = totals + result["totals_checksum"] = checksum(totals) + return result + + async def collect_continuity_baseline( session: AsyncSession, repo_id: uuid.UUID ) -> dict[str, Any]: - workplans = await _id_rows( - session, select(Workplan.id).where(Workplan.repo_id == repo_id) - ) - workplan_uuids = [uuid.UUID(value) for value in workplans] - tasks = await _id_rows( + workplans = await _relationship_rows( session, - select(Task.id).where( + select( + Workplan.id.label("id"), + Workplan.repo_id.label("repo_id"), + Workplan.slug.label("slug"), + ).where(Workplan.repo_id == repo_id), + ("id", "repo_id", "slug"), + ) + workplan_uuids = [uuid.UUID(row["id"]) for row in workplans] + tasks = await _relationship_rows( + session, + select( + Task.id.label("id"), + Task.workplan_id.label("workplan_id"), + Task.record_id.label("record_id"), + ).where( Task.workplan_id.in_(workplan_uuids) if workplan_uuids else False ), + ("id", "workplan_id", "record_id"), ) - task_uuids = [uuid.UUID(value) for value in tasks] - progress = await _id_rows( + task_uuids = [uuid.UUID(row["id"]) for row in tasks] + progress = await _relationship_rows( session, - select(ProgressEvent.id).where( + select( + ProgressEvent.id.label("id"), + ProgressEvent.workplan_id.label("workplan_id"), + ProgressEvent.task_id.label("task_id"), + ProgressEvent.decision_id.label("decision_id"), + ).where( or_( ProgressEvent.workplan_id.in_(workplan_uuids) if workplan_uuids @@ -200,16 +255,31 @@ async def collect_continuity_baseline( ProgressEvent.task_id.in_(task_uuids) if task_uuids else False, ) ), + ("id", "workplan_id", "task_id", "decision_id"), ) - decisions = await _id_rows( + decisions = await _relationship_rows( session, - select(Decision.id).where( + select( + Decision.id.label("id"), + Decision.workplan_id.label("workplan_id"), + ).where( Decision.workplan_id.in_(workplan_uuids) if workplan_uuids else False ), + ("id", "workplan_id"), ) - token_events = await _id_rows( + token_events = await _relationship_rows( session, - select(TokenEvent.id).where( + select( + TokenEvent.id.label("id"), + TokenEvent.repo_id.label("repo_id"), + TokenEvent.workplan_id.label("workplan_id"), + TokenEvent.task_id.label("task_id"), + TokenEvent.tokens_in.label("tokens_in"), + TokenEvent.tokens_out.label("tokens_out"), + TokenEvent.cached_input_tokens.label("cached_input_tokens"), + TokenEvent.reasoning_output_tokens.label("reasoning_output_tokens"), + TokenEvent.raw_total_tokens.label("raw_total_tokens"), + ).where( or_( TokenEvent.repo_id == repo_id, TokenEvent.workplan_id.in_(workplan_uuids) @@ -218,44 +288,141 @@ async def collect_continuity_baseline( TokenEvent.task_id.in_(task_uuids) if task_uuids else False, ) ), + ( + "id", + "repo_id", + "workplan_id", + "task_id", + "tokens_in", + "tokens_out", + "cached_input_tokens", + "reasoning_output_tokens", + "raw_total_tokens", + ), ) - sbom_snapshots = await _id_rows( - session, select(SBOMSnapshot.id).where(SBOMSnapshot.repo_id == repo_id) - ) - sbom_entries = await _id_rows( - session, select(SBOMEntry.id).where(SBOMEntry.repo_id == repo_id) - ) - services = await _id_rows( + token_totals = { + field: sum(int(row[field]) for row in token_events if row[field] is not None) + for field in ( + "tokens_in", + "tokens_out", + "cached_input_tokens", + "reasoning_output_tokens", + "raw_total_tokens", + ) + } + sbom_snapshots = await _relationship_rows( session, - select(ServiceFirstParty.service_id).where(ServiceFirstParty.repo_id == repo_id), + select( + SBOMSnapshot.id.label("id"), + SBOMSnapshot.repo_id.label("repo_id"), + ).where(SBOMSnapshot.repo_id == repo_id), + ("id", "repo_id"), ) - capabilities = await _id_rows( + sbom_entries = await _relationship_rows( session, - select(CapabilityCatalog.id).where(CapabilityCatalog.repo_id == repo_id), + select( + SBOMEntry.id.label("id"), + SBOMEntry.repo_id.label("repo_id"), + SBOMEntry.snapshot_id.label("snapshot_id"), + ).where(SBOMEntry.repo_id == repo_id), + ("id", "repo_id", "snapshot_id"), ) - interface_changes = await _id_rows( + services = await _relationship_rows( session, - select(InterfaceChange.id).where(InterfaceChange.repo_id == repo_id), + select( + ServiceFirstParty.service_id.label("id"), + ServiceFirstParty.repo_id.label("repo_id"), + ).where(ServiceFirstParty.repo_id == repo_id), + ("id", "repo_id"), ) - bindings = await _id_rows( + capabilities = await _relationship_rows( session, - select(Workplan.id).where( + select( + CapabilityCatalog.id.label("id"), + CapabilityCatalog.repo_id.label("repo_id"), + ).where(CapabilityCatalog.repo_id == repo_id), + ("id", "repo_id"), + ) + interface_changes = await _relationship_rows( + session, + select( + InterfaceChange.id.label("id"), + InterfaceChange.repo_id.label("repo_id"), + ).where(InterfaceChange.repo_id == repo_id), + ("id", "repo_id"), + ) + bindings = await _relationship_rows( + session, + select( + Workplan.id.label("id"), + Workplan.repo_id.label("repo_id"), + Workplan.backing_relative_path.label("backing_relative_path"), + Workplan.backing_filename.label("backing_filename"), + ).where( Workplan.repo_id == repo_id, Workplan.backing_relative_path.is_not(None), ), + ("id", "repo_id", "backing_relative_path", "backing_filename"), ) + slug_rows = await _relationship_rows( + session, + select( + RepositorySlug.id.label("id"), + RepositorySlug.repo_id.label("repo_id"), + RepositorySlug.slug.label("slug"), + RepositorySlug.kind.label("kind"), + ).where(RepositorySlug.repo_id == repo_id), + ("id", "repo_id", "slug", "kind"), + ) + known_slugs = [row["slug"] for row in slug_rows] + messages = await _relationship_rows( + session, + select( + AgentMessage.id.label("id"), + AgentMessage.from_agent.label("from_agent"), + AgentMessage.to_agent.label("to_agent"), + AgentMessage.thread_id.label("thread_id"), + ).where( + or_( + AgentMessage.from_agent.in_(known_slugs), + AgentMessage.to_agent.in_(known_slugs), + ) + if known_slugs + else False + ), + ("id", "from_agent", "to_agent", "thread_id"), + ) + active_work = await _active_work(session, repo_id) + active_dispatch = [ + {"id": row["id"], "repo_id": str(repo_id), "kind": "workplan"} + for row in active_work["workplans"] + ] + [ + { + "id": row["id"], + "workplan_id": row["workplan_id"], + "kind": "task", + } + for row in active_work["tasks"] + ] + active_dispatch = sorted(active_dispatch, key=lambda row: canonical_json(row)) records = { - "workplans": _record_summary(workplans), - "tasks": _record_summary(tasks), - "progress_events": _record_summary(progress), - "decisions": _record_summary(decisions), - "token_events": _record_summary(token_events), - "sbom_snapshots": _record_summary(sbom_snapshots), - "sbom_entries": _record_summary(sbom_entries), - "services": _record_summary(services), - "capabilities": _record_summary(capabilities), - "interface_changes": _record_summary(interface_changes), - "workplan_bindings": _record_summary(bindings), + "repository": _relationship_summary( + [{"id": str(repo_id), "repo_id": str(repo_id)}] + ), + "workplans": _relationship_summary(workplans), + "tasks": _relationship_summary(tasks), + "progress_events": _relationship_summary(progress), + "decisions": _relationship_summary(decisions), + "token_events": _relationship_summary(token_events, totals=token_totals), + "sbom_snapshots": _relationship_summary(sbom_snapshots), + "sbom_entries": _relationship_summary(sbom_entries), + "services": _relationship_summary(services), + "capabilities": _relationship_summary(capabilities), + "interface_changes": _relationship_summary(interface_changes), + "workplan_bindings": _relationship_summary(bindings), + "active_dispatch": _relationship_summary(active_dispatch), + "aliases": _relationship_summary(slug_rows), + "messages": _relationship_summary(messages), } records["continuity_checksum"] = checksum(records) return records @@ -373,6 +540,11 @@ async def verify_forge_identity( raise RenamePreconditionFailed(str(exc)) from exc if snapshot is None: raise RenamePreconditionFailed("Forge repository is absent or unreadable") + if snapshot.name != repo.slug: + raise RenamePreconditionFailed( + "Forge identity verification followed a repository redirect", + details={"expected_name": repo.slug, "actual_name": snapshot.name}, + ) if snapshot.repository_id != body.forge_repository_id: raise RenamePreconditionFailed( "Forge repository ID does not match the asserted immutable ID", @@ -536,6 +708,15 @@ async def build_preflight( {"code": "forge_source_absent", "message": "Canonical Forge repository is absent or unreadable"} ) else: + if old_snapshot.name != repo.slug: + blockers.append( + { + "code": "forge_redirected", + "message": "Forge source lookup resolved to another coordinate", + "expected": repo.slug, + "actual": old_snapshot.name, + } + ) if old_snapshot.repository_id != identity.forge_repository_id: blockers.append( { @@ -786,6 +967,15 @@ async def create_operation( "phases": { "preflighted": {"at": now.isoformat(), "resumed": False} }, + "telemetry": { + "started_at": now.isoformat(), + "phase_attempts": {"preflighted": 1}, + "phase_durations_ms": {"preflighted": 0}, + "retries": 0, + "failures": 0, + "rollback_attempts": 0, + "verification_outcome": "pending", + }, }, ) session.add(operation) @@ -860,7 +1050,12 @@ async def list_operations( return list(result.scalars().all()) -def _assert_snapshot(operation: RepositoryRenameOperation, snapshot: ForgeRepositorySnapshot) -> None: +def _assert_snapshot( + operation: RepositoryRenameOperation, + snapshot: ForgeRepositorySnapshot, + *, + expected_name: str | None = None, +) -> None: if snapshot.repository_id != operation.expected_forge_repository_id: raise RenamePreconditionFailed( "Forge repository ID changed", @@ -881,6 +1076,11 @@ def _assert_snapshot(operation: RepositoryRenameOperation, snapshot: ForgeReposi ) if not snapshot.projection_readable or not snapshot.projection_source_present: raise RenamePreconditionFailed("Forge work-record projection is unreadable") + if expected_name is not None and snapshot.name != expected_name: + raise RenamePreconditionFailed( + "Forge response resolved to a different repository coordinate", + details={"expected_name": expected_name, "actual_name": snapshot.name}, + ) async def _forge_at( @@ -904,13 +1104,13 @@ async def _apply_forge_rename( old = await _forge_at(gateway, operation, operation.old_slug) new = await _forge_at(gateway, operation, operation.new_slug) if new is not None: - _assert_snapshot(operation, new) + _assert_snapshot(operation, new, expected_name=operation.new_slug) if old is not None: raise RenamePreconditionFailed("Both old and new Forge names are claimed") return {"resumed": True, "forge": new.as_dict()} if old is None: raise RenamePreconditionFailed("Forge repository is absent at both expected names") - _assert_snapshot(operation, old) + _assert_snapshot(operation, old, expected_name=operation.old_slug) try: renamed = await gateway.rename( instance=operation.expected_forge_instance, @@ -920,7 +1120,7 @@ async def _apply_forge_rename( ) except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc: raise RenamePreconditionFailed(str(exc)) from exc - _assert_snapshot(operation, renamed) + _assert_snapshot(operation, renamed, expected_name=operation.new_slug) return {"resumed": False, "forge": renamed.as_dict()} @@ -1005,6 +1205,7 @@ async def verify_operation( else: add("forge_readable", forge is not None, True, forge is not None) if forge is not None: + 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("source_commit", forge.head_commit == operation.expected_source_commit, operation.expected_source_commit, forge.head_commit) add("default_branch", forge.default_branch == operation.expected_default_branch, operation.expected_default_branch, forge.default_branch) @@ -1022,6 +1223,8 @@ async def verify_operation( baseline = (operation.evidence.get("preflight") or {}).get("baselines") or {} expected_checksum = baseline.get("continuity_checksum", "") missing: dict[str, list[str]] = {} + changed_relationships: dict[str, list[str]] = {} + preserved_relationship_checksums: dict[str, str] = {} counts: dict[str, dict[str, int]] = {} for record_type, baseline_record in baseline.items(): if record_type == "continuity_checksum" or not isinstance(baseline_record, dict): @@ -1030,6 +1233,30 @@ async def verify_operation( absent = sorted(set(baseline_record.get("ids") or []) - set(current_record.get("ids") or [])) if absent: missing[record_type] = absent + if record_type != "aliases": + baseline_rows = { + row["id"]: row + for row in baseline_record.get("relationships") or [] + } + current_rows = { + row["id"]: row + for row in current_record.get("relationships") or [] + } + changed = sorted( + record_id + for record_id, baseline_row in baseline_rows.items() + if current_rows.get(record_id) != baseline_row + ) + if changed: + changed_relationships[record_type] = changed + preserved_rows = [ + current_rows[record_id] + for record_id in baseline_rows + if record_id in current_rows + ] + preserved_relationship_checksums[record_type] = checksum( + sorted(preserved_rows, key=lambda row: canonical_json(row)) + ) counts[record_type] = { "baseline": int(baseline_record.get("count") or 0), "current": int(current_record.get("count") or 0), @@ -1037,13 +1264,54 @@ async def verify_operation( # New telemetry may legitimately arrive while a phased operation is being # executed. Continuity means every baseline identity still exists; it does # not freeze the repository's append-only history at the preflight count. - add("relationship_continuity", not missing, {"missing": {}}, {"missing": missing}) + add( + "relationship_continuity", + not missing and not changed_relationships, + {"missing": {}, "changed": {}}, + {"missing": missing, "changed": changed_relationships}, + ) add( "record_counts_non_decreasing", all(item["current"] >= item["baseline"] for item in counts.values()), {name: item["baseline"] for name, item in counts.items()}, {name: item["current"] for name, item in counts.items()}, ) + baseline_token_totals = (baseline.get("token_events") or {}).get("totals") or {} + current_token_totals = (current.get("token_events") or {}).get("totals") or {} + add( + "token_totals_non_decreasing", + all( + int(current_token_totals.get(name) or 0) >= int(value or 0) + for name, value in baseline_token_totals.items() + ), + baseline_token_totals, + current_token_totals, + ) + slug_result = await session.execute( + select(RepositorySlug).where( + RepositorySlug.repo_id == operation.repo_id, + RepositorySlug.slug.in_((operation.old_slug, operation.new_slug)), + ) + ) + slug_routes = {row.slug: row.kind for row in slug_result.scalars().all()} + if operation.phase in {"preflighted", "forge-renamed"}: + expected_routes = {operation.old_slug: "canonical"} + elif operation.phase == "rolled-back": + expected_routes = { + operation.old_slug: "canonical", + operation.new_slug: "alias", + } + else: + expected_routes = { + operation.old_slug: "alias", + operation.new_slug: "canonical", + } + add( + "slug_routes", + all(slug_routes.get(slug) == kind for slug, kind in expected_routes.items()), + expected_routes, + slug_routes, + ) return { "operation_id": operation.id, "repo_id": operation.repo_id, @@ -1052,12 +1320,63 @@ async def verify_operation( "checks": checks, "baseline_checksum": expected_checksum, "current_checksum": current["continuity_checksum"], + "relationship_checksums": { + "baseline": { + name: value.get("relationship_checksum") + for name, value in baseline.items() + if isinstance(value, dict) and name != "aliases" + }, + "preserved": preserved_relationship_checksums, + }, } +def _telemetry(operation: RepositoryRenameOperation) -> tuple[dict[str, Any], dict[str, Any]]: + journal = deepcopy(operation.evidence or {}) + telemetry = dict(journal.get("telemetry") or {}) + telemetry.setdefault("started_at", operation.preflighted_at.isoformat()) + telemetry.setdefault("phase_attempts", {}) + telemetry.setdefault("phase_durations_ms", {}) + telemetry.setdefault("retries", 0) + telemetry.setdefault("failures", 0) + telemetry.setdefault("rollback_attempts", 0) + telemetry.setdefault("verification_outcome", "pending") + return journal, telemetry + + +def _record_retry(operation: RepositoryRenameOperation, phase: str) -> None: + journal, telemetry = _telemetry(operation) + attempts = dict(telemetry["phase_attempts"]) + attempts[phase] = int(attempts.get(phase) or 0) + 1 + telemetry["phase_attempts"] = attempts + telemetry["retries"] = int(telemetry["retries"]) + 1 + telemetry["last_retry"] = {"phase": phase, "at": utcnow().isoformat()} + journal["telemetry"] = telemetry + operation.evidence = journal + + +def _record_failure( + operation: RepositoryRenameOperation, phase: str, exc: RenameLifecycleError +) -> None: + journal, telemetry = _telemetry(operation) + failures_by_phase = dict(telemetry.get("failures_by_phase") or {}) + failures_by_phase[phase] = int(failures_by_phase.get(phase) or 0) + 1 + telemetry["failures_by_phase"] = failures_by_phase + telemetry["failures"] = int(telemetry["failures"]) + 1 + telemetry["last_failure"] = { + "phase": phase, + "code": exc.code, + "at": utcnow().isoformat(), + } + if phase.startswith("rollback"): + telemetry["rollback_outcome"] = "failed" + journal["telemetry"] = telemetry + operation.evidence = journal + + def _record_phase(operation: RepositoryRenameOperation, phase: str, evidence: dict[str, Any]) -> None: now = utcnow() - journal = deepcopy(operation.evidence or {}) + journal, telemetry = _telemetry(operation) phases = dict(journal.get("phases") or {}) # Verification payloads use native UUID/datetime values for API response # typing. The operation journal is JSONB, so normalize at the boundary @@ -1065,6 +1384,26 @@ def _record_phase(operation: RepositoryRenameOperation, phase: str, evidence: di safe_evidence = json.loads(canonical_json(evidence)) phases[phase] = {"at": now.isoformat(), **safe_evidence} journal["phases"] = phases + attempts = dict(telemetry["phase_attempts"]) + attempts[phase] = int(attempts.get(phase) or 0) + 1 + telemetry["phase_attempts"] = attempts + durations = dict(telemetry["phase_durations_ms"]) + durations[phase] = max( + 0, int((now - operation.phase_changed_at).total_seconds() * 1000) + ) + telemetry["phase_durations_ms"] = durations + telemetry["last_phase"] = phase + if phase == "rollback-preflight": + telemetry["rollback_attempts"] = int(telemetry["rollback_attempts"]) + 1 + telemetry["rollback_outcome"] = "approved" + elif phase == "rolled-back": + telemetry["rollback_outcome"] = "completed" + verification = evidence.get("verification") + if isinstance(verification, dict): + telemetry["verification_outcome"] = ( + "passed" if verification.get("ok") else "failed" + ) + journal["telemetry"] = telemetry operation.evidence = journal operation.phase = phase operation.phase_changed_at = now @@ -1101,6 +1440,9 @@ async def apply_phase( current_index = _phase_index(operation.phase) requested_index = _phase_index(requested_phase) if current_index >= requested_index >= 0: + _record_retry(operation, requested_phase) + await session.commit() + await session.refresh(operation) return operation, True if operation.phase != body.expected_phase: raise RenamePreconditionFailed( @@ -1119,17 +1461,35 @@ async def apply_phase( forge = await _forge_at(gateway, operation, operation.new_slug) if forge is None: raise RenamePreconditionFailed("Forge rename is not observable") - _assert_snapshot(operation, forge) + _assert_snapshot(operation, forge, expected_name=operation.new_slug) evidence = await _apply_statehub_rebind(session, operation) elif requested_phase == "source-synced": - if not body.evidence: + if not body.evidence or body.evidence.get("fresh_clone") is not True: raise RenamePreconditionFailed( - "Source synchronization requires operator evidence" + "Source synchronization requires fresh-clone operator evidence" + ) + clone_repository_id = body.evidence.get("forge_repository_id") + if clone_repository_id != operation.expected_forge_repository_id: + raise RenamePreconditionFailed( + "Fresh clone points at the wrong Forge repository ID", + details={ + "expected": operation.expected_forge_repository_id, + "actual": clone_repository_id, + }, + ) + clone_head = body.evidence.get("head_commit") + if clone_head != operation.expected_source_commit: + raise RenamePreconditionFailed( + "Fresh clone head does not match the rename baseline", + details={ + "expected": operation.expected_source_commit, + "actual": clone_head, + }, ) forge = await _forge_at(gateway, operation, operation.new_slug) if forge is None: raise RenamePreconditionFailed("Renamed Forge repository is absent") - _assert_snapshot(operation, forge) + _assert_snapshot(operation, forge, expected_name=operation.new_slug) evidence = {"forge": forge.as_dict(), "operator_evidence": body.evidence} elif requested_phase == "consumers-verified": if not body.checks or not all(body.checks.values()): @@ -1161,9 +1521,10 @@ async def apply_phase( # phase is precisely what lets the next request discover and resume it. await session.rollback() failed = await load_operation(session, repo_id, operation_id, for_update=True) + _record_failure(failed, requested_phase, exc) failed.error_code = exc.code failed.error_message = str(exc) - failed.error_details = exc.details + failed.error_details = json.loads(canonical_json(exc.details)) failed.error_at = utcnow() await session.commit() raise @@ -1173,9 +1534,10 @@ async def apply_phase( "Repository rename phase lost a database compare-and-set race" ) failed = await load_operation(session, repo_id, operation_id, for_update=True) + _record_failure(failed, requested_phase, failure) failed.error_code = failure.code failed.error_message = str(failure) - failed.error_details = failure.details + failed.error_details = json.loads(canonical_json(failure.details)) failed.error_at = utcnow() await session.commit() raise failure from exc @@ -1225,7 +1587,7 @@ async def rollback_preflight( if operation.phase == "preflighted": if old is not None and new is None: try: - _assert_snapshot(operation, old) + _assert_snapshot(operation, old, expected_name=operation.old_slug) except RenameLifecycleError as exc: blockers.append({"code": exc.code, "message": str(exc), **exc.details}) elif old is None and new is not None: @@ -1233,7 +1595,7 @@ async def rollback_preflight( # principal interruption case the operation ID must recover from. rollback_from_phase = "forge-renamed-unrecorded" try: - _assert_snapshot(operation, new) + _assert_snapshot(operation, new, expected_name=operation.new_slug) except RenameLifecycleError as exc: blockers.append({"code": exc.code, "message": str(exc), **exc.details}) elif old is None: @@ -1247,7 +1609,7 @@ async def rollback_preflight( blockers.append({"code": "renamed_forge_repository_absent"}) else: try: - _assert_snapshot(operation, new) + _assert_snapshot(operation, new, expected_name=operation.new_slug) except RenameLifecycleError as exc: blockers.append({"code": exc.code, "message": str(exc), **exc.details}) irreversible = [ @@ -1262,7 +1624,16 @@ async def rollback_preflight( "irreversible": irreversible, } if blockers: - raise RenamePreconditionFailed("Repository rename rollback is unsafe", details=data) + failure = RenamePreconditionFailed( + "Repository rename rollback is unsafe", details=data + ) + _record_failure(operation, "rollback-preflight", failure) + operation.error_code = failure.code + operation.error_message = str(failure) + operation.error_details = json.loads(canonical_json(failure.details)) + operation.error_at = utcnow() + await session.commit() + raise failure journal = deepcopy(operation.evidence or {}) journal["rollback_preflight"] = data operation.evidence = journal @@ -1319,6 +1690,9 @@ async def apply_rollback( ) -> tuple[RepositoryRenameOperation, bool]: operation = await load_operation(session, repo_id, operation_id, for_update=True) if operation.phase == "rolled-back": + _record_retry(operation, "rolled-back") + await session.commit() + await session.refresh(operation) return operation, True if operation.phase != expected_phase or operation.phase != "rollback-preflight": raise RenamePreconditionFailed( @@ -1330,35 +1704,56 @@ async def apply_rollback( rollback_from = ((operation.evidence or {}).get("rollback_preflight") or {}).get("rollback_from_phase") if not rollback_from: raise RenamePreconditionFailed("Rollback preflight evidence is missing") - old = await _forge_at(gateway, operation, operation.old_slug) - new = await _forge_at(gateway, operation, operation.new_slug) - forge_resumed = False - if rollback_from != "preflighted": - if old is not None: - _assert_snapshot(operation, old) - if new is not None: - raise RenamePreconditionFailed("Both Forge coordinates are claimed during rollback") - forge_resumed = True - else: - if new is None: - raise RenamePreconditionFailed("Forge repository is absent during rollback") - _assert_snapshot(operation, new) - try: - restored = await gateway.rename( - instance=operation.expected_forge_instance, - owner=operation.expected_forge_owner, - old_name=operation.new_slug, - new_name=operation.old_slug, + try: + old = await _forge_at(gateway, operation, operation.old_slug) + new = await _forge_at(gateway, operation, operation.new_slug) + forge_resumed = False + if rollback_from != "preflighted": + if old is not None: + _assert_snapshot(operation, old, expected_name=operation.old_slug) + if new is not None: + raise RenamePreconditionFailed( + "Both Forge coordinates are claimed during rollback" + ) + forge_resumed = True + else: + if new is None: + raise RenamePreconditionFailed( + "Forge repository is absent during rollback" + ) + _assert_snapshot(operation, new, expected_name=operation.new_slug) + try: + restored = await gateway.rename( + instance=operation.expected_forge_instance, + owner=operation.expected_forge_owner, + old_name=operation.new_slug, + new_name=operation.old_slug, + ) + except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc: + raise RenamePreconditionFailed(str(exc)) from exc + _assert_snapshot( + operation, restored, expected_name=operation.old_slug ) - except (ForgeRepositoryUnreadable, ForgeRepositoryConflict) as exc: - raise RenamePreconditionFailed(str(exc)) from exc - _assert_snapshot(operation, restored) - statehub = await _rollback_statehub(session, operation) - _record_phase( - operation, - "rolled-back", - {"rollback_from_phase": rollback_from, "forge_resumed": forge_resumed, "statehub": statehub}, - ) - await session.commit() - await session.refresh(operation) - return operation, False + statehub = await _rollback_statehub(session, operation) + _record_phase( + operation, + "rolled-back", + { + "rollback_from_phase": rollback_from, + "forge_resumed": forge_resumed, + "statehub": statehub, + }, + ) + await session.commit() + await session.refresh(operation) + return operation, False + except RenameLifecycleError as exc: + await session.rollback() + failed = await load_operation(session, repo_id, operation_id, for_update=True) + _record_failure(failed, "rolled-back", exc) + failed.error_code = exc.code + failed.error_message = str(exc) + failed.error_details = json.loads(canonical_json(exc.details)) + failed.error_at = utcnow() + await session.commit() + raise diff --git a/docs/evidence/STATE-WP-0085-T07-repository-rename-recovery-matrix.md b/docs/evidence/STATE-WP-0085-T07-repository-rename-recovery-matrix.md new file mode 100644 index 0000000..3721cb8 --- /dev/null +++ b/docs/evidence/STATE-WP-0085-T07-repository-rename-recovery-matrix.md @@ -0,0 +1,70 @@ +# Repository rename recovery and continuity matrix + +Work item: `STATE-WP-0085-T07` +Scope: State Hub lifecycle only; no live `flex-auth`/`access-engine` mutation. + +## Automated recovery matrix + +| Hazard | Automated proof | Required result | +| --- | --- | --- | +| Interruption/failure after every forward phase | `test_every_forward_phase_failure_is_retry_safe`; `test_interrupt_resume_every_phase_and_preserve_uuid` | Journal remains at the last achieved phase; retry resumes the same operation UUID. | +| Stale preflight/head | `test_stale_head_wrong_id_target_conflict_and_queued_writes_fail_closed` | Operation creation fails before mutation. | +| Double submission | `test_client_operation_id_is_idempotent_and_globally_discoverable` | Same intent and operation UUID returns `no_op`; changed intent fails closed. | +| Conflicting target rename | `test_stale_head_wrong_id_target_conflict_and_queued_writes_fail_closed` | Claimed Forge/State Hub target blocks preflight. | +| Forge redirect | `test_redirected_forge_and_wrong_fresh_clone_identity_fail_closed_then_resume` | A redirected lookup cannot masquerade as the requested coordinate. | +| Old slug unavailable during rollback | `test_rollback_recovers_when_old_statehub_slug_temporarily_unavailable` | Failure is journaled; restoring the protected route permits a safe retry after the external Forge rollback already committed. | +| State Hub outage | `test_state_hub_outage_fails_without_losing_operation_identity_or_leaking_secrets` | CLI reports `state_hub_unavailable`; the client-owned operation ID remains the recovery key. | +| Edge outbox replay | `test_message_history_is_immutable_and_old_slug_write_replays_once`; edge outbox/relay suites | Old-slug write canonicalizes and persists exactly once under its idempotency key. | +| Unreadable private repository | `test_expired_token_bad_confirmation_and_unreadable_forge_fail_closed` | Preflight is unsafe and no token is issued. | +| Fresh clone has wrong immutable Forge ID | `test_redirected_forge_and_wrong_fresh_clone_identity_fail_closed_then_resume` | `source-synced` fails, journals the failure, and succeeds only with matching repository ID and head. | + +## Continuity proof + +`collect_continuity_baseline` records both identity lists and immutable +relationship tuples. Each record family has a `relationship_checksum`; token +events additionally record input, output, cached-input, reasoning-output, and +raw totals. Verification recomputes the relationships for every baseline ID, +permits append-only records, and fails when an original ID is missing or any +repository/workplan/task/snapshot binding changes. + +The matrix covers repository identity, workplans, tasks, progress, decisions, +token events and totals, SBOM snapshots and entries, active dispatch, protected +slug routes, historical messages, and workplan file bindings. Alias verification +is phase-aware because the intended mutation changes the old route from +`canonical` to `alias`; all work-record and telemetry relationships remain +unchanged. + +`test_equal_counts_do_not_hide_detached_work_and_telemetry` deliberately moves +an original workplan and token event to another repository, inserts replacements +with identical counts and token totals, and proves verification still fails. + +## Operational evidence + +The durable operation journal exposes: + +- phase attempt counts and durations in milliseconds; +- total retries and failures plus failures by phase; +- rollback attempts and outcome; +- verification outcome; +- the last failure code, without operator evidence or credentials. + +Credential-free NATS events publish phase, failure, verification, rollback, and +completion outcomes. Event payloads contain immutable IDs, coordinate names, +the expected source commit, metrics, and an operation evidence reference. They +never include preflight tokens, supplied operator evidence, authorization data, +remote URLs, or error details. + +## Reproduction + +```bash +.venv/bin/pytest -q \ + tests/test_repository_rename_api.py \ + tests/test_repository_rename_cli.py \ + tests/test_repository_alias_routing.py \ + tests/test_edge_outbox.py \ + tests/test_edge_relay.py +``` + +Run the full repository suite before accepting this work item. Record the final +test count and revision in the workplan result rather than editing this matrix +with transient local values. diff --git a/docs/nats-event-subjects.md b/docs/nats-event-subjects.md index a3876b9..93a12f1 100644 --- a/docs/nats-event-subjects.md +++ b/docs/nats-event-subjects.md @@ -42,6 +42,11 @@ those publishers from colliding on the same `{noun}.{verb}` shape. | Subject | When | Required attributes | | ------------------------------------ | ------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------- | | `org.statehub.repo.registered` | A new repo is registered via `POST /repos/` | `repo_id`, `repo_slug`, `domain_slug`, `remote_url?`, `local_path?` | +| `org.statehub.repo.rename.phase` | A rename operation starts, advances, or is safely retried | `operation_id`, `repo_id`, `forge_repository_id`, `old_slug`, `new_slug`, `phase`, `actor`, `expected_source_commit`, `outcome`, `no_op`, phase metrics, `evidence_ref` | +| `org.statehub.repo.rename.failed` | A rename or rollback phase fails closed | Rename phase attributes plus `error_code`; never error details or supplied evidence | +| `org.statehub.repo.rename.verified` | An operator/API verification evaluates continuity | Rename phase attributes plus `verification_outcome` | +| `org.statehub.repo.rename.rolled_back` | A rename operation reaches `rolled-back` | Rename phase attributes plus rollback attempts and outcome | +| `org.statehub.repo.renamed` | A rename operation reaches `completed` | Rename phase attributes; repository and Forge immutable IDs remain unchanged | | `org.statehub.workplan.completed` | A workplan transitions to canonical status `finished` | `workplan_id`, `slug`, `title`, `topic_id`, `repo_id?`, `repo_goal_id?` | | ~~`org.statehub.workstream.completed`~~ | **Retired 2026-07-08** (`STATE-WP-0069` T05). Use `org.statehub.workplan.completed`. | — | | `org.statehub.decision.resolved` | A decision is resolved via `POST /decisions/{id}/resolve` | `decision_id`, `title`, `topic_id?`, `workstream_id?`, `decided_by`, `rationale_snippet` | diff --git a/repository_rename_cli.py b/repository_rename_cli.py index b0fb126..02e2fd1 100644 --- a/repository_rename_cli.py +++ b/repository_rename_cli.py @@ -88,7 +88,7 @@ def _redact_text(value: str) -> str: flags=re.IGNORECASE, ) value = re.sub( - r"([?&][^=&#\s]*(?:api[_-]?key|password|secret|signature|token|credential)" + r"((?:^|[\s?&])[^=&#\s]*(?:api[_-]?key|authorization|password|secret|signature|token|credential)" r"[^=&#\s]*=)[^&#\s]+", r"\1[REDACTED]", value, diff --git a/tests/test_repository_rename_api.py b/tests/test_repository_rename_api.py index 5d6d3ae..adcdd97 100644 --- a/tests/test_repository_rename_api.py +++ b/tests/test_repository_rename_api.py @@ -1,16 +1,24 @@ from __future__ import annotations import uuid +from datetime import datetime, timezone import pytest import pytest_asyncio -from sqlalchemy import func, select +from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from api.config import settings from api.main import app from api.models.managed_repo import ManagedRepo from api.models.repository_rename import RepositoryRenameOperation, RepositorySlug +from api.models.agent_message import AgentMessage +from api.models.decision import Decision +from api.models.sbom_entry import Ecosystem, SBOMEntry +from api.models.sbom_snapshot import SBOMSnapshot +from api.models.token_event import TokenEvent +from api.models.workplan import Workplan +from api.routers.repository_renames import repository_rename_events from api.services.forge_repository import ( ForgeRepositorySnapshot, get_forge_repository_gateway, @@ -251,7 +259,17 @@ async def test_interrupt_resume_every_phase_and_preserve_uuid(client, rename_set sequence = [ ("statehub-rebound", "forge-renamed", {}), - ("source-synced", "statehub-rebound", {"evidence": {"clone": "fresh"}}), + ( + "source-synced", + "statehub-rebound", + { + "evidence": { + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + } + }, + ), ("consumers-verified", "source-synced", {"checks": {"routes": True, "security-stack": True}}), ("completed", "consumers-verified", {}), ] @@ -383,7 +401,17 @@ async def test_rollback_restores_old_canonical_and_remains_auditable(client, ren for requested, expected, extra in [ ("forge-renamed", "preflighted", {}), ("statehub-rebound", "forge-renamed", {}), - ("source-synced", "statehub-rebound", {"evidence": {"clone": "fresh"}}), + ( + "source-synced", + "statehub-rebound", + { + "evidence": { + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + } + }, + ), ]: await _phase(client, repo["id"], operation_id, requested, expected, confirmation, **extra) @@ -441,3 +469,463 @@ async def test_rollback_recovers_unrecorded_forge_rename(client, rename_setup): assert rollback.status_code == 200, rollback.text assert rollback.json()["phase"] == "rolled-back" assert forge.name == "flex-auth" + + +@pytest.mark.asyncio +async def test_continuity_evidence_covers_relationships_totals_routes_and_dispatch( + client, test_engine, rename_setup +): + repo, _forge = rename_setup + workplan = await create_test_workplan( + client, repo_id=repo["id"], slug="FLEX-WP-0042", status="active" + ) + task = await client.post( + "/tasks/", + json={ + "workplan_id": workplan["id"], + "title": "Preserve security-stack lineage", + "status": "todo", + "priority": "high", + }, + ) + assert task.status_code == 201, task.text + now = datetime.now(timezone.utc) + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + bound = await session.get(Workplan, uuid.UUID(workplan["id"])) + bound.backing_filename = "FLEX-WP-0042-lineage.md" + bound.backing_relative_path = "workplans/FLEX-WP-0042-lineage.md" + decision = Decision( + workplan_id=bound.id, + title="Retain immutable repository identity", + ) + token = TokenEvent( + repo_id=uuid.UUID(repo["id"]), + workplan_id=bound.id, + task_id=uuid.UUID(task.json()["id"]), + tokens_in=120, + tokens_out=30, + cached_input_tokens=10, + reasoning_output_tokens=5, + raw_total_tokens=165, + ) + snapshot = SBOMSnapshot( + repo_id=uuid.UUID(repo["id"]), + snapshot_at=now, + source="pytest", + entry_count=1, + created_at=now, + ) + message = AgentMessage( + from_agent="security-review", + to_agent="flex-auth", + subject="Historical route", + body="Keep the recorded coordinate", + ) + session.add_all([decision, token, snapshot, message]) + await session.flush() + session.add( + SBOMEntry( + repo_id=uuid.UUID(repo["id"]), + snapshot_id=snapshot.id, + package_name="policy-engine", + package_version="1.0", + ecosystem=Ecosystem.python, + snapshot_at=now, + created_at=now, + ) + ) + await session.commit() + + preflight = await _preflight(client, repo["id"]) + baseline = preflight["baselines"] + assert set(baseline) >= { + "repository", + "workplans", + "tasks", + "progress_events", + "decisions", + "token_events", + "sbom_snapshots", + "active_dispatch", + "aliases", + "messages", + "workplan_bindings", + "continuity_checksum", + } + assert baseline["token_events"]["totals"] == { + "tokens_in": 120, + "tokens_out": 30, + "cached_input_tokens": 10, + "reasoning_output_tokens": 5, + "raw_total_tokens": 165, + } + assert baseline["workplan_bindings"]["relationships"][0]["repo_id"] == repo["id"] + + operation, confirmation = await _operation(client, repo["id"], preflight) + for requested, expected, extra in [ + ("forge-renamed", "preflighted", {}), + ("statehub-rebound", "forge-renamed", {}), + ( + "source-synced", + "statehub-rebound", + { + "evidence": { + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + } + }, + ), + ("consumers-verified", "source-synced", {"checks": {"dispatch": True}}), + ("completed", "consumers-verified", {}), + ]: + await _phase( + client, + repo["id"], + operation["id"], + requested, + expected, + confirmation, + **extra, + ) + + verification = await client.get( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify" + ) + assert verification.status_code == 200, verification.text + evidence = verification.json() + assert evidence["ok"] is True + for name, expected_checksum in evidence["relationship_checksums"]["baseline"].items(): + assert evidence["relationship_checksums"]["preserved"][name] == expected_checksum + assert (await client.get("/repos/flex-auth")).json()["slug_status"] == "alias" + dispatch = await client.get("/repos/flex-auth/dispatch") + assert dispatch.status_code == 200 + assert dispatch.json()["canonical_slug"] == "access-engine" + assert dispatch.json()["active_workplans"][0]["id"] == workplan["id"] + + +@pytest.mark.asyncio +async def test_equal_counts_do_not_hide_detached_work_and_telemetry( + client, test_engine, rename_setup +): + repo, _forge = rename_setup + workplan = await create_test_workplan( + client, repo_id=repo["id"], slug="FLEX-WP-0050", status="active" + ) + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + original_token = TokenEvent( + repo_id=uuid.UUID(repo["id"]), tokens_in=11, tokens_out=7 + ) + session.add(original_token) + await session.commit() + preflight = await _preflight(client, repo["id"]) + operation, _confirmation = await _operation(client, repo["id"], preflight) + + other_domain = await create_test_domain(client, slug="other-domain") + other_repo = await create_test_repo( + client, domain_slug=other_domain["slug"], slug="other-repo" + ) + async with factory() as session: + detached = await session.get(Workplan, uuid.UUID(workplan["id"])) + detached.repo_id = uuid.UUID(other_repo["id"]) + token = await session.get(TokenEvent, original_token.id) + token.repo_id = uuid.UUID(other_repo["id"]) + session.add( + Workplan( + repo_id=uuid.UUID(repo["id"]), + slug="FLEX-WP-0050-RECREATED", + title="Count-preserving replacement", + status="active", + ) + ) + session.add( + TokenEvent(repo_id=uuid.UUID(repo["id"]), tokens_in=11, tokens_out=7) + ) + await session.commit() + + verification = await client.get( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify" + ) + assert verification.status_code == 200, verification.text + result = verification.json() + assert result["ok"] is False + relationship = next( + check for check in result["checks"] if check["name"] == "relationship_continuity" + ) + assert relationship["ok"] is False + assert set(relationship["actual"]["missing"]) >= {"workplans", "token_events"} + counts = next( + check for check in result["checks"] if check["name"] == "record_counts_non_decreasing" + ) + assert counts["actual"]["workplans"] == counts["expected"]["workplans"] + assert counts["actual"]["token_events"] == counts["expected"]["token_events"] + + +@pytest.mark.asyncio +async def test_redirected_forge_and_wrong_fresh_clone_identity_fail_closed_then_resume( + client, rename_setup +): + repo, forge = rename_setup + original_inspect = forge.inspect + + async def redirected(*, instance: str, owner: str, name: str): + if name == "flex-auth": + return forge.snapshot("access-engine") + if name == "access-engine": + return None + return await original_inspect(instance=instance, owner=owner, name=name) + + forge.inspect = redirected + redirected_report = await _preflight(client, repo["id"]) + assert redirected_report["safe_to_apply"] is False + assert {item["code"] for item in redirected_report["blockers"]} >= {"forge_redirected"} + forge.inspect = original_inspect + + 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) + wrong = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/phases/source-synced", + json={ + "expected_phase": "statehub-rebound", + "confirmation": confirmation, + "evidence": { + "fresh_clone": True, + "forge_repository_id": 999, + "head_commit": "a" * 40, + "password": "must-never-reach-an-event", + }, + }, + ) + assert wrong.status_code == 412 + status = (await client.get(f"/repository-renames/operations/{operation['id']}")).json() + assert status["phase"] == "statehub-rebound" + assert status["evidence"]["telemetry"]["failures"] == 1 + resumed = await _phase( + client, + repo["id"], + operation["id"], + "source-synced", + "statehub-rebound", + confirmation, + evidence={ + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + }, + ) + replay = await _phase( + client, + repo["id"], + operation["id"], + "source-synced", + "statehub-rebound", + confirmation, + evidence={ + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + }, + ) + assert resumed["phase"] == "source-synced" + assert replay["no_op"] is True + assert replay["evidence"]["telemetry"]["retries"] == 1 + + +@pytest.mark.asyncio +async def test_rollback_recovers_when_old_statehub_slug_temporarily_unavailable( + client, test_engine, rename_setup +): + repo, forge = rename_setup + preflight = await _preflight(client, repo["id"]) + operation, confirmation = await _operation(client, repo["id"], preflight) + for requested, expected, extra in [ + ("forge-renamed", "preflighted", {}), + ("statehub-rebound", "forge-renamed", {}), + ( + "source-synced", + "statehub-rebound", + { + "evidence": { + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + } + }, + ), + ]: + await _phase(client, repo["id"], operation["id"], requested, expected, confirmation, **extra) + rollback_confirmation = f"rollback:{operation['id']}" + report = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback-preflight", + json={"expected_phase": "source-synced", "confirmation": rollback_confirmation}, + ) + assert report.status_code == 200, report.text + + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + await session.execute(delete(RepositorySlug).where(RepositorySlug.slug == "flex-auth")) + await session.commit() + failed = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback", + json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation}, + ) + assert failed.status_code == 412 + assert forge.name == "flex-auth" + async with factory() as session: + session.add( + RepositorySlug( + repo_id=uuid.UUID(repo["id"]), + slug="flex-auth", + kind="alias", + protected=True, + source_operation_id=uuid.UUID(operation["id"]), + ) + ) + await session.commit() + resumed = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback", + json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation}, + ) + assert resumed.status_code == 200, resumed.text + assert resumed.json()["phase"] == "rolled-back" + assert resumed.json()["evidence"]["telemetry"]["failures"] == 1 + assert resumed.json()["evidence"]["telemetry"]["rollback_outcome"] == "completed" + + +@pytest.mark.asyncio +async def test_rename_events_expose_metrics_without_operator_evidence_or_secrets( + client, test_engine, rename_setup +): + repo, _forge = rename_setup + 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) + replay = await _phase(client, repo["id"], operation["id"], "forge-renamed", "preflighted", confirmation) + assert replay["evidence"]["telemetry"]["retries"] == 1 + + factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False) + async with factory() as session: + persisted = await session.get(RepositoryRenameOperation, uuid.UUID(operation["id"])) + events = repository_rename_events(persisted, outcome="succeeded", no_op=True) + assert [subject for subject, _event in events] == ["org.statehub.repo.rename.phase"] + attributes = events[0][1].attributes + assert attributes["repo_id"] == repo["id"] + assert attributes["phase"] == "forge-renamed" + assert attributes["phase_duration_ms"] >= 0 + assert attributes["retries"] == 1 + assert set(attributes) >= { + "failures", + "rollback_attempts", + "verification_outcome", + "evidence_ref", + } + serialized = events[0][1].model_dump_json().lower() + assert "preflight_token" not in serialized + assert "operator_evidence" not in serialized + assert "password" not in serialized + persisted.phase = "completed" + completion = repository_rename_events(persisted, outcome="succeeded") + assert [subject for subject, _event in completion] == [ + "org.statehub.repo.rename.phase", + "org.statehub.repo.renamed", + ] + persisted.phase = "forge-renamed" + persisted.error_code = "repository_rename_precondition_failed" + persisted.error_details = {"authorization": "Bearer must-not-publish"} + failure = repository_rename_events(persisted, outcome="failed") + failure_json = failure[0][1].model_dump_json() + assert failure[0][0] == "org.statehub.repo.rename.failed" + assert "must-not-publish" not in failure_json + + +@pytest.mark.asyncio +async def test_every_forward_phase_failure_is_retry_safe(client, rename_setup): + repo, forge = rename_setup + preflight = await _preflight(client, repo["id"]) + operation, confirmation = await _operation(client, repo["id"], preflight) + operation_id = operation["id"] + + async def fail_unreadable(requested: str, expected: str) -> None: + forge.unreadable = True + response = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/{requested}", + json={"expected_phase": expected, "confirmation": confirmation}, + ) + forge.unreadable = False + assert response.status_code == 412 + status = await client.get(f"/repository-renames/operations/{operation_id}") + assert status.json()["phase"] == expected + + await fail_unreadable("forge-renamed", "preflighted") + await _phase(client, repo["id"], operation_id, "forge-renamed", "preflighted", confirmation) + await fail_unreadable("statehub-rebound", "forge-renamed") + await _phase(client, repo["id"], operation_id, "statehub-rebound", "forge-renamed", confirmation) + + wrong_clone = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/source-synced", + json={ + "expected_phase": "statehub-rebound", + "confirmation": confirmation, + "evidence": { + "fresh_clone": True, + "forge_repository_id": 999, + "head_commit": "a" * 40, + }, + }, + ) + assert wrong_clone.status_code == 412 + await _phase( + client, + repo["id"], + operation_id, + "source-synced", + "statehub-rebound", + confirmation, + evidence={ + "fresh_clone": True, + "forge_repository_id": 417, + "head_commit": "a" * 40, + }, + ) + + bad_consumers = await client.post( + f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/consumers-verified", + json={ + "expected_phase": "source-synced", + "confirmation": confirmation, + "checks": {"security-stack": False}, + }, + ) + assert bad_consumers.status_code == 412 + await _phase( + client, + repo["id"], + operation_id, + "consumers-verified", + "source-synced", + confirmation, + checks={"security-stack": True}, + ) + + await fail_unreadable("completed", "consumers-verified") + completed = await _phase( + client, + repo["id"], + operation_id, + "completed", + "consumers-verified", + confirmation, + ) + assert completed["evidence"]["telemetry"]["failures"] == 5 + assert completed["evidence"]["telemetry"]["verification_outcome"] == "passed" + assert set(completed["evidence"]["telemetry"]["failures_by_phase"]) == { + "forge-renamed", + "statehub-rebound", + "source-synced", + "consumers-verified", + "completed", + } diff --git a/tests/test_repository_rename_cli.py b/tests/test_repository_rename_cli.py index f68332b..753a93b 100644 --- a/tests/test_repository_rename_cli.py +++ b/tests/test_repository_rename_cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import stat import sys +import urllib.error import uuid import pytest @@ -240,6 +241,27 @@ def test_rollback_requires_separate_execute_flag(monkeypatch, capsys): assert [method for method, _path, _body in calls] == ["GET", "POST"] +def test_state_hub_outage_fails_without_losing_operation_identity_or_leaking_secrets( + monkeypatch, +): + def unavailable(*_args, **_kwargs): + raise urllib.error.URLError( + "connection refused token=super-secret authorization=Bearer-secret" + ) + + monkeypatch.setattr(rename_cli.urllib.request, "urlopen", unavailable) + with pytest.raises(rename_cli.RenameCLIError) as raised: + rename_cli._api_request( + "http://127.0.0.1:8000", + "GET", + f"/repository-renames/operations/{OPERATION_ID}", + ) + assert raised.value.code == "state_hub_unavailable" + assert str(raised.value) == "State Hub is unavailable" + assert "super-secret" not in json.dumps(raised.value.details) + assert OPERATION_ID not in raised.value.details.get("reason", "") + + def test_rollback_execute_reaches_terminal_state(monkeypatch, capsys): calls = [] diff --git a/workplans/STATE-WP-0085-repository-lineage-preserving-rename.md b/workplans/STATE-WP-0085-repository-lineage-preserving-rename.md index cd0b18e..fa34cf1 100644 --- a/workplans/STATE-WP-0085-repository-lineage-preserving-rename.md +++ b/workplans/STATE-WP-0085-repository-lineage-preserving-rename.md @@ -476,7 +476,7 @@ dashboard retains the pre-existing `/docs/intakes` broken-link warning. ```task id: STATE-WP-0085-T07 -status: todo +status: done priority: high state_hub_task_id: "5160ad75-d985-5559-895c-7f8012ea76bc" ``` @@ -503,6 +503,27 @@ Acceptance: - rename events and operational metrics expose phase duration, retries, failures, rollback, and verification outcome without secrets. +Result (2026-08-29): continuity baselines now retain immutable relationship +tuples and checksums for repository identity, workplans, tasks, progress, +decisions, token events and totals, SBOM, active dispatch, protected routes, +historical messages, and file bindings. Verification compares every original +relationship while permitting append-only history, and phase-aware route checks +prove the old slug remains an alias. Equal-count workplan/token replacements +attached to another repository are rejected. Fresh-clone evidence must attest +the expected Forge numeric ID and baseline head; Forge redirects are rejected +as the wrong coordinate. The operation journal records phase durations, +attempts, retries, failures by phase, rollback attempts/outcome, and verification +outcome. Credential-free phase/failure/verification/rollback/completion events +reference the durable journal without publishing tokens, operator evidence, +URLs, or error details. Automated coverage injects failure after every forward +phase and covers stale evidence, duplicate/conflicting requests, redirects, +rollback route loss and recovery, State Hub outage, edge replay, unreadable +Forge state, and wrong-clone identity. Evidence matrix: +`docs/evidence/STATE-WP-0085-T07-repository-rename-recovery-matrix.md`. +Verification: 810 Python tests and the 70-page dashboard build pass. One known +dashboard `/docs/intakes` to `/suggestions` broken-link warning and one existing +SQLAlchemy async cancellation warning remain outside this task. + ## Document operations and repository-boundary handoffs ```task