diff --git a/.github/product-version-operations/reset-revalidation-remediation-2.7.23.json b/.github/product-version-operations/reset-revalidation-remediation-2.7.23.json new file mode 100644 index 0000000..a9cba51 --- /dev/null +++ b/.github/product-version-operations/reset-revalidation-remediation-2.7.23.json @@ -0,0 +1,16 @@ +{ + "baseline_version": "2.7.22", + "classification_rationale": "user-authorized-reset-revalidation-remediation", + "component": "product", + "event_lineage": "user-authorized-reset-revalidation-remediation", + "expected_head": "0da4ffae94f1834b957a04ea31d7ff2010fd072a", + "operation_id": "reset-revalidation-remediation-2.7.23", + "policy_revision": "forge-bootstrap-release-cadence-v2", + "product": "forge", + "projection_paths": "product-version.json", + "release_class": "PATCH", + "requested_bump": "patch", + "requested_version": null, + "schema_version": "1", + "target_version": "2.7.23" +} diff --git a/docs/architecture/FORGE_OPERATIONAL_RESET_V1.md b/docs/architecture/FORGE_OPERATIONAL_RESET_V1.md index 7a33f77..ad5eeee 100644 --- a/docs/architecture/FORGE_OPERATIONAL_RESET_V1.md +++ b/docs/architecture/FORGE_OPERATIONAL_RESET_V1.md @@ -29,6 +29,23 @@ There is no `--force`, arbitrary table list, peer SQL or automatic rollback. The stable machine envelope is `contract_version=operational-reset-v1` and contains no bearer values or raw credential verifiers. +The semantic plan contract is `forge-operational-reset-plan-1.1` under reset +policy version `2`. A general preview reports current availability separately +from the approved reset meaning. Consequently, the owning prepare transition +may add its exact maintenance operation, audit entries and verified backup +without changing the plan digest. A general preview still reports +`MAINTENANCE_ALREADY_ACTIVE` and cannot start or join an operation. + +Only `revalidate` may inspect an already prepared operation. It requires the +exact operation, plan, request and backup digests and re-proves the current +operator/authority, physical database binding, schema, writer-fence owner, +source-row contents, preserved security/configuration, effect set, generation, +product/implementation provenance and verified backup. It creates a distinct +read-only revalidation digest; it neither rewrites the approved plan nor emits +a new backup or authority. Plan 1.0 operations are not reinterpreted under +these rules and must be cancelled before apply or handled by their original +installed implementation. + ## Schema-owned data classification Schema version 38 owns this complete table mapping. Any additional application @@ -78,6 +95,7 @@ touches repositories, source, documents, other workspaces or other instances. read-only PREVIEW -> PREPARED (durable writer fence + exact authority/request binding) -> BACKUP_VERIFIED + -> read-only operation-bound REVALIDATE -> DATABASE_APPLIED (one domain transaction) -> APPLIED (every filesystem step reconciled) -> VERIFIED @@ -94,6 +112,12 @@ same-root mutation lock serializes runtime service ticks and maintenance commands. A raw/peer SQL writer is unsupported and does not become a product interface. +Successful revalidation is not a durable permission token. `apply` repeats +the operation, operator/authority, database identity, lifecycle, fence, +generation, meaningful-source and backup checks after entering its actual +`BEGIN IMMEDIATE` mutation boundary. Fence loss or drift after revalidation +therefore remains blocking. + `resume` rereads the same operation. A failure before the SQLite commit leaves the whole operational population. A failure after commit continues forward. Each external item has a durable step; if removal happened before its step write, diff --git a/docs/operations/FORGE_INSTALLED_UPDATE_RUNBOOK.md b/docs/operations/FORGE_INSTALLED_UPDATE_RUNBOOK.md index 44878e0..31a41ed 100644 --- a/docs/operations/FORGE_INSTALLED_UPDATE_RUNBOOK.md +++ b/docs/operations/FORGE_INSTALLED_UPDATE_RUNBOOK.md @@ -1,7 +1,8 @@ # Forge installed update controller -Status: bounded product-owned maintenance provisioner for the selected -Forge 2.7.21 to 2.7.22 transition. +Status: bounded product-owned maintenance provisioner for the selected Forge +2.7.21 to 2.7.22 schema-37-to-38 transition and the selected 2.7.22 to 2.7.23 +same-schema corrective transition. This controller closes one concrete product provisioning gap. It is not the universal Forge Platform installer, an installer UI, a new service supervisor, @@ -17,8 +18,9 @@ cannot be called by the installed runtime as a second self-installer. ## Supported operation -The controller accepts only one explicitly bound existing installation and one -exact `forge-autonomy` 2.7.22 wheel. Every invocation binds: +The controller accepts only one explicitly bound existing installation and an +exact `forge-autonomy` wheel for one of those two transitions. Every invocation +binds: - operation, runtime, installation, and peer-configuration identities; - data root, runtime root, current command resolver, legacy interpreter, and @@ -37,7 +39,10 @@ interpreter, version, and bytes. 1. Read the wheel once through a no-follow descriptor; validate its digest, canonical RECORD, purelib tag, package metadata, member allowlist, and the - exact terminal release/reconciliation receipt without importing it. + exact terminal release receipt without importing it. The historical + 2.7.22 reconciliation receipt retains its dedicated validation; 2.7.23 must + have the normal protected release-complete publication/readback/cleanup + shape and cannot be presented to the older transition. 2. Create an isolated versioned runtime slot outside the source checkout with the explicit Python interpreter. Extract only the already validated bytes into a pip-free virtual environment, then verify every installed file and @@ -56,10 +61,11 @@ interpreter, version, and bytes. 6. Take a SQLite backup through the backup API while the writer fence is held. Verify `integrity_check`, foreign keys, digest, and the complete logical pre-migration snapshot. -7. Migrate an isolated copy of that backup through the staged candidate's - normal `forge ... server init` path. Require schema 38, the exact new reset - table set, an idle reset control row, and byte-logical preservation of every - pre-existing domain table and protected metadata/binding. +7. Re-open an isolated copy of that backup through the staged candidate's + normal `forge ... server init` path. The 37-to-38 route requires exactly the + reset-table additions and a fresh idle control row. The 38-to-38 route + permits no table additions and requires the complete reset state and every + domain/security/configuration row to remain byte-logically unchanged. 8. Point the stable resolver at a maintenance fence. Take an exclusive SQLite writer boundary, prove the live database is still byte-logically identical to the backed-up snapshot, and atomically install the already Forge-migrated @@ -83,11 +89,11 @@ authorized read-only Forge-to-EP check. Re-run the same exact operation ID and arguments. A conflicting request is rejected. -- Before atomic database replacement, failure restores the retained 2.7.21 +- Before atomic database replacement, failure restores the retained prior command route only when the complete live snapshot still equals `before`. A hard interruption may leave the explicit maintenance fence; resuming the same operation reconciles it from durable evidence. -- From the first schema-38 readback or any ambiguous partial state onward, the +- From the first target-schema readback or any ambiguous partial state onward, the old binary is never selected. Any caught activation/readback failure selects the maintenance fence; replay reconciles the exact protected candidate. - A completed receipt is idempotently returned only after revalidating its @@ -103,7 +109,7 @@ separate authority and compatibility proof. ## Qualification boundary `tests/test_installed_forge_update.py` covers exact release binding, target and -writer rejection, the real Forge schema-37 to schema-38 migrator, preservation +writer rejection, both bounded schema transitions, preservation of Missions, allocations, reviews, execution receipts, governance grants, configuration and identity, concurrent-operation exclusion, resolver adoption, canonical receipt shape, path safety, exact slot contents, exclusive atomic diff --git a/docs/operations/FORGE_OPERATIONAL_RESET_RUNBOOK.md b/docs/operations/FORGE_OPERATIONAL_RESET_RUNBOOK.md index bcfa65e..2e97cae 100644 --- a/docs/operations/FORGE_OPERATIONAL_RESET_RUNBOOK.md +++ b/docs/operations/FORGE_OPERATIONAL_RESET_RUNBOOK.md @@ -44,7 +44,25 @@ consistent backup and returns `request_digest` plus verified `backup.digest`. If backup creation fails, leave the instance in maintenance and use `status`; do not start another operation. -## 3. Apply the reset — DESTRUCTIVE +## 3. Revalidate this prepared operation — READ-ONLY + +Do not repeat the general preview as the same-operation gate. It correctly +blocks any new reset while maintenance is active. Instead, bind the exact +prepared receipt: + +```text +forge --data-root "/absolute/Forge Server" server reset revalidate \ + --operation-id "forge-reset-" \ + --plan-digest "sha256:" \ + --request-digest "sha256:" \ + --backup-digest "sha256:" +``` + +Require `allowed=true`, no blockers, the unchanged plan/relevant-source and +preserved-bindings digests, the exact fence owner and a revalidation digest. +This command must not change owning status, data, authority or backup bytes. + +## 4. Apply the reset — DESTRUCTIVE ```text forge --data-root "/absolute/Forge Server" server reset apply \ @@ -54,7 +72,7 @@ forge --data-root "/absolute/Forge Server" server reset apply \ --backup-digest "sha256:" ``` -## 4. Verify — MAINTENANCE REMAINS ACTIVE +## 5. Verify — MAINTENANCE REMAINS ACTIVE ```text forge --data-root "/absolute/Forge Server" server reset verify \ @@ -67,7 +85,7 @@ forge --data-root "/absolute/Forge Server" server reset verify \ Require `state=VERIFIED`, all operational counts zero, all integrity checks green, unchanged target/bindings and a non-empty `verification_digest`. -## 5. Interrupted operation +## 6. Interrupted operation ```text forge --data-root "/absolute/Forge Server" server reset status \ @@ -85,7 +103,7 @@ before a backup digest exists, omit `--backup-digest`; the same PREPARED operati finishes its backup first. Do not automatically restore or create a replacement operation. -## 6. Release maintenance — MUTATING +## 7. Release maintenance — MUTATING ```text forge --data-root "/absolute/Forge Server" server reset finish \ @@ -113,7 +131,8 @@ HTTP authentication/preflight; it must not create a Mission or submission. if the normal allocator cannot produce the desired display label. 3. Prepare both owning operations; confirm both durable maintenance states and both verified backups. -4. Recheck both exact plans under their writer fences. +4. Invoke both owning operation-bound `revalidate` commands under their writer + fences. Do not substitute general preview. 5. Apply each owning reset sequentially. If either fails, keep both products in maintenance and resume the same owning operation; never auto-resume the first. 6. Verify both empty operational generations, preserved identities/peer binding, diff --git a/docs/reports/FORGE_OPERATIONAL_RESET_DELIVERY_STATUS.md b/docs/reports/FORGE_OPERATIONAL_RESET_DELIVERY_STATUS.md index 57f8395..bb9845e 100644 --- a/docs/reports/FORGE_OPERATIONAL_RESET_DELIVERY_STATUS.md +++ b/docs/reports/FORGE_OPERATIONAL_RESET_DELIVERY_STATUS.md @@ -1,7 +1,9 @@ # Forge operational reset delivery status The bounded Forge-owned operational-history reset is implemented in source with -schema 38 and the `forge server reset` command family. It is independent from +schema 38 and the `forge server reset` command family. Patch 2.7.23 separates +general availability preview from operation-bound revalidation and adds the +explicit installed `revalidate` command. It is independent from the planned Operations Console, full export/import/relocation product and Forge Server daemon. Those broader nodes remain planned. @@ -11,8 +13,8 @@ Status dimensions are deliberately separate: | --- | --- | | Architecture/contract | IMPLEMENTED in `FORGE_OPERATIONAL_RESET_V1.md` | | Application service and CLI | IMPLEMENTED | -| Isolated synthetic positive/negative/crash qualification | PASSED locally: 23 reset regressions within `scripts/validate.sh` | -| Full repository source gate | PASSED locally: 694 tests, product-version and offline-projection validation | +| Same-operation revalidation regressions | PASSED locally: stable semantic plan, real row-content drift, backup, authority and fence checks | +| Joint subprocess candidate sequence | PASSED locally with non-editable candidate wheels and the real coordinator; protected installed-release repetition remains required | | Independent protected-candidate review/checks | NOT YET EVIDENCED | | Protected review/merge | NOT YET EVIDENCED in this source branch | | Stable release publication | NOT YET EVIDENCED | @@ -21,9 +23,13 @@ Status dimensions are deliberately separate: | Production reset | NOT AUTHORIZED / NOT PERFORMED | | Mission 3 | NOT STARTED | -The intended stable patch release is `2.7.22` under +The corrective stable patch release is `2.7.23` under `forge-bootstrap-release-cadence-v2`: protected exact-head review/merge, one clean wheel/sdist build, full qualification bound to their SHA-256 values, draft GitHub release, PyPI publication, registry digest readback, installed-wheel -smoke, durable publication/cleanup receipt and completion. No branch-local test -or wheel build is presented as published or installed evidence. +smoke, durable publication/cleanup receipt and completion. The earlier +qualification missed the defect because the coordinator revalidation path was +covered by a fixture state machine and product reset tests invoked the owning +services directly; no installed two-product sequence performed general preview +after both prepares. No branch-local test or wheel build is presented as +published or installed evidence. diff --git a/forge/__main__.py b/forge/__main__.py index 8481723..23f90d0 100644 --- a/forge/__main__.py +++ b/forge/__main__.py @@ -101,6 +101,7 @@ def main(argv: list[str] | None = None) -> int: prepare_reset.add_argument("--plan-digest", required=True) prepare_reset.add_argument("--acknowledge-operational-fk", action="append", default=[]) for name, help_text in ( + ("revalidate", "revalidate the exact prepared operation without changing it"), ("apply", "apply the authorized destructive reset"), ("verify", "verify reset integrity and preserved bindings"), ): @@ -179,7 +180,7 @@ def main(argv: list[str] | None = None) -> int: expected_plan_digest=args.plan_digest, acknowledge_operational_fk=args.acknowledge_operational_fk, ) - elif args.reset_command in {"apply", "verify"}: + elif args.reset_command in {"revalidate", "apply", "verify"}: result = getattr(service, args.reset_command)( operation_id=args.operation_id, plan_digest=args.plan_digest, request_digest=args.request_digest, backup_digest=args.backup_digest, diff --git a/forge/runtime/operational_reset.py b/forge/runtime/operational_reset.py index 662c2e0..80a13d6 100644 --- a/forge/runtime/operational_reset.py +++ b/forge/runtime/operational_reset.py @@ -31,7 +31,8 @@ RESET_PROFILE = "forge-operational-history-v1" -RESET_POLICY_VERSION = "1" +RESET_POLICY_VERSION = "2" +RESET_PLAN_CONTRACT = "forge-operational-reset-plan-1.1" _OPERATION_ID = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9._-]{7,127}\Z") _DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z") @@ -511,10 +512,18 @@ def _inventory(self, connection: sqlite3.Connection) -> dict[str, Any]: ).fetchone() if state is not None: generation, active_operation = int(state[0]), state[1] + database_stat = self.database_path.stat() + database_binding = { + "path": str(self.database_path.resolve()), + "device": int(database_stat.st_dev), + "inode": int(database_stat.st_ino), + } target = { "product": "forge", "runtime_id": runtime_id, "data_root": str(self.data_root), "database": str(self.database_path), - "database_binding_digest": _digest(str(self.database_path.resolve())), + "database_binding_digest": _digest(database_binding), + "database_device": database_binding["device"], + "database_inode": database_binding["inode"], "marker_digest": _file_digest(self.marker_path), "schema_version": schema, "dataset_generation": generation, } @@ -580,8 +589,12 @@ def _inventory(self, connection: sqlite3.Connection) -> dict[str, Any]: and active_permits == 0 ) source_revision = self._source_revision() + semantic_classifications = { + category: entries for category, entries in classifications.items() + if category != "MAINTENANCE_AUDIT" + } plan_core = { - "contract": "forge-operational-reset-plan-1.0", "profile": RESET_PROFILE, + "contract": RESET_PLAN_CONTRACT, "profile": RESET_PROFILE, "policy_version": RESET_POLICY_VERSION, "product_version": canonical_version(), "source_revision": source_revision, "source_revision_kind": ( @@ -590,7 +603,8 @@ def _inventory(self, connection: sqlite3.Connection) -> dict[str, Any]: ), "implementation_digest": self._implementation_digest(), "target": target, - "relevant_revision": relevant_revision, "classifications": classifications, + "relevant_revision": relevant_revision, + "classifications": semantic_classifications, "preserved": preserved, "effect_set": effect_set, "effect_set_digest": _digest(effect_set), "foreign_key_issues": foreign_keys, "required_fk_acknowledgements": [item["issue_id"] for item in operational_fk], @@ -601,7 +615,7 @@ def _inventory(self, connection: sqlite3.Connection) -> dict[str, Any]: "preserved_external_digest": _digest(preserved_external), "unknown_external": unknown_external, "unknown_tables": unknown_tables, "missing_tables": missing_tables, - "blockers": blockers, "no_op": no_op, + "no_op": no_op, "namespace": { "repository_document": "missions/MISSION-0003.md", "repository_document_kind": "HISTORICAL_REPOSITORY_DOCUMENT", @@ -610,7 +624,17 @@ def _inventory(self, connection: sqlite3.Connection) -> dict[str, Any]: }, } return { - **plan_core, "plan_digest": _digest(plan_core), + **plan_core, + # Availability findings remain visible and fail closed, but are + # not part of the approved semantic effect plan. In particular, + # the owning prepare transition necessarily makes maintenance + # active without redefining the reset that was authorized. + "blockers": blockers, + "plan_digest": _digest(plan_core), + "maintenance_observation": { + "active_operation_id": active_operation, + "tables": classifications["MAINTENANCE_AUDIT"], + }, "external_control_entries": external_controls, } @@ -999,8 +1023,79 @@ def _bound_operation( raise OperationalResetError("source revision changed after authorization") if row["implementation_digest"] != self._implementation_digest(): raise OperationalResetError("maintenance service revision changed after authorization") + actor_reference, authority_digest = self._operator_authority(connection) + if ( + row["actor_reference"] != actor_reference + or row["authority_digest"] != authority_digest + ): + raise OperationalResetError("operator or reset authority changed after authorization") return row + def _pre_apply_revalidation( + self, connection: sqlite3.Connection, operation: sqlite3.Row, + ) -> dict[str, Any]: + """Re-prove one prepared operation without redefining its approved plan.""" + if operation["state"] != "BACKUP_VERIFIED": + raise OperationalResetError("operation is not in the prepared revalidation phase") + operation_id = str(operation["operation_id"]) + document = json.loads(operation["document"]) + plan = document["plan"] + current = self._inventory(connection) + expected_maintenance = [{ + "code": "MAINTENANCE_ALREADY_ACTIVE", "operation_id": operation_id, + }] + if current["blockers"] != expected_maintenance: + raise OperationalResetError("prepared operation has new or missing blocking findings") + if current["target"] != plan["target"]: + raise OperationalResetError("prepared operation target or database binding changed") + if current["relevant_revision"] != operation["relevant_revision"]: + raise OperationalResetError("meaningful source data changed after the approved preview") + for key in ( + "effect_set_digest", "external_inventory_digest", "preserved_external_digest", + "required_fk_acknowledgements", "foreign_key_issues", "policy_version", + "source_revision", "implementation_digest", + ): + if current[key] != plan[key]: + raise OperationalResetError(f"prepared operation binding changed: {key}") + if int(current["target"]["dataset_generation"]) != int(operation["generation_before"]): + raise OperationalResetError("dataset generation changed after authorization") + evidence = { + "operation_id": operation_id, + "state": str(operation["state"]), + "writer_fence_owner": operation_id, + "actor_reference": str(operation["actor_reference"]), + "authority_digest": str(operation["authority_digest"]), + "target_binding_digest": str(plan["target"]["database_binding_digest"]), + "plan_digest": str(operation["plan_digest"]), + "request_digest": str(operation["request_digest"]), + "effect_set_digest": str(plan["effect_set_digest"]), + "relevant_revision_digest": str(operation["relevant_revision"]), + "backup_digest": str(operation["backup_digest"]), + "dataset_generation": int(operation["generation_before"]), + "source_revision": str(operation["source_revision"]), + "implementation_digest": str(operation["implementation_digest"]), + } + evidence["revalidation_digest"] = _digest(evidence) + return evidence + + def revalidate( + self, *, operation_id: str, plan_digest: str, request_digest: str, + backup_digest: str, + ) -> dict[str, Any]: + """Read-only proof for the already prepared, exact owning operation.""" + with self._lock.acquire(): + with self._connect(read_only=True) as connection: + row = self._bound_operation( + connection, operation_id, plan_digest=plan_digest, + request_digest=request_digest, backup_digest=backup_digest, + ) + evidence = self._pre_apply_revalidation(connection, row) + self._verify_operation_backup(row) + receipt = self._operation_receipt(operation_id) + receipt["revalidation"] = evidence + receipt["execution_allowed"] = True + return receipt + @staticmethod def _tombstone_values(connection: sqlite3.Connection) -> list[tuple[str, str]]: queries: tuple[tuple[str, str], ...] = ( @@ -1036,6 +1131,7 @@ def _apply_database(self, row: sqlite3.Row) -> None: # and every DELETE. A stale BACKUP_VERIFIED label can never # authorize a purge after recovery bytes were changed. self._verify_operation_backup(row) + self._pre_apply_revalidation(connection, row) current = self._relevant_revision(connection) if current != row["relevant_revision"]: raise OperationalResetError("meaningful source data changed after the approved preview") @@ -1487,6 +1583,11 @@ def operator_envelope(self, command: str, payload: Mapping[str, Any]) -> dict[st state = str(payload.get("state") or payload.get("status") or "UNKNOWN") backup_digest = payload.get("backup_digest") backup_reference = payload.get("backup_reference") + observed_generation = payload.get("generation_after") + if observed_generation is None: + observed_generation = payload.get("dataset_generation") + if observed_generation is None: + observed_generation = target.get("dataset_generation") return { "contract_version": "operational-reset-v1", "product": "forge", @@ -1503,9 +1604,7 @@ def operator_envelope(self, command: str, payload: Mapping[str, Any]) -> dict[st "schema_version": target.get("schema_version", payload.get("schema_version")), }, "profile": payload.get("profile", RESET_PROFILE), - "dataset_generation": target.get( - "dataset_generation", payload.get("generation_after", payload.get("dataset_generation")) - ), + "dataset_generation": observed_generation, "plan_digest": payload.get("plan_digest"), "relevant_revision_digest": ( None if plan is None else plan.get("relevant_revision") diff --git a/product-version.json b/product-version.json index c200171..0efeb93 100644 --- a/product-version.json +++ b/product-version.json @@ -1,5 +1,5 @@ { "product": "forge", "schema_version": 1, - "version": "2.7.22" + "version": "2.7.23" } diff --git a/scripts/update_installed_forge.py b/scripts/update_installed_forge.py index 03398c1..7aabb95 100644 --- a/scripts/update_installed_forge.py +++ b/scripts/update_installed_forge.py @@ -45,8 +45,10 @@ CONTRACT_VERSION = "forge-installed-update/v1" -SCHEMA_BEFORE = 37 -SCHEMA_AFTER = 38 +SUPPORTED_TRANSITIONS = { + ("2.7.21", "2.7.22"): (37, 38), + ("2.7.22", "2.7.23"): (38, 38), +} PHASE_ORDER = { phase: index for index, phase in enumerate(( "PREPARED", "STAGED", "ADOPTED", "BACKED_UP", "MIGRATION_QUALIFIED", @@ -75,6 +77,15 @@ class InstalledForgeUpdateError(RuntimeError): """The selected installation cannot be updated without weakening a gate.""" +def transition_schemas(request: "UpdateRequest") -> tuple[int, int]: + try: + return SUPPORTED_TRANSITIONS[(request.existing_version, request.version)] + except KeyError as error: + raise InstalledForgeUpdateError( + "this bounded controller does not support the selected Forge version transition" + ) from error + + def _now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") @@ -281,8 +292,7 @@ def validate(self) -> None: for value in identifiers ): raise InstalledForgeUpdateError("operation and installation identities must be filesystem-safe") - if self.version != "2.7.22" or self.existing_version != "2.7.21": - raise InstalledForgeUpdateError("this bounded controller supports only the selected 2.7.21 to 2.7.22 update") + transition_schemas(self) for label, digest in ( ("wheel", self.wheel_sha256), ("qualification receipt", self.qualification_receipt_sha256), ("controller", self.controller_sha256), ("resolver", self.resolver_sha256), @@ -366,6 +376,75 @@ def _validated_wheel(request: UpdateRequest) -> tuple[bytes, dict[str, str]]: return wheel_bytes, manifest +def _normal_release_evidence( + request: UpdateRequest, receipt: Mapping[str, Any], manifest: Mapping[str, str], + receipt_path: Path, +) -> dict[str, Any]: + """Validate the normal 2.7.23 release route without weakening 2.7.22 recovery.""" + expected_name = f"forge_autonomy-{request.version}-py3-none-any.whl" + sdist_name = f"forge_autonomy-{request.version}.tar.gz" + qualification = receipt.get("qualification") + artifacts = receipt.get("artifacts") + publication = receipt.get("publication_receipt") + cleanup = receipt.get("cleanup") + expected_top = { + "product", "component", "version", "source_revision", "operation_id", "policy_revision", + "artifacts", "state", "qualification", "publication_receipt", "cleanup", + } + sdist_digest = artifacts.get("sdist") if isinstance(artifacts, Mapping) else None + exact_artifacts = {"wheel": request.wheel_sha256, "sdist": sdist_digest} + exact_qualified = { + f"dist/{expected_name}": request.wheel_sha256, + f"dist/{sdist_name}": sdist_digest, + } + exact_observed = {expected_name: request.wheel_sha256, sdist_name: sdist_digest} + if ( + (request.existing_version, request.version) != ("2.7.22", "2.7.23") + or set(receipt) != expected_top + or receipt.get("state") != "RELEASE_COMPLETE" + or receipt.get("product") != "forge" + or receipt.get("component") != "forge-autonomy" + or receipt.get("version") != request.version + or receipt.get("source_revision") != request.product_source + or receipt.get("operation_id") != f"forge-release-{request.version}-{request.product_source}" + or receipt.get("policy_revision") != "forge-bootstrap-release-cadence-v2" + or not isinstance(artifacts, Mapping) + or set(artifacts) != {"wheel", "sdist"} + or dict(artifacts) != exact_artifacts + or not isinstance(sdist_digest, str) + or re.fullmatch(r"sha256:[0-9a-f]{64}", sdist_digest) is None + or not isinstance(qualification, Mapping) + or set(qualification) != {"artifact_digests", "exact_main_sha", "qualification"} + or qualification.get("exact_main_sha") != request.product_source + or qualification.get("qualification") != "forge-production-distribution" + or qualification.get("artifact_digests") != exact_qualified + or not isinstance(publication, Mapping) + or set(publication) != {"observed_artifact_digests", "readback", "registry"} + or publication.get("observed_artifact_digests") != exact_observed + or publication.get("readback") != "PASS" + or publication.get("registry") != "pypi" + or not isinstance(cleanup, Mapping) + or set(cleanup) != {"github_release", "result", "temporary_paths"} + or cleanup.get("result") != "COMPLETE" + or cleanup.get("temporary_paths") != [ + "published-readback", "published-input/dist", "pending-readback", + ] + or cleanup.get("github_release") != {"draft": False} + ): + raise InstalledForgeUpdateError( + "normal release-complete publication, policy, or cleanup lineage is noncanonical" + ) + return { + "wheel": str(Path(request.wheel)), + "wheel_sha256": request.wheel_sha256, + "wheel_manifest_digest": _digest_bytes(_json_bytes(dict(manifest))), + "receipt": str(receipt_path), + "receipt_sha256": request.qualification_receipt_sha256, + "release_operation_id": receipt.get("operation_id"), + "release_route": "NORMAL", + } + + def _qualified_artifact(request: UpdateRequest) -> tuple[dict[str, Any], bytes, dict[str, str]]: wheel_bytes, manifest = _validated_wheel(request) expected_name = f"forge_autonomy-{request.version}-py3-none-any.whl" @@ -380,6 +459,12 @@ def _qualified_artifact(request: UpdateRequest) -> tuple[dict[str, Any], bytes, raise InstalledForgeUpdateError("qualification receipt is malformed") from error if not isinstance(receipt, dict): raise InstalledForgeUpdateError("qualification receipt is malformed") + if (request.existing_version, request.version) == ("2.7.22", "2.7.23"): + return ( + _normal_release_evidence(request, receipt, manifest, receipt_path), + wheel_bytes, + manifest, + ) qualification = receipt.get("qualification") artifacts = receipt.get("artifacts") publication = receipt.get("publication_receipt") @@ -575,7 +660,8 @@ def assert_selected_installation(request: UpdateRequest, snapshot: Mapping[str, raise InstalledForgeUpdateError("selected data root belongs to a different runtime") if metadata.get("installation_id") != request.installation_id: raise InstalledForgeUpdateError("selected data root belongs to a different installation") - if snapshot.get("user_version") not in {SCHEMA_BEFORE, SCHEMA_AFTER}: + schema_before, schema_after = transition_schemas(request) + if snapshot.get("user_version") not in {schema_before, schema_after}: raise InstalledForgeUpdateError("selected runtime schema is outside the bounded update path") if not isinstance(peer, Mapping) or peer.get("configuration_digest") != request.peer_configuration_digest: raise InstalledForgeUpdateError("selected peer configuration changed") @@ -619,11 +705,14 @@ def assert_quiescent(snapshot: Mapping[str, Any]) -> None: raise InstalledForgeUpdateError("Forge operational reset maintenance is active") -def assert_completed_schema(snapshot: Mapping[str, Any], installed_readback: Mapping[str, Any]) -> None: +def assert_completed_schema( + snapshot: Mapping[str, Any], installed_readback: Mapping[str, Any], request: UpdateRequest, +) -> None: tables = snapshot.get("tables") expected_digest = installed_readback.get("database_schema_digest") + _schema_before, schema_after = transition_schemas(request) if ( - snapshot.get("user_version") != SCHEMA_AFTER + snapshot.get("user_version") != schema_after or not isinstance(tables, Mapping) or not NEW_SCHEMA_38_TABLES.issubset(tables) or not isinstance(expected_digest, str) @@ -633,10 +722,13 @@ def assert_completed_schema(snapshot: Mapping[str, Any], installed_readback: Map def verify_preservation(before: Mapping[str, Any], after: Mapping[str, Any], request: UpdateRequest) -> dict[str, Any]: + schema_before, schema_after = transition_schemas(request) + if before.get("user_version") != schema_before: + raise InstalledForgeUpdateError("source runtime schema is outside the selected update path") if after.get("integrity_check") != "ok" or after.get("foreign_key_check") != []: raise InstalledForgeUpdateError("migrated runtime failed SQLite integrity validation") - if after.get("user_version") != SCHEMA_AFTER: - raise InstalledForgeUpdateError("Forge owning migration did not reach schema 38") + if after.get("user_version") != schema_after: + raise InstalledForgeUpdateError("Forge owning migration did not reach the selected target schema") before_metadata, after_metadata = before.get("metadata"), after.get("metadata") if not isinstance(before_metadata, Mapping) or not isinstance(after_metadata, Mapping): raise InstalledForgeUpdateError("migration metadata readback is incomplete") @@ -658,17 +750,21 @@ def verify_preservation(before: Mapping[str, Any], after: Mapping[str, Any], req if after_tables.get(table) != metric: raise InstalledForgeUpdateError(f"migration changed historical table contents: {table}") new_tables = set(after_tables) - set(before_tables) - if new_tables != set(NEW_SCHEMA_38_TABLES): - raise InstalledForgeUpdateError("migration produced an unexpected schema-38 table set") - expected_counts = {table: 0 for table in NEW_SCHEMA_38_TABLES} - expected_counts["operational_reset_state"] = 1 - if any(after_tables[table]["count"] != count for table, count in expected_counts.items()): - raise InstalledForgeUpdateError("migration initialized unexpected operational-reset data") + expected_new_tables = set(NEW_SCHEMA_38_TABLES) if schema_before == 37 else set() + if new_tables != expected_new_tables: + raise InstalledForgeUpdateError("migration produced an unexpected target-schema table set") reset = after.get("writer_state", {}).get("operational_reset", []) - if reset != [{"dataset_generation": 0, "active_operation_id": None, "state": "IDLE"}]: - raise InstalledForgeUpdateError("schema-38 reset state is not an idle, fresh control record") + if schema_before == 37: + expected_counts = {table: 0 for table in NEW_SCHEMA_38_TABLES} + expected_counts["operational_reset_state"] = 1 + if any(after_tables[table]["count"] != count for table, count in expected_counts.items()): + raise InstalledForgeUpdateError("migration initialized unexpected operational-reset data") + if reset != [{"dataset_generation": 0, "active_operation_id": None, "state": "IDLE"}]: + raise InstalledForgeUpdateError("schema-38 reset state is not an idle, fresh control record") + elif reset != before.get("writer_state", {}).get("operational_reset", []): + raise InstalledForgeUpdateError("same-schema update changed operational-reset state") return { - "status": "PASS", "from_schema": SCHEMA_BEFORE, "to_schema": SCHEMA_AFTER, + "status": "PASS", "from_schema": schema_before, "to_schema": schema_after, "preserved_table_count": len(before_tables) - 1, "added_tables": sorted(new_tables), "protected_metadata_digest": after.get("protected_metadata_digest"), @@ -735,14 +831,16 @@ def _copy_sqlite_backup(source: Path, destination: Path) -> dict[str, Any]: } -def _candidate_migrate(executable: Path, data_root: Path, *, cwd: Path) -> dict[str, Any]: +def _candidate_migrate( + executable: Path, data_root: Path, *, cwd: Path, target_schema: int, +) -> dict[str, Any]: result = _run((str(executable), "--data-root", str(data_root), "server", "init"), cwd=cwd) try: output = json.loads(result.stdout) except json.JSONDecodeError as error: raise InstalledForgeUpdateError("candidate migration readback is malformed") from error - if output.get("storage_schema") != str(SCHEMA_AFTER) or output.get("initialized") is not True: - raise InstalledForgeUpdateError("candidate migration did not return schema-38 installed readback") + if output.get("storage_schema") != str(target_schema) or output.get("initialized") is not True: + raise InstalledForgeUpdateError("candidate migration did not return the target-schema installed readback") return output @@ -1049,6 +1147,7 @@ def _adopt_resolver(self, state: dict[str, Any]) -> dict[str, Any]: }) def _backup(self, state: dict[str, Any], before: Mapping[str, Any]) -> dict[str, Any]: + schema_before, _schema_after = transition_schemas(self.request) existing = state.get("backup") if isinstance(existing, Mapping): if existing.get("path") != str(self.backup_path) or file_digest(self.backup_path) != existing.get("sha256"): @@ -1060,7 +1159,7 @@ def _backup(self, state: dict[str, Any], before: Mapping[str, Any]) -> dict[str, elif self.backup_path.exists() and not self.backup_path.is_symlink(): recovered = database_snapshot(self.backup_path) if ( - recovered.get("user_version") != SCHEMA_BEFORE + recovered.get("user_version") != schema_before or recovered.get("integrity_check") != "ok" or recovered.get("foreign_key_check") != [] or recovered.get("content_digest") != before.get("content_digest") @@ -1123,7 +1222,11 @@ def _qualify_copy(self, state: dict[str, Any], before: Mapping[str, Any]) -> dic copy_before = database_snapshot(database) if copy_before.get("content_digest") != before.get("content_digest"): raise InstalledForgeUpdateError("isolated qualification copy does not match the consistent backup") - candidate_output = _candidate_migrate(self.slot / "bin" / "forge", root, cwd=self.runtime_root) + _schema_before, schema_after = transition_schemas(self.request) + candidate_output = _candidate_migrate( + self.slot / "bin" / "forge", root, cwd=self.runtime_root, + target_schema=schema_after, + ) copy_after = database_snapshot(database) qualification = verify_preservation(copy_before, copy_after, self.request) qualification.update({ @@ -1148,6 +1251,7 @@ def _restore_legacy_before_migration(self, state: dict[str, Any], error: Excepti ) def _install_qualified_database(self, before: Mapping[str, Any]) -> dict[str, Any]: + schema_before, _schema_after = transition_schemas(self.request) source = self.operation_root / "qualification-copy" / "forge.db" qualified = database_snapshot(source) verify_preservation(before, qualified, self.request) @@ -1167,7 +1271,7 @@ def _install_qualified_database(self, before: Mapping[str, Any]) -> dict[str, An live_connection.execute("BEGIN EXCLUSIVE") locked_live = database_snapshot(self.database, existing_connection=live_connection) if ( - locked_live.get("user_version") != SCHEMA_BEFORE + locked_live.get("user_version") != schema_before or locked_live.get("content_digest") != before.get("content_digest") ): raise InstalledForgeUpdateError("live runtime changed after the qualified backup") @@ -1215,8 +1319,15 @@ def _install_qualified_database(self, before: Mapping[str, Any]) -> dict[str, An return installed def _migrate_live(self, state: dict[str, Any], before: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + schema_before, schema_after = transition_schemas(self.request) current = database_snapshot(self.database) - if current.get("user_version") == SCHEMA_BEFORE: + if ( + current.get("user_version") == schema_before + and ( + schema_before != schema_after + or current.get("content_digest") == before.get("content_digest") + ) + ): state = self._fence(state) self._interrupt("fence") after = self._install_qualified_database(before) @@ -1227,11 +1338,11 @@ def _migrate_live(self, state: dict[str, Any], before: Mapping[str, Any]) -> tup "application_mode": "ATOMIC_PRODUCT_MIGRATED_COPY", "before_snapshot_digest": before["snapshot_digest"], "after_snapshot_digest": after["snapshot_digest"], - }, safety_disposition="CANDIDATE_REQUIRED_SCHEMA_38", + }, safety_disposition=f"CANDIDATE_REQUIRED_SCHEMA_{schema_after}", ) self._interrupt("migration") return state, after - if current.get("user_version") == SCHEMA_AFTER: + if current.get("user_version") == schema_after: preservation = verify_preservation(before, current, self.request) os.chmod(self.database, 0o400) if state.get("phase") not in {"MIGRATED", "ACTIVATING", "ACTIVATED", "COMPLETE"}: @@ -1240,12 +1351,13 @@ def _migrate_live(self, state: dict[str, Any], before: Mapping[str, Any]) -> tup **preservation, "reconciled_at": _now(), "before_snapshot_digest": before["snapshot_digest"], "after_snapshot_digest": current["snapshot_digest"], - }, safety_disposition="CANDIDATE_REQUIRED_SCHEMA_38", + }, safety_disposition=f"CANDIDATE_REQUIRED_SCHEMA_{schema_after}", ) return state, current raise InstalledForgeUpdateError("live runtime schema changed outside the bounded operation") def _activate(self, state: dict[str, Any], after: Mapping[str, Any]) -> dict[str, Any]: + _schema_before, schema_after = transition_schemas(self.request) state = self._advance(state, "ACTIVATING", safety_disposition="CANDIDATE_ACTIVATION_IN_PROGRESS") candidate = self.slot / "bin" / "forge" _replace_symlink(self.current, os.path.relpath(candidate, self.runtime_root)) @@ -1270,7 +1382,7 @@ def _activate(self, state: dict[str, Any], after: Mapping[str, Any]) -> dict[str or status.get("product_version") != self.request.version or Path(str(status.get("data_root", ""))).resolve() != self.data_root.resolve() or status.get("instance_id") != self.request.runtime_id - or status.get("storage_schema") != str(SCHEMA_AFTER) + or status.get("storage_schema") != str(schema_after) ): raise InstalledForgeUpdateError("activated Forge CLI readback does not match the selected installation") final_snapshot = database_snapshot(self.database) @@ -1296,8 +1408,9 @@ def _secure_failure(self, state: dict[str, Any], error: Exception) -> None: except Exception: current = {} before = state.get("before") + schema_before, _schema_after = transition_schemas(self.request) if ( - current.get("user_version") == SCHEMA_BEFORE + current.get("user_version") == schema_before and isinstance(before, Mapping) and current.get("content_digest") == before.get("content_digest") and self.legacy_entrypoint.exists() @@ -1311,6 +1424,7 @@ def _secure_failure(self, state: dict[str, Any], error: Exception) -> None: ) def _verify_complete(self, state: Mapping[str, Any], receipt: Mapping[str, Any]) -> None: + _schema_before, schema_after = transition_schemas(self.request) if state.get("phase") != "COMPLETE" or state.get("request_digest") != self.request.digest: raise InstalledForgeUpdateError("completed operation state conflicts with this request") if state.get("receipt_sha256") != file_digest(self.receipt_path): @@ -1355,7 +1469,7 @@ def _verify_complete(self, state: Mapping[str, Any], receipt: Mapping[str, Any]) or version != self.request.version or status.get("product_version") != self.request.version or status.get("instance_id") != self.request.runtime_id - or status.get("storage_schema") != str(SCHEMA_AFTER) + or status.get("storage_schema") != str(schema_after) or Path(str(status.get("data_root", ""))).resolve() != self.data_root.resolve() ): raise InstalledForgeUpdateError("completed installed CLI identity changed") @@ -1380,7 +1494,7 @@ def _verify_complete(self, state: Mapping[str, Any], receipt: Mapping[str, Any]) or installed_readback["preservation"].get("status") != "PASS" ): raise InstalledForgeUpdateError("completed operation lacks successful installation readback") - assert_completed_schema(final_snapshot, installed_readback) + assert_completed_schema(final_snapshot, installed_readback, self.request) def _restore_database_writable(self) -> None: _assert_no_symlink_components(self.database) @@ -1412,16 +1526,21 @@ def run(self) -> dict[str, Any]: assert_selected_installation(self.request, live) assert_quiescent(live) try: + schema_before, _schema_after = transition_schemas(self.request) state = self._adopt_resolver(state) self._interrupt("adoption") before = state.get("before") if not isinstance(before, Mapping): - if live.get("user_version") != SCHEMA_BEFORE: - raise InstalledForgeUpdateError("schema 38 lacks this operation's pre-migration snapshot") + if live.get("user_version") != schema_before: + raise InstalledForgeUpdateError( + "selected schema lacks this operation's pre-migration snapshot" + ) before = live assert_selected_installation(self.request, before) - if before.get("user_version") != SCHEMA_BEFORE: - raise InstalledForgeUpdateError("durable pre-migration snapshot is not schema 37") + if before.get("user_version") != schema_before: + raise InstalledForgeUpdateError( + "durable pre-migration snapshot has the wrong source schema" + ) state = self._backup(state, before) self._interrupt("backup") state = self._qualify_copy(state, before) diff --git a/tests/test_installed_forge_update.py b/tests/test_installed_forge_update.py index 6eaaa70..5cc3216 100644 --- a/tests/test_installed_forge_update.py +++ b/tests/test_installed_forge_update.py @@ -199,6 +199,14 @@ def _controller(self) -> object: self.request, process_reader=lambda: (), ) + def _same_schema_request(self) -> object: + return update.UpdateRequest(**{ + **self.request.__dict__, + "operation_id": "forge-update-2723-test-001", + "version": "2.7.23", + "existing_version": "2.7.22", + }) + def _qualified_schema38_copy(self, controller: object, before: dict[str, object]) -> None: copy_root = controller.operation_root / "qualification-copy" (copy_root / "instance").mkdir(parents=True) @@ -222,6 +230,85 @@ def test_exact_release_complete_artifact_is_accepted_and_mismatch_rejected(self) with self.assertRaisesRegex(update.InstalledForgeUpdateError, "wheel digest"): update.validate_qualified_artifact(self.request) + def test_normal_2723_release_receipt_is_accepted_only_for_the_new_transition(self) -> None: + version = "2.7.23" + wheel = self.root / f"forge_autonomy-{version}-py3-none-any.whl" + dist_info = f"forge_autonomy-{version}.dist-info" + members = { + "forge/__init__.py": f"__version__ = '{version}'\n".encode(), + f"{dist_info}/METADATA": ( + f"Metadata-Version: 2.4\nName: forge-autonomy\nVersion: {version}\n\n" + ).encode(), + f"{dist_info}/WHEEL": ( + b"Wheel-Version: 1.0\nRoot-Is-Purelib: true\nTag: py3-none-any\n\n" + ), + } + record_name = f"{dist_info}/RECORD" + record = "".join( + f"{name},sha256={base64.urlsafe_b64encode(sha256(payload).digest()).rstrip(b'=').decode()},{len(payload)}\n" + for name, payload in members.items() + ) + f"{record_name},,\n" + with zipfile.ZipFile(wheel, "w") as archive: + for name, payload in members.items(): + archive.writestr(name, payload) + archive.writestr(record_name, record) + wheel_digest = update.file_digest(wheel) + source = "e" * 40 + receipt = self.root / "normal-release-complete.json" + receipt.write_text(json.dumps({ + "state": "RELEASE_COMPLETE", + "product": "forge", + "component": "forge-autonomy", + "version": version, + "source_revision": source, + "operation_id": f"forge-release-{version}-{source}", + "policy_revision": "forge-bootstrap-release-cadence-v2", + "artifacts": {"wheel": wheel_digest, "sdist": self.sdist_digest}, + "qualification": { + "exact_main_sha": source, + "qualification": "forge-production-distribution", + "artifact_digests": { + f"dist/forge_autonomy-{version}-py3-none-any.whl": wheel_digest, + f"dist/forge_autonomy-{version}.tar.gz": self.sdist_digest, + }, + }, + "publication_receipt": { + "registry": "pypi", + "readback": "PASS", + "observed_artifact_digests": { + f"forge_autonomy-{version}-py3-none-any.whl": wheel_digest, + f"forge_autonomy-{version}.tar.gz": self.sdist_digest, + }, + }, + "cleanup": { + "result": "COMPLETE", + "temporary_paths": [ + "published-readback", "published-input/dist", "pending-readback", + ], + "github_release": {"draft": False}, + }, + }, sort_keys=True), encoding="utf-8") + request = update.UpdateRequest(**{ + **self.request.__dict__, + "operation_id": "forge-update-2723-test-002", + "version": version, + "product_source": source, + "wheel": str(wheel), + "wheel_sha256": wheel_digest, + "qualification_receipt": str(receipt), + "qualification_receipt_sha256": update.file_digest(receipt), + "existing_version": "2.7.22", + }) + + evidence = update.validate_qualified_artifact(request) + + self.assertEqual(evidence["release_route"], "NORMAL") + wrong_transition = update.UpdateRequest(**{ + **request.__dict__, "existing_version": "2.7.21", + }) + with self.assertRaises(update.InstalledForgeUpdateError): + update.validate_qualified_artifact(wrong_transition) + def test_real_schema37_to_38_migration_preserves_history_and_bindings(self) -> None: self._installed_schema37() before = update.database_snapshot(self.data_root / "forge.db") @@ -460,14 +547,35 @@ def test_completed_replay_requires_exact_activated_schema38_fingerprint(self) -> "schema_digest": "sha256:" + "1" * 64, } readback = {"database_schema_digest": snapshot["schema_digest"]} - update.assert_completed_schema(snapshot, readback) + update.assert_completed_schema(snapshot, readback, self.request) with self.assertRaisesRegex(update.InstalledForgeUpdateError, "schema changed"): - update.assert_completed_schema({**snapshot, "user_version": 37}, readback) + update.assert_completed_schema( + {**snapshot, "user_version": 37}, readback, self.request, + ) with self.assertRaisesRegex(update.InstalledForgeUpdateError, "schema changed"): update.assert_completed_schema( - {**snapshot, "schema_digest": "sha256:" + "2" * 64}, readback, + {**snapshot, "schema_digest": "sha256:" + "2" * 64}, + readback, + self.request, ) + def test_schema38_to_38_transition_preserves_all_runtime_content(self) -> None: + self._installed_schema37() + database = RuntimeBootstrap(data_root=self.data_root, forge_version="2.7.22").open() + database.close() + before = update.database_snapshot(self.data_root / "forge.db") + request = self._same_schema_request() + + preservation = update.verify_preservation(before, before, request) + + self.assertEqual(preservation["from_schema"], 38) + self.assertEqual(preservation["to_schema"], 38) + self.assertEqual(preservation["added_tables"], []) + changed = json.loads(json.dumps(before)) + changed["tables"]["operational_reset_audit"]["digest"] = "sha256:" + "9" * 64 + with self.assertRaisesRegex(update.InstalledForgeUpdateError, "historical table"): + update.verify_preservation(before, changed, request) + def test_atomic_product_migrated_copy_rejects_late_live_mutation(self) -> None: self._installed_schema37() controller = self._controller() diff --git a/tests/test_operational_reset.py b/tests/test_operational_reset.py index fbe0f03..5103d11 100644 --- a/tests/test_operational_reset.py +++ b/tests/test_operational_reset.py @@ -296,6 +296,144 @@ def test_stale_plan_wrong_request_and_operation_id_reuse_are_rejected(self) -> N ) self.service().finish(operation_id="forge-reset-test-0001", cancel_before_apply=True) + def test_prepared_operation_revalidates_without_redefining_the_plan(self) -> None: + self.seed_operational_state(external=False) + service = self.service() + receipt = self.prepare(service) + before = service.status(operation_id="forge-reset-test-0001") + + generic_preview = service.preview() + self.assertEqual(generic_preview["status"], "BLOCKED") + self.assertEqual( + [{"code": "MAINTENANCE_ALREADY_ACTIVE", "operation_id": "forge-reset-test-0001"}], + generic_preview["blockers"], + ) + self.assertEqual(receipt["plan_digest"], generic_preview["plan_digest"]) + self.assertEqual( + "forge-reset-test-0001", + generic_preview["maintenance_observation"]["active_operation_id"], + ) + + revalidated = service.revalidate( + operation_id="forge-reset-test-0001", plan_digest=str(receipt["plan_digest"]), + request_digest=str(receipt["request_digest"]), + backup_digest=str(receipt["backup_digest"]), + ) + + self.assertEqual(receipt["plan_digest"], revalidated["plan_digest"]) + self.assertEqual("forge-reset-test-0001", revalidated["revalidation"]["writer_fence_owner"]) + self.assertEqual( + revalidated["revalidation"]["relevant_revision_digest"], + generic_preview["relevant_revision"], + ) + self.assertRegex(revalidated["revalidation"]["revalidation_digest"], r"^sha256:[0-9a-f]{64}$") + self.assertEqual(before, service.status(operation_id="forge-reset-test-0001")) + service.finish(operation_id="forge-reset-test-0001", cancel_before_apply=True) + + def test_operation_revalidation_rejects_same_count_drift_and_tampered_backup(self) -> None: + service = self.service() + receipt = self.prepare(service) + privileged = sqlite3.connect(self.root / "forge.db") + privileged.create_function("forge_maintenance_write_permitted", 0, lambda: 1) + privileged.execute( + "UPDATE operational_reset_state SET active_operation_id=NULL WHERE singleton=1" + ) + privileged.execute( + "UPDATE runtime_metadata SET value='changed-with-same-row-count' WHERE key='status'" + ) + privileged.execute( + "UPDATE operational_reset_state SET active_operation_id='forge-reset-test-0001' WHERE singleton=1" + ) + privileged.commit() + privileged.close() + with self.assertRaisesRegex(OperationalResetError, "meaningful source data changed"): + service.revalidate( + operation_id="forge-reset-test-0001", plan_digest=str(receipt["plan_digest"]), + request_digest=str(receipt["request_digest"]), + backup_digest=str(receipt["backup_digest"]), + ) + + # A separate prepared operation proves backup binding independently. + other_root = self.root / "backup-revalidation" + other_database = RuntimeBootstrap(data_root=other_root, forge_version="2.7.21").open() + self.addCleanup(other_database.close) + InstallationOperatorService(other_database, lambda: IDENTITY).first_bind() + other = ForgeOperationalResetService(other_root, identity_resolver=lambda: IDENTITY) + other_plan = other.preview() + other_receipt = other.prepare( + operation_id="forge-reset-test-0002", + expected_plan_digest=str(other_plan["plan_digest"]), + ) + (other_root / str(other_receipt["backup_reference"]) / "runtime-instance.json").write_text( + "tampered\n", encoding="utf-8", + ) + with self.assertRaisesRegex(OperationalResetError, "backup digest"): + other.revalidate( + operation_id="forge-reset-test-0002", + plan_digest=str(other_receipt["plan_digest"]), + request_digest=str(other_receipt["request_digest"]), + backup_digest=str(other_receipt["backup_digest"]), + ) + + def test_apply_rechecks_authority_and_fence_after_successful_revalidation(self) -> None: + service = self.service() + receipt = self.prepare(service) + service.revalidate( + operation_id="forge-reset-test-0001", plan_digest=str(receipt["plan_digest"]), + request_digest=str(receipt["request_digest"]), + backup_digest=str(receipt["backup_digest"]), + ) + privileged = sqlite3.connect(self.root / "forge.db") + privileged.create_function("forge_maintenance_write_permitted", 0, lambda: 1) + privileged.create_function("forge_governance_write_permitted", 0, lambda: 1) + privileged.execute( + "UPDATE operational_reset_state SET active_operation_id=NULL WHERE singleton=1" + ) + privileged.execute("DROP TRIGGER governance_authority_immutable_delete") + privileged.execute( + "DELETE FROM governance_authority WHERE capability='SECURITY_APPROVAL'" + ) + privileged.execute( + "UPDATE operational_reset_state SET active_operation_id='forge-reset-test-0001' WHERE singleton=1" + ) + privileged.commit() + privileged.close() + with self.assertRaisesRegex(OperationalResetError, "requires current Security"): + service.apply( + operation_id="forge-reset-test-0001", plan_digest=str(receipt["plan_digest"]), + request_digest=str(receipt["request_digest"]), + backup_digest=str(receipt["backup_digest"]), + ) + + other_root = self.root / "fence-revalidation" + other_database = RuntimeBootstrap(data_root=other_root, forge_version="2.7.21").open() + self.addCleanup(other_database.close) + InstallationOperatorService(other_database, lambda: IDENTITY).first_bind() + other = ForgeOperationalResetService(other_root, identity_resolver=lambda: IDENTITY) + other_plan = other.preview() + other_receipt = other.prepare( + operation_id="forge-reset-test-0002", + expected_plan_digest=str(other_plan["plan_digest"]), + ) + other.revalidate( + operation_id="forge-reset-test-0002", plan_digest=str(other_receipt["plan_digest"]), + request_digest=str(other_receipt["request_digest"]), + backup_digest=str(other_receipt["backup_digest"]), + ) + fence = sqlite3.connect(other_root / "forge.db") + fence.create_function("forge_maintenance_write_permitted", 0, lambda: 1) + fence.execute( + "UPDATE operational_reset_state SET active_operation_id=NULL,state='IDLE' WHERE singleton=1" + ) + fence.commit() + fence.close() + with self.assertRaisesRegex(OperationalResetError, "does not own durable maintenance"): + other.apply( + operation_id="forge-reset-test-0002", plan_digest=str(other_receipt["plan_digest"]), + request_digest=str(other_receipt["request_digest"]), + backup_digest=str(other_receipt["backup_digest"]), + ) + def test_insufficient_disk_leaves_durable_safe_maintenance_and_can_cancel(self) -> None: service = self.service(disk_usage=lambda _path: Disk(1, 1, 0)) plan = service.preview() @@ -575,6 +713,17 @@ def test_operator_envelope_has_stable_cross_product_fields(self) -> None: self.assertEqual(set(envelope["target"]), {"instance_id", "database_path", "database_identity", "schema_version"}) self.assertIn("preserved_bindings_digest", envelope) + def test_operator_envelope_reports_the_observed_post_apply_generation(self) -> None: + plan = self.service().preview() + payload = { + **plan, + "state": "VERIFIED", + "operation_id": "forge-reset-generation-fixture-001", + "generation_after": 1, + } + envelope = self.service().operator_envelope("verify", payload) + self.assertEqual(envelope["dataset_generation"], 1) + def test_cli_error_retains_stable_cross_product_envelope(self) -> None: result = subprocess.run( ( @@ -610,7 +759,7 @@ def test_packaged_cli_exposes_real_read_only_preview_and_command_family(self) -> text=True, capture_output=True, check=False, ) self.assertEqual(help_result.returncode, 0) - for command in ("preview", "prepare", "apply", "status", "resume", "verify", "finish"): + for command in ("preview", "prepare", "revalidate", "apply", "status", "resume", "verify", "finish"): self.assertIn(command, help_result.stdout) def marker_id(self) -> str: