From e1a3b52f143021bd4439baa2d19cfb8de3b8d262 Mon Sep 17 00:00:00 2001 From: zhnt Date: Wed, 23 Sep 2026 01:22:21 +0000 Subject: [PATCH 1/2] feat(harness): project package GC operator status Refs #509 --- .../architecture/harness/plugin/README.md | 9 +- ...gin-lifecycle-coding-pluginization-plan.md | 3 + .../plugin/plugin-lifecycle-plc9-inventory.md | 12 +- .../plugin-lifecycle-plc9d1-contract.md | 56 +++++ .../harness/plugin_management/__init__.py | 10 + .../harness/plugin_management/package_gc.py | 193 ++++++++++++++++++ .../test_package_lifecycle.py | 48 +++++ 7 files changed, 326 insertions(+), 5 deletions(-) create mode 100644 docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d1-contract.md create mode 100644 src/loushang/harness/plugin_management/package_gc.py diff --git a/docs/internals/architecture/harness/plugin/README.md b/docs/internals/architecture/harness/plugin/README.md index b8128b4b5..c1cad4976 100644 --- a/docs/internals/architecture/harness/plugin/README.md +++ b/docs/internals/architecture/harness/plugin/README.md @@ -10,8 +10,10 @@ are implementation evidence. - Implementation status: PLC0--PLC8 are complete and terminally reviewed; PLC9 is partially implemented through the explicit Linux and Windows AMD64 - Coding Worker canaries. The remaining platform gaps are summarized by - `architecture.md` and tracked in the lifecycle plan. + Coding Worker canaries. PLC9D1 adds a local-candidate, read-only Package GC + operator projection; physical artifact GC remains absent. The remaining + platform gaps are summarized by `architecture.md` and tracked in the + lifecycle plan. - Owner: `loushang.harness` Plugin architecture scope; contribution runtime authority remains with each exact domain owner. @@ -51,6 +53,9 @@ Neither may silently override a narrower implemented owner contract. - [Resource Catalog And Source Pluginization Plan](resource-catalog-pluginization-plan.md) owns the Resource/Skill catalog convergence and the rule that mechanisms may be Plugin components while individual Skills remain Resources. +- [PLC9D1 Package GC Operator Projection Contract](plugin-lifecycle-plc9d1-contract.md) + bounds the current internal GC/cleanup read-model candidate; it does not + authorize physical deletion or claim PLC9D completion. ## Frozen Contracts diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md index a3bf3638e..849ee6c00 100644 --- a/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md @@ -66,6 +66,9 @@ mechanics/rejection, C5.4 Linux Coding Product canary, and C5.5b/c Windows LPAC containment and Coding Product canary. Both canaries are explicit and default-dark; Current remains the default and unlisted routes remain closed. + [PLC9D1](plugin-lifecycle-plc9d1-contract.md) is a local candidate for an + internal, read-only Package GC/cleanup-debt projection. It does not reserve + or delete an artifact and does not close PLC9D. - Scope: one delivery order for the common Plugin lifecycle, ordinary Definition / Provider / Consumer authoring primitives, `coding.lsp`, `coding.base`, `coding.arch`, management control, pre-LSP internal Resource/ diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md index 69a34178c..966d42a40 100644 --- a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md @@ -9,6 +9,10 @@ [Plugin Lifecycle PLC9.0 Baseline](plugin-lifecycle-plc9-baseline.md). - PLC9B.0 refinement: [Safe Package Boundary Contract](plugin-lifecycle-plc9b-contract.md). +- PLC9D1 refinement: + [Package GC Operator Projection Contract](plugin-lifecycle-plc9d1-contract.md) + adds an internal all-revision read model over existing retention evidence. + It grants no Store deletion or GC reservation authority. - PLC9B1 refinement: the dark internal Owner Kernel now supplies versioned inert records, classification, journal CAS, retry/cancel/status, and disabled refusal. It has no production composition or artifact capability; all @@ -835,8 +839,9 @@ publication outside those exact canaries. | Current seam | Exact source owner or symbol | Current fact | PLC9 disposition and gate | | --- | --- | --- | --- | -| Cleanup attempts and repair | `src/loushang/harness/plugin_management/package_lifecycle.py::PluginPackageLifecycleLedger` | Derives `pending`, `retryable_failure`, `terminal_failure`, `retry_permitted`, `succeeded`, and `safe_abandoned` from durable attempts/decisions | Retain; PLC9D adds operator projection and exact deletion execution without releasing debt implicitly | -| GC candidate | `src/loushang/harness/plugin_management/package_lifecycle.py::PluginPackageGcCandidateV1` | Binds desired, Instance, package-journal, and recovery-barrier revisions | Retain and recheck immediately before exact revision deletion; desired absence alone is insufficient | +| Cleanup attempts and repair | `src/loushang/harness/plugin_management/package_lifecycle.py::PluginPackageLifecycleLedger` | Derives `pending`, `retryable_failure`, `terminal_failure`, `retry_permitted`, `succeeded`, and `safe_abandoned` from durable attempts/decisions | Retain; PLC9D1 projects this evidence, while later deletion execution must not release debt implicitly | +| Package GC operator projection | `src/loushang/harness/plugin_management/package_gc.py::PluginPackageGcReadModel` | PLC9D1 projects every known revision, exact candidate or blocker codes, and durable cleanup/repair state without a mutation port | Retain as internal read-only evidence; future executable GC must add an exclusive reservation, Store-owned rooted deletion, and a durable result/debt receipt | +| GC candidate | `src/loushang/harness/plugin_management/package_lifecycle.py::PluginPackageGcCandidateV1` | Binds desired, Instance, package-journal, and recovery-barrier revisions | Retain; later executable GC must reserve against new references and recheck under the owner fence before exact revision deletion; desired absence alone is insufficient | | Coding private roots | `src/loushang/coding/_plugin_lifecycle.py::CodingPluginLifecycleStateLayout` | Separates private lifecycle state and package data bases and prepares private directory permissions | Retain path containment; path ownership is not deletion authorization | | Continuity deletion authorization | `src/loushang/harness/plugin_management/continuity_mutation.py::PluginContinuityDeletionAuthority` | Serializes one exact deletion, durably authorizes it, and settles terminal receipt/cancel evidence; it does not perform the source mutation | Retain as Product authorization/settlement precedent; never elevate it into a generic destructive executor | | Continuity destructive commit | `src/loushang/harness/continuity/mutation.py::AuthorizedContinuityDeletionLease._commit_complete_and_release` over the source-owned `PreparedContinuityDeletion.commit` port, prepared by `src/loushang/harness/continuity/plugin_provider.py::PluginContinuityProvider._prepare_delete` | Calls the source/data-domain candidate commit first, validates its receipt, then asks the Product authority to settle | Retain the plan -> authorization -> source commit -> receipt settlement order for any future domain deletion contract | @@ -879,7 +884,8 @@ PLC9A1 contract: domain generation publication, and recovery/rollback composition; C5.0 documents and guards these absences but implements none of them; - `remote_service` topology contract and client; -- executable artifact-GC owner/receipt; +- executable artifact-GC owner/reservation/result receipt (PLC9D1 adds only the + internal operator projection); - generic Plugin-private data deletion command/receipt; and - correlated backup-retention projection. diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d1-contract.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d1-contract.md new file mode 100644 index 000000000..27468790c --- /dev/null +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d1-contract.md @@ -0,0 +1,56 @@ +# PLC9D1 Package GC Operator Projection Contract + +## Status And Scope + +- Tracking: PLC9 issue `#509`; this is a narrow implementation candidate, not + PLC9D or PLC9 completion evidence. +- Owner: `PluginPackageLifecycleLedger` remains the sole retention-evidence + owner. `PluginPackageGcReadModel` is a read-only internal projection. +- Effect: no artifact deletion, GC lease/reservation, Store mutation, private + data deletion, backup-expiry claim, or new management command. + +PLC9D1 makes existing durable retention and cleanup evidence inspectable for +**every known Package revision**, including a revision present only in history. +The existing management Installation view is keyed by current Installations; +it cannot by itself list an orphaned revision that might be GC eligible. + +## Operator Query + +`PluginPackageGcReadModel.snapshot()` captures one +`PluginPackageLifecycleSnapshotV1` and returns a versioned projection sorted by +exact Package revision. Each row carries the exact revision, zero or more +blocker codes, an optional GC candidate, and that revision's cleanup task +states. Cleanup summaries expose attempt count, last result code, retry time, +last repair action, and whether the lease is still open; they do not invent an +operator repair decision. + +The blocker codes are `startup_recovery`, `desired_installation`, +`nonretired_instance`, `runtime_family`, `retention_pin`, `cleanup_lease`, and +`terminal_cleanup_failure`. The candidate is present exactly when the existing +retention owner reports no blocker. Its identity binds the desired, Instance, +Package journal, and recovery-barrier revisions. It proves retention eligibility, +not that a published Store tree currently exists. A terminal cleanup failure +retains a lease and stays blocked until the existing durable repair/attempt +sequence reaches success or an explicit `safe_abandon` decision. + +The operator projection grants no deletion authority. A later executable GC +slice must first add a durable exclusive reservation that prevents new +references during deletion, bind the exact published Store tree and native +identity, recheck the candidate under the appropriate Product/owner fence, +perform rooted deletion through the Store owner, and durably settle success or +retryable debt. Calling `recheck_gc_candidate()` alone cannot close the race +between a read and physical deletion. + +## Caller Inventory And Verification + +| Seam | Before | PLC9D1 | +| --- | --- | --- | +| Package lifecycle ledger | retention snapshots, candidate generation/recheck, cleanup attempts and repair evidence | unchanged authority and journal format | +| Internal GC operator | no all-revision projection | `plugin_management.package_gc.PluginPackageGcReadModel` consumes only `snapshot()`; no command or Store dependency | +| Coding/CLI/RPC/UI | no artifact GC route | unchanged; no new Product or transport caller | + +The focused durable-ledger regression covers recovery gating, an exact +candidate, stale candidate rejection after a new pin, visible terminal cleanup +debt, and explicit safe abandonment. Static checks keep the read model free of +deletion or Store imports. No physical deletion or crash-restart GC claim is +made by this slice. diff --git a/src/loushang/harness/plugin_management/__init__.py b/src/loushang/harness/plugin_management/__init__.py index 06a0e5b84..5916f15dd 100644 --- a/src/loushang/harness/plugin_management/__init__.py +++ b/src/loushang/harness/plugin_management/__init__.py @@ -98,6 +98,12 @@ PluginManagementOperationEventV1, PluginManagementOperationResultV1, ) +from loushang.harness.plugin_management.package_gc import ( + PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION, + PluginPackageGcOperatorProjectionV1, + PluginPackageGcOperatorRowV1, + PluginPackageGcReadModel, +) from loushang.harness.plugin_management.package_lifecycle import ( PLUGIN_PACKAGE_GC_CANDIDATE_VERSION, PluginCleanupTaskSnapshotV1, @@ -237,6 +243,7 @@ "PLUGIN_MIGRATION_FENCE_VERSION", "PLUGIN_PACKAGE_REVISION_REF_VERSION", "PLUGIN_PACKAGE_GC_CANDIDATE_VERSION", + "PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION", "PLUGIN_PACKAGE_LIFECYCLE_EVENT_VERSION", "PLUGIN_PACKAGE_PIN_RELEASE_VERSION", "PLUGIN_PACKAGE_PIN_VERSION", @@ -329,6 +336,9 @@ "PluginMigrationFenceV1", "PluginPackageRevisionRefV1", "PluginPackageGcCandidateV1", + "PluginPackageGcOperatorProjectionV1", + "PluginPackageGcOperatorRowV1", + "PluginPackageGcReadModel", "PluginPackageLifecycleError", "PluginPackageLifecycleEventV1", "PluginPackageLifecycleLedger", diff --git a/src/loushang/harness/plugin_management/package_gc.py b/src/loushang/harness/plugin_management/package_gc.py new file mode 100644 index 000000000..80b26731d --- /dev/null +++ b/src/loushang/harness/plugin_management/package_gc.py @@ -0,0 +1,193 @@ +"""Read-only PLC9D package-GC and cleanup-debt projection. + +The projection exposes durable owner evidence, not a deletion capability. A +candidate must still be rechecked and reserved by a future Store GC owner. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + +from loushang.harness.plugin_management.package_lifecycle import ( + PluginCleanupTaskSnapshotV1, + PluginPackageGcCandidateV1, + PluginPackageLifecycleSnapshotV1, + PluginPackageRetentionSnapshotV1, +) +from loushang.harness.plugin_management.records import PluginPackageRevisionRefV1 + +PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION = 1 + + +class PluginPackageGcSnapshotSourcePort(Protocol): + def snapshot(self) -> PluginPackageLifecycleSnapshotV1: ... + + +@dataclass(frozen=True, slots=True) +class PluginPackageGcOperatorRowV1: + package_revision: PluginPackageRevisionRefV1 + blocker_codes: tuple[str, ...] + candidate: PluginPackageGcCandidateV1 | None + cleanup_tasks: tuple[PluginCleanupTaskSnapshotV1, ...] + + def __post_init__(self) -> None: + if self.blocker_codes != tuple(sorted(set(self.blocker_codes))): + raise ValueError("Package GC blocker codes must be sorted and unique") + if (self.candidate is None) != bool(self.blocker_codes): + raise ValueError("Package GC candidate and blockers disagree") + if self.candidate is not None and ( + self.candidate.package_revision != self.package_revision + ): + raise ValueError("Package GC candidate names another revision") + if self.cleanup_tasks != tuple( + sorted(self.cleanup_tasks, key=lambda item: item.task.cleanup_id) + ): + raise ValueError("Package GC cleanup tasks must be sorted") + if any( + item.task.package_revision != self.package_revision + for item in self.cleanup_tasks + ): + raise ValueError("Package GC cleanup task names another revision") + + def to_dict(self) -> dict[str, object]: + candidate = self.candidate + return { + "blockerCodes": list(self.blocker_codes), + "candidate": ( + None + if candidate is None + else { + "candidateId": candidate.candidate_id, + "candidateVersion": candidate.candidate_version, + "desiredInventoryRevision": candidate.desired_inventory_revision, + "instanceRuntimeRevision": candidate.instance_runtime_revision, + "packageJournalRevision": candidate.package_journal_revision, + "recoveryBarrierId": candidate.recovery_barrier_id, + } + ), + "cleanupTasks": [ + { + "cleanupId": item.task.cleanup_id, + "state": item.state, + "attemptCount": len(item.attempts), + "leaseOpen": item.lease_open, + "lastResultCode": ( + None if not item.attempts else item.attempts[-1].result_code + ), + "retryNotBeforeEpochMs": ( + None + if not item.attempts + else item.attempts[-1].retry_not_before_epoch_ms + ), + "lastRepairAction": ( + None + if not item.repair_decisions + else item.repair_decisions[-1].action + ), + } + for item in self.cleanup_tasks + ], + "packageRevision": self.package_revision.to_dict(), + } + + +@dataclass(frozen=True, slots=True) +class PluginPackageGcOperatorProjectionV1: + startup_id: str + package_journal_revision: int + recovery_complete: bool + packages: tuple[PluginPackageGcOperatorRowV1, ...] + projection_version: int = PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION + + def __post_init__(self) -> None: + if self.projection_version != PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION: + raise ValueError("Unsupported Package GC operator projection") + if self.packages != tuple( + sorted( + self.packages, key=lambda item: _package_sort_key(item.package_revision) + ) + ): + raise ValueError("Package GC operator rows must be sorted") + if len({item.package_revision for item in self.packages}) != len(self.packages): + raise ValueError("Package GC operator rows must be unique") + + def to_dict(self) -> dict[str, object]: + return { + "packageJournalRevision": self.package_journal_revision, + "packages": [item.to_dict() for item in self.packages], + "projectionVersion": self.projection_version, + "recoveryComplete": self.recovery_complete, + "startupId": self.startup_id, + } + + +class PluginPackageGcReadModel: + """Project every known revision, including history-only GC candidates.""" + + def __init__(self, lifecycle: PluginPackageGcSnapshotSourcePort) -> None: + self._lifecycle = lifecycle + + def snapshot(self) -> PluginPackageGcOperatorProjectionV1: + source = self._lifecycle.snapshot() + tasks_by_package: dict[ + PluginPackageRevisionRefV1, list[PluginCleanupTaskSnapshotV1] + ] = {} + for item in source.cleanup_tasks: + tasks_by_package.setdefault(item.task.package_revision, []).append(item) + return PluginPackageGcOperatorProjectionV1( + startup_id=source.startup_id, + package_journal_revision=source.journal_revision, + recovery_complete=source.startup_recovered, + packages=tuple( + _row(item, source, tasks_by_package.get(item.package_revision, [])) + for item in source.packages + ), + ) + + +def _row( + retained: PluginPackageRetentionSnapshotV1, + source: PluginPackageLifecycleSnapshotV1, + cleanup_tasks: list[PluginCleanupTaskSnapshotV1], +) -> PluginPackageGcOperatorRowV1: + blockers = set() + if not source.startup_recovered: + blockers.add("startup_recovery") + for values, code in ( + (retained.desired_installations, "desired_installation"), + (retained.nonretired_instances, "nonretired_instance"), + (retained.open_runtime_family_ids, "runtime_family"), + (retained.open_pin_ids, "retention_pin"), + (retained.open_cleanup_ids, "cleanup_lease"), + (retained.terminal_failure_cleanup_ids, "terminal_cleanup_failure"), + ): + if values: + blockers.add(code) + return PluginPackageGcOperatorRowV1( + package_revision=retained.package_revision, + blocker_codes=tuple(sorted(blockers)), + candidate=retained.gc_candidate, + cleanup_tasks=tuple(cleanup_tasks), + ) + + +def _package_sort_key( + package_revision: PluginPackageRevisionRefV1, +) -> tuple[str, str, str, str, str]: + return ( + package_revision.plugin_id, + package_revision.plugin_version or "", + package_revision.package_content_digest, + package_revision.dependency_lock_digest, + package_revision.package_source_identity, + ) + + +__all__ = [ + "PLUGIN_PACKAGE_GC_OPERATOR_PROJECTION_VERSION", + "PluginPackageGcOperatorProjectionV1", + "PluginPackageGcOperatorRowV1", + "PluginPackageGcReadModel", + "PluginPackageGcSnapshotSourcePort", +] diff --git a/tests/harness/plugin_management/test_package_lifecycle.py b/tests/harness/plugin_management/test_package_lifecycle.py index 1421b1a1c..6078e5af5 100644 --- a/tests/harness/plugin_management/test_package_lifecycle.py +++ b/tests/harness/plugin_management/test_package_lifecycle.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ast import asyncio import json from collections.abc import Callable @@ -37,6 +38,7 @@ PluginManagementAction, PluginManagementCommandV1, ) +from loushang.harness.plugin_management.package_gc import PluginPackageGcReadModel from loushang.harness.plugin_management.package_lifecycle import ( PluginPackageGcCandidateV1, PluginPackageLifecycleError, @@ -73,6 +75,22 @@ from loushang.harness.resources.plugins.selection import PluginInstanceRevisionRef +def test_plc9d1_gc_projection_cannot_import_store_or_deletion_authority() -> None: + source = Path("src/loushang/harness/plugin_management/package_gc.py") + tree = ast.parse(source.read_text(encoding="utf-8")) + imported = { + node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) + } + assert imported <= { + "__future__", + "dataclasses", + "typing", + "loushang.harness.plugin_management.package_lifecycle", + "loushang.harness.plugin_management.records", + } + assert not any(isinstance(node, ast.Import) for node in ast.walk(tree)) + + def test_continuity_publication_security_close_hands_off_package_cleanup( tmp_path: Path, ) -> None: @@ -518,6 +536,16 @@ def test_cleanup_retry_terminal_repair_and_safe_abandon(tmp_path: Path) -> None: ).state == "terminal_failure" ) + operator_row = PluginPackageGcReadModel(context.packages).snapshot().packages[0] + assert "terminal_cleanup_failure" in operator_row.blocker_codes + assert operator_row.candidate is None + host_status = next( + item + for item in operator_row.to_dict()["cleanupTasks"] + if item["cleanupId"] == host_task.cleanup_id + ) + assert host_status["state"] == "terminal_failure" + assert host_status["lastResultCode"] == "cleanup.host-terminal" abandon = PluginCleanupRepairDecisionV1.create( cleanup_id=host_task.cleanup_id, repair_sequence=1, @@ -530,6 +558,13 @@ def test_cleanup_retry_terminal_repair_and_safe_abandon(tmp_path: Path) -> None: abandoned = context.packages.record_repair_decision(abandon) assert abandoned.state == "safe_abandoned" assert not abandoned.lease_open + operator_row = PluginPackageGcReadModel(context.packages).snapshot().packages[0] + assert "terminal_cleanup_failure" not in operator_row.blocker_codes + assert "cleanup_lease" not in operator_row.blocker_codes + assert any( + item["state"] == "safe_abandoned" and item["lastRepairAction"] == "safe_abandon" + for item in operator_row.to_dict()["cleanupTasks"] + ) def test_gc_candidate_requires_every_source_zero_and_revision_recheck( @@ -587,6 +622,10 @@ def test_gc_candidate_requires_every_source_zero_and_revision_recheck( with pytest.raises(PluginPackageLifecycleError) as caught: context.packages.gc_candidates() assert caught.value.code == "plugin_package_recovery_incomplete" + operator = PluginPackageGcReadModel(context.packages) + before_recovery = operator.snapshot() + assert not before_recovery.recovery_complete + assert before_recovery.packages[0].blocker_codes == ("startup_recovery",) context.packages.complete_startup_recovery( operation_id="recover-a", idempotency_key="recover-request-a", @@ -594,6 +633,12 @@ def test_gc_candidate_requires_every_source_zero_and_revision_recheck( ) candidate = context.packages.gc_candidates()[0] assert context.packages.recheck_gc_candidate(candidate) == candidate + ready = operator.snapshot() + assert ready.packages[0].candidate == candidate + assert ready.packages[0].blocker_codes == () + assert ready.to_dict()["packages"][0]["candidate"]["candidateId"] == ( + candidate.candidate_id + ) pin = context.packages.acquire_pin( prepared.active.package_revision, @@ -605,6 +650,9 @@ def test_gc_candidate_requires_every_source_zero_and_revision_recheck( with pytest.raises(PluginPackageLifecycleError) as caught: context.packages.recheck_gc_candidate(candidate) assert caught.value.code == "invalid_plugin_package_lifecycle_transition" + pinned = operator.snapshot().packages[0] + assert pinned.candidate is None + assert pinned.blocker_codes == ("retention_pin",) context.packages.release_pin(_pin_release(pin, suffix="late")) replacement = context.packages.gc_candidates()[0] assert replacement.package_revision == candidate.package_revision From 00c619e8dbbf79b339224126309ae7b06ca1d03e Mon Sep 17 00:00:00 2001 From: zhnt Date: Wed, 23 Sep 2026 02:54:28 +0000 Subject: [PATCH 2/2] feat(harness): reserve package GC candidates Add a dark durable reservation journal and optional reference-writer gate for desired, Instance, and Package ledgers. Keep physical deletion and Product composition out of scope; legacy unbound writers remain an explicit D3 blocker. Refs #509 --- .../architecture/harness/plugin/README.md | 11 +- ...gin-lifecycle-coding-pluginization-plan.md | 4 +- .../plugin/plugin-lifecycle-plc9-inventory.md | 9 +- .../plugin-lifecycle-plc9d2-contract.md | 69 +++ .../harness/plugin_management/gc_fence.py | 23 + .../plugin_management/instance_runtime.py | 71 ++- .../harness/plugin_management/ledger.py | 55 +- .../package_gc_reservation.py | 500 ++++++++++++++++++ .../plugin_management/package_lifecycle.py | 119 +++-- .../harness/plugin_management/service.py | 68 +-- .../test_package_lifecycle.py | 198 ++++++- 11 files changed, 1024 insertions(+), 103 deletions(-) create mode 100644 docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d2-contract.md create mode 100644 src/loushang/harness/plugin_management/gc_fence.py create mode 100644 src/loushang/harness/plugin_management/package_gc_reservation.py diff --git a/docs/internals/architecture/harness/plugin/README.md b/docs/internals/architecture/harness/plugin/README.md index c1cad4976..90495f296 100644 --- a/docs/internals/architecture/harness/plugin/README.md +++ b/docs/internals/architecture/harness/plugin/README.md @@ -10,10 +10,10 @@ are implementation evidence. - Implementation status: PLC0--PLC8 are complete and terminally reviewed; PLC9 is partially implemented through the explicit Linux and Windows AMD64 - Coding Worker canaries. PLC9D1 adds a local-candidate, read-only Package GC - operator projection; physical artifact GC remains absent. The remaining - platform gaps are summarized by `architecture.md` and tracked in the - lifecycle plan. + Coding Worker canaries. PLC9D1/D2 add local-candidate Package GC operator + projection and dark reservation/fence mechanics; physical artifact GC remains + absent. The remaining platform gaps are summarized by `architecture.md` and + tracked in the lifecycle plan. - Owner: `loushang.harness` Plugin architecture scope; contribution runtime authority remains with each exact domain owner. @@ -56,6 +56,9 @@ Neither may silently override a narrower implemented owner contract. - [PLC9D1 Package GC Operator Projection Contract](plugin-lifecycle-plc9d1-contract.md) bounds the current internal GC/cleanup read-model candidate; it does not authorize physical deletion or claim PLC9D completion. +- [PLC9D2 Dark Package GC Reservation Contract](plugin-lifecycle-plc9d2-contract.md) + bounds opt-in reference fencing and durable reservation replay; no Product or + Store deletion route is active. ## Frozen Contracts diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md index 849ee6c00..f465c6ea5 100644 --- a/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-coding-pluginization-plan.md @@ -68,7 +68,9 @@ default-dark; Current remains the default and unlisted routes remain closed. [PLC9D1](plugin-lifecycle-plc9d1-contract.md) is a local candidate for an internal, read-only Package GC/cleanup-debt projection. It does not reserve - or delete an artifact and does not close PLC9D. + or delete an artifact. [PLC9D2](plugin-lifecycle-plc9d2-contract.md) adds a + dark, opt-in durable reservation and reference-writer fence; it has no + Product/Store deletion route and does not close PLC9D. - Scope: one delivery order for the common Plugin lifecycle, ordinary Definition / Provider / Consumer authoring primitives, `coding.lsp`, `coding.base`, `coding.arch`, management control, pre-LSP internal Resource/ diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md index 966d42a40..ab2c136dd 100644 --- a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9-inventory.md @@ -13,6 +13,10 @@ [Package GC Operator Projection Contract](plugin-lifecycle-plc9d1-contract.md) adds an internal all-revision read model over existing retention evidence. It grants no Store deletion or GC reservation authority. +- PLC9D2 refinement: + [Dark Package GC Reservation Contract](plugin-lifecycle-plc9d2-contract.md) + adds a durable, opt-in reference-writer fence. It is not composed by a + Product and grants no Store deletion authority. - PLC9B1 refinement: the dark internal Owner Kernel now supplies versioned inert records, classification, journal CAS, retry/cancel/status, and disabled refusal. It has no production composition or artifact capability; all @@ -389,6 +393,7 @@ implemented. | Instance runtime | `src/loushang/harness/plugin_management/instance_runtime.py::PluginInstanceRuntimeLedger` | Durable Instance activation, lease-family, drain, revocation, and retirement state | Retain; do not replace with Worker/process state | | Security retirement acceptance | `src/loushang/harness/plugin_management/security_acceptance.py::PluginInstanceSecurityRetirementJournal` | Durable acceptance evidence for security retirement | Retain; keep distinct from graceful retirement and generic management auth | | Package retention and cleanup | `src/loushang/harness/plugin_management/package_lifecycle.py::PluginPackageLifecycleLedger` | Durable pins, cleanup leases/attempts/repair decisions, recovery barrier, retention snapshots, and GC candidates | Retain as lifecycle evidence; PLC9D must add deletion execution/result without weakening candidate recheck | +| Dark GC reservation and reference fence | `src/loushang/harness/plugin_management/package_gc_reservation.py::PluginPackageGcReservationJournal` and `src/loushang/harness/plugin_management/gc_fence.py::PluginPackageGcReferenceGatePort` | PLC9D2 journals exact reservation/cancellation, replays active fences, and guards opt-in desired/Instance/Package reference writers in one lock order | Retain dark; D3 needs Product-wide writer binding and downgrade exclusion before Store-owned rooted deletion; a reservation alone is not authorization | | Coding Product composition | `src/loushang/coding/_plugin_lifecycle.py::CodingPluginLifecycle` | Product adapter composes the generic ledgers under one workspace identity and coordination lock | Retain as an outer Product adapter until common application ports replace Product-specific call sites; it must not become a second generic owner | | Management application command adapter | `src/loushang/harness/plugin_management/application.py::PluginManagementCommandApplication` | A1-1 preserves correlation around the durable operation identity and delegates every mutation to `PluginManagementService` | Retain as the transport-neutral command boundary; transports cannot import the service or desired-state ledger directly | | Management query projector | `src/loushang/harness/plugin_management/application.py::PluginManagementReadModelProjector` | A1-1 joins independently revisioned desired, operation, migration, Source, Instance, Package, and retirement snapshots without persisting another clock | Retain as the common read boundary; optional owners remain explicitly unsupported/unknown and forward/reverse skew remains observable | @@ -884,8 +889,8 @@ PLC9A1 contract: domain generation publication, and recovery/rollback composition; C5.0 documents and guards these absences but implements none of them; - `remote_service` topology contract and client; -- executable artifact-GC owner/reservation/result receipt (PLC9D1 adds only the - internal operator projection); +- executable artifact-GC owner and Store result/debt receipt (PLC9D1/D2 add + only the operator projection and dark reservation/fence mechanics); - generic Plugin-private data deletion command/receipt; and - correlated backup-retention projection. diff --git a/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d2-contract.md b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d2-contract.md new file mode 100644 index 000000000..0fbe6705b --- /dev/null +++ b/docs/internals/architecture/harness/plugin/plugin-lifecycle-plc9d2-contract.md @@ -0,0 +1,69 @@ +# PLC9D2 Dark Package GC Reservation Contract + +## Status And Boundary + +- Tracking: PLC9 `#509`. This is a local implementation candidate, not an + accepted executable-GC or PLC9D completion claim. +- Owner: `PluginPackageGcReservationJournal` durably owns only reservation and + cancellation evidence. The Package lifecycle ledger still owns retention + candidates; Desired State and Instance Runtime still own their references. +- Activation: dark. No Coding/Product, CLI, RPC, UI, author SDK, Store, or + physical deletion route is added. + +An executable GC owner cannot rely on a read-time candidate recheck: another +writer could select or pin the same revision before deletion. D2 introduces a +shared, explicitly injected guard over the three reference-writing ledgers. +It is a prerequisite for deletion, not deletion authorization. + +## Reservation Protocol + +An opt-in owner graph binds **one journal object** to the desired-state, +Instance-runtime, and Package-lifecycle ledgers. `reserve()` refuses an +unbound graph, rechecks the exact D1 candidate while holding the journal's +exclusive cross-process lock, then appends a versioned reservation. Repeating +the same operation rechecks the candidate and converges. Changed operation or +idempotency identities and another active reservation for the same revision +fail closed. A cancellation +requires the exact reservation ID, a new operation identity, and a reason; D2 +permits cancellation because it never starts physical deletion. + +Bound reference writers acquire the reservation guard **before** their owner +locks. The management service also takes it before its operation journal lock, +so its nested desired-state commit and recovery follow the same order. A +reentrant in-process guard avoids reacquiring the file lock in that nested +path. A reserved revision cannot be newly selected by desired-state commit or +update, activated or leased by the Instance runtime, or pinned by Package +retention. Unrelated revisions and reference release remain available. + +The journal replays active reservations after restart, repairs only an +incomplete tail, and rejects a complete corrupt record. Bound writers fail +closed when the reservation journal cannot be reconstructed. The operator can +inspect active reservations separately from the D1 retention projection; an +ordinary D1 candidate remains a retention candidate, not a deletion receipt. + +## Explicit Limits And Next Gate + +An older or separately constructed writer that omits the injected guard can +still write the same desired-state journal. The regression freezes this +counterexample and proves that a bound Instance runtime refuses activation of +its newly selected revision. Consequently **D2 does not authorize deletion**. +Before D3, one Product composition must bind every supported writer and prove +a minimum-version/downgrade fence excludes legacy writers; D3 must recheck +desired, Instance, Package, and reservation evidence under that authority. + +D3 must also bind the exact published Store tree/native identity, execute +rooted deletion through the Store owner, and persist a result or retryable +debt. Neither reservation cancellation nor Plugin remove may invoke private +data deletion or assert backup expiry. + +| Seam | Old caller | D2 caller and authority | +| --- | --- | --- | +| Desired selection | management service or direct desired ledger | optional gate before operation/desired locks; same desired owner | +| Instance activation/family | runtime ledger | optional gate before operation/runtime locks; same Instance owner | +| Package pin | lifecycle ledger | optional gate before Package lock; same retention owner | +| Reservation | absent | internal `package_gc_reservation.py` journal only; no Product or Store caller | + +Focused durable-ledger tests cover exact replay, cancellation, stale candidate, +new pin/selection refusal, management lock order, restart and partial-tail +repair, corrupt-log fail-closed behavior, unbound-graph refusal, and the legacy +writer counterexample. No native deletion or end-to-end GC test is claimed. diff --git a/src/loushang/harness/plugin_management/gc_fence.py b/src/loushang/harness/plugin_management/gc_fence.py new file mode 100644 index 000000000..fa59cee84 --- /dev/null +++ b/src/loushang/harness/plugin_management/gc_fence.py @@ -0,0 +1,23 @@ +"""Neutral reference-writer fence for a future executable Package GC owner.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager, nullcontext +from typing import Protocol + +from loushang.harness.plugin_management.records import PluginPackageRevisionRefV1 + + +class PluginPackageGcReferenceGatePort(Protocol): + def guard( + self, + ) -> AbstractContextManager[frozenset[PluginPackageRevisionRefV1]]: ... + + +def gc_reference_guard( + gate: PluginPackageGcReferenceGatePort | None, +) -> AbstractContextManager[frozenset[PluginPackageRevisionRefV1]]: + return nullcontext(frozenset()) if gate is None else gate.guard() + + +__all__ = ["PluginPackageGcReferenceGatePort", "gc_reference_guard"] diff --git a/src/loushang/harness/plugin_management/instance_runtime.py b/src/loushang/harness/plugin_management/instance_runtime.py index 340484405..5ab5df232 100644 --- a/src/loushang/harness/plugin_management/instance_runtime.py +++ b/src/loushang/harness/plugin_management/instance_runtime.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import ExitStack from dataclasses import dataclass, replace from pathlib import Path from typing import Protocol @@ -14,6 +15,10 @@ journal_file_lock, load_jsonl, ) +from loushang.harness.plugin_management.gc_fence import ( + PluginPackageGcReferenceGatePort, + gc_reference_guard, +) from loushang.harness.plugin_management.instance_records import ( PLUGIN_INSTANCE_RUNTIME_EVENT_CODEC, PluginInstanceActivationV1, @@ -277,6 +282,7 @@ def __init__( retirement_intents: PluginInstanceRetirementIntentSourcePort, retirement_sets: PluginInstanceRetirementSetSourcePort, security_acceptances: PluginInstanceSecurityAcceptanceSourcePort, + gc_gate: PluginPackageGcReferenceGatePort | None = None, ) -> None: # Lock sidecars are derived from these stored paths, so normalize once # before any equality check or cross-process operation gate is used. @@ -287,6 +293,7 @@ def __init__( self._retirement_sets = retirement_sets self._validate_security_acceptance_source(security_acceptances) self._security_acceptances = security_acceptances + self._gc_gate = gc_gate journal_paths = { self._path.resolve(), self._operation_path.resolve(), @@ -304,6 +311,10 @@ def __init__( def path(self) -> Path: return self._path + @property + def gc_gate(self) -> PluginPackageGcReferenceGatePort | None: + return self._gc_gate + @property def management_operation_journal_path(self) -> Path: return self._operation_path @@ -352,11 +363,15 @@ def activate_current( _require_nonempty(operation_id, name="activation operation id") _require_nonempty(idempotency_key, name="activation idempotency key") _require_nonempty(direct_host_reference, name="direct host reference") - with journal_file_lock( - self._operation_path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context( + journal_file_lock( + self._operation_path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) sources = self._load_sources() with journal_file_lock( self._path, @@ -399,6 +414,12 @@ def activate_current( self._path, "Plugin Installation has no current enabled Instance", ) + if selection.package_revision in reserved: + raise PluginInstanceRuntimeError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) self._require_not_security_accepted((selection.instance_revision_ref,)) if selection.instance_revision_ref in replayed.instances: raise _transition_error( @@ -452,11 +473,15 @@ def acquire_current_family( (holder_reference, "lease holder reference"), ): _require_nonempty(value, name=name) - with journal_file_lock( - self._operation_path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context( + journal_file_lock( + self._operation_path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) sources = self._load_sources() with journal_file_lock( self._path, @@ -501,6 +526,12 @@ def acquire_current_family( ) for installation_key in keys ) + if any(package in reserved for _, _, package in subjects): + raise PluginInstanceRuntimeError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) family = PluginInstanceLeaseFamilyV1.create( lease_kind=lease_kind, operation_id=operation_id, @@ -537,11 +568,15 @@ def derive_agent_membership( (holder_reference, "lease holder reference"), ): _require_nonempty(value, name=name) - with journal_file_lock( - self._operation_path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context( + journal_file_lock( + self._operation_path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) sources = self._load_sources() with journal_file_lock( self._path, @@ -600,6 +635,12 @@ def derive_agent_membership( ) for member in parent.family.members ) + if any(package in reserved for _, _, package in subjects): + raise PluginInstanceRuntimeError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) if any( replayed.instances[instance_ref].state not in {"ACTIVE", "DRAINING"} for _, instance_ref, _ in subjects diff --git a/src/loushang/harness/plugin_management/ledger.py b/src/loushang/harness/plugin_management/ledger.py index 5b785cb94..91351dc3c 100644 --- a/src/loushang/harness/plugin_management/ledger.py +++ b/src/loushang/harness/plugin_management/ledger.py @@ -2,6 +2,7 @@ import secrets from collections.abc import Callable +from contextlib import ExitStack from dataclasses import dataclass, replace from pathlib import Path @@ -16,6 +17,10 @@ journal_file_lock, load_jsonl, ) +from loushang.harness.plugin_management.gc_fence import ( + PluginPackageGcReferenceGatePort, + gc_reference_guard, +) from loushang.harness.plugin_management.journal_codecs import ( PLUGIN_DESIRED_STATE_JOURNAL_CODEC, PluginDesiredStateJournalTransition, @@ -96,8 +101,8 @@ class PluginDesiredStateLedger: """Durable, inert desired-selection and staged-cutover authority for PLC2. This ledger owns only management intent and Instance Revision identity. It - has no live Plugin, Product Session, Graph, registration, or package-GC - dependency. + has no live Plugin, Product Session, Graph, registration, or executable-GC + authority. An optional neutral guard can fence new references during GC. """ def __init__( @@ -105,9 +110,11 @@ def __init__( path: str | Path, *, instance_id_factory: PluginInstanceIdFactory | None = None, + gc_gate: PluginPackageGcReferenceGatePort | None = None, ) -> None: self._path = Path(path) self._instance_id_factory = instance_id_factory or _new_instance_id + self._gc_gate = gc_gate self._unlocked_durability = replace( DURABLE_LOCKED_JOURNAL, locking=False, @@ -118,6 +125,10 @@ def __init__( def path(self) -> Path: return self._path + @property + def gc_gate(self) -> PluginPackageGcReferenceGatePort | None: + return self._gc_gate + def snapshot(self) -> PluginDesiredStateSnapshotV1: with journal_file_lock( self._path, @@ -157,11 +168,15 @@ def commit( ) -> PluginDesiredStateTransitionV1: if not isinstance(mutation, PluginDesiredStateMutationV1): raise TypeError("Plugin desired-state mutation is required") - with journal_file_lock( - self._path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context( + journal_file_lock( + self._path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) replayed = self._load_and_replay_unlocked() repeated = self._repeat_result(replayed, mutation) if repeated is not None: @@ -208,6 +223,12 @@ def commit( code="invalid_plugin_lifecycle_transition", path=self._path, ) from exc + if committed.selection.package_revision in reserved: + raise PluginLifecycleError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) if fresh_instance_id is not None: owner = replayed.instance_owners.get(fresh_instance_id) if owner is not None: @@ -239,11 +260,15 @@ def commit_update( ) -> PluginDesiredStateUpdateTransitionV2: if not isinstance(mutation, PluginDesiredStateUpdateMutationV1): raise TypeError("Plugin desired-state update mutation is required") - with journal_file_lock( - self._path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context( + journal_file_lock( + self._path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) replayed = self._load_and_replay_unlocked() repeated = self._repeat_result(replayed, mutation) if repeated is not None: @@ -279,6 +304,12 @@ def commit_update( code="invalid_plugin_lifecycle_transition", path=self._path, ) from exc + if committed.selection.package_revision in reserved: + raise PluginLifecycleError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) transition = PluginDesiredStateUpdateTransitionV2( inventory_revision=head + 1, diff --git a/src/loushang/harness/plugin_management/package_gc_reservation.py b/src/loushang/harness/plugin_management/package_gc_reservation.py new file mode 100644 index 000000000..357f806ce --- /dev/null +++ b/src/loushang/harness/plugin_management/package_gc_reservation.py @@ -0,0 +1,500 @@ +"""Dark, durable PLC9D2 reservation of one retention-eligible Package revision. + +This journal has no Store or deletion capability. It fences only owner graphs +that explicitly bind the same journal to desired, Instance, and Package writers. +""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, replace +from hashlib import sha256 +from pathlib import Path +from typing import Literal, cast + +from loushang.harness.journal import ( + DURABLE_LOCKED_JOURNAL, + SORTED_UNICODE_JSONL_FORMAT, + FunctionalJournalRecordCodec, + JournalCodecError, + JournalFileError, + JournalLoadPolicy, + JsonlSnapshot, + append_jsonl_record, + journal_file_lock, + load_jsonl, +) +from loushang.harness.plugin_management.package_lifecycle import ( + PluginPackageGcCandidateV1, + PluginPackageLifecycleLedger, +) +from loushang.harness.plugin_management.records import PluginPackageRevisionRefV1 + +PLUGIN_PACKAGE_GC_RESERVATION_EVENT_VERSION = 1 +GcReservationEventKind = Literal["reserved", "cancelled"] + + +class PluginPackageGcReservationError(RuntimeError): + def __init__(self, message: str, *, code: str, path: Path) -> None: + super().__init__(message) + self.code = code + self.path = path + + +class _ReservationCodecError(JournalCodecError): + pass + + +@dataclass(frozen=True, slots=True) +class PluginPackageGcReservationEventV1: + journal_revision: int + kind: GcReservationEventKind + reservation_id: str + candidate: PluginPackageGcCandidateV1 | None + operation_id: str + idempotency_key: str + reason_code: str | None = None + event_version: int = PLUGIN_PACKAGE_GC_RESERVATION_EVENT_VERSION + + def __post_init__(self) -> None: + if type(self.journal_revision) is not int or self.journal_revision < 1: + raise ValueError("GC reservation journal revision must be positive") + if self.kind not in {"reserved", "cancelled"}: + raise ValueError("Unsupported GC reservation event kind") + if not _is_sha256(self.reservation_id): + raise ValueError("GC reservation identity is invalid") + if not self.operation_id or not self.idempotency_key: + raise ValueError("GC reservation operation identity is required") + if self.kind == "reserved": + if self.candidate is None or self.reason_code is not None: + raise ValueError("GC reservation requires only a candidate") + if self.reservation_id != _reservation_id( + self.candidate, self.operation_id, self.idempotency_key + ): + raise ValueError("GC reservation identity does not match") + elif self.candidate is not None or not self.reason_code: + raise ValueError("GC cancellation requires only a reason") + if self.event_version != PLUGIN_PACKAGE_GC_RESERVATION_EVENT_VERSION: + raise ValueError("Unsupported GC reservation event version") + + def to_dict(self) -> dict[str, object]: + candidate = self.candidate + return { + "candidate": ( + None + if candidate is None + else { + "candidateId": candidate.candidate_id, + "candidateVersion": candidate.candidate_version, + "desiredInventoryRevision": candidate.desired_inventory_revision, + "instanceRuntimeRevision": candidate.instance_runtime_revision, + "packageJournalRevision": candidate.package_journal_revision, + "packageRevision": candidate.package_revision.to_dict(), + "recoveryBarrierId": candidate.recovery_barrier_id, + } + ), + "eventVersion": self.event_version, + "idempotencyKey": self.idempotency_key, + "journalRevision": self.journal_revision, + "kind": self.kind, + "operationId": self.operation_id, + "reasonCode": self.reason_code, + "reservationId": self.reservation_id, + } + + @classmethod + def from_dict(cls, value: object) -> PluginPackageGcReservationEventV1: + try: + item = _exact_dict( + value, + { + "candidate", + "eventVersion", + "idempotencyKey", + "journalRevision", + "kind", + "operationId", + "reasonCode", + "reservationId", + }, + ) + candidate_value = item["candidate"] + candidate = ( + None if candidate_value is None else _decode_candidate(candidate_value) + ) + return cls( + journal_revision=_integer(item["journalRevision"]), + kind=_kind(item["kind"]), + reservation_id=_string(item["reservationId"]), + candidate=candidate, + operation_id=_string(item["operationId"]), + idempotency_key=_string(item["idempotencyKey"]), + reason_code=( + None if item["reasonCode"] is None else _string(item["reasonCode"]) + ), + event_version=_integer(item["eventVersion"]), + ) + except (JournalCodecError, TypeError, ValueError) as exc: + raise _ReservationCodecError( + "Invalid GC reservation record", + code="invalid_plugin_package_gc_reservation_record", + ) from exc + + +_EVENT_CODEC = FunctionalJournalRecordCodec( + encoder=PluginPackageGcReservationEventV1.to_dict, + decoder=PluginPackageGcReservationEventV1.from_dict, +) + + +@dataclass(frozen=True, slots=True) +class PluginPackageGcReservationSnapshotV1: + journal_revision: int + active: tuple[PluginPackageGcReservationEventV1, ...] + + +class PluginPackageGcReservationJournal: + """One explicit gate shared by every writer in an opted-in owner graph.""" + + def __init__(self, path: str | Path) -> None: + self._path = Path(path).resolve() + self._thread_lock = threading.RLock() + self._lock_depth = 0 + self._unlocked_durability = replace(DURABLE_LOCKED_JOURNAL, locking=False) + self._load_policy = JournalLoadPolicy(partial_tail="repair") + + @property + def path(self) -> Path: + return self._path + + @contextmanager + def guard(self) -> Iterator[frozenset[PluginPackageRevisionRefV1]]: + with self._thread_lock: + if self._lock_depth: + self._lock_depth += 1 + try: + yield self._active_packages_unlocked() + finally: + self._lock_depth -= 1 + return + with journal_file_lock( + self._path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ): + self._lock_depth = 1 + try: + yield self._active_packages_unlocked() + finally: + self._lock_depth = 0 + + def snapshot(self) -> PluginPackageGcReservationSnapshotV1: + with self.guard(): + events, active = self._replay_unlocked() + return PluginPackageGcReservationSnapshotV1( + journal_revision=len(events), + active=tuple( + sorted(active.values(), key=lambda item: item.reservation_id) + ), + ) + + def reserve( + self, + candidate: PluginPackageGcCandidateV1, + *, + lifecycle: PluginPackageLifecycleLedger, + operation_id: str, + idempotency_key: str, + ) -> PluginPackageGcReservationEventV1: + if not isinstance(candidate, PluginPackageGcCandidateV1): + raise TypeError("GC candidate is required") + if not isinstance(lifecycle, PluginPackageLifecycleLedger): + raise TypeError("Package lifecycle owner is required") + if not lifecycle.gc_reservation_graph_bound_to(self): + raise self._error( + "GC owner graph is not fully fenced", "plugin_package_gc_graph_unbound" + ) + if self._path in { + path.resolve() for path in lifecycle.gc_reservation_peer_journal_paths + }: + raise self._error( + "GC journal overlaps an owner", "plugin_package_gc_graph_unbound" + ) + _require_identity(operation_id, idempotency_key) + with self.guard(): + events, active = self._replay_unlocked() + repeated = _repeated(events, operation_id, idempotency_key, path=self._path) + if repeated is not None: + if repeated.kind != "reserved" or repeated.candidate != candidate: + raise self._error( + "GC operation identity was reused", "plugin_package_gc_conflict" + ) + if repeated.reservation_id not in active: + raise self._error( + "GC reservation was cancelled", "plugin_package_gc_stale" + ) + lifecycle.recheck_gc_candidate(candidate) + return repeated + if any( + item.candidate is not None + and item.candidate.package_revision == candidate.package_revision + for item in active.values() + ): + raise self._error( + "Package revision already reserved", "plugin_package_gc_conflict" + ) + lifecycle.recheck_gc_candidate(candidate) + event = PluginPackageGcReservationEventV1( + journal_revision=len(events) + 1, + kind="reserved", + reservation_id=_reservation_id( + candidate, operation_id, idempotency_key + ), + candidate=candidate, + operation_id=operation_id, + idempotency_key=idempotency_key, + ) + self._append_unlocked(event) + return event + + def cancel( + self, + reservation_id: str, + *, + operation_id: str, + idempotency_key: str, + reason_code: str, + ) -> PluginPackageGcReservationEventV1: + _require_identity(operation_id, idempotency_key) + if not _is_sha256(reservation_id) or not reason_code: + raise ValueError( + "Exact GC reservation and cancellation reason are required" + ) + with self.guard(): + events, active = self._replay_unlocked() + repeated = _repeated(events, operation_id, idempotency_key, path=self._path) + if repeated is not None: + if ( + repeated.kind != "cancelled" + or repeated.reservation_id != reservation_id + or repeated.reason_code != reason_code + ): + raise self._error( + "GC operation identity was reused", "plugin_package_gc_conflict" + ) + return repeated + if reservation_id not in active: + raise self._error( + "GC reservation is not active", "plugin_package_gc_stale" + ) + event = PluginPackageGcReservationEventV1( + journal_revision=len(events) + 1, + kind="cancelled", + reservation_id=reservation_id, + candidate=None, + operation_id=operation_id, + idempotency_key=idempotency_key, + reason_code=reason_code, + ) + self._append_unlocked(event) + return event + + def _active_packages_unlocked(self) -> frozenset[PluginPackageRevisionRefV1]: + _, active = self._replay_unlocked() + return frozenset( + item.candidate.package_revision + for item in active.values() + if item.candidate is not None + ) + + def _replay_unlocked( + self, + ) -> tuple[ + tuple[PluginPackageGcReservationEventV1, ...], + dict[str, PluginPackageGcReservationEventV1], + ]: + if not self._path.exists(): + return (), {} + try: + loaded: JsonlSnapshot[None, PluginPackageGcReservationEventV1] = load_jsonl( + self._path, + record_codec=_EVENT_CODEC, + format_profile=SORTED_UNICODE_JSONL_FORMAT, + durability=self._unlocked_durability, + load_policy=self._load_policy, + ) + events = loaded.records + except JournalFileError as exc: + raise self._error( + "GC reservation journal is corrupt", "plugin_package_gc_journal_corrupt" + ) from exc + active: dict[str, PluginPackageGcReservationEventV1] = {} + active_packages: set[PluginPackageRevisionRefV1] = set() + operations: set[str] = set() + idempotency: set[str] = set() + for revision, event in enumerate(events, start=1): + if ( + event.journal_revision != revision + or event.operation_id in operations + or event.idempotency_key in idempotency + ): + raise self._error( + "GC reservation history is corrupt", + "plugin_package_gc_journal_corrupt", + ) + if event.kind == "reserved": + candidate = event.candidate + if ( + candidate is None + or event.reservation_id in active + or candidate.package_revision in active_packages + ): + raise self._error( + "GC reservation is duplicated", + "plugin_package_gc_journal_corrupt", + ) + active[event.reservation_id] = event + active_packages.add(candidate.package_revision) + elif event.reservation_id not in active: + raise self._error( + "GC cancellation lacks a reservation", + "plugin_package_gc_journal_corrupt", + ) + else: + prior = active.pop(event.reservation_id) + if prior.candidate is None: + raise self._error( + "GC reservation lacks its candidate", + "plugin_package_gc_journal_corrupt", + ) + active_packages.remove(prior.candidate.package_revision) + operations.add(event.operation_id) + idempotency.add(event.idempotency_key) + return events, active + + def _append_unlocked(self, event: PluginPackageGcReservationEventV1) -> None: + try: + append_jsonl_record( + self._path, + event, + record_codec=_EVENT_CODEC, + format_profile=SORTED_UNICODE_JSONL_FORMAT, + durability=self._unlocked_durability, + ) + except JournalFileError as exc: + raise self._error( + "GC reservation append failed", "plugin_package_gc_journal_corrupt" + ) from exc + + def _error(self, message: str, code: str) -> PluginPackageGcReservationError: + return PluginPackageGcReservationError(message, code=code, path=self._path) + + +def _repeated( + events: tuple[PluginPackageGcReservationEventV1, ...], + operation_id: str, + idempotency_key: str, + *, + path: Path, +) -> PluginPackageGcReservationEventV1 | None: + by_operation = next( + (item for item in events if item.operation_id == operation_id), None + ) + by_key = next( + (item for item in events if item.idempotency_key == idempotency_key), None + ) + if by_operation != by_key: + raise PluginPackageGcReservationError( + "GC operation and idempotency identities diverge", + code="plugin_package_gc_conflict", + path=path, + ) + return by_operation + + +def _reservation_id( + candidate: PluginPackageGcCandidateV1, + operation_id: str, + idempotency_key: str, +) -> str: + payload = json.dumps( + [candidate.candidate_id, operation_id, idempotency_key], + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + return sha256(b"plugin-package-gc-reservation-v1\0" + payload).hexdigest() + + +def _decode_candidate(value: object) -> PluginPackageGcCandidateV1: + item = _exact_dict( + value, + { + "candidateId", + "candidateVersion", + "desiredInventoryRevision", + "instanceRuntimeRevision", + "packageJournalRevision", + "packageRevision", + "recoveryBarrierId", + }, + ) + return PluginPackageGcCandidateV1( + candidate_id=_string(item["candidateId"]), + candidate_version=_integer(item["candidateVersion"]), + desired_inventory_revision=_integer(item["desiredInventoryRevision"]), + instance_runtime_revision=_integer(item["instanceRuntimeRevision"]), + package_journal_revision=_integer(item["packageJournalRevision"]), + package_revision=PluginPackageRevisionRefV1.from_dict(item["packageRevision"]), + recovery_barrier_id=_string(item["recoveryBarrierId"]), + ) + + +def _exact_dict(value: object, fields: set[str]) -> dict[str, object]: + if type(value) is not dict or set(value) != fields: + raise ValueError("GC reservation record fields are invalid") + return value + + +def _integer(value: object) -> int: + if type(value) is not int: + raise ValueError("GC reservation integer field is invalid") + return value + + +def _string(value: object) -> str: + if type(value) is not str or not value: + raise ValueError("GC reservation string field is invalid") + return value + + +def _kind(value: object) -> GcReservationEventKind: + if value not in {"reserved", "cancelled"}: + raise ValueError("GC reservation event kind is invalid") + return cast(GcReservationEventKind, value) + + +def _is_sha256(value: object) -> bool: + return ( + type(value) is str + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _require_identity(operation_id: str, idempotency_key: str) -> None: + if type(operation_id) is not str or not operation_id: + raise ValueError("GC operation id is required") + if type(idempotency_key) is not str or not idempotency_key: + raise ValueError("GC idempotency key is required") + + +__all__ = [ + "PLUGIN_PACKAGE_GC_RESERVATION_EVENT_VERSION", + "PluginPackageGcReservationError", + "PluginPackageGcReservationEventV1", + "PluginPackageGcReservationJournal", + "PluginPackageGcReservationSnapshotV1", +] diff --git a/src/loushang/harness/plugin_management/package_lifecycle.py b/src/loushang/harness/plugin_management/package_lifecycle.py index a5b50688d..58d520c38 100644 --- a/src/loushang/harness/plugin_management/package_lifecycle.py +++ b/src/loushang/harness/plugin_management/package_lifecycle.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import ExitStack from dataclasses import dataclass, replace from hashlib import sha256 from pathlib import Path @@ -15,6 +16,10 @@ journal_file_lock, load_jsonl, ) +from loushang.harness.plugin_management.gc_fence import ( + PluginPackageGcReferenceGatePort, + gc_reference_guard, +) from loushang.harness.plugin_management.instance_records import ( PluginInstanceLeaseFamilyReleaseV1, PluginInstanceLeaseFamilyV1, @@ -212,9 +217,9 @@ class PluginPackageRetentionSnapshotV1: gc_candidate: PluginPackageGcCandidateV1 | None def __post_init__(self) -> None: - if self.desired_installations != tuple(sorted(self.desired_installations)) or len( - self.desired_installations - ) != len(set(self.desired_installations)): + if self.desired_installations != tuple( + sorted(self.desired_installations) + ) or len(self.desired_installations) != len(set(self.desired_installations)): raise ValueError( "Plugin Package desired Installations must be sorted and unique" ) @@ -232,9 +237,7 @@ def __post_init__(self) -> None: ): if values != tuple(sorted(values)) or len(values) != len(set(values)): raise ValueError(f"Plugin Package {name} must be sorted and unique") - if not set(self.terminal_failure_cleanup_ids).issubset( - self.open_cleanup_ids - ): + if not set(self.terminal_failure_cleanup_ids).issubset(self.open_cleanup_ids): raise ValueError("Terminal cleanup failure must retain its lease") blocked = any( ( @@ -246,9 +249,7 @@ def __post_init__(self) -> None: self.terminal_failure_cleanup_ids, ) ) - if (self.gc_candidate is not None) != ( - self.recovery_complete and not blocked - ): + if (self.gc_candidate is not None) != (self.recovery_complete and not blocked): raise ValueError("Plugin Package GC candidate contradicts retention") if ( self.gc_candidate is not None @@ -277,7 +278,9 @@ def __post_init__(self) -> None: and self.recovery_barrier.startup_id != self.startup_id ): raise ValueError("Plugin Package recovery barrier startup does not match") - if self.open_pins != tuple(sorted(self.open_pins, key=lambda item: item.pin_id)): + if self.open_pins != tuple( + sorted(self.open_pins, key=lambda item: item.pin_id) + ): raise ValueError("Open Plugin Package pins must be sorted") if len({item.pin_id for item in self.open_pins}) != len(self.open_pins): raise ValueError("Open Plugin Package pins must be unique") @@ -290,7 +293,9 @@ def __post_init__(self) -> None: ): raise ValueError("Plugin cleanup task snapshots must be unique") if self.packages != tuple( - sorted(self.packages, key=lambda item: _package_sort_key(item.package_revision)) + sorted( + self.packages, key=lambda item: _package_sort_key(item.package_revision) + ) ): raise ValueError("Plugin Package retention snapshots must be sorted") if len({item.package_revision for item in self.packages}) != len(self.packages): @@ -360,6 +365,7 @@ def __init__( desired_state: PluginPackageDesiredStateSourcePort, instance_runtime: PluginPackageInstanceRuntimeSourcePort, retirement_sets: PluginPackageRetirementSetSourcePort, + gc_gate: PluginPackageGcReferenceGatePort | None = None, ) -> None: self._path = Path(path) _require_nonempty(startup_id, name="Plugin Package startup id") @@ -367,6 +373,7 @@ def __init__( self._desired_state = desired_state self._instance_runtime = instance_runtime self._retirement_sets = retirement_sets + self._gc_gate = gc_gate paths = { self._path.resolve(), desired_state.path.resolve(), @@ -382,6 +389,19 @@ def __init__( def path(self) -> Path: return self._path + @property + def gc_gate(self) -> PluginPackageGcReferenceGatePort | None: + return self._gc_gate + + def gc_reservation_graph_bound_to( + self, gate: PluginPackageGcReferenceGatePort + ) -> bool: + return ( + self._gc_gate is gate + and getattr(self._desired_state, "gc_gate", None) is gate + and getattr(self._instance_runtime, "gc_gate", None) is gate + ) + @property def startup_id(self) -> str: return self._startup_id @@ -392,6 +412,21 @@ def instance_runtime_journal_path(self) -> Path: return self._instance_runtime.path + @property + def gc_reservation_peer_journal_paths(self) -> tuple[Path, ...]: + operation_path = getattr( + self._instance_runtime, "management_operation_journal_path", None + ) + if not isinstance(operation_path, Path): + raise TypeError("GC reservation requires the exact Instance runtime owner") + return ( + self._path, + self._desired_state.path, + self._instance_runtime.path, + self._retirement_sets.path, + operation_path, + ) + def acquire_pin( self, package_revision: PluginPackageRevisionRefV1, @@ -410,12 +445,16 @@ def acquire_pin( idempotency_key=idempotency_key, holder_reference=holder_reference, ) - with self._exclusive_lock(): + with ExitStack() as locks: + reserved = locks.enter_context(gc_reference_guard(self._gc_gate)) + locks.enter_context(self._exclusive_lock()) replayed = self._load_and_replay_unlocked() repeated = self._existing_operation(replayed, pin) if repeated is not None: if repeated.pin != pin: - raise _conflict(self._path, "Plugin Package pin identity was reused") + raise _conflict( + self._path, "Plugin Package pin identity was reused" + ) mutable = replayed.pins[pin.pin_id] if mutable.release is not None: raise _transition_error( @@ -425,6 +464,12 @@ def acquire_pin( return pin if pin.pin_id in replayed.pins: raise _conflict(self._path, "Plugin Package pin id was reused") + if package_revision in reserved: + raise PluginPackageLifecycleError( + "Plugin Package Revision is reserved for GC", + code="plugin_package_gc_reserved", + path=self._path, + ) self._append_and_apply_unlocked(replayed, pin) return pin @@ -610,10 +655,13 @@ def record_repair_decision( self._path, "Plugin cleanup repair sequence is not contiguous", ) - if _derive_cleanup_state( - tuple(current.attempts), - tuple(current.repair_decisions), - ) != "terminal_failure": + if ( + _derive_cleanup_state( + tuple(current.attempts), + tuple(current.repair_decisions), + ) + != "terminal_failure" + ): raise _transition_error( self._path, "Plugin cleanup repair requires terminal failure", @@ -1065,10 +1113,13 @@ def _apply_event( raise _corrupt(path, "Plugin cleanup repair has no task") if decision.repair_sequence != len(current_cleanup.repair_decisions) + 1: raise _corrupt(path, "Plugin cleanup repair is not contiguous") - if _derive_cleanup_state( - tuple(current_cleanup.attempts), - tuple(current_cleanup.repair_decisions), - ) != "terminal_failure": + if ( + _derive_cleanup_state( + tuple(current_cleanup.attempts), + tuple(current_cleanup.repair_decisions), + ) + != "terminal_failure" + ): raise _corrupt(path, "Plugin cleanup repair transition is invalid") current_cleanup.repair_decisions.append(decision) else: @@ -1149,17 +1200,17 @@ def _validate_cleanup_evidence( ) if len(repairs) > len(terminal_attempts): raise ValueError("Plugin cleanup repair has no terminal attempt") - if any( - attempt.disposition == "succeeded" for attempt in attempts[:-1] - ): + if any(attempt.disposition == "succeeded" for attempt in attempts[:-1]): raise ValueError("Successful Plugin cleanup attempt must be final") for index, terminal_attempt in enumerate(terminal_attempts): has_later_attempt = terminal_attempt.attempt < len(attempts) repair = repairs[index] if index < len(repairs) else None if has_later_attempt and (repair is None or repair.action != "retry"): raise ValueError("Plugin cleanup retry lacks a repair decision") - if repair is not None and repair.action == "safe_abandon" and ( - has_later_attempt or index != len(repairs) - 1 + if ( + repair is not None + and repair.action == "safe_abandon" + and (has_later_attempt or index != len(repairs) - 1) ): raise ValueError("Safe-abandoned Plugin cleanup cannot continue") @@ -1267,8 +1318,7 @@ def _retention_snapshot( family.family_id for family in sources.runtime_snapshot.open_families if any( - member.package_revision == package_revision - for member in family.members + member.package_revision == package_revision for member in family.members ) ) ) @@ -1286,9 +1336,7 @@ def _retention_snapshot( if current.task.package_revision == package_revision ) open_cleanup_ids = tuple( - sorted( - item.task.cleanup_id for item in cleanup_snapshots if item.lease_open - ) + sorted(item.task.cleanup_id for item in cleanup_snapshots if item.lease_open) ) terminal_failure_cleanup_ids = tuple( sorted( @@ -1373,8 +1421,7 @@ def _instance_coordination_at( state = "ACTIVE" elif ( event.retirement_intent is not None - and event.retirement_intent.instance_revision_ref - == instance_revision_ref + and event.retirement_intent.instance_revision_ref == instance_revision_ref ): state = "DRAINING" coordination_id = event.retirement_intent.retirement_id @@ -1440,8 +1487,10 @@ def _require_nonempty(value: str, *, name: str) -> None: def _require_sha256(value: str, *, name: str) -> None: - if not isinstance(value, str) or len(value) != 64 or any( - character not in "0123456789abcdef" for character in value + if ( + not isinstance(value, str) + or len(value) != 64 + or any(character not in "0123456789abcdef" for character in value) ): raise ValueError(f"{name} must be a lowercase SHA-256 digest") diff --git a/src/loushang/harness/plugin_management/service.py b/src/loushang/harness/plugin_management/service.py index 8973fa536..2cd5daabb 100644 --- a/src/loushang/harness/plugin_management/service.py +++ b/src/loushang/harness/plugin_management/service.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import ExitStack from dataclasses import dataclass, replace from pathlib import Path from typing import Protocol @@ -14,6 +15,7 @@ journal_file_lock, load_jsonl, ) +from loushang.harness.plugin_management.gc_fence import gc_reference_guard from loushang.harness.plugin_management.journal_codecs import ( PLUGIN_MANAGEMENT_OPERATION_JOURNAL_CODEC, PluginDesiredStateJournalTransition, @@ -187,11 +189,17 @@ def submit( command, (PluginManagementCommandV1, PluginManagementUpdateCommandV2) ): raise TypeError("Plugin management command is required") - with journal_file_lock( - self._path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + locks.enter_context( + gc_reference_guard(getattr(self._desired_state, "gc_gate", None)) + ) + locks.enter_context( + journal_file_lock( + self._path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) replayed = self._load_and_replay_unlocked() existing = self._existing_operation(replayed, command) if existing is not None: @@ -226,11 +234,17 @@ def submit( def recover(self) -> tuple[PluginManagementOperationEvent, ...]: """Recover accepted/running operations in original acceptance order.""" - with journal_file_lock( - self._path, - "exclusive", - lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, - ): + with ExitStack() as locks: + locks.enter_context( + gc_reference_guard(getattr(self._desired_state, "gc_gate", None)) + ) + locks.enter_context( + journal_file_lock( + self._path, + "exclusive", + lock_suffix=DURABLE_LOCKED_JOURNAL.lock_suffix, + ) + ) replayed = self._load_and_replay_unlocked() pending = tuple( sorted( @@ -389,9 +403,7 @@ def _execute_update_unlocked( mutation, preparation_error = self._prepare_update_mutation(command, fence) if preparation_error is not None: - result = PluginUpdateOperationResultV2.failed( - error_code=preparation_error - ) + result = PluginUpdateOperationResultV2.failed(error_code=preparation_error) else: if mutation is None: raise AssertionError("Prepared Plugin update mutation is missing") @@ -526,8 +538,7 @@ def _reject_busy_installation( key: PluginInstallationKeyV1, ) -> None: if any( - event.status != "terminal" - and _installation_key(event.command) == key + event.status != "terminal" and _installation_key(event.command) == key for event in replayed.latest_by_operation.values() ): raise PluginManagementError( @@ -540,14 +551,12 @@ def _load_and_replay_unlocked(self) -> _ReplayedOperations: if not self._path.exists(): return _empty_replay() try: - snapshot: JsonlSnapshot[None, PluginManagementOperationEvent] = ( - load_jsonl( - self._path, - record_codec=PLUGIN_MANAGEMENT_OPERATION_JOURNAL_CODEC, - format_profile=SORTED_UNICODE_JSONL_FORMAT, - durability=self._unlocked_durability, - load_policy=self._load_policy, - ) + snapshot: JsonlSnapshot[None, PluginManagementOperationEvent] = load_jsonl( + self._path, + record_codec=PLUGIN_MANAGEMENT_OPERATION_JOURNAL_CODEC, + format_profile=SORTED_UNICODE_JSONL_FORMAT, + durability=self._unlocked_durability, + load_policy=self._load_policy, ) except JournalFileError as exc: code = ( @@ -584,12 +593,10 @@ def _validate_terminal_results(self, replayed: _ReplayedOperations) -> None: } retirement_snapshot = self._retirement_intents.snapshot() retirement_by_operation = { - intent.source_operation_id: intent - for intent in retirement_snapshot.intents + intent.source_operation_id: intent for intent in retirement_snapshot.intents } retirement_by_id = { - intent.retirement_id: intent - for intent in retirement_snapshot.intents + intent.retirement_id: intent for intent in retirement_snapshot.intents } for operation_id, intent in retirement_by_operation.items(): if desired_by_operation.get(operation_id) != intent.source_transition: @@ -621,9 +628,7 @@ def _validate_terminal_results(self, replayed: _ReplayedOperations) -> None: expected_retirement = ( None if desired is None else retirement_intent_for_transition(desired) ) - actual_retirement = retirement_by_operation.get( - event.command.operation_id - ) + actual_retirement = retirement_by_operation.get(event.command.operation_id) if committed and actual_retirement != expected_retirement: raise _retirement_corrupt( self._retirement_intents.path, @@ -637,8 +642,7 @@ def _validate_terminal_results(self, replayed: _ReplayedOperations) -> None: retirement_set_snapshot = self._retirement_sets.snapshot() retirement_sets = { - item.intent.retirement_id: item - for item in retirement_set_snapshot.sets + item.intent.retirement_id: item for item in retirement_set_snapshot.sets } for retirement_id, retirement_set in retirement_sets.items(): if retirement_by_id.get(retirement_id) != retirement_set.intent: diff --git a/tests/harness/plugin_management/test_package_lifecycle.py b/tests/harness/plugin_management/test_package_lifecycle.py index 6078e5af5..2214677e8 100644 --- a/tests/harness/plugin_management/test_package_lifecycle.py +++ b/tests/harness/plugin_management/test_package_lifecycle.py @@ -30,15 +30,23 @@ PluginInstanceRevocationV1, ) from loushang.harness.plugin_management.instance_runtime import ( + PluginInstanceRuntimeError, PluginInstanceRuntimeLedger, PluginInstanceRuntimeSnapshotV1, ) -from loushang.harness.plugin_management.ledger import PluginDesiredStateLedger +from loushang.harness.plugin_management.ledger import ( + PluginDesiredStateLedger, + PluginLifecycleError, +) from loushang.harness.plugin_management.operations import ( PluginManagementAction, PluginManagementCommandV1, ) from loushang.harness.plugin_management.package_gc import PluginPackageGcReadModel +from loushang.harness.plugin_management.package_gc_reservation import ( + PluginPackageGcReservationError, + PluginPackageGcReservationJournal, +) from loushang.harness.plugin_management.package_lifecycle import ( PluginPackageGcCandidateV1, PluginPackageLifecycleError, @@ -91,6 +99,23 @@ def test_plc9d1_gc_projection_cannot_import_store_or_deletion_authority() -> Non assert not any(isinstance(node, ast.Import) for node in ast.walk(tree)) +def test_plc9d2_gc_reservation_has_no_store_or_deletion_import() -> None: + source = Path("src/loushang/harness/plugin_management/package_gc_reservation.py") + tree = ast.parse(source.read_text(encoding="utf-8")) + imported = { + node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) + } + assert not any( + module is not None and module.startswith("loushang.harness.resources.packages") + for module in imported + ) + assert not any( + isinstance(node, ast.Import) + and any(alias.name in {"os", "shutil"} for alias in node.names) + for node in ast.walk(tree) + ) + + def test_continuity_publication_security_close_hands_off_package_cleanup( tmp_path: Path, ) -> None: @@ -660,6 +685,167 @@ def test_gc_candidate_requires_every_source_zero_and_revision_recheck( assert replacement.candidate_id != candidate.candidate_id +def test_gc_reservation_fences_new_references_and_survives_restart( + tmp_path: Path, +) -> None: + journal = PluginPackageGcReservationJournal(tmp_path / "gc-reservations.jsonl") + context = _context(tmp_path, gc_gate=journal) + key = _key("plugin.a") + context.service.submit(_command(key, "install", revision=0, operation=1)) + context.service.submit(_command(key, "remove", revision=1, operation=2)) + context.packages.complete_startup_recovery( + operation_id="recover-gc", + idempotency_key="recover-gc-request", + recovery_reference="recovery:gc", + ) + candidate = context.packages.gc_candidates()[0] + reservation = journal.reserve( + candidate, + lifecycle=context.packages, + operation_id="reserve-gc", + idempotency_key="reserve-gc-request", + ) + assert ( + journal.reserve( + candidate, + lifecycle=context.packages, + operation_id="reserve-gc", + idempotency_key="reserve-gc-request", + ) + == reservation + ) + with pytest.raises(PluginPackageGcReservationError) as competing: + journal.reserve( + candidate, + lifecycle=context.packages, + operation_id="reserve-competing", + idempotency_key="reserve-competing-request", + ) + assert competing.value.code == "plugin_package_gc_conflict" + + with pytest.raises(PluginPackageLifecycleError) as pin_error: + context.packages.acquire_pin( + candidate.package_revision, + pin_kind="forensic_retention", + operation_id="pin-during-gc", + idempotency_key="pin-during-gc-request", + holder_reference="forensic:gc", + ) + assert pin_error.value.code == "plugin_package_gc_reserved" + with pytest.raises(PluginLifecycleError) as install_error: + context.desired.commit(_mutation(key, "install", revision=2, operation=3)) + assert install_error.value.code == "plugin_package_gc_reserved" + with pytest.raises(PluginLifecycleError) as service_error: + context.service.submit(_command(key, "install", revision=2, operation=4)) + assert service_error.value.code == "plugin_package_gc_reserved" + + # A legacy writer without the injected gate can still select the revision. + # The bound runtime refuses to turn that stale selection into a new Instance. + legacy_writer = PluginDesiredStateLedger(context.desired.path) + legacy_writer.commit(_mutation(key, "install", revision=2, operation=5)) + legacy_writer.commit(_mutation(key, "enable", revision=3, operation=6)) + with pytest.raises( + PluginInstanceRuntimeError, match="reserved for GC" + ) as runtime_error: + context.runtime.activate_current( + key, + operation_id="activate-during-gc", + idempotency_key="activate-during-gc-request", + direct_host_reference="host:gc", + ) + assert runtime_error.value.code == "plugin_package_gc_reserved" + + with journal.path.open("ab") as handle: + handle.write(b'{"eventVersion":') + reopened = PluginPackageGcReservationJournal(journal.path) + assert reopened.snapshot().active == (reservation,) + restarted = _context(tmp_path, startup_id="startup-reopened", gc_gate=reopened) + with pytest.raises(PluginPackageLifecycleError) as restarted_error: + restarted.packages.acquire_pin( + candidate.package_revision, + pin_kind="dependency_lock", + operation_id="pin-after-restart", + idempotency_key="pin-after-restart-request", + holder_reference="dependency:gc", + ) + assert restarted_error.value.code == "plugin_package_gc_reserved" + cancelled = reopened.cancel( + reservation.reservation_id, + operation_id="cancel-gc", + idempotency_key="cancel-gc-request", + reason_code="operator.cancelled", + ) + assert ( + reopened.cancel( + reservation.reservation_id, + operation_id="cancel-gc", + idempotency_key="cancel-gc-request", + reason_code="operator.cancelled", + ) + == cancelled + ) + assert reopened.snapshot().active == () + restarted.packages.acquire_pin( + candidate.package_revision, + pin_kind="dependency_lock", + operation_id="pin-after-cancel", + idempotency_key="pin-after-cancel-request", + holder_reference="dependency:gc", + ) + with pytest.raises(PluginPackageLifecycleError): + reopened.reserve( + candidate, + lifecycle=restarted.packages, + operation_id="reserve-stale", + idempotency_key="reserve-stale-request", + ) + + +def test_gc_reservation_refuses_unbound_graph(tmp_path: Path) -> None: + context = _context(tmp_path) + key = _key("plugin.a") + context.service.submit(_command(key, "install", revision=0, operation=1)) + context.service.submit(_command(key, "remove", revision=1, operation=2)) + context.packages.complete_startup_recovery( + operation_id="recover-gc", + idempotency_key="recover-gc-request", + recovery_reference="recovery:gc", + ) + journal = PluginPackageGcReservationJournal(tmp_path / "gc-reservations.jsonl") + with pytest.raises(PluginPackageGcReservationError) as caught: + journal.reserve( + context.packages.gc_candidates()[0], + lifecycle=context.packages, + operation_id="reserve-unbound", + idempotency_key="reserve-unbound-request", + ) + assert caught.value.code == "plugin_package_gc_graph_unbound" + assert not journal.path.exists() + + +def test_gc_reservation_corruption_fails_closed_for_bound_writers( + tmp_path: Path, +) -> None: + journal = PluginPackageGcReservationJournal(tmp_path / "gc-reservations.jsonl") + context = _context(tmp_path, gc_gate=journal) + journal.path.write_bytes(b"{}\n") + key = _key("plugin.a") + with pytest.raises(PluginPackageGcReservationError) as desired_error: + context.desired.commit(_mutation(key, "install", revision=0, operation=1)) + with pytest.raises(PluginPackageGcReservationError) as pin_error: + context.packages.acquire_pin( + _package(key.plugin_id, "a"), + pin_kind="dependency_lock", + operation_id="pin-corrupt", + idempotency_key="pin-corrupt-request", + holder_reference="dependency:corrupt", + ) + assert desired_error.value.code == "plugin_package_gc_journal_corrupt" + assert pin_error.value.code == "plugin_package_gc_journal_corrupt" + assert context.desired.snapshot().inventory_revision == 0 + assert context.packages.snapshot().journal_revision == 0 + + def test_security_cleanup_uses_revocation_without_graceful_target( tmp_path: Path, ) -> None: @@ -837,11 +1023,17 @@ def release_family(self, release): return self._runtime.release_family(release) -def _context(tmp_path: Path, *, startup_id: str = "startup-1") -> _Context: +def _context( + tmp_path: Path, + *, + startup_id: str = "startup-1", + gc_gate: PluginPackageGcReservationJournal | None = None, +) -> _Context: operation_path = tmp_path / "operations.jsonl" desired = PluginDesiredStateLedger( tmp_path / "desired.jsonl", instance_id_factory=_instance_id_factory(), + gc_gate=gc_gate, ) intents = PluginRetirementIntentLedger(tmp_path / "intents.jsonl") sets = PluginRetirementSetLedger( @@ -865,6 +1057,7 @@ def _context(tmp_path: Path, *, startup_id: str = "startup-1") -> _Context: retirement_intents=intents, retirement_sets=sets, security_acceptances=security_acceptances, + gc_gate=gc_gate, ) packages = PluginPackageLifecycleLedger( tmp_path / "packages.jsonl", @@ -872,6 +1065,7 @@ def _context(tmp_path: Path, *, startup_id: str = "startup-1") -> _Context: desired_state=desired, instance_runtime=runtime, retirement_sets=sets, + gc_gate=gc_gate, ) return _Context( desired,