Expose generic web rollback planner route

control_plane/drivers/registry.py

@@ -312,7 +312,7 @@ def _route_alias(
             capability_id="image_deployable",
             label="Image deployable",
             description="Deploy immutable container images and record stable-lane deployment evidence.",
-            actions=("stable_deploy", "prod_promotion"),
+            actions=("stable_deploy", "prod_promotion", "prod_rollback_plan"),
             panels=("lane_health", "deployment_evidence", "promotion_evidence"),
         ),
         DriverCapabilityDescriptor(
@@ -373,6 +373,16 @@ def _route_alias(
             authz_action="generic_web_prod_promotion.dispatch",
             writes_records=(),
         ),
+        _action(
+            "prod_rollback_plan",
+            "Plan prod rollback",
+            "Build and persist a generic-web rollback plan from a previous good deployment record.",
+            safety="safe_write",
+            scope="instance",
+            route_path="/v1/drivers/generic-web/prod-rollback-plan",
+            authz_action="generic_web_prod_rollback.plan",
+            writes_records=("generic_web_rollback_plan",),
+        ),
         _action(
             "stable_verification",
             "Record stable verification",

control_plane/service.py

@@ -75,6 +75,10 @@
 from control_plane.contracts.every_code_summary_read_model import (
     build_every_code_summary_read_model,
 )
+from control_plane.contracts.generic_web_rollback import (
+    GenericWebRollbackPlanRequest,
+    execute_generic_web_rollback_plan,
+)
 from control_plane.contracts.idempotency_record import LaunchplaneIdempotencyRecord
 from control_plane.contracts.idempotency_record import build_launchplane_idempotency_record_id
 from control_plane.contracts.merge_train_batch import (
@@ -887,6 +891,28 @@ def _validate_alignment(self) -> "GenericWebPromotionWorkflowEnvelope":
 )
 
 
+class GenericWebRollbackPlanEnvelope(_ProductRouteEnvelope):
+    schema_version: int = Field(default=1, ge=1)
+    rollback_plan: GenericWebRollbackPlanRequest
+
+    @model_validator(mode="after")
+    def _validate_alignment(self) -> "GenericWebRollbackPlanEnvelope":
+        if not self.product.strip():
+            raise ValueError("generic web rollback plan requires product")
+        if self.product.strip() != self.rollback_plan.product.strip():
+            raise ValueError("generic web rollback plan requires matching product values")
+        return self
+
+
+_GENERIC_WEB_ROLLBACK_PLAN_ROUTE = _DriverRouteExecutionMetadata(
+    route_path="/v1/drivers/generic-web/prod-rollback-plan",
+    envelope_model=GenericWebRollbackPlanEnvelope,
+    denial_message=(
+        "Workflow cannot plan the generic web prod rollback for the requested product/context."
+    ),
+)
+
+
 class GenericWebStableVerificationRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -1270,6 +1296,7 @@ def _odoo_preview_identifier(value: str, *, suffix: str) -> str:
         _GENERIC_WEB_DEPLOY_ROUTE.route_path,
         _GENERIC_WEB_PROD_PROMOTION_ROUTE.route_path,
         _GENERIC_WEB_PROD_PROMOTION_WORKFLOW_ROUTE.route_path,
+        _GENERIC_WEB_ROLLBACK_PLAN_ROUTE.route_path,
         _GENERIC_WEB_STABLE_VERIFICATION_ROUTE.route_path,
     }
 )
@@ -4903,6 +4930,7 @@ def _accepted_payload(
         "odoo_stable_bootstrap_operation_id",
         "odoo_stable_target_replacement_operation_id",
         "runner_host_hygiene_audit_record_key",
+        "generic_web_rollback_plan_id",
     }
     records: dict[str, object] = {}
     for key, value in result.items():
@@ -11577,6 +11605,49 @@ def product_action_allowed(
                     request=generic_web_workflow_request.workflow,
                 )
                 result = driver_result.model_dump(mode="json")
