From 9b94d8d0f52d1cbfb4168d8322e9c12723a938b8 Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Fri, 9 Oct 2026 21:29:20 +0000 Subject: [PATCH] Publish curated Unraid-AppOps-Gateway source from c61e0e1e292914893b40354f904f23dd9fd4a958 --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 66 +- backend/app/mcp_server.py | 6 +- backend/app/services/action_execution.py | 107 +- backend/app/services/action_registry.py | 6 + .../controller_artifact_helper_spec.py | 2 +- backend/app/services/deployment_broker.py | 14 +- backend/app/services/mcp_operations.py | 10 +- backend/app/services/policy_engine.py | 10 + backend/tests/test_container_restart.py | 6 + .../test_controller_artifact_bootstrap.py | 23 +- .../test_controller_artifact_promoter.py | 78 +- .../test_controller_hub_release_volume.py | 33 +- .../tests/test_mcp_rag_n8n_integrations.py | 39 + .../test_runner_acceptance_quiescence.py | 931 ++++++++++++++++++ .../tests/test_runner_acceptance_surface.py | 260 +++++ .../tests/test_runner_resource_envelope.py | 82 +- ...unner_specialization_controller_surface.py | 51 + 18 files changed, 1651 insertions(+), 75 deletions(-) create mode 100644 backend/tests/test_runner_acceptance_quiescence.py create mode 100644 backend/tests/test_runner_acceptance_surface.py diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index a74207f..58a0084 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `1350d8bf6527e085bbfa950ec0d1ee3e973d618e`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `c61e0e1e292914893b40354f904f23dd9fd4a958`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index f530c75..97e2fd5 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "1350d8bf6527e085bbfa950ec0d1ee3e973d618e", + "sourceRevision": "c61e0e1e292914893b40354f904f23dd9fd4a958", "files": [ { "path": ".dockerignore", @@ -404,8 +404,8 @@ }, { "path": "backend/app/mcp_server.py", - "sha256": "065bc0c1f9779f0df30cb6d19eee129b0cf3c4b66a190687e9bfc682c239c94a", - "bytes": 51192 + "sha256": "1f7351ce29000dc0e2a41fd1f66a33001a64d7e5cd4a9ed689c62b447e55d570", + "bytes": 51299 }, { "path": "backend/app/runner_statistics_mcp.py", @@ -474,13 +474,13 @@ }, { "path": "backend/app/services/action_execution.py", - "sha256": "4c4099328c3931ef3811f7d9043d7ad69c2bd27c1e7b6b1bdc139fbb5ba3c29a", - "bytes": 100242 + "sha256": "31e54c9a8fec9d79ad95045679ed3ace944c44e1c123921088c2d3b709d8de8d", + "bytes": 106444 }, { "path": "backend/app/services/action_registry.py", - "sha256": "3e0f83e62f6d8dfdf0bb4bc6c09b58d612bb6ff2a4b5fcf619dba5aa8d009fd1", - "bytes": 21446 + "sha256": "4c7493f28f237c1020bf1fbfc22be93b11b9815440d38cfac99afde4e6c3d844", + "bytes": 21721 }, { "path": "backend/app/services/ai_remediation_planner.py", @@ -524,7 +524,7 @@ }, { "path": "backend/app/services/controller_artifact_helper_spec.py", - "sha256": "573a1a79ff75940a1a0a49cab5718ad62ddd20703369d47167f1ab796f2e5cae", + "sha256": "ad26cb686288c0757588c4cc97a4fec4a21e97416f241781829b1268f7f9465e", "bytes": 6029 }, { @@ -539,8 +539,8 @@ }, { "path": "backend/app/services/deployment_broker.py", - "sha256": "4370e327b0b608b46be446159f59b9975fb99381967282c9661dc1c7837f16bd", - "bytes": 45377 + "sha256": "7eaf302c579a3084e1084415236f16291cbeea1bbc80c7f78f7995dc5e22d4c6", + "bytes": 46123 }, { "path": "backend/app/services/deployment_control_plane.py", @@ -659,8 +659,8 @@ }, { "path": "backend/app/services/mcp_operations.py", - "sha256": "f04792f20727c049a80047afd6b4352b8b4ade34832db8d1f0dfb44a02c1b42a", - "bytes": 14589 + "sha256": "1a598fb4055b98321ee0d2e44fa7d3e0d054af893ea86fc508284278f7b1bf85", + "bytes": 14959 }, { "path": "backend/app/services/media_codec_dashboard_deployment_reconciliation.py", @@ -704,8 +704,8 @@ }, { "path": "backend/app/services/policy_engine.py", - "sha256": "23ede379cf2078de600e0cecd0d54ec5f921e331ca3996d64b5a068cc7ecda1c", - "bytes": 13119 + "sha256": "e22bad183a57ac31e8d2e021bf2e958ce84120878a08988bd2a1840e203361d5", + "bytes": 13812 }, { "path": "backend/app/services/portfolio_deployment_enrollment.py", @@ -944,13 +944,13 @@ }, { "path": "backend/tests/test_container_restart.py", - "sha256": "6bbe8c44a07a0609873148c5b8a0775d375c92c2a5f44ffbfeafb1960d462cab", - "bytes": 108190 + "sha256": "faf4f9068f668a0a743a54d1da9f063c501ead709056a6790b38b3f95cf47468", + "bytes": 108593 }, { "path": "backend/tests/test_controller_artifact_bootstrap.py", - "sha256": "179c50723fc29524088b9991af9f7d924337fe11585f4fe6456917c1932efefa", - "bytes": 82644 + "sha256": "ff7b6e484926fa82336def1e2ef1353adbb3988bb895c4fe74af31aed38df066", + "bytes": 82720 }, { "path": "backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py", @@ -964,8 +964,8 @@ }, { "path": "backend/tests/test_controller_artifact_promoter.py", - "sha256": "41b10414669b592f6055488f2dc51f1a0034e2aaf8506153c180ab93c4121560", - "bytes": 7570 + "sha256": "f139565d2508fe9f71113bc52041e2b0545a202f0529417ddf7516bddcf955da", + "bytes": 8667 }, { "path": "backend/tests/test_controller_artifact_promotion_diagnostics.py", @@ -989,8 +989,8 @@ }, { "path": "backend/tests/test_controller_hub_release_volume.py", - "sha256": "07a28688018b027d0590e68e65ca619fe223f21d4f8f58196d95eb344b087d55", - "bytes": 165301 + "sha256": "abdc5eab49e59dc62fedc925bbd93fdffe907df7522e6a3c538fb2711906984b", + "bytes": 167055 }, { "path": "backend/tests/test_controller_root_procedure_contract.py", @@ -1294,8 +1294,8 @@ }, { "path": "backend/tests/test_mcp_rag_n8n_integrations.py", - "sha256": "587ff5dac2861f47224f0ec174ad0ef2472980f2ba41d28d7f2d72eb4015abd7", - "bytes": 57461 + "sha256": "75bd5eec23e50ef4e62cbb941e404539653f098117fbc7ce68f184f81288c7b8", + "bytes": 60283 }, { "path": "backend/tests/test_media_codec_dashboard_deployment_reconciliation.py", @@ -1582,6 +1582,16 @@ "sha256": "7ea8409b3129938fd57390db26ba2e6068af90cf46db31204a0f0cd567c254ed", "bytes": 3242 }, + { + "path": "backend/tests/test_runner_acceptance_quiescence.py", + "sha256": "c7212b67f2995bf2c09d88ef6df38498fcfe7f3d513d43fd00bb3f8b62d4673d", + "bytes": 33359 + }, + { + "path": "backend/tests/test_runner_acceptance_surface.py", + "sha256": "4362211eea644b0eab88fce7eb9d909991d550685a6e4e40a0144587237232ce", + "bytes": 10699 + }, { "path": "backend/tests/test_runner_admission_controller.py", "sha256": "dcfb8e7a1d1ee86d6bfadef71bbd8be629a7f70a26baafb37bb134c61e78491d", @@ -1639,8 +1649,8 @@ }, { "path": "backend/tests/test_runner_resource_envelope.py", - "sha256": "4970625e2fb1d95750f7a0acc1ce328865711abf3450e089a8a2304b21c20d9f", - "bytes": 46791 + "sha256": "0181c36c9560c5833fb653b50d8739677b45df22b856ff514dd9cba69b732a3d", + "bytes": 51403 }, { "path": "backend/tests/test_runner_scheduler_liveness_recovery.py", @@ -1649,8 +1659,8 @@ }, { "path": "backend/tests/test_runner_specialization_controller_surface.py", - "sha256": "dc26476686237318953bf335865ad0901635044629826690af0e1d98d786612b", - "bytes": 9174 + "sha256": "a6caa41c68478e1ded3509e44ec319cfb3ca6a060f0129c0d6e090f8929dbf6a", + "bytes": 12349 }, { "path": "backend/tests/test_runner_statistics_binding.py", diff --git a/backend/app/mcp_server.py b/backend/app/mcp_server.py index f7a5e44..40f5a69 100644 --- a/backend/app/mcp_server.py +++ b/backend/app/mcp_server.py @@ -343,7 +343,8 @@ def _mcp_action_registry(session: Session) -> dict[str, Any]: target = str(item.get("target") or "") capability = str(item.get("capability") or "operator") mcp_policy = evaluate_policy( - session, action=action, target=target, source="automation" + session, action=action, target=target, + source=mcp_operations.mcp_policy_source(action, target) ) original_decision = str(mcp_policy.get("decision") or "blocked") @@ -417,7 +418,8 @@ def preview_action(action: str, target: str) -> dict[str, Any]: safe_action = action[:160] with Session(engine) as session: decision = evaluate_policy( - session, action=safe_action, target=target[:240], source="automation" + session, action=safe_action, target=target[:240], + source=mcp_operations.mcp_policy_source(safe_action, target[:240]) ) mcp_mode, mcp_supported = _mcp_preview_mode( safe_action, str(decision["decision"]) diff --git a/backend/app/services/action_execution.py b/backend/app/services/action_execution.py index 5e48ee1..f5474bd 100644 --- a/backend/app/services/action_execution.py +++ b/backend/app/services/action_execution.py @@ -55,6 +55,8 @@ APPROVAL_EXECUTOR_ACTIONS = { "runner.fleet.upgrade_v2", "runner.fleet.specialize", "runner.statistics.configure", + "runner.acceptance.preflight", + "runner.acceptance.quiesce", "hub.migration.prepare_permissions", "deployment.targets.sync", "deployment.runnerless.provision", @@ -125,7 +127,12 @@ _DIRECT_STEPS = { def action_execution_profile( action: str, policy_decision: str, capability: str ) -> dict[str, Any]: - if policy_decision == "blocked": + if action in {"runner.acceptance.preflight", "runner.acceptance.quiesce"} and policy_decision != "blocked": + # The fixed MCP typed authority is separate from Command Center's + # manual route; do not advertise an unsupported manual executor. + mode = "governance_only" + surface = None + elif policy_decision == "blocked": mode = "blocked" surface = None elif action in GENERIC_EXECUTOR_ACTIONS: @@ -1669,6 +1676,31 @@ def _execute_controller_artifact_promote( return operation_result +def _execute_runner_acceptance(session: Session, *, action: str, target: str, source: str, + risk: str, reason: str, original_policy_decision: str) -> dict[str, Any]: + from ..db.models import GatewayUser + from .user_access import actor_identity + + actor = actor_identity.get() + admin = False + if actor: + provider, _, subject = actor.partition(":") + user = session.exec(select(GatewayUser).where(GatewayUser.provider == provider, + GatewayUser.subject == subject)).first() + admin = bool(user and user.enabled and user.role == "admin") + authorized = (source == "chatgpt-mcp" and actor is None) or admin + if (target != "tower/ci-acceptance-residue" or not authorized + or source not in {"manual", "chatgpt-mcp", "approved_operator"} + or original_policy_decision not in {"allowed", "approval_required"}): + return {"success": False, "status": "blocked", "result": "blocked", + "message": "Fixed CI maintenance requires the authenticated administrator boundary.", + "policyDecision": "blocked"} + return _execute_runner_fleet_upgrade_v2( + session, source=source, risk=risk, reason=reason, + original_policy_decision=original_policy_decision, acceptance=action.rsplit(".", 1)[-1], + ) + + def _execute_runner_fleet_upgrade_v2( session: Session, *, @@ -1678,12 +1710,15 @@ def _execute_runner_fleet_upgrade_v2( original_policy_decision: str, specialized: bool = False, statistics_configuration: bool = False, + acceptance: str | None = None, ) -> dict[str, Any]: from .deployment_broker import DeploymentBrokerClient - action = ("runner.statistics.configure" if statistics_configuration else + action = ("runner.acceptance." + acceptance if acceptance else + "runner.statistics.configure" if statistics_configuration else "runner.fleet.specialize" if specialized else "runner.fleet.upgrade_v2") - target = ("tower/projectbrain-runner-statistics" if statistics_configuration else + target = ("tower/ci-acceptance-residue" if acceptance else + "tower/projectbrain-runner-statistics" if statistics_configuration else "tower/runner-fleet-specialized" if specialized else "tower/runner-fleet-v2") steps = [ "Validate exact fixed runner-fleet target", @@ -1697,6 +1732,9 @@ def _execute_runner_fleet_upgrade_v2( steps = ["Verify fixed read-only statistics target", "Prove protected controller/runtime source", "Lock production configuration", "Provision private independent credential with durable CAS", "Configure ProjectBrain for its governed delivery"] + if acceptance: + steps = ["Prove protected controller/runtime source", "Attest finite CI origin and complete private identities", + "Verify dispatch barrier and production exclusion", "Preserve all resources and data"] job = create_job( session, action_id=action, @@ -1710,26 +1748,57 @@ def _execute_runner_fleet_upgrade_v2( set_job_progress(session, job, 25, f"Requesting {action} through the fixed controller endpoint.") try: broker = DeploymentBrokerClient() - result = (broker.runner_statistics_configure(reason=reason) if statistics_configuration else + result = (broker.runner_acceptance(reason=reason, preflight=acceptance == "preflight") if acceptance else + broker.runner_statistics_configure(reason=reason) if statistics_configuration else broker.runner_fleet_specialize(reason=reason) if specialized else broker.runner_fleet_upgrade_v2(reason=reason)) + if acceptance: + allowed_blockers = { + "terminal_origin_claim_missing", "terminal_origin_task_binding_unavailable", + "terminal_origin_unavailable", "active_product_receipt_exclusion_unavailable", + "dispatch_barrier_guard_unverified", "preflight_evidence_rejected", + "acceptance_evidence_rejected", "term_timeout", "controller_response_untrusted", + "maintenance_outcome_unknown", "maintenance_receipt_untrusted", "maintenance_execution_deadline", + } + blockers = result.get("blockers", []) + clean_blockers = (blockers if isinstance(blockers, list) and len(blockers) <= 20 + and all(isinstance(code, str) and code in allowed_blockers for code in blockers) + else ["controller_response_untrusted"]) + result = { + "status": result.get("status") if result.get("status") in { + "ready", "blocked", "partial", "quiesced", "already_quiesced", "unknown", "in_progress" + } else "blocked", + "revision": result.get("revision") if isinstance(result.get("revision"), str) + and len(result["revision"]) == 40 and all(char in "0123456789abcdef" for char in result["revision"]) else None, + "blockers": clean_blockers, + "preservedContainers": 18 if result.get("preservedContainers") == 18 else None, + "effectsPerformed": result.get("effectsPerformed") if type(result.get("effectsPerformed")) is bool else None, + "noMutation": result.get("noMutation") if type(result.get("noMutation")) is bool else None, + "maintenancePhase": result.get("maintenance", {}).get("phase") if isinstance(result.get("maintenance"), dict) + and result["maintenance"].get("phase") in {"guarding", "quiescing", "restoring", "complete"} else result.get("phase") + if result.get("phase") in {"guarding", "quiescing", "restoring", "complete"} else None, + } status = str(result.get("status") or "error") - success = status in ({"configured", "already_configured"} if statistics_configuration else + success = status in ({"ready"} if acceptance == "preflight" else {"quiesced", "already_quiesced"} if acceptance else + {"configured", "already_configured"} if statistics_configuration else {"specialized", "already_applied"} if specialized else {"upgraded", "already_applied"}) message = str( result.get("reason") or ( - ("ProjectBrain read-only statistics configuration verified." if statistics_configuration else + ("Finite CI acceptance maintenance verified." if acceptance else + "ProjectBrain read-only statistics configuration verified." if statistics_configuration else "Tower runner specialization verified." if specialized else "Tower runner-fleet v2 migration verified.") if success else - ("ProjectBrain statistics configuration was rejected or failed." if statistics_configuration else + ("Maintenance outcome is unknown or still active; use read-only preflight status before any explicit resume." if acceptance and status in {"unknown", "in_progress", "partial"} else + "Finite CI acceptance maintenance is blocked by required evidence." if acceptance else + "ProjectBrain statistics configuration was rejected or failed." if statistics_configuration else "Tower runner specialization was rejected or failed." if specialized else "Tower runner-fleet v2 migration was rejected or failed.") ) ) except Exception as exc: - result = {"status": "error"} - status = "error" + result = {"status": "unknown" if acceptance == "quiesce" else "error", "effectsPerformed": None} + status = result["status"] success = False message = f"{type(exc).__name__}: fixed controller operation failed" @@ -1749,7 +1818,7 @@ def _execute_runner_fleet_upgrade_v2( risk=risk, policy_decision="allowed", reason=reason, - result="success" if success else "failed", + result="success" if success else "unknown" if acceptance and status in {"unknown", "in_progress", "partial"} else "failed", execution_plan={"steps": steps, "jobId": completed.id}, after_state={ "status": status, @@ -1768,7 +1837,7 @@ def _execute_runner_fleet_upgrade_v2( return { "success": success, "status": "success" if success else status, - "result": "success" if success else "failed", + "result": "success" if success else "unknown" if acceptance and status in {"unknown", "in_progress", "partial"} else "failed", "message": message, "jobId": completed.id, "auditId": audit.id, @@ -1778,6 +1847,11 @@ def _execute_runner_fleet_upgrade_v2( "revision": result.get("revision"), "verifiedRunners": result.get("verifiedRunners"), "projectbrainDeliveryRequired": result.get("projectbrainDeliveryRequired"), + "blockers": result.get("blockers") if acceptance else None, + "preservedContainers": result.get("preservedContainers") if acceptance else None, + "effectsPerformed": result.get("effectsPerformed") if acceptance else None, + "noMutation": result.get("noMutation") if acceptance else None, + "maintenancePhase": result.get("maintenancePhase") if acceptance else None, } @@ -2065,6 +2139,10 @@ def execute_governed_action( "originalPolicyDecision": original_policy_decision, } + if action in {"runner.acceptance.preflight", "runner.acceptance.quiesce"}: + return _execute_runner_acceptance(session, action=action, target=target, source=source, + risk=risk, reason=reason, original_policy_decision=original_policy_decision) + if action == "controller.artifact.promote": if target.strip().lower() != "tower/deployment-controller": return { @@ -2603,6 +2681,13 @@ def execute_approved_action(session: Session, approval: Approval) -> dict[str, A original_policy_decision="approval_required", ) + if approval.action in {"runner.acceptance.preflight", "runner.acceptance.quiesce"}: + return _execute_runner_acceptance( + session, action=approval.action, target=approval.target, source="approved_operator", + risk=approval.risk, reason=approval.reason or f"Approved through {approval.id}.", + original_policy_decision="approval_required", + ) + if approval.action == "runner.fleet.upgrade_v2": if approval.target.strip().lower() != "tower/runner-fleet-v2": return { diff --git a/backend/app/services/action_registry.py b/backend/app/services/action_registry.py index c2c3357..5e12434 100644 --- a/backend/app/services/action_registry.py +++ b/backend/app/services/action_registry.py @@ -9,6 +9,8 @@ ACTIONS = [ 'runner.fleet.upgrade_v2', 'runner.fleet.specialize', 'runner.statistics.configure', + 'runner.acceptance.preflight', + 'runner.acceptance.quiesce', 'controller.artifact.promote', 'controller.managed-storage.initialize', 'hub.migration.prepare_permissions', @@ -222,6 +224,8 @@ ACTION_RISK = { 'runner.fleet.upgrade_v2': 'high', 'runner.fleet.specialize': 'high', 'runner.statistics.configure': 'high', + 'runner.acceptance.preflight': 'low', + 'runner.acceptance.quiesce': 'high', 'controller.artifact.promote': 'high', 'controller.managed-storage.initialize': 'high', 'hub.migration.prepare_permissions': 'high', @@ -553,6 +557,8 @@ def action_default_target(action: str) -> str: return 'tower/runner-fleet-v2' if action == 'runner.fleet.specialize': return 'tower/runner-fleet-specialized' + if action in {'runner.acceptance.preflight', 'runner.acceptance.quiesce'}: + return 'tower/ci-acceptance-residue' if action == 'runner.statistics.configure': return 'tower/projectbrain-runner-statistics' if action == 'deployment.targets.sync': diff --git a/backend/app/services/controller_artifact_helper_spec.py b/backend/app/services/controller_artifact_helper_spec.py index c75e8f3..ab8478b 100644 --- a/backend/app/services/controller_artifact_helper_spec.py +++ b/backend/app/services/controller_artifact_helper_spec.py @@ -5,7 +5,7 @@ from typing import Any SOURCE_REPOSITORY = "jens/unraid-appops-gateway" HELPER_NAME = "itworx-controller-artifact-promoter-modern" -CONTROLLER_VERSION = "2.1.20" +CONTROLLER_VERSION = "2.1.21" HELPER_LABEL = "itworx.controller.promoter" _REVISION = re.compile(r"^[0-9a-f]{40}$") _IMAGE_ID = re.compile(r"^sha256:[0-9a-f]{64}$") diff --git a/backend/app/services/deployment_broker.py b/backend/app/services/deployment_broker.py index c6a0a50..8d5fa74 100644 --- a/backend/app/services/deployment_broker.py +++ b/backend/app/services/deployment_broker.py @@ -174,6 +174,7 @@ class DeploymentBrokerClient: secret: str, payload: dict[str, Any] | None = None, error_response_fields: frozenset[str] = frozenset(), + request_timeout: httpx.Timeout | None = None, ) -> dict[str, Any]: if not self.is_configured(): return {"status": "missing_config", "reason": self.configuration_error()} @@ -194,7 +195,7 @@ class DeploymentBrokerClient: headers["Content-Type"] = "application/json" try: with httpx.Client( - timeout=self.timeout_seconds, + timeout=request_timeout or self.timeout_seconds, verify=self._verify_option(), ) as client: response = client.request(method, f"{self.base_url}{path}", content=body, headers=headers) @@ -990,6 +991,17 @@ class DeploymentBrokerClient: secret=settings.mcp_token or "", payload={"reason": reason[:1000]}, ) + def runner_acceptance(self, *, reason: str, preflight: bool) -> dict[str, Any]: + result = self._request( + "POST", "/v1/appops/runner-acceptance/preflight" if preflight else "/v1/appops/runner-acceptance/quiesce", + secret=settings.mcp_token or "", payload={"reason": reason[:1000]}, + request_timeout=httpx.Timeout(connect=5, write=5, pool=5, read=350), + ) + if not preflight and result.get("status") in {"unreachable", "error", "rejected"}: + return {"status": "unknown", "noMutation": False, "effectsPerformed": None, + "blockers": ["maintenance_outcome_unknown"]} + return result + def hub_migration_prepare_permissions(self, *, reason: str) -> dict[str, Any]: """Prepare only the fixed Hub backup and controller-state permissions.""" return self._request( diff --git a/backend/app/services/mcp_operations.py b/backend/app/services/mcp_operations.py index 99b77ec..04df9f1 100644 --- a/backend/app/services/mcp_operations.py +++ b/backend/app/services/mcp_operations.py @@ -59,6 +59,14 @@ def _reconciliation_required_response( } +def mcp_policy_source(action: str, target: str) -> str: + """Normalize only the fixed acceptance authority at the authenticated MCP boundary.""" + if (action in {"runner.acceptance.preflight", "runner.acceptance.quiesce"} + and target == "tower/ci-acceptance-residue"): + return "chatgpt-mcp" + return "automation" + + def request_operation( session: Session, *, @@ -97,7 +105,7 @@ def request_operation( session, action=safe_action, target=safe_target, - source="automation", + source=mcp_policy_source(safe_action, safe_target), ) risk = str(policy.get("risk") or ACTION_RISK.get(safe_action, "critical")) if capability == "read_only": diff --git a/backend/app/services/policy_engine.py b/backend/app/services/policy_engine.py index 2f7e99f..42492df 100644 --- a/backend/app/services/policy_engine.py +++ b/backend/app/services/policy_engine.py @@ -210,6 +210,16 @@ def evaluate_policy( "scope_decision": "allowed" if allowed else None, } + if action in {'runner.acceptance.preflight', 'runner.acceptance.quiesce'}: + # Dedicated finite maintenance authority; wildcard and database rows + # cannot redirect this source-qualified action to other containers. + eligible = target == 'tower/ci-acceptance-residue' and source in {'manual', 'chatgpt-mcp'} + return { + 'decision': ('allowed' if action.endswith('.preflight') else 'approval_required') if eligible else 'blocked', + 'reason': 'Fixed CI acceptance maintenance requires current native origin, dispatch and product-exclusion proofs.', + 'risk': ACTION_RISK[action], 'policy_id': None, 'scope_decision': None, + } + if action in _DEFAULT_DECISIONS: reason = _DEFAULT_REASONS.get(action, 'Action uses default governance policy.') return { diff --git a/backend/tests/test_container_restart.py b/backend/tests/test_container_restart.py index 3a32b5d..d7a0cec 100644 --- a/backend/tests/test_container_restart.py +++ b/backend/tests/test_container_restart.py @@ -191,6 +191,12 @@ def test_action_registry_reports_real_execution_modes(api_client): assert items['stack.redeploy']['executionMode'] == 'governance_only' assert items['jellyfin.scan_library']['executionMode'] == 'blocked' assert items['deployment.target.promote_workmesh']['executionMode'] == 'approval_then_execute' + for action in ('runner.acceptance.preflight', 'runner.acceptance.quiesce'): + assert items[action]['target'] == 'tower/ci-acceptance-residue' + assert items[action]['executionMode'] == 'governance_only' + assert items[action]['executionSupported'] is False + assert items[action]['commandExecutionSupported'] is False + assert items[action]['executionSurface'] is None assert payload['summary']['commandExecutable'] >= 1 assert payload['summary']['approvalExecutable'] == 5 diff --git a/backend/tests/test_controller_artifact_bootstrap.py b/backend/tests/test_controller_artifact_bootstrap.py index 113436e..36167eb 100644 --- a/backend/tests/test_controller_artifact_bootstrap.py +++ b/backend/tests/test_controller_artifact_bootstrap.py @@ -589,7 +589,7 @@ def _health( *, revision: str = REVISION, tree: str = TREE, - version: str = "2.1.20", + version: str = "2.1.21", artifact_sha256: str = "1" * 64, ) -> dict[str, Any]: return { @@ -1347,7 +1347,7 @@ def test_current_214_baseline_is_verified_and_bound_to_new_helper(): "itworx.controller.promoter": "true", "itworx.source.repository": "jens/unraid-appops-gateway", "itworx.source.revision": REVISION, - "itworx.controller.version": "2.1.20", + "itworx.controller.version": "2.1.21", "itworx.controller.bootstrap.target.id": MODERN_CONTROLLER_CONTAINER_ID, "itworx.controller.bootstrap.target.image": CONTROLLER_IMAGE_ID, "itworx.controller.bootstrap.target.revision": MODERN_CONTROLLER_REVISION, @@ -1826,7 +1826,7 @@ def test_successful_child_requires_exact_source_tree_version_and_health(): assert result["status"] == "promoted" assert result["revision"] == REVISION - assert result["controllerVersion"] == "2.1.20" + assert result["controllerVersion"] == "2.1.21" assert result["artifactSha256"] == "1" * 64 assert docker.started == [MODERN_HELPER_ID] assert docker.created[0]["Image"] == CONTROLLER_IMAGE_ID @@ -2086,32 +2086,33 @@ def test_modern_upgrade_path_still_requires_private_archive_and_image_id( def test_current_helper_version_is_distinct_from_fixed_controller_baselines(): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() namespace = bootstrap._safe_modern_spec_namespace(source) - assert namespace["CONTROLLER_VERSION"] == "2.1.20" - assert bootstrap._verified_historical_helper_version(source) == "2.1.20" + assert namespace["CONTROLLER_VERSION"] == "2.1.21" + assert bootstrap._verified_historical_helper_version(source) == "2.1.21" assert [baseline.version for baseline in bootstrap.CONTROLLER_BASELINES] == ["2.1.12", "2.1.14"] with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(RETIRED_MODERN_HELPER_SOURCE.encode("utf-8")) assert bootstrap._verified_historical_helper_version(RETIRED_MODERN_HELPER_SOURCE.encode("utf-8")) == "2.1.14" -@pytest.mark.parametrize("version", ["2.1.16", "2.1.21", "9.9.9"]) +@pytest.mark.parametrize("version", ["2.1.16", "2.1.22", "9.9.9"]) def test_current_and_historical_helper_reject_unreviewed_target_versions(version): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() - foreign = source.replace(b'CONTROLLER_VERSION = "2.1.20"', f'CONTROLLER_VERSION = "{version}"'.encode()) + foreign = source.replace(b'CONTROLLER_VERSION = "2.1.21"', f'CONTROLLER_VERSION = "{version}"'.encode()) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(foreign) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._verified_historical_helper_version(foreign) -@pytest.mark.parametrize("version", ["2.1.14", "2.1.16", "2.1.21"]) +@pytest.mark.parametrize("version", ["2.1.14", "2.1.16", "2.1.22"]) def test_promoted_health_cannot_substitute_a_baseline_or_unknown_version(version): assert not bootstrap._health_matches(_health(version=version), REVISION, TREE) -def test_previous_reviewed_helper_is_historical_only(): +@pytest.mark.parametrize("version", ["2.1.19", "2.1.20"]) +def test_previous_reviewed_helper_is_historical_only(version): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() - prior = source.replace(b'CONTROLLER_VERSION = "2.1.20"', b'CONTROLLER_VERSION = "2.1.19"') + prior = source.replace(b'CONTROLLER_VERSION = "2.1.21"', f'CONTROLLER_VERSION = "{version}"'.encode()) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(prior) - assert bootstrap._verified_historical_helper_version(prior) == "2.1.19" + assert bootstrap._verified_historical_helper_version(prior) == version diff --git a/backend/tests/test_controller_artifact_promoter.py b/backend/tests/test_controller_artifact_promoter.py index 9f77f37..8d58404 100644 --- a/backend/tests/test_controller_artifact_promoter.py +++ b/backend/tests/test_controller_artifact_promoter.py @@ -14,20 +14,33 @@ def test_promoter_archive_covers_every_required_installer_source(): spec = helper_container_spec(revision="a" * 40, image_id="sha256:" + "b" * 64) command = spec["Cmd"][2].replace("\\\n", " ") - archive = re.search(r'archive --format=tar "\$revision"(.*?)> "\$work/controller.tar"', command, re.S) + archive = re.search( + r'archive --format=tar "\$revision"(.*?)> "\$work/controller.tar"', + command, + re.S, + ) assert archive is not None packaged = shlex.split(archive.group(1)) - installer = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text(encoding="utf-8") + installer = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text( + encoding="utf-8" + ) required_block = re.search(r"(?m)^source_paths=\(\n(.*?)\n\)", installer, re.S) assert required_block is not None required = shlex.split(required_block.group(1)) assert required - assert all(any(path == prefix or path.startswith(prefix + "/") for prefix in packaged) for path in required) + assert all( + any(path == prefix or path.startswith(prefix + "/") for prefix in packaged) + for path in required + ) def test_controller_promoter_action_is_fixed_and_registered(): - registry = (ROOT / "backend/app/services/action_registry.py").read_text(encoding="utf-8") - execution = (ROOT / "backend/app/services/action_execution.py").read_text(encoding="utf-8") + registry = (ROOT / "backend/app/services/action_registry.py").read_text( + encoding="utf-8" + ) + execution = (ROOT / "backend/app/services/action_execution.py").read_text( + encoding="utf-8" + ) assert "'controller.artifact.promote'" in registry assert "return 'tower/deployment-controller'" in registry assert 'action == "controller.artifact.promote"' in execution @@ -46,7 +59,7 @@ def test_controller_promoter_helper_spec_is_fixed_and_bounded(): spec = helper_container_spec(revision=revision, image_id=image_id) assert HELPER_NAME == "itworx-controller-artifact-promoter-modern" - assert CONTROLLER_VERSION == "2.1.20" + assert CONTROLLER_VERSION == "2.1.21" assert re.fullmatch(r"2\.1\.\d+", CONTROLLER_VERSION) assert spec["Image"] == image_id assert spec["HostConfig"]["Privileged"] is False @@ -68,23 +81,29 @@ def test_controller_promoter_helper_spec_is_fixed_and_bounded(): command = spec["Cmd"][2] assert "mktemp -d -p /tmp itworx-controller-artifact-promoter-" in command assert f"itworx-controller-artifact-promoter-{revision[:12]}.XXXXXX" in command - assert "/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-" not in command + assert ( + "/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-" not in command + ) assert "work='/mnt/cache" not in command - assert "rm -rf -- \"$work\"" in command - assert "scripts/upgrade_deployment_controller.sh > \"$work/controller.tar\"" in command - assert "chmod 0600 \"$work/controller.tar\"" in command + assert 'rm -rf -- "$work"' in command + assert ( + 'scripts/upgrade_deployment_controller.sh > "$work/controller.tar"' in command + ) + assert 'chmod 0600 "$work/controller.tar"' in command assert "scripts/prepare_controller_hub_release_volume.py" in command assert "scripts/run_repopulse_adoption.py" in command assert "UPGRADE_DEPLOYMENT_CONTROLLER_V2" in command assert revision in command assert "--image-id" in command assert CONTROLLER_VERSION in command - assert "version='2.1.20'" in command + assert "version='2.1.21'" in command assert "/gitea-repositories/jens/unraid-appops-gateway.git" in command archive_start = command.index('git --git-dir="$repo" archive') archive_end = command.index('> "$work/controller.tar"', archive_start) assert "scripts/vendor" in command[archive_start:archive_end] - assert "scripts/upgrade_deployment_controller.sh > \"$work/controller.tar\"" in command + assert ( + 'scripts/upgrade_deployment_controller.sh > "$work/controller.tar"' in command + ) assert '--image-id "$image_id"' in command @@ -110,7 +129,7 @@ def test_controller_promoter_bootstrap_child_spec_binds_legacy_target_identity() "itworx.controller.promoter": "true", "itworx.source.repository": "jens/unraid-appops-gateway", "itworx.source.revision": revision, - "itworx.controller.version": "2.1.20", + "itworx.controller.version": "2.1.21", "itworx.controller.bootstrap.target.id": target_id, "itworx.controller.bootstrap.target.image": image_id, "itworx.controller.bootstrap.target.revision": target_revision, @@ -142,6 +161,7 @@ def test_controller_promoter_rejects_partial_or_mutable_bootstrap_binding(): current_controller_version="2.1.12", ) + @pytest.mark.parametrize( "revision,image_id", [ @@ -151,7 +171,9 @@ def test_controller_promoter_rejects_partial_or_mutable_bootstrap_binding(): ("a" * 40, "sha256:" + "z" * 64), ], ) -def test_controller_promoter_rejects_mutable_or_invalid_identity(revision: str, image_id: str): +def test_controller_promoter_rejects_mutable_or_invalid_identity( + revision: str, image_id: str +): from backend.app.services.controller_artifact_promoter import helper_container_spec with pytest.raises(ValueError): @@ -159,7 +181,9 @@ def test_controller_promoter_rejects_mutable_or_invalid_identity(revision: str, def test_controller_promoter_uses_only_protected_main_and_push_gate(): - source = (ROOT / "backend/app/services/controller_artifact_promoter.py").read_text(encoding="utf-8") + source = (ROOT / "backend/app/services/controller_artifact_promoter.py").read_text( + encoding="utf-8" + ) assert 'branch = "main"' in source assert 'REQUIRED_PUSH_CONTEXT = "Managed validation / full (push)"' in source assert "controller_promoter_main_not_validated" in source @@ -167,7 +191,29 @@ def test_controller_promoter_uses_only_protected_main_and_push_gate(): def test_deployment_broker_exposes_bounded_controller_health(): - source = (ROOT / "backend/app/services/deployment_broker.py").read_text(encoding="utf-8") + source = (ROOT / "backend/app/services/deployment_broker.py").read_text( + encoding="utf-8" + ) assert "def controller_health(self)" in source assert 'f"{self.base_url}/health"' in source assert "_MAX_CONTROLLER_HEALTH_BYTES" in source + + +def test_modern_promoter_21_binds_actual_installed_19_without_widening_bootstrap_baselines(): + from backend.app.services.controller_artifact_promoter import helper_container_spec + + actual_revision = "0f89e1506065c5365c48968b1d7845d666a18be8" + spec = helper_container_spec( + revision="a" * 40, + image_id="sha256:" + "b" * 64, + current_controller_id="c" * 64, + current_controller_revision=actual_revision, + current_controller_tree="d" * 40, + current_controller_version="2.1.19", + ) + labels = spec["Labels"] + assert labels["itworx.controller.version"] == "2.1.21" + assert labels["itworx.controller.bootstrap.target.revision"] == actual_revision + assert labels["itworx.controller.bootstrap.target.version"] == "2.1.19" + assert labels["itworx.controller.bootstrap.target.id"] == "c" * 64 + assert labels["itworx.controller.bootstrap.target.tree"] == "d" * 40 diff --git a/backend/tests/test_controller_hub_release_volume.py b/backend/tests/test_controller_hub_release_volume.py index 5826043..3a2d91a 100644 --- a/backend/tests/test_controller_hub_release_volume.py +++ b/backend/tests/test_controller_hub_release_volume.py @@ -2441,6 +2441,34 @@ def test_storage_command_diagnostics_never_retain_private_paths(stderr, conditio assert 'fixture-secret' not in diagnostic +@pytest.mark.parametrize('condition', [ + 'image_metadata_failed', 'inspect_failed', 'identity_invalid', 'reinspect_failed', + 'identity_changed', 'remove_failed', 'remove_storage_full', 'remove_read_only', + 'remove_permission_denied', 'remove_daemon_unavailable', 'remove_busy', + 'remove_container_missing', 'remove_api_version_unsupported', 'absence_failed', 'unverified', +]) +def test_probe_child_diagnostics_survive_promotion_without_raw_output(monkeypatch, condition): + stderr = ('resource_cgroup_probe_cleanup_' + condition + '\n').encode() + monkeypatch.setattr(module.subprocess, 'run', lambda *a, **kw: types.SimpleNamespace( + returncode=1, stderr=stderr, stdout=b'Bearer fixture-secret')) + with pytest.raises(module.VolumeTransitionError) as error: + module._run(['python3', 'qualified-validator'], diagnostic_action='probe_cleanup', max_output=4096) + assert module._safe_command_diagnostic(error.value.command_failure) == ( + 'probe_cleanup_exit_1_probe_' + condition) + assert 'fixture-secret' not in str(error.value) + + +@pytest.mark.parametrize('stderr', [ + b'resource_cgroup_probe_cleanup_remove_failed Bearer fixture-secret', + b'resource_cgroup_probe_cleanup_remove_failed\nBearer fixture-secret', + b'resource_cgroup_probe_cleanup_fixture-secret', b'permission denied fixture-secret', + b'resource_cgroup_probe_cleanup_\xff', + b'resource_cgroup_probe_cleanup_remove_failed' + b'x' * 4096, +]) +def test_probe_child_rejects_untrusted_or_multiline_diagnostics(stderr): + assert module._command_stderr_condition(stderr, 'probe_cleanup', max_bytes=4096) is None + + @pytest.mark.parametrize('read_only,free_bytes', [(True, 1024**3), (False, 0)]) def test_stop_rejects_failed_storage_before_recording_intent_or_stopping(monkeypatch, read_only, free_bytes): receipt = {'phase': 'prepared'} @@ -2628,6 +2656,7 @@ def test_installer_release_order_matches_renderer_contract(): assert module.LEGACY_RELEASE_FILES_WITH_CONFIG_ROOT == HISTORICAL_CONFIG_ROOT_RELEASE_FILES assert module._KNOWN_RELEASE_FILE_SETS == ( module.RELEASE_FILES, + module.LEGACY_RELEASE_FILES_PRE_QUIESCENCE, module.LEGACY_RELEASE_FILES_PRE_COMPONENT, module.LEGACY_RELEASE_FILES_WITH_ROLE_HELPER, module.LEGACY_RELEASE_FILES_WITHOUT_ROLE_HELPER, @@ -2656,7 +2685,7 @@ def test_installer_release_order_matches_renderer_contract(): ) assert module.LEGACY_RELEASE_FILES_WITH_ROLE_HELPER == tuple( name for name in module.RELEASE_FILES - if name not in {"forge_backup_paths.py", "forge_secret_bundle.py", "immutable_service_promotion.py", "registered_legacy_adoption.py", "registered_legacy_transport.py", "registered_build_context.py"} + if name not in {"forge_backup_paths.py", "forge_secret_bundle.py", "immutable_service_promotion.py", "registered_legacy_adoption.py", "registered_legacy_transport.py", "registered_build_context.py", "runner_acceptance_quiescence.py"} ) assert module.LEGACY_RELEASE_FILES_WITHOUT_ROLE_HELPER == tuple( name for name in module.LEGACY_RELEASE_FILES_WITH_ROLE_HELPER @@ -3605,7 +3634,7 @@ def test_promotion_cleanup_uses_incoming_code_but_only_attested_previous_source( assert args[0] == "python3" assert Path(args[1]).name == "runner_resource_envelope.py" assert args[2:] == ["--cleanup-stopped-host-probe", IMAGE_ID, old_source] - assert kw == {"max_output": 4096} + assert kw == {"max_output": 4096, "diagnostic_action": "probe_cleanup"} events.append("cleanup") return b"" monkeypatch.setattr(module, "_run", run) diff --git a/backend/tests/test_mcp_rag_n8n_integrations.py b/backend/tests/test_mcp_rag_n8n_integrations.py index a9c8e90..3c21979 100644 --- a/backend/tests/test_mcp_rag_n8n_integrations.py +++ b/backend/tests/test_mcp_rag_n8n_integrations.py @@ -1482,3 +1482,42 @@ def test_mcp_sdk_redaction_preserves_schema_valid_calls_and_domain_denials(): assert "policy_denied" in denied.content[0].text asyncio.run(exercise()) assert calls == [True, False] + + +def test_public_mcp_acceptance_list_preview_request_and_replay_are_consistent(integration_client, monkeypatch): + from backend.app import mcp_server + from backend.app.services import deployment_broker + from backend.app.db.models import Approval, AuditEvent + _, engine = integration_client + calls = [] + def broker(*args, **kwargs): + calls.append(kwargs) + return {"status": "ready" if kwargs["preflight"] else "quiesced", "revision": "d"*40, + "preservedContainers": 18, "private": "Bearer SECRET-CANARY"} + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", broker) + async def exercise(): + async with Client(mcp_server.appops_mcp) as client: + listed = (await client.call_tool("get_action_registry", {})).structured_content + by_id = {item["id"]: item for item in listed["items"]} + for action in ("runner.acceptance.preflight", "runner.acceptance.quiesce"): + assert by_id[action]["mcpExecutionMode"] == "direct_typed" + assert by_id[action]["manualExecutionMode"] == "governance_only" + preview = (await client.call_tool("preview_action", {"action": action, "target": "tower/ci-acceptance-residue"})).structured_content + assert preview["mcpExecutionSupported"] is True and preview["executed"] is False + args = {"action": action, "target": "tower/ci-acceptance-residue", + "reason": "Preserve finite residue. Authorization: Bearer SECRET-CANARY", + "idempotencyKey": "public-acceptance-"+action} + result = (await client.call_tool("server.operation.request", args)).structured_content + assert result["executed"] is True and result["executionSupported"] is True + assert result["controllerStatus"] == ("ready" if action.endswith("preflight") else "quiesced") + assert result["originalPolicyDecision"] == ("allowed" if action.endswith("preflight") else "approval_required") + replay = (await client.call_tool("server.operation.request", args)).structured_content + assert replay["idempotentReplay"] is True and replay["jobId"] == result["jobId"] + assert "SECRET-CANARY" not in json.dumps(result) + denied = (await client.call_tool("preview_action", {"action": "container.stop", "target": "tower/ci-acceptance-residue"})).structured_content + assert denied["policyDecision"] == "blocked" + asyncio.run(exercise()) + assert len(calls) == 2 + with Session(engine) as session: + assert session.exec(select(Approval)).all() == [] + assert "SECRET-CANARY" not in json.dumps([row.model_dump(mode="json") for row in session.exec(select(AuditEvent)).all()]) diff --git a/backend/tests/test_runner_acceptance_quiescence.py b/backend/tests/test_runner_acceptance_quiescence.py new file mode 100644 index 0000000..91d8ac9 --- /dev/null +++ b/backend/tests/test_runner_acceptance_quiescence.py @@ -0,0 +1,931 @@ +from __future__ import annotations + +import copy +import importlib.util +import sys +from contextlib import nullcontext +from pathlib import Path + +import pytest + +BROKER = Path(__file__).resolve().parents[2] / "ops/gitea-deploy-broker" + + +@pytest.fixture +def module(monkeypatch): + monkeypatch.syspath_prepend(str(BROKER)) + spec = importlib.util.spec_from_file_location( + "acceptance_quiescence_test", BROKER / "runner_acceptance_quiescence.py" + ) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + monkeypatch.setattr(module.fleet, "_migration_lock", lambda **kwargs: nullcontext()) + return module + + +class Driver: + def __init__(self, module): + self.module = module + self.state = { + "daemon": { + "legacyAttested": True, + "engineId": "fixed-private-engine", + "outerId": "a" * 64, + "dockerRootDir": module.ROOTLESS_STORAGE, + "rootless": True, + }, + "productExclusionVerified": True, + "dispatchBarrierVerified": True, + "resources": { + "attachmentsVerified": True, + "ownershipVerified": True, + "networks": ["fixed-network"], + "volumes": ["preserved-data"], + }, + "rows": copy.deepcopy(module._fixed_rows()), + "origins": {}, + } + for row in self.state["rows"]: + row["ownerLabelDigest"] = "b" * 64 if row["ownerLabelPresent"] else None + row["labelDigest"] = module._digest(row["composeLabels"]) + for project, claim in module.ORIGINS.items(): + if claim is None: + self.state["origins"][project] = None + continue + attempt = claim["attempt"] if claim["attempt"] is not None else 0 + self.state["origins"][project] = { + "repository": "jens/workmesh", + "run": claim["run"], + "head": claim["head"], + "status": "completed", + "conclusion": "failure", + "attempt": attempt, + "task": claim["task"], + "taskHead": claim["head"], + "taskAttempt": attempt, + "jobsTerminal": True, + } + self.runners = [ + { + "id": 9, + "name": "legacy-linux-1", + "labels": ["linux", "docker"], + "busy": False, + "disabled": False, + }, + { + "id": 22, + "name": "legacy-linux-2", + "labels": ["linux", "docker"], + "busy": False, + "disabled": False, + }, + { + "id": 700, + "name": "windows-preserved", + "labels": ["windows"], + "busy": False, + "disabled": False, + }, + ] + self.effects = [] + self.pending = [] + self.timeout = False + self.after_term = None + + def maintenance(self): + return nullcontext() + + def capture_guards(self): + return {"runner": {"synthetic": True}, "host": {"synthetic": True}} + + def install_guards(self, receipt, persist): + for name in ("runner", "host"): + if receipt["guardProgress"].get(name) in { + "installed", + "restore_intent", + "restored", + }: + continue + receipt["guardProgress"][name] = "install_intent" + persist() + receipt["guardProgress"][name] = "installed" + persist() + + def restore_guards(self, receipt, persist): + for name in ("runner", "host"): + if receipt["guardProgress"].get(name) == "restored": + continue + receipt["guardProgress"][name] = "restore_intent" + persist() + receipt["guardProgress"][name] = "restored" + persist() + + def snapshot(self): + return copy.deepcopy(self.state) + + def registrations(self): + return copy.deepcopy(self.runners) + + def attest_registrations(self, rows): + assert {row["id"] for row in rows if "linux" in row["labels"]} == {9, 22} + + def pending_jobs(self): + return copy.deepcopy(self.pending) + + def pause(self): + pass + + def set_registration_disabled(self, original, value): + self.effects.append(("registration", original["id"], value)) + next(row for row in self.runners if row["id"] == original["id"])["disabled"] = ( + value + ) + + def disable_restart(self, cid): + self.effects.append(("restart", cid)) + next(row for row in self.state["rows"] if row["id"] == cid)["restartPolicy"] = { + "Name": "no", + "MaximumRetryCount": 0, + } + + def term(self, cid): + row = next(row for row in self.state["rows"] if row["id"] == cid) + if row["state"]["Running"]: + self.effects.append(("TERM", cid)) + if not self.timeout: + row["state"].update( + { + "Running": False, + "Status": "exited", + "Pid": 0, + "FinishedAt": "2026-10-09T20:00:00Z", + } + ) + if self.after_term: + self.after_term(self) + + def wait_exited(self, cid): + return not self.timeout + + +@pytest.fixture +def proven(module, monkeypatch): + receipt = {"value": None} + monkeypatch.setattr( + module, "_read_receipt", lambda: copy.deepcopy(receipt["value"]) + ) + + def save(value, expected): + current = receipt["value"] + actual = module._digest(current) if current is not None else None + module._require(actual == expected, "maintenance_receipt_cas_failed") + receipt["value"] = copy.deepcopy(value) + return module._digest(value) + + monkeypatch.setattr(module, "_save_receipt", save) + return Driver(module), receipt + + +def test_missing_backup_origin_never_becomes_authority(module, monkeypatch): + monkeypatch.setitem(module.ORIGINS, "workmesh-backup-exercise-39a1a242e4b9", None) + monkeypatch.setattr(module, "_read_receipt", lambda: None) + driver = Driver(module) + result = module.preflight(revision="d" * 40, projects={}, driver=driver) + assert result["status"] == "blocked" + assert result["effectsPerformed"] is False + assert driver.effects == [] + + +def test_positive_preserves_resources_and_windows_then_idempotent(module, proven): + driver, receipt = proven + data_before = copy.deepcopy(driver.state["resources"]) + ids_before = {row["id"] for row in driver.state["rows"]} + assert ( + module.preflight(revision="d" * 40, projects={}, driver=driver)["status"] + == "ready" + ) + result = module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert result["status"] == "quiesced" + assert result["preservedData"] is True + assert driver.state["resources"] == data_before + assert {row["id"] for row in driver.state["rows"]} == ids_before + assert all(not row["state"]["Running"] for row in driver.state["rows"]) + assert sum(effect[0] == "TERM" for effect in driver.effects) == 12 + assert driver.runners[2]["disabled"] is False + assert all( + effect[0] in {"registration", "restart", "TERM"} for effect in driver.effects + ) + assert receipt["value"]["phase"] == "complete" + count = len(driver.effects) + assert ( + module.quiesce(revision="d" * 40, projects={}, driver=driver)["status"] + == "already_quiesced" + ) + assert len(driver.effects) == count + with pytest.raises(module.QuiescenceError, match="maintenance_owner_changed"): + module.quiesce(revision="e" * 40, projects={}, driver=driver) + + +@pytest.mark.parametrize("kind", ["queued", "in_progress", "pending"]) +def test_initial_new_work_refuses_all_effects(module, proven, kind): + driver, _ = proven + driver.pending = [{"id": 99, "status": kind}] + with pytest.raises(module.QuiescenceError, match="global_ci_work_present"): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert driver.effects == [] + + +@pytest.mark.parametrize("kind", ["queued", "in_progress", "pending"]) +def test_new_work_during_execution_blocks_next_container(module, proven, kind): + driver, receipt = proven + driver.after_term = lambda self: setattr( + self, "pending", [{"id": 99, "status": kind}] + ) + with pytest.raises(module.QuiescenceError, match="global_ci_work_present"): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert sum(e[0] == "restart" for e in driver.effects) == 1 + assert receipt["value"]["phase"] != "complete" + assert all(row["disabled"] for row in driver.runners[:2]) + + +@pytest.mark.parametrize( + "change", + [ + lambda s: s["daemon"].update({"rootless": False}), + lambda s: s["daemon"].update({"dockerRootDir": "/srv/example/appdata/production"}), + lambda s: s["rows"][0].update({"id": "f" * 64}), + lambda s: s["rows"][0].update({"imageId": "sha256:" + "f" * 64}), + lambda s: s["rows"][0]["composeLabels"].update( + {"com.docker.compose.config-hash": "f" * 64} + ), + lambda s: s["rows"][0]["composeLabels"].update( + {"com.docker.compose.project.working_dir": "/production"} + ), + lambda s: s["rows"][0]["mounts"].append( + {"Type": "bind", "Source": "/production", "RW": True} + ), + lambda s: s["rows"][0]["networks"].update({"foreign": "f" * 64}), + lambda s: s["resources"].update({"attachmentsVerified": False}), + lambda s: s["resources"].update({"ownershipVerified": False}), + lambda s: s.update({"productExclusionVerified": False}), + lambda s: s.update({"dispatchBarrierVerified": False}), + lambda s: s["rows"].append(copy.deepcopy(s["rows"][0])), + lambda s: s["rows"][0].update({"ownerLabelDigest": "e" * 64}), + ], +) +def test_identity_or_scope_drift_has_no_effect(module, proven, change): + driver, _ = proven + change(driver.state) + with pytest.raises(module.QuiescenceError): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert driver.effects == [] + + +@pytest.mark.parametrize( + "field,value", + [ + ("attempt", None), + ("attempt", 2), + ("taskAttempt", 3), + ("task", 999), + ("head", "e" * 40), + ("jobsTerminal", False), + ], +) +def test_origin_drift_refuses_effects(module, proven, field, value): + driver, _ = proven + driver.state["origins"]["workmesh-product-21300-7bace5f2edf0d259"][field] = value + with pytest.raises(module.QuiescenceError): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert driver.effects == [] + + +def test_legacy_attempt_zero_is_actual_sentinel_and_missing_is_not_assumed( + module, proven +): + driver, _ = proven + module.ORIGINS["workmesh-runtime-19421"]["attempt"] = 0 + driver.state["origins"]["workmesh-runtime-19421"].update(attempt=0, taskAttempt=0) + module.attest_snapshot(driver.snapshot()) + driver.state["origins"]["workmesh-runtime-19421"]["attempt"] = None + with pytest.raises(module.QuiescenceError, match="attempt_unavailable"): + module.attest_snapshot(driver.snapshot()) + + +def test_timeout_retains_owned_guard_and_resumes_only_own_transition(module, proven): + driver, receipt = proven + driver.timeout = True + result = module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert result["status"] == "partial" and result["guardHeld"] is True + assert receipt["value"]["progress"][driver.state["rows"][0]["id"]] == "term_intent" + driver.timeout = False + result = module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert result["status"] == "quiesced" + + +def test_registration_drift_after_partial_never_restores_foreign_identity( + module, proven +): + driver, _ = proven + driver.timeout = True + module.quiesce(revision="d" * 40, projects={}, driver=driver) + before = len(driver.effects) + driver.runners[0]["labels"].append("foreign-project") + with pytest.raises(module.QuiescenceError, match="legacy_registration_changed"): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert len(driver.effects) == before + + +def test_no_arbitrary_or_destructive_docker_effect(module, monkeypatch): + calls = [] + + class Result: + returncode = 0 + + monkeypatch.setattr( + module.fleet, "_run", lambda args, **kwargs: calls.append(args) or Result() + ) + driver = module.ControllerDriver({}) + monkeypatch.setattr(driver, "assert_barrier", lambda: None) + monkeypatch.setattr(driver, "_guard_binding", lambda: {}) + monkeypatch.setattr(driver, "_effect_admission", lambda cid, args: None) + cid = module._fixed_rows()[0]["id"] + driver.disable_restart(cid) + assert calls == [ + [ + "docker", + "exec", + "-e", + "DOCKER_HOST=tcp://127.0.0.1:2375", + "gitea-runner-dind", + "docker", + "update", + "--restart=no", + cid, + ] + ] + with pytest.raises(module.QuiescenceError, match="container_target_invalid"): + driver.disable_restart("user-selected") + + +def test_public_preflight_never_serializes_private_canary(module): + class BadDriver: + def maintenance(self): + return nullcontext() + + def pending_jobs(self): + raise ValueError("Bearer SECRET-CANARY") + + def proof_gaps(self): + return ["terminal_origin_unavailable"] + + result = module.preflight(revision="d" * 40, projects={}, driver=BadDriver()) + assert "SECRET-CANARY" not in str(result) + assert result["effectsPerformed"] is False + + +@pytest.mark.parametrize("boundary", range(1, 93)) +def test_every_durable_write_boundary_resumes_without_adoption( + module, proven, monkeypatch, boundary +): + driver, receipt = proven + save = module._save_receipt + count = 0 + + def crash(value, expected): + nonlocal count + count += 1 + result = save(value, expected) + if count == boundary: + raise OSError("simulated power loss after durable commit") + return result + + monkeypatch.setattr(module, "_save_receipt", crash) + try: + module.quiesce(revision="d" * 40, projects={}, driver=driver) + except OSError: + pass + monkeypatch.setattr(module, "_save_receipt", save) + result = module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert result["status"] in {"quiesced", "already_quiesced"} + assert receipt["value"]["phase"] == "complete" + assert all(not row["state"]["Running"] for row in driver.state["rows"]) + assert sum(e[0] == "TERM" for e in driver.effects) == 12 + + +@pytest.mark.parametrize( + "effect", ["set_registration_disabled", "disable_restart", "term"] +) +def test_effect_commit_before_receipt_completion_is_owned_resume( + module, proven, monkeypatch, effect +): + driver, _ = proven + original = getattr(driver, effect) + crashed = False + + def wrapper(*args): + nonlocal crashed + original(*args) + if not crashed: + crashed = True + raise OSError("simulated loss after external effect") + + monkeypatch.setattr(driver, effect, wrapper) + with pytest.raises(OSError): + module.quiesce(revision="d" * 40, projects={}, driver=driver) + monkeypatch.setattr(driver, effect, original) + assert ( + module.quiesce(revision="d" * 40, projects={}, driver=driver)["status"] + == "quiesced" + ) + + +def test_current_fixed_origins_include_reviewed_backup_and_attempts(module): + assert module.ControllerDriver({}).proof_gaps() == [] + assert module.ORIGINS["workmesh-backup-exercise-39a1a242e4b9"]["run"] == 20184 + assert all(row["attempt"] == 1 for row in module.ORIGINS.values()) + + +def test_malformed_complete_receipt_cannot_reenable_maintenance(module, proven): + driver, receipt = proven + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert module.maintenance_fenced() is False + receipt["value"]["progress"].pop(next(iter(receipt["value"]["progress"]))) + assert module.maintenance_fenced() is True + + +def test_failure_projection_never_misreports_partial_effects(module, proven): + driver, _ = proven + assert module.failure_projection("d" * 40)["effectsPerformed"] is False + driver.timeout = True + module.quiesce(revision="d" * 40, projects={}, driver=driver) + result = module.failure_projection("d" * 40) + assert result["status"] == "partial" + assert result["effectsPerformed"] is True and result["guardHeld"] is True + + +@pytest.fixture +def native_origins(module, monkeypatch): + dates = { + 19421: ("2026-09-29T00:53:46Z", "2026-09-29T00:57:44Z", 768), + 19617: ("2026-09-29T07:36:50Z", "2026-09-29T07:41:39Z", 834), + 20184: ("2026-09-29T22:34:11Z", "2026-09-29T22:35:35Z", 941), + 21300: ("2026-10-01T20:53:43Z", "2026-10-01T20:54:47Z", 1128), + } + tasks, runs, jobs = [], {}, {} + for claim in module.ORIGINS.values(): + start, end, number = dates[claim["run"]] + conclusion = "cancelled" if claim["run"] == 20184 else "failure" + tasks.append( + { + "id": claim["task"], + "run_number": number, + "head_sha": claim["head"], + "workflow_id": "runtime-acceptance.yml", + "status": conclusion, + "created_at": start, + "run_started_at": start, + "updated_at": end, + "url": f"http://127.0.0.1:3000/Jens/WorkMesh/actions/runs/{claim['run']}", + } + ) + runs[claim["run"]] = { + "id": claim["run"], + "run_number": number, + "head_sha": claim["head"], + "status": "completed", + "conclusion": conclusion, + "run_attempt": 1, + "run_started_at": None, + "updated_at": end, + } + jobs[claim["run"]] = [ + { + "id": claim["job"], + "run_id": claim["run"], + "head_sha": claim["head"], + "run_attempt": 1, + "runner_id": claim["runner"], + "started_at": start, + "status": "completed", + "conclusion": conclusion, + } + ] + # Pagination includes unrelated metadata but never returns it publicly. + tasks.extend({"id": i} for i in range(50000, 50055)) + calls = [] + + def api(path, **kwargs): + calls.append(path) + if "/actions/tasks?" in path: + page = int(path.rsplit("=", 1)[-1]) + return { + "total_count": len(tasks), + "workflow_runs": copy.deepcopy(tasks[(page - 1) * 50 : page * 50]), + } + run = int(path.split("/runs/")[1].split("/")[0]) + return ( + {"total_count": len(jobs[run]), "jobs": copy.deepcopy(jobs[run])} + if "/jobs?" in path + else copy.deepcopy(runs[run]) + ) + + monkeypatch.setattr(module.fleet, "_api_json", api) + return tasks, runs, jobs, calls + + +def test_native_documented_task_binding_current_attempt_and_null_run_start( + module, native_origins +): + result = module.ControllerDriver({})._origins() + module._attest_origins(result) + assert all(row["attempt"] == 1 and row["jobsTerminal"] for row in result.values()) + assert len([path for path in native_origins[3] if "/tasks?" in path]) == 2 + + +@pytest.mark.parametrize( + "change", + [ + "task-url", + "task-head", + "task-workflow", + "task-time", + "foreign-job", + "duplicate-job-start", + "later-attempt", + "active-job", + "missing-job", + "missing-attempt", + "terminal-task-changed", + "duplicate-task", + ], +) +def test_native_origin_drift_refuses_authority(module, native_origins, change): + tasks, runs, jobs, _ = native_origins + claim = module.ORIGINS[next(iter(module.ORIGINS))] + task = next(row for row in tasks if row["id"] == claim["task"]) + job = jobs[claim["run"]][0] + if change == "task-url": + task["url"] = task["url"].replace("127.0.0.1", "foreign.invalid") + elif change == "task-head": + task["head_sha"] = "f" * 40 + elif change == "task-workflow": + task["workflow_id"] = "other.yml" + elif change == "task-time": + task["created_at"] = "2026-10-02T00:00:00Z" + elif change == "foreign-job": + job["runner_id"] = 800 + elif change == "duplicate-job-start": + jobs[claim["run"]].append(copy.deepcopy(job)) + elif change == "later-attempt": + runs[claim["run"]]["run_attempt"] = 2 + elif change == "active-job": + job.update(status="in_progress", conclusion=None) + elif change == "missing-job": + jobs[claim["run"]] = [] + elif change == "missing-attempt": + runs[claim["run"]].pop("run_attempt") + elif change == "terminal-task-changed": + task["status"] = "success" + elif change == "duplicate-task": + tasks[-1] = copy.deepcopy(tasks[0]) + with pytest.raises(module.QuiescenceError): + module._attest_origins(module.ControllerDriver({})._origins()) + + +def test_exact_inspect_inventory_cannot_omit_foreign_volume_attachment( + module, monkeypatch +): + driver = module.ControllerDriver({}) + ids = ["a" * 64, "b" * 64] + monkeypatch.setattr(driver, "_docker", lambda *args: [{"Id": ids[0]}]) + with pytest.raises(module.QuiescenceError, match="legacy_inventory_unavailable"): + driver._snapshot_from_ids({}, ids) + + +@pytest.fixture +def product(module, monkeypatch): + source = "/mnt/cache/appdata/gitea-deploy-runners/sources/workmesh" + names = [ + "workmesh-postgres-1", + "workmesh-api-1", + "workmesh-n8n-1", + "workmesh-workmesh-1", + ] + target = { + "slug": "workmesh", + "branch": "main", + "exact_source": True, + "source_dir": source, + "env_source": "/srv/example/appdata/workmesh/.env", + "compose_files": ["compose.yaml"], + "compose_project": "workmesh", + "containers": names, + } + rows = { + name: { + "Id": str(i + 1) * 64, + "Name": "/" + name, + "Image": "sha256:" + str(i + 1) * 64, + "Created": "2026-09-01T00:00:00Z", + "State": {"Running": True, "Health": {"Status": "healthy"}}, + "Config": { + "Labels": { + "com.docker.compose.project": "workmesh", + "com.docker.compose.service": name.removeprefix( + "workmesh-" + ).removesuffix("-1"), + "com.docker.compose.project.working_dir": source, + "com.docker.compose.project.config_files": source + "/compose.yaml", + } + }, + "Mounts": [ + { + "Type": "volume", + "Name": "production_data", + "Source": "/var/lib/docker/volumes/production_data/_data", + "Destination": "/data", + "RW": True, + } + ], + "NetworkSettings": { + "Networks": {"workmesh_default": {"NetworkID": "e" * 64}} + }, + } + for i, name in enumerate(names) + } + info = {"ID": "host-engine", "DockerRootDir": "/var/lib/docker"} + inventory = [row["Id"] for row in rows.values()] + + def run(argv, **kw): + import json + + if argv[1] == "info": + return json.dumps(info) + if argv[1] == "inspect": + return json.dumps([rows[argv[2]]]) + assert argv[1:3] == ["ps", "-a"] + return "\n".join(inventory) + + monkeypatch.setattr(module.fleet, "_run", run) + return target, rows, info, inventory + + +def test_manual_product_observation_protects_exact_healthy_target_without_fake_receipt( + module, product +): + target, _, _, _ = product + proof = module.ControllerDriver({"jens/workmesh": target})._product_binding( + {"engineId": "private"}, [] + ) + assert ( + len(proof["rows"]) == 4 + and proof["receiptStatus"] == "legacy-physical-observation" + ) + + +@pytest.mark.parametrize( + "change", + ["source", "engine", "root", "cid", "health", "mount", "network", "extra-product"], +) +def test_product_disjointness_fails_on_changed_identity_or_overlap( + module, product, change +): + target, rows, info, inventory = product + row = next(iter(rows.values())) + private = [ + { + "id": "a" * 64, + "mounts": [{"Source": "/private-data"}], + "networks": {"private": "b" * 64}, + } + ] + if change == "source": + row["Config"]["Labels"]["com.docker.compose.project.config_files"] = ( + "/foreign/compose.yaml" + ) + elif change == "engine": + info["ID"] = "private" + elif change == "root": + info["DockerRootDir"] = module.ROOTLESS_STORAGE + elif change == "cid": + row["Id"] = "a" * 64 + elif change == "health": + row["State"]["Health"]["Status"] = "unhealthy" + elif change == "mount": + row["Mounts"][0]["Source"] = "/private-data" + elif change == "network": + row["NetworkSettings"]["Networks"]["workmesh_default"]["NetworkID"] = "b" * 64 + elif change == "extra-product": + inventory.append("f" * 64) + with pytest.raises(module.QuiescenceError): + module.ControllerDriver({"jens/workmesh": target})._product_binding( + {"engineId": "private"}, private + ) + + +@pytest.fixture +def real_guards(module, proven, monkeypatch, tmp_path): + import hashlib + + driver = module.ControllerDriver({}) + monkeypatch.setattr(driver, "assert_barrier", lambda: None) + monkeypatch.setattr( + driver, "registrations", lambda: copy.deepcopy(proven[0].runners) + ) + monkeypatch.setattr(driver, "attest_registrations", lambda rows: None) + monkeypatch.setattr(driver, "pending_jobs", lambda: []) + sources = {} + for name in ("runner", "host"): + stanza = ( + "if ! flock -n 8; then\nexit 0\nfi" + if name == "runner" + else "flock -n 8 || exit 0" + ) + raw = ( + '#!/bin/bash\nreadonly MAINTENANCE_LOCK="/srv/example/appdata/gitea-deploy-runners/state/runner-maintenance.lock"\nexec 8>"$MAINTENANCE_LOCK"\n' + + stanza + + "\n# private SECRET-CANARY\n" + ).encode() + path = tmp_path / name + path.write_bytes(raw) + sources[name] = (path, hashlib.sha256(raw).hexdigest()) + monkeypatch.setattr(module, "GUARDS", sources) + monkeypatch.setattr( + module.fleet, "_read_source_file", lambda path, **kw: path.read_bytes() + ) + + def write(path, raw, *, mode, expected_sha256): + module._require( + hashlib.sha256(path.read_bytes()).hexdigest() == expected_sha256, + "guard_cas_failed", + ) + path.write_bytes(raw) + + monkeypatch.setattr(module.fleet, "_atomic_write_owned", write) + # Only the filesystem-independent binding is exercised on Windows here. + monkeypatch.setattr( + module.os, "fstat", lambda fd: type("Info", (), {"st_dev": 1, "st_ino": 2})() + ) + return driver, sources, proven[1] + + +def test_actual_guard_wrapper_owned_cas_and_exact_restore(module, real_guards): + driver, sources, store = real_guards + captured = driver.capture_guards() + value = { + "guards": captured, + "guardProgress": {}, + "registrations": driver.registrations(), + "registrationProgress": {}, + "progress": {row["id"]: "quiesced" for row in module._fixed_rows()}, + } + store["value"] = copy.deepcopy(value) + + def persist(): + store["value"] = copy.deepcopy(value) + + driver.install_guards(value, persist) + assert all( + module.GUARD_MARKER.encode() in path.read_bytes() + for path, _ in sources.values() + ) + value["registrationProgress"] = {"9": "restored", "22": "restored"} + persist() + driver.restore_guards(value, persist) + assert value["guardProgress"] == {"runner": "restored", "host": "restored"} + assert all( + module.hashlib.sha256(path.read_bytes()).hexdigest() == digest + for path, digest in sources.values() + ) + + +@pytest.mark.parametrize( + "stage", [None, "install_intent", "installed", "restore_intent", "restored"] +) +def test_foreign_guard_changes_are_never_adopted_or_overwritten( + module, real_guards, stage +): + driver, sources, store = real_guards + captured = driver.capture_guards() + store["value"] = { + "guards": captured, + "guardProgress": {"runner": stage} if stage else {}, + } + path = sources["runner"][0] + path.write_bytes(path.read_bytes() + b"foreign-change") + before = path.read_bytes() + with pytest.raises(module.QuiescenceError, match="dispatch_guard_changed"): + driver._guard_binding() + assert path.read_bytes() == before + + +@pytest.mark.parametrize( + "failure", ["short-write", "temp-fsync", "before-link", "after-link"] +) +def test_initial_receipt_atomic_publication_faults( + module, monkeypatch, tmp_path, failure +): + import json + + monkeypatch.setattr(module, "STATE_PATH", tmp_path / "receipt.json") + monkeypatch.setattr(module.fleet, "_trusted_directory", lambda path: None) + monkeypatch.setattr( + module.fleet, "_trusted_private_file", lambda path, **kw: path.read_bytes() + ) + receipt = { + "schema": module.SCHEMA, + "revision": "d" * 40, + "manifestDigest": module._digest(module.OBSERVATION), + "phase": "guarding", + "baseline": {}, + "registrations": [], + "registrationProgress": {}, + "progress": {}, + "guards": {"runner": {}, "host": {}}, + "guardProgress": {}, + } + real_open, real_link, real_fsync = module.os.fdopen, module.os.link, module.os.fsync + if failure == "short-write": + + class Short: + def __init__(self, fd, mode): + self.stream = real_open(fd, mode) + + def __enter__(self): + return self + + def __exit__(self, *args): + self.stream.close() + + def write(self, raw): + return self.stream.write(raw[:1]) + + monkeypatch.setattr(module.os, "fdopen", Short) + elif failure == "temp-fsync": + monkeypatch.setattr( + module.os, "fsync", lambda fd: (_ for _ in ()).throw(OSError("fault")) + ) + else: + + def link(src, dst, **kw): + if failure == "after-link": + real_link(src, dst, **kw) + raise OSError("fault") + + monkeypatch.setattr(module.os, "link", link) + with pytest.raises((OSError, module.QuiescenceError)): + module._save_receipt(receipt, None) + if failure == "after-link": + assert json.loads(module.STATE_PATH.read_bytes()) == receipt + assert module._read_receipt() == receipt + else: + assert not module.STATE_PATH.exists() + assert list(tmp_path.glob(".acceptance-receipt-*")) == [] + + +def test_deadline_expires_before_native_read_or_effect(module, monkeypatch): + driver = module.ControllerDriver({}) + driver._deadline = module.time.monotonic() - 1 + monkeypatch.setattr( + module.fleet, "_run", lambda *a, **kw: pytest.fail("no effect after deadline") + ) + monkeypatch.setattr( + module.fleet, + "_api_json", + lambda *a, **kw: pytest.fail("no read after deadline"), + ) + with pytest.raises(module.QuiescenceError, match="maintenance_execution_deadline"): + driver._run(["docker", "info"]) + with pytest.raises(module.QuiescenceError, match="maintenance_execution_deadline"): + driver._api_json("/admin/actions/runners") + + +def test_readonly_poll_reports_real_partial_and_complete_without_resuming( + module, proven +): + driver, receipt = proven + driver.timeout = True + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert module.operation_status()["status"] == "partial" + assert module.operation_status(active=True)["status"] == "in_progress" + before = len(driver.effects) + assert module.operation_status()["phase"] == "quiescing" + assert len(driver.effects) == before + driver.timeout = False + module.quiesce(revision="d" * 40, projects={}, driver=driver) + assert module.operation_status()["status"] == "quiesced" + assert module.operation_status()["completedContainers"] == 18 + + +def test_external_migration_lock_is_nonblocking_and_readonly_busy(module, proven, monkeypatch): + def busy(**kwargs): + assert kwargs == {"blocking": False} + raise module.fleet.RunnerFleetSpecializeError("runner_fleet_specialization_recovery_busy") + monkeypatch.setattr(module.fleet, "_migration_lock", busy) + driver, _ = proven + result = module.preflight(revision="d"*40, projects={}, driver=driver) + assert result["status"] == "in_progress" and result["effectsPerformed"] is False + assert driver.effects == [] diff --git a/backend/tests/test_runner_acceptance_surface.py b/backend/tests/test_runner_acceptance_surface.py new file mode 100644 index 0000000..c24909c --- /dev/null +++ b/backend/tests/test_runner_acceptance_surface.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +import pytest +from sqlmodel import Session, SQLModel, create_engine, select + +from backend.app.db.models import AuditEvent, GatewayUser, JobRun, Policy +from backend.app.services import ( + action_execution, + deployment_broker, + policy_engine, + user_access, +) + + +@pytest.fixture +def session(tmp_path): + engine = create_engine(f"sqlite:///{tmp_path / 'acceptance.db'}") + SQLModel.metadata.create_all(engine) + with Session(engine) as value: + yield value + + +@pytest.mark.parametrize( + "action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"] +) +@pytest.mark.parametrize( + "target,source,expected", + [ + ("tower/ci-acceptance-residue", "chatgpt-mcp", True), + ("tower/ci-acceptance-residue", "manual", True), + ("tower/ci-acceptance-residue", "voice", False), + ("other-target", "chatgpt-mcp", False), + ("*", "manual", False), + ], +) +def test_finite_policy_never_inherits_wildcard_or_voice( + session, action, target, source, expected +): + session.add( + Policy( + id="wildcard", + action=action, + target="*", + manual="allowed", + voice="allowed", + risk="low", + ) + ) + session.commit() + result = policy_engine.evaluate_policy( + session, action=action, target=target, source=source + ) + assert (result["decision"] != "blocked") is expected + assert result["policy_id"] is None + if expected and action.endswith(".quiesce"): + assert result["decision"] == "approval_required" + assert ( + policy_engine.evaluate_policy( + session, action="container.stop", target=target, source=source + )["decision"] + == "blocked" + ) + + +@pytest.mark.parametrize("role", ["operator", "read_only", "custom"]) +@pytest.mark.parametrize( + "action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"] +) +def test_human_nonadmin_cannot_call_new_executor(session, monkeypatch, role, action): + user = GatewayUser( + id="human", + provider="authentik", + issuer="https://issuer", + subject="jens", + role=role, + ) + session.add(user) + session.commit() + actor = user_access.actor_identity.set("authentik:jens") + try: + monkeypatch.setattr( + deployment_broker.DeploymentBrokerClient, + "runner_acceptance", + lambda *a, **k: pytest.fail("must not call broker"), + ) + result = action_execution.execute_governed_action( + session, + action=action, + target="tower/ci-acceptance-residue", + source="manual", + risk="high", + reason="inspect fixture", + original_policy_decision="allowed", + ) + assert result["policyDecision"] == "blocked" + assert session.exec(select(JobRun)).all() == [] + finally: + user_access.actor_identity.reset(actor) + + +@pytest.mark.parametrize( + "action,status,success", + [ + ("runner.acceptance.preflight", "ready", True), + ("runner.acceptance.preflight", "blocked", False), + ("runner.acceptance.quiesce", "quiesced", True), + ("runner.acceptance.quiesce", "already_quiesced", True), + ("runner.acceptance.quiesce", "partial", False), + ("runner.acceptance.quiesce", "blocked", False), + ("runner.acceptance.quiesce", "ready", False), + ], +) +def test_typed_mcp_records_actual_controller_result( + session, monkeypatch, action, status, success +): + monkeypatch.setattr( + deployment_broker.DeploymentBrokerClient, + "runner_acceptance", + lambda *_, **kw: { + "status": status, + "revision": "a" * 40, + "blockers": ["terminal_origin_claim_missing"], + }, + ) + result = action_execution.execute_governed_action( + session, + action=action, + target="tower/ci-acceptance-residue", + source="chatgpt-mcp", + risk="high", + reason="inspect fixture", + original_policy_decision="allowed", + ) + assert result["success"] is success + assert session.get(JobRun, result["jobId"]).status == ( + "succeeded" if success else "failed" + ) + assert ( + session.get(AuditEvent, result["auditId"]).target + == "tower/ci-acceptance-residue" + ) + assert result["blockers"] == ["terminal_origin_claim_missing"] + + +@pytest.mark.parametrize( + "target,source,decision", + [ + ("other", "chatgpt-mcp", "allowed"), + ("tower/ci-acceptance-residue", "voice", "allowed"), + ("tower/ci-acceptance-residue", "chatgpt-mcp", "blocked"), + ("tower/ci-acceptance-residue", "manual", "allowed"), + ], +) +def test_unsupported_target_or_auth_refuses_before_job( + session, monkeypatch, target, source, decision +): + monkeypatch.setattr( + deployment_broker.DeploymentBrokerClient, + "runner_acceptance", + lambda *a, **k: pytest.fail("must not call broker"), + ) + result = action_execution.execute_governed_action( + session, + action="runner.acceptance.quiesce", + target=target, + source=source, + risk="high", + reason="inspect fixture", + original_policy_decision=decision, + ) + assert result["policyDecision"] == "blocked" + assert session.exec(select(JobRun)).all() == [] + + +@pytest.mark.parametrize( + "preflight,path", + [ + (True, "/v1/appops/runner-acceptance/preflight"), + (False, "/v1/appops/runner-acceptance/quiesce"), + ], +) +def test_broker_has_fixed_signed_path_and_bounded_body(monkeypatch, preflight, path): + calls = [] + monkeypatch.setattr( + deployment_broker.DeploymentBrokerClient, + "_request", + lambda _, *a, **kw: calls.append((a, kw)) or {}, + ) + deployment_broker.DeploymentBrokerClient().runner_acceptance( + reason="x" * 1100, preflight=preflight + ) + assert calls[0][0] == ("POST", path) + assert calls[0][1]["payload"] == {"reason": "x" * 1000} + assert "secret" in calls[0][1] + + +def test_controller_canary_is_discarded_before_job_and_audit(session, monkeypatch): + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **k: { + "status": "Bearer SECRET-CANARY", "reason": "Bearer SECRET-CANARY", "revision": "SECRET-CANARY", + "blockers": ["SECRET-CANARY"], "verifiedRunners": ["SECRET-CANARY"], "projectbrainDeliveryRequired": "SECRET-CANARY", + }) + result = action_execution.execute_governed_action(session, action="runner.acceptance.quiesce", target="tower/ci-acceptance-residue", + source="chatgpt-mcp", risk="high", reason="inspect fixture", + original_policy_decision="allowed") + assert result["success"] is False + assert result["blockers"] == ["controller_response_untrusted"] + assert "SECRET-CANARY" not in str(result) + assert "SECRET-CANARY" not in str(session.get(JobRun, result["jobId"])) + assert "SECRET-CANARY" not in str(session.get(AuditEvent, result["auditId"])) + + +def test_acceptance_transport_loss_is_unknown_never_effect_free(monkeypatch): + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "_request", lambda *a, **kw: {"status": "unreachable"}) + result = deployment_broker.DeploymentBrokerClient().runner_acceptance(reason="preserve residue", preflight=False) + assert result["status"] == "unknown" and result["noMutation"] is False + assert result["effectsPerformed"] is None + + +def test_acceptance_unknown_transport_is_preserved_in_job_result_and_audit(session, monkeypatch): + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **kw: + {"status": "unknown", "noMutation": False, "effectsPerformed": None, + "blockers": ["maintenance_outcome_unknown"]}) + result = action_execution.execute_governed_action(session, action="runner.acceptance.quiesce", + target="tower/ci-acceptance-residue", source="chatgpt-mcp", risk="high", reason="preserve residue", + original_policy_decision="allowed") + assert result["controllerStatus"] == "unknown" and result["result"] == "unknown" + assert result["noMutation"] is False and result["effectsPerformed"] is None + + +@pytest.mark.parametrize("action", ["runner.acceptance.preflight", "runner.acceptance.quiesce"]) +def test_actual_mcp_request_preserves_actor_guard_and_never_grants_automation(session, monkeypatch, action): + from backend.app.services import mcp_operations + session.add(GatewayUser(id="operator", provider="authentik", issuer="test", subject="operator", role="operator")) + session.commit() + token = user_access.actor_identity.set("authentik:operator") + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", + lambda *a, **k: pytest.fail("nonadmin must not reach broker")) + try: + result = mcp_operations.request_operation(session, action=action, target="tower/ci-acceptance-residue", + reason="preserve finite residue", idempotency_key="nonadmin-"+action) + finally: + user_access.actor_identity.reset(token) + assert result["policyDecision"] == "blocked" + assert session.exec(select(JobRun)).all() == [] + assert policy_engine.evaluate_policy(session, action=action, target="tower/ci-acceptance-residue", source="automation")["decision"] == "blocked" + profile = action_execution.action_execution_profile(action, "allowed", "operator") + assert profile["executionMode"] == "governance_only" and profile["commandExecutionSupported"] is False + + +@pytest.mark.parametrize("action,target", [("runner.acceptance.quiesce", "other-target"), + ("runner.acceptance.preflight", "*"), + ("container.stop", "tower/ci-acceptance-residue")]) +def test_actual_mcp_request_keeps_wrong_target_and_generic_stop_denied(session, monkeypatch, action, target): + from backend.app.services import mcp_operations + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_acceptance", lambda *a, **k: pytest.fail("must remain denied")) + result = mcp_operations.request_operation(session, action=action, target=target, + reason="preserve finite residue", idempotency_key="deny-"+action) + assert result["policyDecision"] == "blocked" and result["executed"] is False + assert session.exec(select(JobRun)).all() == [] + assert mcp_operations.mcp_policy_source(action, target) == "automation" diff --git a/backend/tests/test_runner_resource_envelope.py b/backend/tests/test_runner_resource_envelope.py index 403d264..cf8a68f 100644 --- a/backend/tests/test_runner_resource_envelope.py +++ b/backend/tests/test_runner_resource_envelope.py @@ -4,6 +4,7 @@ import copy import importlib.util import json import os +import subprocess import sys from pathlib import Path from types import SimpleNamespace @@ -649,7 +650,11 @@ def test_inherited_image_labels_are_exact_and_owned_probe_cleanup_is_bounded(mon inspected = iter([probe, copy.deepcopy(probe), None]) monkeypatch.setattr(envelope, "_inspect_host_probe_by_name", lambda: next(inspected)) calls = [] - monkeypatch.setattr(envelope, "_run", lambda args, **kw: calls.append(args)) + def remove(args, **kw): + calls.append(args) + assert kw == {"check": False, "capture_output": True, "timeout": 10} + return SimpleNamespace(returncode=0, stderr=b"") + monkeypatch.setattr(envelope.subprocess, "run", remove) assert envelope.cleanup_stopped_host_probe(image_id=image_id, code_source=code_source) assert calls == [["docker", "rm", probe["Id"]]] @@ -671,10 +676,85 @@ def test_probe_cleanup_refuses_foreign_running_or_changed_identity(monkeypatch, inspected = iter([probe, fresh]) monkeypatch.setattr(envelope, "_inspect_host_probe_by_name", lambda: next(inspected)) monkeypatch.setattr(envelope, "_run", lambda *a, **kw: pytest.fail("No removal or probe execution")) + monkeypatch.setattr(envelope, "_remove_cleanup_probe", lambda *a: pytest.fail("No removal")) with pytest.raises(RuntimeError, match="resource_cgroup_probe_cleanup_identity_"): envelope.cleanup_stopped_host_probe(image_id=image_id, code_source=code_source) +@pytest.mark.parametrize("stderr,condition", [ + (b"private-canary: no space left on device", "remove_storage_full"), + (b"private-canary: read-only file system", "remove_read_only"), + (b"private-canary: permission denied", "remove_permission_denied"), + (b"Cannot connect to the Docker daemon at private-canary", "remove_daemon_unavailable"), + (b"private-canary: device or resource busy", "remove_busy"), + (b"Removal of container private-canary is already in progress", "remove_busy"), + (b"No such container: private-canary", "remove_container_missing"), + (b"private-canary: client version is too old", "remove_api_version_unsupported"), + (b"removal of container private-canary failed unexpectedly", "remove_failed"), + (b"private-canary", "remove_failed"), + (b"no space left on device" + b"x" * 4096, "remove_failed"), +]) +def test_probe_removal_has_finite_diagnostics_and_never_forces(monkeypatch, stderr, condition): + envelope = _module() + identifier = "d" * 64 + def run(args, **kw): + assert args == ["docker", "rm", identifier] + assert kw == {"check": False, "capture_output": True, "timeout": 10} + return SimpleNamespace(returncode=1, stderr=stderr, stdout=b"private-canary") + monkeypatch.setattr(envelope.subprocess, "run", run) + with pytest.raises(envelope.HostProbeCleanupError) as error: + envelope._remove_cleanup_probe(identifier) + assert str(error.value) == "resource_cgroup_probe_cleanup_" + condition + assert "private-canary" not in str(error.value) + assert error.value.__cause__ is None + + +@pytest.mark.parametrize("identifier", ["probe", "d" * 12, "d" * 64 + " --force", "../private-canary"]) +def test_probe_removal_rejects_selectors_before_any_command(monkeypatch, identifier): + envelope = _module() + monkeypatch.setattr(envelope.subprocess, "run", lambda *a, **kw: pytest.fail("No command")) + with pytest.raises(envelope.HostProbeCleanupError, match="cleanup_identity_invalid"): + envelope._remove_cleanup_probe(identifier) + + +@pytest.mark.parametrize("stage", ["image_metadata_failed", "inspect_failed", "reinspect_failed", "absence_failed"]) +def test_probe_read_failures_drop_untrusted_exception_text(stage): + envelope = _module() + def fail(): + raise RuntimeError("Bearer private-canary") + with pytest.raises(envelope.HostProbeCleanupError) as error: + envelope._cleanup_probe_read(stage, fail) + assert str(error.value) == "resource_cgroup_probe_cleanup_" + stage + assert error.value.__cause__ is None + + +@pytest.mark.skipif(os.name == "nt", reason="Docker CLI shim uses POSIX executable semantics") +@pytest.mark.parametrize("failure,condition", [("read-only file system private-canary", "remove_read_only"), + ("Bearer private-canary", "remove_failed")]) +def test_real_probe_cleanup_cli_exposes_only_finite_child_failure(tmp_path, failure, condition): + envelope = _module() + probe, image_id, code_source = _host_probe_fixture(envelope) + fixture = tmp_path / "fixture.json" + fixture.write_text(json.dumps({"probe": probe, "image": image_id, "failure": failure})) + shim = tmp_path / "docker" + shim.write_text("#!" + sys.executable + "\n" + + "import json,sys,pathlib\n" + + "v=json.loads(pathlib.Path(" + repr(str(fixture)) + ").read_text())\n" + + "a=sys.argv[1:]\n" + + "if a[:2]==['image','inspect']: print(json.dumps([{'Id':v['image'],'Config':{'Labels':{}}}]))\n" + + "elif a[:2]==['ps','-a']: print(v['probe']['Id'])\n" + + "elif a[0]=='inspect': print(json.dumps([v['probe']]))\n" + + "elif a==['rm',v['probe']['Id']]: print(v['failure'],file=sys.stderr); sys.exit(1)\n" + + "else: sys.exit(99)\n") + shim.chmod(0o700) + result = subprocess.run([sys.executable, str(ENVELOPE_PATH), "--cleanup-stopped-host-probe", + image_id, code_source], capture_output=True, timeout=15, + env={**os.environ, "PATH": str(tmp_path) + os.pathsep + os.environ.get("PATH", "")}) + assert result.returncode == 1 + assert result.stdout == b"" + assert result.stderr == ("resource_cgroup_probe_cleanup_" + condition + "\n").encode() + + @pytest.mark.parametrize("labels", [None, {}, {"org.opencontainers.image.title": "runner"}]) def test_probe_label_metadata_comes_from_exact_image(monkeypatch, labels): envelope = _module() diff --git a/backend/tests/test_runner_specialization_controller_surface.py b/backend/tests/test_runner_specialization_controller_surface.py index a89f2a9..aed3c1f 100644 --- a/backend/tests/test_runner_specialization_controller_surface.py +++ b/backend/tests/test_runner_specialization_controller_surface.py @@ -156,3 +156,54 @@ def test_statistics_configuration_caller_cannot_select_credential_or_endpoint(co monkeypatch.setattr(controller, 'configure_runner_statistics_binding', lambda **kwargs: pytest.fail('must not write configuration')) instance.do_POST() assert responses == [(400, {'status': 'invalid_request'})] + + +@pytest.mark.parametrize("stage", ["preflight", "quiesce"]) +def test_acceptance_uses_fixed_signed_route_and_protected_source(controller, monkeypatch, stage): + import runner_acceptance_quiescence + instance, responses, signatures = handler(controller, monkeypatch) + instance.path = getattr(controller, "RUNNER_ACCEPTANCE_" + stage.upper() + "_PATH") + calls = [] + monkeypatch.setattr(runner_acceptance_quiescence, stage, + lambda **kw: calls.append(kw) or {"status": "blocked", "blockers": ["terminal_origin_claim_missing"]}) + instance.do_POST() + assert signatures == [{"method": "POST", "path": instance.path}] + assert calls == [{"revision": "a" * 40, "projects": {controller.SELF_UPGRADE_REPOSITORY: {"branch": "main"}}}] + assert responses == [(200, {"status": "blocked", "blockers": ["terminal_origin_claim_missing"]})] + + +@pytest.mark.parametrize("stage", ["preflight", "quiesce"]) +@pytest.mark.parametrize("gate", ["untrusted", "invalid_signature", "source_mismatch", "push_pending", "wrong_head"]) +def test_acceptance_security_gate_refuses_before_executor(controller, monkeypatch, stage, gate): + import runner_acceptance_quiescence + instance, responses, _ = handler(controller, monkeypatch) + instance.path = getattr(controller, "RUNNER_ACCEPTANCE_" + stage.upper() + "_PATH") + monkeypatch.setattr(runner_acceptance_quiescence, stage, lambda **_: pytest.fail("no executor before source/auth gate")) + if gate == "untrusted": + monkeypatch.setattr(instance, "_trusted_client", lambda: False) + elif gate == "invalid_signature": + monkeypatch.setattr(controller, "_self_upgrade_signature_is_valid", lambda *_, **__: False) + elif gate == "source_mismatch": + monkeypatch.setattr(controller, "_require_controller_release", lambda: {"sourceRevision": "b" * 40, "sourceRepository": controller.SELF_UPGRADE_REPOSITORY}) + elif gate == "push_pending": + monkeypatch.setattr(controller, "_self_upgrade_preflight", lambda _: {"status": "preflight_rejected"}) + else: + monkeypatch.setattr(controller, "_is_current_branch_head", lambda *args: False) + instance.do_POST() + assert responses[0][0] in {403, 409} + + +@pytest.mark.parametrize("field", ["containerId", "project", "revision", "daemon", "command", "credential", "path"]) +def test_acceptance_caller_cannot_choose_any_effect_identity(controller, monkeypatch, field): + instance, responses, _ = handler(controller, monkeypatch, {"reason": "inspect fixture", field: "SECRET-CANARY"}) + instance.path = controller.RUNNER_ACCEPTANCE_QUIESCE_PATH + instance.do_POST() + assert responses == [(400, {"status": "invalid_request"})] + assert "SECRET-CANARY" not in str(responses) + + +def test_pending_acceptance_receipt_fences_legacy_maintenance(controller, monkeypatch): + import runner_acceptance_quiescence + monkeypatch.setattr(runner_acceptance_quiescence, "maintenance_fenced", lambda: True) + monkeypatch.setattr(controller.RunnerFleetService, "automatic_cycle", lambda _: pytest.fail("no re-enable during partial receipt")) + assert controller._runner_maintenance_cycle() == "acceptance_maintenance_owned"