From dfcba1112382cfd2f5faf343eeb185a1e962e5e3 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Mon, 2 Nov 2026 09:55:53 -0600 Subject: [PATCH] fix(governance): reconcile expired released reservation claims --- .echo/sdk.json | 1 + .../certforge-capability-surface.v2.json | 1 + contracts/certforge-sdk-capabilities.v1.json | 32 ++- docs/RELEASED_RESERVATION_RECONCILIATION.md | 95 ++++++++ scripts/certification_journey.py | 4 +- scripts/master_acceptance.py | 11 +- scripts/register_certforge_caps.sql | 2 + scripts/register_certforge_sdk_schemas.sql | 5 +- src/echo_certification_forge/service.py | 21 ++ src/echo_certification_forge/subscriber.py | 117 +++++++++- tests/test_echo_sdk_manifest.py | 2 +- tests/test_master_acceptance.py | 18 +- ...est_released_reservation_reconciliation.py | 211 ++++++++++++++++++ tests/test_sdk_capability_contract.py | 2 +- 14 files changed, 510 insertions(+), 12 deletions(-) create mode 100644 docs/RELEASED_RESERVATION_RECONCILIATION.md create mode 100644 tests/test_released_reservation_reconciliation.py diff --git a/.echo/sdk.json b/.echo/sdk.json index f856e69..f9d4fb1 100644 --- a/.echo/sdk.json +++ b/.echo/sdk.json @@ -29,6 +29,7 @@ "echo.certforge.evidence", "echo.certforge.findings", "echo.certforge.governance.get", + "echo.certforge.governance.reconcile_reservation", "echo.certforge.governance.update", "echo.certforge.health", "echo.certforge.list_runs", diff --git a/contracts/certforge-capability-surface.v2.json b/contracts/certforge-capability-surface.v2.json index a1851fe..85c5d22 100644 --- a/contracts/certforge-capability-surface.v2.json +++ b/contracts/certforge-capability-surface.v2.json @@ -34,6 +34,7 @@ "POST /v1/subscriber/subscription/cancel": "echo.certforge.subscription.cancel", "GET /v1/subscriber/governance": "echo.certforge.governance.get", "PUT /v1/subscriber/governance": "echo.certforge.governance.update", + "POST /v1/subscriber/certifications/{run_id}/reconcile-reservation": "echo.certforge.governance.reconcile_reservation", "POST /v1/subscriber/policy-packs": "echo.certforge.policy_packs.create", "GET /v1/subscriber/policy-packs": "echo.certforge.policy_packs.list", "POST /v1/subscriber/private-workers": "echo.certforge.private_workers.create", diff --git a/contracts/certforge-sdk-capabilities.v1.json b/contracts/certforge-sdk-capabilities.v1.json index 66e2950..65125cc 100644 --- a/contracts/certforge-sdk-capabilities.v1.json +++ b/contracts/certforge-sdk-capabilities.v1.json @@ -871,6 +871,36 @@ "type": "object" } }, + "echo.certforge.governance.reconcile_reservation": { + "input_schema": { + "additionalProperties": false, + "properties": { + "command": { + "maxLength": 64, + "minLength": 1, + "type": "string" + }, + "reason": { + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$", + "type": "string" + }, + "run_id": { + "type": "string" + } + }, + "required": [ + "command", + "reason", + "run_id" + ], + "type": "object" + }, + "operation": "POST /v1/subscriber/certifications/{run_id}/reconcile-reservation", + "output_schema": { + "additionalProperties": true, + "type": "object" + } + }, "echo.certforge.governance.update": { "input_schema": { "additionalProperties": false, @@ -2322,7 +2352,7 @@ } } }, - "capability_count": 61, + "capability_count": 62, "policy": "Every Certification Forge SDK action has a command-bearing strict input schema and an output schema derived from its production route contract.", "schema_version": "1.0.0" } diff --git a/docs/RELEASED_RESERVATION_RECONCILIATION.md b/docs/RELEASED_RESERVATION_RECONCILIATION.md new file mode 100644 index 0000000..af68262 --- /dev/null +++ b/docs/RELEASED_RESERVATION_RECONCILIATION.md @@ -0,0 +1,95 @@ +# Reconcile expired ownership on a released terminal reservation + +Historical terminal reconciliation could set a reservation to `RELEASED` while +retaining its claim and lease. Those fields do not authorize a released worker; +their retention is a metadata-cleanup defect. The initial maintenance guard used +their presence as a conservative proxy for possible orphan work. A separately +reviewed activity classifier can establish quiescence from canonical terminal, +lease, dispatch and process-ownership evidence while retaining the anomaly and +cleanup obligation. This change does not alter that guard or waive cleanup. +Normal expired-claim recovery selects only `EXECUTING`/`STARTED`, and an ordinary +release is already a no-op for `RELEASED`, so neither repairs this residue. + +Normal terminal reconciliation now clears claim and lease fields in the same +existing transaction that releases the reservation. No run state, dispatch, +usage charge, compensation, historical heartbeat or signing field is changed by +that additional cleanup. + +For existing residue, the supported owner API is: + +```http +POST /v1/subscriber/certifications/{run_id}/reconcile-reservation +X-Tenant-ID: +Authorization: Bearer +Content-Type: application/json + +{"reason":"expired_terminal_claim_owner_repair"} +``` + +This uses the existing `governance:manage` permission and its existing role and +key-scope policy. It does not create a permission, credential or scope grant. +The source declares `echo.certforge.governance.reconcile_reservation` with the +existing `certforge.governance.manage` SDK scope, tier2 and server-side Vault +credential reference. Its strict command-bearing schema is generated through +the existing contract generator. This declaration is not live registration. +The exact source-declared SDK set increases from 61 to 62 capabilities; manifest, +generated schemas and exact-count validators stay synchronized and strict. +The library entry point is +`SubscriberGovernance.reconcile_released_reservation(principal, run_id=..., reason=...)`. +It repeats live key, membership, role, organization, subscription and scope +authorization under the same immediate write transaction as the repair. + +The method accepts exactly one reservation joined to a run in the principal's +tenant. It requires reservation `RELEASED` and run `COMPLETED`, `CANCELLED` or +`INFRASTRUCTURE_FAILURE`. An existing dispatch must be terminal and unclaimed. +For residual ownership, the stored lease must be parseable, timezone-aware and +expired under the canonical clock. A claim without proof of lease expiry is +refused. Active, unexpired, nonterminal, ambiguous or cross-tenant rows cannot +be repaired through this operation. + +The transaction clears only `claim_token`, `lease_expires_at` and the reservation +update time, then reads back the cleared fields. SQL selects only the boolean +presence of a claim; neither API output nor audit contains a claim token or +idempotency key. It preserves run/outcome, dispatch, usage/compensation and all +other reservations. The existing audit chain records one +`certification.released_claim_reconcile` event with reason, terminal state, +prior ownership-presence flags and `meter_compensated:false`. Audit failure rolls +back the cleanup. A clean repeat returns `reconciled:false` with unchanged +reservation metadata and creates no second domain reconciliation event. Normal +API request-outcome auditing still applies to each request. + +The response schema is +`echo.certforge.released-reservation-reconciliation.v1`. It reports tenant/run, +terminal run state, released reservation state, whether cleanup occurred, +claim/lease absence, durable reservation update time and uncompensated metering. +Unknown or other-tenant run IDs return the same `run_reservation_not_found`. + +## Admission and operation boundaries + +This source candidate is based on installed commit +`d8ad9fd881365801354b78a758632d2a31ab6828`; it is not an installed repair. The +existing d8ad API does not acquire this endpoint merely because source tests +pass. No SDK registration SQL is applied by this source change. + +Normal delivery requires independent source acceptance, the required local and +hosted checks, exact-candidate CertForge and Certification App evidence, and +the normal owner release/readiness and deployment/rollback admission applicable +to the changed service. Any installed source change invalidates old exact-source +commitments: refresh the installed source closure, readiness and owner guard +bindings rather than relabeling d8ad evidence. Until then, do not import this +candidate into the live service, autoload it, mutate the live database directly, +or weaken the quiet guard. + +Once the exact implementation and owner route are admitted, the owner must first +read the exact tenant/run/reservation/dispatch, prove no active work, preserve the +pending-work projection and metering/history fingerprints, then use the ordinary +authenticated API. Read back the exact response, canonical row/audit state and +all protected projections; repeat once to prove idempotence. Existing grants +must cover `governance:manage`; absent permission is an admission dependency, +never a reason to mint or broaden a key implicitly. + +Rollback of this source candidate is an ordinary scoped revert or retention of +the previously admitted service revision. A later successful terminal-ownership +cleanup must not be undone by restoring an old claim token or rewriting history. +Live service rollback and any database recovery require their separately tested +owner procedures; this source change performs neither. diff --git a/scripts/certification_journey.py b/scripts/certification_journey.py index bf48bc6..dfa199c 100644 --- a/scripts/certification_journey.py +++ b/scripts/certification_journey.py @@ -62,8 +62,8 @@ def main() -> int: sdk = json.loads((ROOT / ".echo" / "sdk.json").read_text(encoding="utf-8")) capabilities = sdk.get("capabilities", []) - require(len(capabilities) == 61, "SDK capability declaration must contain exactly 61 entries") - require(len(set(capabilities)) == 61, "SDK capability declaration contains duplicates") + require(len(capabilities) == 62, "SDK capability declaration must contain exactly 62 entries") + require(len(set(capabilities)) == 62, "SDK capability declaration contains duplicates") require(all(isinstance(item, str) and item.startswith("echo.") for item in capabilities), "SDK capability declaration contains an invalid identifier") diff --git a/scripts/master_acceptance.py b/scripts/master_acceptance.py index 4048db6..50407ec 100644 --- a/scripts/master_acceptance.py +++ b/scripts/master_acceptance.py @@ -72,10 +72,10 @@ def _p7(value: dict[str, Any]) -> None: def _sdk(value: dict[str, Any]) -> None: capabilities = value.get("capabilities") - _require(value.get("capability_count") == 61, "SDK contract must contain exactly 61 capabilities") - _require(isinstance(capabilities, dict) and len(capabilities) == 61, "SDK capability rows are incomplete") + _require(value.get("capability_count") == 62, "SDK contract must contain exactly 62 capabilities") + _require(isinstance(capabilities, dict) and len(capabilities) == 62, "SDK capability rows are incomplete") names = list(capabilities) - _require(len(set(names)) == 61, "SDK capability IDs are not unique") + _require(len(set(names)) == 62, "SDK capability IDs are not unique") _require(all(isinstance(name, str) and name.startswith("echo.cert") for name in names), "SDK capability namespace drift") _require( all( @@ -87,6 +87,11 @@ def _sdk(value: dict[str, Any]) -> None: ), "SDK capability contract rows are malformed", ) + _require( + capabilities.get("echo.certforge.governance.reconcile_reservation", {}).get("operation") + == "POST /v1/subscriber/certifications/{run_id}/reconcile-reservation", + "SDK released-reservation owner route is missing or mismatched", + ) def _ci(value: dict[str, Any], source_commit: str) -> None: diff --git a/scripts/register_certforge_caps.sql b/scripts/register_certforge_caps.sql index 6969503..937065a 100644 --- a/scripts/register_certforge_caps.sql +++ b/scripts/register_certforge_caps.sql @@ -274,6 +274,8 @@ VALUES '{"type":"object","additionalProperties":false}'::jsonb, 'certforge.governance.read', 0, '{"X-Tenant-ID":"org-echo-sovereign","Authorization":"vault:certforge.desktop_admin_api_key"}'::jsonb, 15, 'active', 'unknown'), ('echo.certforge.governance.update', 'Update tenant governance with optimistic version control.', 'http', 'http://127.0.0.1:8309/v1/subscriber/governance', 'PUT', 'json_body', 'forge', '{"type":"object","required":["expected_version","config"],"properties":{"expected_version":{"type":"integer"},"config":{"type":"object"}},"additionalProperties":false}'::jsonb, 'certforge.governance.manage', 2, '{"X-Tenant-ID":"org-echo-sovereign","Authorization":"vault:certforge.desktop_admin_api_key"}'::jsonb, 20, 'active', 'unknown'), + ('echo.certforge.governance.reconcile_reservation', 'Reconcile only expired residual claim and lease metadata on one tenant-owned released terminal reservation. Existing governance authority is rechecked transactionally; no execution, state transition or quota compensation.', 'http', 'http://127.0.0.1:8309/v1/subscriber/certifications/{run_id}/reconcile-reservation', 'POST', 'path', 'forge', + '{"type":"object","required":["run_id","reason"],"properties":{"run_id":{"type":"string","minLength":1,"maxLength":128},"reason":{"type":"string","pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"}},"additionalProperties":false}'::jsonb, 'certforge.governance.manage', 2, '{"X-Tenant-ID":"org-echo-sovereign","Authorization":"vault:certforge.desktop_admin_api_key"}'::jsonb, 20, 'active', 'unknown'), ('echo.certforge.policy_packs.create', 'Create a versioned tenant policy pack.', 'http', 'http://127.0.0.1:8309/v1/subscriber/policy-packs', 'POST', 'json_body', 'forge', '{"type":"object","required":["name","version","manifest"],"properties":{"name":{"type":"string"},"version":{"type":"string"},"manifest":{"type":"object"}},"additionalProperties":false}'::jsonb, 'certforge.policy_pack.manage', 2, '{"X-Tenant-ID":"org-echo-sovereign","Authorization":"vault:certforge.desktop_admin_api_key"}'::jsonb, 20, 'active', 'unknown'), ('echo.certforge.policy_packs.list', 'List tenant policy packs.', 'http', 'http://127.0.0.1:8309/v1/subscriber/policy-packs', 'GET', 'query', 'forge', diff --git a/scripts/register_certforge_sdk_schemas.sql b/scripts/register_certforge_sdk_schemas.sql index 78f876b..9e714ac 100644 --- a/scripts/register_certforge_sdk_schemas.sql +++ b/scripts/register_certforge_sdk_schemas.sql @@ -33,6 +33,7 @@ INSERT INTO certforge_sdk_contract(id, input_schema_json, output_schema_json) VA ('echo.certforge.evidence', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"run_id":{"type":"string"}},"required":["command","run_id"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.findings', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"run_id":{"type":"string"}},"required":["command","run_id"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.governance.get', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"}},"required":["command"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), + ('echo.certforge.governance.reconcile_reservation', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"reason":{"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$","type":"string"},"run_id":{"type":"string"}},"required":["command","reason","run_id"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.governance.update', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"config":{"additionalProperties":false,"properties":{"allowed_policy_ids":{"items":{"type":"string"},"type":"array"},"customer_managed_signing":{"type":"boolean"},"customer_signing_key_id":{"anyOf":[{"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$","type":"string"},{"type":"null"}]},"local_only_execution":{"type":"boolean"},"private_worker_only":{"type":"boolean"},"report_brand_name":{"anyOf":[{"type":"string"},{"type":"null"}]},"report_logo_url":{"anyOf":[{"type":"string"},{"type":"null"}]},"retention_days":{"type":"integer"}},"required":["allowed_policy_ids","retention_days","private_worker_only","report_brand_name","report_logo_url","customer_managed_signing","local_only_execution"],"type":"object"},"expected_version":{"minimum":1.0,"type":"integer"}},"required":["command","config","expected_version"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.health', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"}},"required":["command"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.list_runs', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"}},"required":["command"],"type":"object"}'::jsonb, '{"items":{"additionalProperties":true,"type":"object"},"type":"array"}'::jsonb), @@ -75,8 +76,8 @@ BEGIN FROM arcanum_sdk.sdk_capabilities WHERE id LIKE 'echo.certforge.%' OR id = 'echo.certification_forge.r5.run_negative_controls'; - IF live_count <> 61 THEN - RAISE EXCEPTION 'Certification Forge SDK surface drift: live %, contract 61', live_count; + IF live_count <> 62 THEN + RAISE EXCEPTION 'Certification Forge SDK surface drift: live %, contract 62', live_count; END IF; IF EXISTS ( SELECT 1 FROM certforge_sdk_contract c diff --git a/src/echo_certification_forge/service.py b/src/echo_certification_forge/service.py index 422cdf3..1165ae6 100644 --- a/src/echo_certification_forge/service.py +++ b/src/echo_certification_forge/service.py @@ -122,6 +122,12 @@ class RerunRequest(BaseModel): idempotency_key: str = Field(min_length=8, max_length=128) +class ReleasedReservationReconcileRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + command: str | None = Field(default=None, min_length=1, max_length=64) + reason: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") + + class ApiKeyCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") command: str | None = Field(default=None, min_length=1, max_length=64) @@ -636,6 +642,21 @@ def cancel_certification( raise HTTPException(status_code=409, detail="run_not_cancellable") from exc return project_run(context.store, context.store.get_run(run_id, tenant_id)) + @app.post("/v1/subscriber/certifications/{run_id}/reconcile-reservation") + def reconcile_released_reservation( + run_id: str, + request: ReleasedReservationReconcileRequest, + x_tenant_id: str | None = Header(default=None), + authorization: str | None = Header(default=None), + ) -> dict[str, Any]: + principal = subscriber_principal( + x_tenant_id, authorization, Permission.GOVERNANCE_MANAGE, + "certifications.reservations.reconcile", + ) + return context.subscribers.reconcile_released_reservation( + principal, run_id=run_id, reason=request.reason + ) + @app.get("/v1/certifications/{run_id}/findings") def get_findings( run_id: str, diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index 3c69fb4..49bff66 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -3158,7 +3158,8 @@ def _reconcile_run_reservations( connection.execute( f""" UPDATE subscriber_run_reservations - SET state = 'RELEASED', updated_at = ? + SET state = 'RELEASED', claim_token = NULL, + lease_expires_at = NULL, updated_at = ? WHERE organization_id = ? AND state IN ('BOUND', 'EXECUTING', 'STARTED') AND run_id IN ( SELECT run_id FROM runs WHERE tenant_id = ? AND state IN ({",".join("?" for _ in _TERMINAL_RUN_STATES)}) ) @@ -3166,6 +3167,120 @@ def _reconcile_run_reservations( (now, organization_id, organization_id, *_TERMINAL_RUN_STATES), ) + def reconcile_released_reservation( + self, principal: SubscriberPrincipal, *, run_id: str, reason: str + ) -> dict[str, Any]: + """Clear expired residual ownership only for one already terminal run. + + Existing governance authority is revalidated under the writer lock. No + execution, state transition, quota compensation or token read/export is + performed. Clean repeats return the same durable metadata without a + second reconciliation audit event. + """ + try: + require_identifier(run_id, "run_id") + require_identifier(reason, "reason") + except ValueError as exc: + raise SubscriberError(422, "reservation_reconcile_input_invalid") from exc + with self._connection(immediate=True) as connection: + self._require_with_connection( + connection, principal, Permission.GOVERNANCE_MANAGE + ) + now_value = self._now() + rows = connection.execute( + """ + SELECT r.idempotency_key, r.state, r.updated_at, + r.claim_token IS NOT NULL AS claim_present, + r.lease_expires_at, x.state AS run_state + FROM subscriber_run_reservations r + JOIN runs x ON x.run_id = r.run_id + AND x.tenant_id = r.organization_id + WHERE r.organization_id = ? AND r.run_id = ? + """, + (principal.organization_id, run_id), + ).fetchall() + if not rows: + raise SubscriberError(404, "run_reservation_not_found") + if len(rows) != 1: + raise SubscriberError(409, "run_reservation_not_unique") + row = rows[0] + if row["state"] != "RELEASED": + raise SubscriberError(409, "reservation_not_released") + if row["run_state"] not in _TERMINAL_RUN_STATES: + raise SubscriberError(409, "run_not_terminal") + dispatch = connection.execute( + """ + SELECT state, claim_token IS NOT NULL AS claim_present, + lease_owner IS NOT NULL AS owner_present, lease_expires_at + FROM subscriber_run_dispatches + WHERE run_id = ? AND organization_id = ? + """, + (run_id, principal.organization_id), + ).fetchone() + if dispatch is not None and ( + dispatch["state"] not in ("COMPLETE", "FAILED", "CANCELLED") + or dispatch["claim_present"] or dispatch["owner_present"] + or dispatch["lease_expires_at"] is not None + ): + raise SubscriberError(409, "dispatch_not_terminal_and_unclaimed") + changed = bool(row["claim_present"] or row["lease_expires_at"] is not None) + updated_at = str(row["updated_at"]) + if changed: + if row["lease_expires_at"] is None: + raise SubscriberError(409, "released_claim_expiry_unproven") + try: + expires = parse_utc_iso(str(row["lease_expires_at"])) + except ValueError as exc: + raise SubscriberError(409, "released_claim_expiry_invalid") from exc + if expires > now_value: + raise SubscriberError(409, "released_claim_not_expired") + updated_at = to_utc_iso(now_value) + cursor = connection.execute( + """ + UPDATE subscriber_run_reservations + SET claim_token = NULL, lease_expires_at = NULL, updated_at = ? + WHERE organization_id = ? AND run_id = ? AND idempotency_key = ? + AND state = 'RELEASED' AND lease_expires_at = ? + """, + (updated_at, principal.organization_id, run_id, + row["idempotency_key"], row["lease_expires_at"]), + ) + if cursor.rowcount != 1: + raise SubscriberError(409, "released_reservation_reconcile_conflict") + self._append_audit( + connection, + organization_id=principal.organization_id, + actor_ref=principal.user_id, + action="certification.released_claim_reconcile", + resource_type="certification", + resource_id=run_id, + outcome="allowed", + details={"reason": reason, "run_state": str(row["run_state"]), + "prior_claim_present": bool(row["claim_present"]), + "prior_lease_present": True, "meter_compensated": False}, + ) + readback = connection.execute( + """ + SELECT state, claim_token IS NOT NULL AS claim_present, + lease_expires_at, updated_at + FROM subscriber_run_reservations + WHERE organization_id = ? AND run_id = ? AND idempotency_key = ? + """, + (principal.organization_id, run_id, row["idempotency_key"]), + ).fetchone() + if (readback is None or readback["state"] != "RELEASED" + or readback["claim_present"] or readback["lease_expires_at"] is not None + or readback["updated_at"] != updated_at): + raise SubscriberError(409, "released_reservation_readback_failed") + return { + "schema": "echo.certforge.released-reservation-reconciliation.v1", + "organization_id": principal.organization_id, "run_id": run_id, + "run_state": str(row["run_state"]), "reservation_state": "RELEASED", + "reconciled": changed, "residual_claim_present": False, + "lease_present": False, "reservation_updated_at": updated_at, + "meter_compensated": False, + } + def reserve_certification_run( self, principal: SubscriberPrincipal, diff --git a/tests/test_echo_sdk_manifest.py b/tests/test_echo_sdk_manifest.py index 9848804..a91d3fe 100644 --- a/tests/test_echo_sdk_manifest.py +++ b/tests/test_echo_sdk_manifest.py @@ -13,7 +13,7 @@ def test_echo_sdk_manifest_matches_the_authoritative_contract() -> None: contract = json.loads(contract_path.read_text(encoding="utf-8")) assert manifest["version"] == 1 - assert len(manifest["capabilities"]) == contract["capability_count"] == 61 + assert len(manifest["capabilities"]) == contract["capability_count"] == 62 assert set(manifest["capabilities"]) == set(contract["capabilities"]) assert len(manifest["capabilities"]) == len(set(manifest["capabilities"])) assert all(capability.startswith("echo.") for capability in manifest["capabilities"]) diff --git a/tests/test_master_acceptance.py b/tests/test_master_acceptance.py index b9826f1..938921a 100644 --- a/tests/test_master_acceptance.py +++ b/tests/test_master_acceptance.py @@ -6,6 +6,8 @@ import sys from pathlib import Path +import pytest + from echo_certification_forge.master_acceptance import ( MasterAcceptanceInspector, MasterAcceptanceRunner, @@ -59,7 +61,7 @@ def test_master_fixture_discovery_and_defects_are_complete() -> None: assert by_code["vulnerable_dependency"].blocks_release is True -def test_final_sdk_gate_accepts_checked_exact_60_capability_map() -> None: +def test_final_sdk_gate_accepts_checked_exact_62_capability_map() -> None: contract = json.loads( (ROOT / "contracts" / "certforge-sdk-capabilities.v1.json").read_text( encoding="utf-8" @@ -69,6 +71,20 @@ def test_final_sdk_gate_accepts_checked_exact_60_capability_map() -> None: _master_script_module()._sdk(contract) +@pytest.mark.parametrize("mutation", ["wrong_count", "missing_owner_route", "wrong_owner_operation"]) +def test_final_sdk_gate_rejects_owner_route_drift(mutation) -> None: + contract = json.loads((ROOT / "contracts/certforge-sdk-capabilities.v1.json").read_text(encoding="utf-8")) + name = "echo.certforge.governance.reconcile_reservation" + if mutation == "wrong_count": + contract["capability_count"] = 61 + elif mutation == "missing_owner_route": + contract["capabilities"]["echo.certforge.unrelated"] = contract["capabilities"].pop(name) + else: + contract["capabilities"][name]["operation"] = "POST /v1/unrelated" + with pytest.raises(ValueError): + _master_script_module()._sdk(contract) + + def test_master_acceptance_full_chain_with_docker_protocol_stub(tmp_path: Path) -> None: stub = tmp_path / "docker_stub.py" stub.write_text( diff --git a/tests/test_released_reservation_reconciliation.py b/tests/test_released_reservation_reconciliation.py new file mode 100644 index 0000000..ca41990 --- /dev/null +++ b/tests/test_released_reservation_reconciliation.py @@ -0,0 +1,211 @@ +"""Terminal claim repair stays scoped, authorized, audited and non-executing.""" +from datetime import UTC, datetime, timedelta +import json +import sqlite3 + +import pytest + +from echo_certification_forge.subscriber import Permission, SubscriberError +from test_p7_subscriber_governance import ( + _headers, _owner, _project, _provision, _stack, _submit_body, +) + +NOW = datetime(2026, 11, 2, 15, 0, tzinfo=UTC) +STALE_CLAIM = "synthetic-expired-test-claim" + + +def _subject(tmp_path, manifest): + store, governance, client = _stack(tmp_path, manifest) + governance._clock = lambda: NOW + organization = _provision(governance, "claim-repair", plan_code="enterprise") + headers = _headers(organization.organization_id, organization.bootstrap_api_key) + project = _project(client, organization.organization_id, organization.bootstrap_api_key).json() + request = _submit_body(manifest, organization.organization_id, project["project_id"], "released-repair-0001") + submitted = client.post("/v1/certifications", headers=headers, json=request) + assert submitted.status_code == 201 + run_id = submitted.json()["run_id"] + cancelled = client.post(f"/v1/certifications/{run_id}/cancel", headers=headers) + assert cancelled.status_code == 200 + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE subscriber_run_reservations SET state='RELEASED',claim_token=?," + "lease_expires_at=?,updated_at=? WHERE run_id=?", + (STALE_CLAIM, (NOW - timedelta(days=1)).isoformat(), + (NOW - timedelta(days=1, seconds=5)).isoformat(), run_id)) + connection.execute("UPDATE subscriber_run_dispatches SET state='FAILED' WHERE run_id=?", (run_id,)) + # A prior terminal credit is closed history, not another compensation + # opportunity for the metadata repair. + connection.execute("INSERT INTO subscriber_usage_events(organization_id,meter,quantity,idempotency_key,metadata_json,occurred_at) " + "SELECT u.organization_id,u.meter,-1,'prior-terminal-credit:'||u.idempotency_key,?,u.occurred_at " + "FROM subscriber_usage_events u JOIN subscriber_run_reservations r " + "ON r.organization_id=u.organization_id AND r.idempotency_key=u.idempotency_key " + "WHERE r.run_id=? AND u.quantity=1 LIMIT 1", + (json.dumps({"reason": "previous_terminal_credit"}), run_id)) + principal = _owner(governance, organization.organization_id, organization.bootstrap_api_key, + Permission.GOVERNANCE_MANAGE) + return store, governance, client, organization, headers, principal, run_id, project + + +def _snapshot(store): + with sqlite3.connect(store.db_path) as connection: + return {table: connection.execute(f"SELECT * FROM {table} ORDER BY rowid").fetchall() + for table in ("runs", "subscriber_run_reservations", "subscriber_run_dispatches", + "subscriber_usage_events")} + + +def _repair(governance, principal, run_id): + return governance.reconcile_released_reservation(principal, run_id=run_id, + reason="expired_terminal_claim_owner_repair") + + +def test_owner_api_repairs_only_expired_terminal_residual_and_is_idempotent(tmp_path, manifest): + store, governance, client, org, headers, principal, run_id, project = _subject(tmp_path, manifest) + for index in range(13): + body = _submit_body(manifest, org.organization_id, project["project_id"], f"pending-preserved-{index:02}") + assert client.post("/v1/certifications", headers=headers, json=body).status_code == 201 + # Normal intake already terminalizes old rows but historically leaves the + # RELEASED residual unchanged. Preserve exactly what the owner sees now. + before = _snapshot(store) + with sqlite3.connect(store.db_path) as connection: + connection.row_factory = sqlite3.Row + reservations_before = {row["run_id"]: dict(row) for row in connection.execute("SELECT * FROM subscriber_run_reservations")} + route = f"/v1/subscriber/certifications/{run_id}/reconcile-reservation" + response = client.post(route, headers=headers, json={"command": "reconcile", "reason": "expired_terminal_claim_owner_repair"}) + assert response.status_code == 200 + result = response.json() + assert result["reconciled"] is True and result["run_state"] == "CANCELLED" + assert result["reservation_state"] == "RELEASED" and result["meter_compensated"] is False + assert result["residual_claim_present"] is False and result["lease_present"] is False + assert STALE_CLAIM not in response.text and "idempotency_key" not in response.text + after = _snapshot(store) + for table in ("runs", "subscriber_run_dispatches", "subscriber_usage_events"): + assert before[table] == after[table] + with sqlite3.connect(store.db_path) as connection: + connection.row_factory = sqlite3.Row + reservation = dict(connection.execute("SELECT * FROM subscriber_run_reservations WHERE run_id=?", (run_id,)).fetchone()) + reservations_after = {row["run_id"]: dict(row) for row in connection.execute("SELECT * FROM subscriber_run_reservations")} + assert reservation["claim_token"] is None and reservation["lease_expires_at"] is None + assert connection.execute("SELECT COUNT(*) FROM subscriber_run_dispatches WHERE state='PENDING'").fetchone()[0] == 13 + audits = connection.execute("SELECT details_json FROM subscriber_audit_events WHERE action='certification.released_claim_reconcile'").fetchall() + assert len(audits) == 1 + reservations_before[run_id].update(claim_token=None, lease_expires_at=None, + updated_at=result["reservation_updated_at"]) + assert reservations_after == reservations_before + assert json.loads(audits[0][0])["meter_compensated"] is False + assert STALE_CLAIM not in audits[0][0] + second = client.post(route, headers=headers, json={"reason": "expired_terminal_claim_owner_repair"}) + assert second.status_code == 200 and second.json()["reconciled"] is False + assert _snapshot(store) == after + assert governance.verify_audit(principal)["valid"] is True + with sqlite3.connect(store.db_path) as connection: + assert connection.execute("SELECT COUNT(*) FROM subscriber_audit_events WHERE action='certification.released_claim_reconcile'").fetchone()[0] == 1 + + +def test_normal_terminal_reconciliation_clears_claim_when_it_releases_reservation(tmp_path, manifest): + store, _, client, org, headers, _, run_id, project = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE subscriber_run_reservations SET state='EXECUTING' WHERE run_id=?", (run_id,)) + body = _submit_body(manifest, org.organization_id, project["project_id"], "normal-reconcile-0002") + assert client.post("/v1/certifications", headers=headers, json=body).status_code == 201 + with sqlite3.connect(store.db_path) as connection: + assert connection.execute("SELECT state,claim_token,lease_expires_at FROM subscriber_run_reservations WHERE run_id=?", (run_id,)).fetchone() == ("RELEASED", None, None) + + +@pytest.mark.parametrize("field,value,error", [ + ("state", "BOUND", "reservation_not_released"), + ("state", "EXECUTING", "reservation_not_released"), + ("state", "STARTED", "reservation_not_released"), + ("lease_expires_at", (NOW + timedelta(seconds=1)).isoformat(), "released_claim_not_expired"), + ("lease_expires_at", None, "released_claim_expiry_unproven"), + ("lease_expires_at", "not-a-timestamp", "released_claim_expiry_invalid"), + ("lease_expires_at", "2026-11-01T10:00:00", "released_claim_expiry_invalid"), +]) +def test_unsafe_or_unproven_residual_is_unchanged(tmp_path, manifest, field, value, error): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute(f"UPDATE subscriber_run_reservations SET {field}=? WHERE run_id=?", (value, run_id)) + before = _snapshot(store) + with pytest.raises(SubscriberError, match=error): + _repair(governance, principal, run_id) + assert _snapshot(store) == before + + +@pytest.mark.parametrize("state", ["QUEUED", "EXECUTING_TESTS"]) +def test_nonterminal_run_is_not_reconciled(tmp_path, manifest, state): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE runs SET state=? WHERE run_id=?", (state, run_id)) + before = _snapshot(store) + with pytest.raises(SubscriberError, match="run_not_terminal"): + _repair(governance, principal, run_id) + assert _snapshot(store) == before + + +def test_other_tenant_cannot_discover_or_reconcile_target(tmp_path, manifest): + store, governance, client, _, _, _, run_id, _ = _subject(tmp_path, manifest) + other = _provision(governance, "claim-other-owner") + before = _snapshot(store) + response = client.post(f"/v1/subscriber/certifications/{run_id}/reconcile-reservation", + headers=_headers(other.organization_id, other.bootstrap_api_key), + json={"reason": "expired_terminal_claim_owner_repair"}) + assert response.status_code == 404 + assert response.json()["detail"] == "run_reservation_not_found" + assert _snapshot(store) == before + + +def test_live_permission_is_rechecked_inside_same_write_transaction(tmp_path, manifest): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE subscriber_api_keys SET scopes_json=? WHERE key_id=?", + (json.dumps([Permission.RUN_READ.value]), principal.key_id)) + before = _snapshot(store) + with pytest.raises(SubscriberError, match="permission_denied"): + _repair(governance, principal, run_id) + assert _snapshot(store) == before + + +def test_active_dispatch_is_not_reconciled(tmp_path, manifest): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE subscriber_run_dispatches SET state='CLAIMED',claim_token=?,lease_owner='test-owner',lease_expires_at=? WHERE run_id=?", + ("synthetic-dispatch-token", (NOW + timedelta(seconds=30)).isoformat(), run_id)) + before = _snapshot(store) + with pytest.raises(SubscriberError, match="dispatch_not_terminal_and_unclaimed"): + _repair(governance, principal, run_id) + assert _snapshot(store) == before + + +@pytest.mark.parametrize("terminal", ["CANCELLED", "COMPLETED", "INFRASTRUCTURE_FAILURE"]) +def test_all_existing_terminal_states_support_expired_lease_only_cleanup(tmp_path, manifest, terminal): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE runs SET state=? WHERE run_id=?", (terminal, run_id)) + connection.execute("UPDATE subscriber_run_reservations SET claim_token=NULL,lease_expires_at=? WHERE run_id=?", + (NOW.isoformat(), run_id)) + result = _repair(governance, principal, run_id) + assert result["reconciled"] is True and result["run_state"] == terminal + + +def test_audit_failure_rolls_back_ownership_cleanup(tmp_path, manifest, monkeypatch): + store, governance, _, _, _, principal, run_id, _ = _subject(tmp_path, manifest) + before = _snapshot(store) + def fail(*args, **kwargs): + raise RuntimeError("synthetic_audit_storage_failure") + monkeypatch.setattr(governance, "_append_audit", fail) + with pytest.raises(RuntimeError, match="synthetic_audit_storage_failure"): + _repair(governance, principal, run_id) + assert _snapshot(store) == before + + +def test_api_requires_existing_governance_scope_and_valid_fixed_body(tmp_path, manifest): + store, _, client, _, headers, principal, run_id, _ = _subject(tmp_path, manifest) + route = f"/v1/subscriber/certifications/{run_id}/reconcile-reservation" + before = _snapshot(store) + with sqlite3.connect(store.db_path) as connection: + connection.execute("UPDATE subscriber_api_keys SET scopes_json=? WHERE key_id=?", + (json.dumps([Permission.RUN_READ.value]), principal.key_id)) + denied = client.post(route, headers=headers, json={"reason": "owner_repair"}) + assert denied.status_code == 403 + assert _snapshot(store) == before + invalid = client.post(route, headers=headers, json={"reason": "owner repair", "claim_token": "forbidden"}) + assert invalid.status_code == 422 + assert _snapshot(store) == before diff --git a/tests/test_sdk_capability_contract.py b/tests/test_sdk_capability_contract.py index a0afa0e..ada7bd9 100644 --- a/tests/test_sdk_capability_contract.py +++ b/tests/test_sdk_capability_contract.py @@ -32,7 +32,7 @@ def test_generated_sdk_contract_is_current() -> None: def test_every_certforge_capability_has_command_and_output_schema() -> None: contract = json.loads(CONTRACT.read_text(encoding="utf-8")) capabilities = contract["capabilities"] - assert contract["capability_count"] == 61 == len(capabilities) + assert contract["capability_count"] == 62 == len(capabilities) for capability, item in capabilities.items(): input_schema = item["input_schema"] output_schema = item["output_schema"]