+            elif path == _GENERIC_WEB_ROLLBACK_PLAN_ROUTE.route_path:
+                generic_web_rollback_request = (
+                    _GENERIC_WEB_ROLLBACK_PLAN_ROUTE.envelope_model.model_validate(payload)
+                )
+                resolved_driver_context = _resolve_descriptor_product_driver_context(
+                    record_store=record_store,
+                    route_path=path,
+                    product=generic_web_rollback_request.rollback_plan.product,
+                    instance=generic_web_rollback_request.rollback_plan.instance,
+                    require_profile=True,
+                )
+                if resolved_driver_context.profile is None or resolved_driver_context.lane is None:
+                    raise ProductDriverMismatchError(
+                        "Generic web rollback plan requires a product profile lane."
+                    )
+                authorization_response = _driver_route_authorization_response(
+                    authz_policy=authz_policy,
+                    identity=identity,
+                    route_path=path,
+                    product=resolved_driver_context.profile.product,
+                    context=resolved_driver_context.lane.context,
+                    denial_message=_GENERIC_WEB_ROLLBACK_PLAN_ROUTE.denial_message,
+                    start_response=start_response,
+                    trace_id=request_trace_id,
+                )
+                if authorization_response is not None:
+                    return authorization_response
+                idempotent_response = _check_idempotent_request(
+                    record_store=record_store,
+                    scope=request_scope,
+                    route_path=path,
+                    idempotency_key=request_idempotency_key,
+                    request_fingerprint=request_fingerprint,
+                    start_response=start_response,
+                    trace_id=request_trace_id,
+                )
+                if idempotent_response is not None:
+                    return idempotent_response
+                driver_result = execute_generic_web_rollback_plan(
+                    record_store=record_store,
+                    request=generic_web_rollback_request.rollback_plan,
+                )
+                result = {"generic_web_rollback_plan_id": driver_result.plan_id}
             elif path in _PREVIEW_DESIRED_STATE_ROUTE_PATHS:
                 generic_web_desired_state_request, profile, authorization_response = (
                     _authorize_generic_web_preview_route(

docs/dokploy-service-deployments.md

@@ -173,7 +173,8 @@ exist, Launchplane verifies the source and destination lane health around the
 deployment and writes promotion evidence.
 
 Rollback begins with a Launchplane-owned rollback plan. The generic-web planner
-is a safe-write contract: it reads the product profile, destination lane, a
+is exposed through `POST /v1/drivers/generic-web/prod-rollback-plan` as a
+safe-write contract: it reads the product profile, destination lane, a
 Launchplane deployment record selected as the rollback target, and optional
 backup-gate evidence, then writes a rollback-plan record. It does not mutate
 Dokploy or trigger a product workflow.

docs/driver-descriptors.md

@@ -119,6 +119,14 @@ prod inventory after successful verified deploys. Product-specific drivers such
 as VeriReel or Odoo can wrap this common action when they need additional gates
 such as backups, migrations, rollout checks, or tenant-specific validation.
 
+The `prod_rollback_plan` action routes to
+`POST /v1/drivers/generic-web/prod-rollback-plan`. It is a safe-write planner:
+Launchplane reads the product profile, destination lane, selected deployment
+record, and optional backup gate evidence, then writes a
+`GenericWebRollbackPlanRecord`. It does not mutate the provider. A future
+explicit apply action can consume a ready plan and call the normal generic-web
+deploy path.
+
 The `stable_verification` action routes to
 `POST /v1/drivers/generic-web/stable-verification`. Product workflows submit the
 deployment record, optional promotion record, checked URLs, and pass/fail status;

docs/service-boundary.md

@@ -90,6 +90,7 @@ VeriReel product paths:
   - `POST /v1/drivers/generic-web/deploy`
   - `POST /v1/drivers/generic-web/prod-promotion`
   - `POST /v1/drivers/generic-web/prod-promotion-workflow`
+  - `POST /v1/drivers/generic-web/prod-rollback-plan`
   - `POST /v1/drivers/generic-web/stable-verification`
   - `POST /v1/drivers/generic-web/preview-desired-state`
   - `POST /v1/drivers/generic-web/preview-refresh`

tests/test_driver_descriptors.py

@@ -208,6 +208,11 @@ def test_generic_web_descriptor_is_provider_neutral_base_driver(self) -> None:
         actions = {action.action_id: action for action in descriptor.actions}
         self.assertEqual(actions["stable_deploy"].route_path, "/v1/drivers/generic-web/deploy")
         self.assertEqual(actions["stable_deploy"].safety, "mutation")
+        self.assertEqual(
+            actions["prod_rollback_plan"].route_path,
+            "/v1/drivers/generic-web/prod-rollback-plan",
+        )
+        self.assertEqual(actions["prod_rollback_plan"].safety, "safe_write")
         self.assertEqual(
             actions["stable_verification"].route_path,
             "/v1/drivers/generic-web/stable-verification",
@@ -375,6 +380,11 @@ def test_generic_web_execution_metadata_matches_descriptors(self) -> None:
                     control_plane_service.GenericWebPromotionWorkflowEnvelope,
                     "prod promotion workflow",
                 ),
+                "prod_rollback_plan": (
+                    control_plane_service._GENERIC_WEB_ROLLBACK_PLAN_ROUTE,
+                    control_plane_service.GenericWebRollbackPlanEnvelope,
+                    "rollback",
+                ),
                 "stable_verification": (
                     control_plane_service._GENERIC_WEB_STABLE_VERIFICATION_ROUTE,
                     control_plane_service.GenericWebStableVerificationEnvelope,

tests/test_service.py

@@ -13768,6 +13768,143 @@ def test_generic_web_deploy_route_accepts_padded_lane_context(self) -> None:
         self.assertEqual(payload["records"]["deployment_record_id"], "deployment-syo-testing")
         deploy.assert_called_once()
 
+    def test_generic_web_rollback_plan_route_writes_plan_record(self) -> None:
+        with TemporaryDirectory() as temporary_directory_name:
+            root = Path(temporary_directory_name)
+            state_dir = root / "state"
+            store = FilesystemRecordStore(state_dir=state_dir)
+            store.write_product_profile_record(
+                LaunchplaneProductProfileRecord.model_validate(_product_profile_payload_with_prod())
+            )
+            deployment_record = DeploymentRecord(
+                record_id="deployment-syo-prod-previous",
+                artifact_identity=ArtifactIdentityReference(
+                    artifact_id="ghcr.io/cbusillo/sellyouroutboard@sha256:abc123"
+                ),
+                context="sellyouroutboard-testing",
+                instance="prod",
+                source_git_ref="abc123",
+                destination_health=HealthcheckEvidence(status="pass"),
+                resolved_target=ResolvedTargetEvidence(
+                    target_type="application",
+                    target_id="app-prod",
+                    target_name="syo-prod-app",
+                ),
+                deploy=DeploymentEvidence(
+                    target_name="syo-prod-app",
+                    target_type="application",
+                    deploy_mode="dokploy-application-api",
+                    deployment_id="deployment-provider-1",
+                    status="pass",
+                    started_at="2026-05-25T12:00:00Z",
+                    finished_at="2026-05-25T12:01:00Z",
+                ),
+            )
+            store.write_deployment_record(deployment_record)
+            policy = LaunchplaneAuthzPolicy.model_validate(
+                {
+                    "github_actions": [
+                        {
+                            "repository": "every/verireel",
+                            "workflow_refs": [
+                                "every/verireel/.github/workflows/preview-control-plane.yml@refs/heads/main"
+                            ],
+                            "event_names": ["pull_request"],
+                            "products": ["sellyouroutboard"],
+                            "contexts": ["sellyouroutboard-testing"],
+                            "actions": ["generic_web_prod_rollback.plan"],
+                        }
+                    ]
+                }
+            )
+            app = create_launchplane_service_app(
+                state_dir=state_dir,
+                verifier=_StubVerifier(_identity()),
+                authz_policy=policy,
+                control_plane_root_path=root,
+            )
+
+            status_code, payload = _invoke_app(
+                app,
+                method="POST",
+                path="/v1/drivers/generic-web/prod-rollback-plan",
+                payload={
+                    "schema_version": 1,
+                    "product": "sellyouroutboard",
+                    "rollback_plan": {
+                        "schema_version": 1,
+                        "product": "sellyouroutboard",
+                        "instance": "prod",
+                        "rollback_deployment_record_id": "deployment-syo-prod-previous",
+                    },
+                },
+                headers={"Idempotency-Key": "generic-web-rollback-plan-syo-prod"},
+            )
+
+            plans = store.list_generic_web_rollback_plan_records(
+                context_name="sellyouroutboard-testing",
+                instance_name="prod",
+                limit=1,
+            )
+            plan = plans[0]
+
+        self.assertEqual(status_code, 202)
+        self.assertEqual(payload["records"]["generic_web_rollback_plan_id"], plan.plan_id)
+        self.assertEqual(plan.status, "ready")
+        self.assertEqual(plan.product, "sellyouroutboard")
+        self.assertEqual(plan.context, "sellyouroutboard-testing")
+        self.assertEqual(plan.rollback_deployment_record_id, "deployment-syo-prod-previous")
+
+    def test_generic_web_rollback_plan_route_rejects_unauthorized_context(self) -> None:
+        with TemporaryDirectory() as temporary_directory_name:
+            root = Path(temporary_directory_name)
+            state_dir = root / "state"
+            store = FilesystemRecordStore(state_dir=state_dir)
+            store.write_product_profile_record(
+                LaunchplaneProductProfileRecord.model_validate(_product_profile_payload_with_prod())
+            )
+            policy = LaunchplaneAuthzPolicy.model_validate(
+                {
+                    "github_actions": [
+                        {
+                            "repository": "every/verireel",
+                            "workflow_refs": [
+                                "every/verireel/.github/workflows/preview-control-plane.yml@refs/heads/main"
+                            ],
+                            "event_names": ["pull_request"],
+                            "products": ["sellyouroutboard"],
+                            "contexts": ["other-context"],
+                            "actions": ["generic_web_prod_rollback.plan"],
+                        }
+                    ]
+                }
+            )
+            app = create_launchplane_service_app(
+                state_dir=state_dir,
+                verifier=_StubVerifier(_identity()),
+                authz_policy=policy,
+                control_plane_root_path=root,
+            )
+
+            status_code, payload = _invoke_app(
+                app,
+                method="POST",
+                path="/v1/drivers/generic-web/prod-rollback-plan",
+                payload={
+                    "schema_version": 1,
+                    "product": "sellyouroutboard",
+                    "rollback_plan": {
+                        "schema_version": 1,
+                        "product": "sellyouroutboard",
+                        "instance": "prod",
+                        "rollback_deployment_record_id": "deployment-syo-prod-previous",
+                    },
+                },
+            )
+
+        self.assertEqual(status_code, 403)
+        self.assertEqual(payload["error"]["code"], "authorization_denied")
+
     def test_generic_web_deploy_route_resolves_literal_generic_web_profile(self) -> None:
         with TemporaryDirectory() as temporary_directory_name:
             root = Path(temporary_directory_name)