feat: implement destroy_deployment with S3 remote state backend

- Add TF_STATE_BUCKET/DYNAMODB support (bucket + table created on shared AWS account)
- TerraformRunner.destroy() now retries with 10s/30s/60s backoff (AWS eventual consistency)
- TerraformRunner.init() adds -input=false and -migrate-state -force-copy (fixes silent hangs and interactive prompts under non-interactive docker exec)
- DeployOpsAgent.destroy_deployment() + _describe_remaining_resources() for post-failure diagnostics
- New status 'destroyed' on Deployment model + Alembic migration
- POST /api/jobs/{job_id}/destroy: owner-only auth, server-side service_name confirmation, no false success on failure
- Bind-mount alembic/ in docker-compose.yml so future migrations are visible without rebuilding the image
- Verified end-to-end: real terraform destroy succeeded against AWS, confirmed via ECS describe

Author: hbibbensalem

alembic/versions/b7f3a9c1d2e4_0002_add_destroyed_deployment_status.py

@@ -0,0 +1,40 @@
+"""0002 add destroyed deployment status
+
+Revision ID: b7f3a9c1d2e4
+Revises: 848480a9e6e6
+Create Date: 2026-08-18 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b7f3a9c1d2e4'
+down_revision: Union[str, None] = '848480a9e6e6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 'destroyed' is distinct from 'rolled_back': a rollback leaves a
+    # previous ECS task-definition revision running, a destroy leaves
+    # nothing -- feature/destroy-deployment needs its own terminal status
+    # rather than overloading 'rolled_back'.
+    op.drop_constraint('ck_deployments_status', 'deployments', type_='check')
+    op.create_check_constraint(
+        'ck_deployments_status',
+        'deployments',
+        "status IN ('pending', 'applying', 'succeeded', 'failed', 'rolled_back', 'destroyed')",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('ck_deployments_status', 'deployments', type_='check')
+    op.create_check_constraint(
+        'ck_deployments_status',
+        'deployments',
+        "status IN ('pending', 'applying', 'succeeded', 'failed', 'rolled_back')",
+    )

infrastructure/docker-compose.yml

@@ -70,14 +70,15 @@ services:
       REDIS_URL: redis://redis:6379/2
       QDRANT_URL: http://qdrant:6333
       # Persisted across container recreation (unlike bare /tmp), so a
-      # terraform.tfstate from a real apply survives a backend rebuild —
+      # terraform.tfstate from a real apply survives a backend rebuild —
       # lost one on 2026-08-12 mid-session, had to hand-delete AWS resources.
       DEPLOYOPS_WORKSPACE_ROOT: /data/deployops
     ports:
       - "${BACKEND_PORT:-8000}:8000"
     volumes:
       - deployops_workspace:/data/deployops
       - ../src:/app/src
+      - ../alembic:/app/alembic
       - /var/run/docker.sock:/var/run/docker.sock
     depends_on:
       postgres:

src/agents/deployops/agent.py

@@ -471,6 +471,130 @@ async def rollback(self, job_id: str, payload: Dict[str, Any]) -> Dict[str, Any]
             self.logger.error(f"Rollback failed: {e}")
             return {"status": "failed", "error": str(e)}
 
+    @xray_recorder.capture("destroy_deployment")  # type: ignore[reportCallIssue]
+    async def destroy_deployment(
+        self,
+        job_id: str,
+        aws_config: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Destroy a job's deployed infrastructure via `terraform destroy`,
+        reusing the persisted remote state in TF_STATE_BUCKET (see
+        _write_remote_state_backend). Falls back to reporting "no state" for
+        deployments made before remote state was enabled, rather than
+        pretending to succeed. On failure, describes what's still live in
+        AWS (mirrors rollback()/the /monitoring endpoint) so the caller can
+        tell the user precisely what to clean up manually -- per the
+        feature/destroy-deployment design decisions.
+        """
+        workspace_dir = self._workspace_dir(job_id)
+        tf_dir = workspace_dir / "terraform"
+        tf_dir.mkdir(parents=True, exist_ok=True)
+        self._write_remote_state_backend(tf_dir, job_id=job_id)
+
+        if not (tf_dir / "backend.tf").exists():
+            self.logger.warning(f"[{job_id}] Destroy: TF_STATE_BUCKET not configured, cannot recover state")
+            return {
+                "status": "no_state",
+                "job_id": job_id,
+                "message": (
+                    "TF_STATE_BUCKET is not configured; this deployment has no "
+                    "recoverable Terraform state. Manual AWS cleanup required."
+                ),
+                "remaining_resources": await self._describe_remaining_resources(job_id, aws_config),
+            }
+
+        tf_runner = TerraformRunner(tf_dir)
+
+        if not tf_runner.init():
+            self.logger.error(f"[{job_id}] Destroy: terraform init failed")
+            return {
+                "status": "failed",
+                "job_id": job_id,
+                "error": "terraform_init_failed",
+                "message": "Could not initialize Terraform with the remote state backend.",
+                "remaining_resources": await self._describe_remaining_resources(job_id, aws_config),
+            }
+
+        # A job with no prior real `apply` (or one applied before
+        # TF_STATE_BUCKET existed) has no state key in S3 -- init() still
+        # succeeds against an empty backend, so check for an actual state
+        # explicitly rather than trusting init() alone.
+        state_output = tf_runner.output()
+        if not state_output:
+            self.logger.info(f"[{job_id}] Destroy: no Terraform state found in remote backend")
+            return {
+                "status": "no_state",
+                "job_id": job_id,
+                "message": (
+                    "No Terraform state found for this job (deployed before "
+                    "remote state was enabled, or state was lost). Nothing to "
+                    "destroy via Terraform -- check AWS directly for leftover "
+                    "resources."
+                ),
+                "remaining_resources": await self._describe_remaining_resources(job_id, aws_config),
+            }
+
+        try:
+            destroyed = tf_runner.destroy()
+        except Exception as exc:
+            self.logger.error(f"[{job_id}] terraform destroy failed after retries: {exc}")
+            return {
+                "status": "partial_failure",
+                "job_id": job_id,
+                "error": str(exc),
+                "remaining_resources": await self._describe_remaining_resources(job_id, aws_config),
+            }
+
+        if not destroyed:
+            return {
+                "status": "partial_failure",
+                "job_id": job_id,
+                "error": "terraform destroy did not report success",
+                "remaining_resources": await self._describe_remaining_resources(job_id, aws_config),
+            }
+
+        self.logger.info(f"[{job_id}] Destroy successful")
+        return {"status": "success", "job_id": job_id}
+
+    async def _describe_remaining_resources(
+        self, job_id: str, aws_config: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Best-effort snapshot of what's still live in AWS after a failed,
+        partial, or unrecoverable-state destroy -- so the caller can tell the
+        user precisely what to clean up manually and where (mirrors the read
+        pattern already used by rollback() and the /monitoring endpoint)."""
+        region = aws_config.get("region") or "us-east-1"
+        cluster = aws_config.get("ecs_cluster")
+        service_name = aws_config.get("service_name")
+        remaining: Dict[str, Any] = {
+            "ecs_service": None,
+            "target_groups": [],
+            "error": None,
+        }
+        if not cluster or not service_name:
+            remaining["error"] = "no ecs_cluster/service_name available to check"
+            return remaining
+        try:
+            aws = AWSClient(region=region)
+            ecs = aws.ecs()
+            desc = ecs.describe_services(cluster=cluster, services=[service_name])
+            services = desc.get("services") or []
+            if services and services[0].get("status") != "INACTIVE":
+                svc = services[0]
+                remaining["ecs_service"] = {
+                    "cluster": cluster,
+                    "service_name": service_name,
+                    "status": svc.get("status"),
+                    "running_count": svc.get("runningCount"),
+                }
+                elbv2 = aws.session.client("elbv2", region_name=region, config=RETRY_CONFIG)
+                for lb in svc.get("loadBalancers", []):
+                    tg_arn = lb.get("targetGroupArn")
+                    if tg_arn:
+                        remaining["target_groups"].append(tg_arn)
+        except Exception as exc:  # noqa: BLE001 - best-effort diagnostic, never raise
+            remaining["error"] = str(exc)
+        return remaining
     @xray_recorder.capture("rollback_deployment")  # type: ignore[reportCallIssue]
     async def rollback_deployment(
         self,
@@ -720,7 +844,7 @@ def _prepare_docker_config(self, docker_config: str) -> str:
         Writes an empty-credsStore config.json (avoids macOS credential-helper
         hangs / `osxkeychain` prompts) and, critically, symlinks the real
         ``~/.docker/cli-plugins`` (buildx et al.) into it. Pointing
-        DOCKER_CONFIG at a throwaway directory hides the CLI plugin directory —
+        DOCKER_CONFIG at a throwaway directory hides the CLI plugin directory —
         ``docker buildx`` then fails with "unknown command: docker buildx"
         (reproduced: ``docker buildx build --load`` runs fine from an
         interactive shell but errors identically from the backend subprocess

src/backend/api/jobs.py

@@ -346,7 +346,7 @@ def edit_artifacts(
             ),
         )
 
-    # Validate every edit up front so a batch fails atomically — never a
+    # Validate every edit up front so a batch fails atomically — never a
     # partial write of broken files.
     for edit in body.files:
         if not allowed_file_path(edit.file_path):
@@ -520,6 +520,101 @@ def rollback_job(
     }
 
 
+class DestroyRequest(BaseModel):
+    confirm_service_name: str
+
+
+def _extract_aws_config(deployment: models.Deployment) -> dict:
+    """Resolve ecs_cluster/service_name/region from a deployment's
+    infrastructure_json, trying the DeployOps-native `aws_config` shape
+    first and falling back to the `terraform_outputs` shape (with the same
+    suffix-derivation workaround as the /monitoring endpoint) for older
+    deployments where ecs_cluster_name was stored empty."""
+    infra = deployment.infrastructure_json or {}
+    aws_config = infra.get("aws_config") or {}
+    region = aws_config.get("region") or deployment.aws_region or "us-east-1"
+    ecs_cluster = aws_config.get("ecs_cluster")
+    service_name = aws_config.get("service_name")
+
+    if not ecs_cluster or not service_name:
+        tf_outputs = infra.get("terraform_outputs") or {}
+        service_name = service_name or tf_outputs.get("service_name")
+        ecs_cluster = ecs_cluster or tf_outputs.get("ecs_cluster_name")
+        if not ecs_cluster and service_name and "-" in service_name:
+            suffix = service_name.rsplit("-", 1)[-1]
+            ecs_cluster = f"devguard-cluster-{suffix}"
+
+    return {
+        "region": region,
+        "ecs_cluster": ecs_cluster,
+        "service_name": service_name,
+    }
+
+
+@router.post("/{job_id}/destroy")
+def destroy_job(
+    job_id: str,
+    body: DestroyRequest,
+    db: Session = Depends(get_db),
+    current_user: models.User = Depends(get_current_user),
+):
+    """Destroy a job's deployed infrastructure. Requires the caller to type
+    the exact service_name as confirmation (feature/destroy-deployment
+    decision #2) -- enforced server-side, not just via a disabled UI button.
+    """
+    run = _get_owned_run(db, job_id, current_user.id)
+
+    deployment = (
+        db.query(models.Deployment)
+        .filter(models.Deployment.run_id == job_id)
+        .order_by(models.Deployment.created_at.desc())
+        .first()
+    )
+    if not deployment:
+        raise HTTPException(status_code=404, detail="No deployment found for this job")
+
+    aws_config = _extract_aws_config(deployment)
+    service_name = aws_config.get("service_name")
+
+    if not service_name:
+        raise HTTPException(
+            status_code=400,
+            detail="Deployment has no service_name on record; cannot confirm destroy target.",
+        )
+
+    if body.confirm_service_name != service_name:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Confirmation text does not match. Expected the service name '{service_name}'.",
+        )
+
+    try:
+        from src.agents.deployops.agent import DeployOpsAgent
+
+        agent = DeployOpsAgent()
+        result = asyncio.run(agent.destroy_deployment(job_id=job_id, aws_config=aws_config))
+    except Exception as exc:
+        logger.error(f"Destroy failed for {job_id}: {exc}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Destroy failed: {exc}") from exc
+
+    status = result.get("status")
+    if status == "success":
+        deployment.status = "destroyed"
+        db.commit()
+        if run.status not in ("completed", "failed", "rejected"):
+            run.status = "destroyed"
+            run.completed_at = datetime.utcnow()
+            db.commit()
+        publish_results_ready(job_id)
+        return {"job_id": job_id, "status": "destroyed", "result": result}
+
+    # no_state / failed / partial_failure: don't touch deployment.status --
+    # leaving it as-is is more honest than silently marking it destroyed
+    # when resources may still be live. remaining_resources lets the UI
+    # show exactly what's left and where (feature/destroy-deployment
+    # decision #3).
+    return {"job_id": job_id, "status": status, "result": result}
+
 @router.get("/{job_id}/deployments/revisions")
 def list_deployment_revisions(
     job_id: str,

src/backend/models.py

@@ -85,6 +85,7 @@ class DeploymentStatus(str, PyEnum):
     SUCCEEDED = "succeeded"
     FAILED = "failed"
     ROLLED_BACK = "rolled_back"
+    DESTROYED = "destroyed"
 
 
 class NotificationType(str, PyEnum):
@@ -317,7 +318,7 @@ class Deployment(Base):
             name="ck_deployments_environment"
         ),
         CheckConstraint(
-            "status IN ('pending', 'applying', 'succeeded', 'failed', 'rolled_back')",
+            "status IN ('pending', 'applying', 'succeeded', 'failed', 'rolled_back', 'destroyed')",
             name="ck_deployments_status"
         ),
         Index("idx_deployments_run_id", "run_id"),

src/lib/terraform/runner.py

@@ -12,15 +12,24 @@ def __init__(self, working_dir: Path):
         self.working_dir = working_dir
         self.working_dir.mkdir(parents=True, exist_ok=True)
 
-    def _retry_with_backoff(self, func: Callable[[], Any], max_attempts: int = 3, base_delay: float = 2.0) -> Any:
+    def _retry_with_backoff(
+        self,
+        func: Callable[[], Any],
+        max_attempts: int = 3,
+        base_delay: float = 2.0,
+        delays: Optional[List[float]] = None,
+    ) -> Any:
         last_exception = None
         for attempt in range(1, max_attempts + 1):
             try:
                 return func()
             except Exception as e:
                 last_exception = e
                 if attempt < max_attempts:
-                    delay = base_delay * (2 ** (attempt - 1))
+                    if delays:
+                        delay = delays[min(attempt - 1, len(delays) - 1)]
+                    else:
+                        delay = base_delay * (2 ** (attempt - 1))
                     logger.warning(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...")
                     time.sleep(delay)
                 else:
@@ -60,7 +69,18 @@ def _run(self, cmd: List[str]) -> subprocess.CompletedProcess:
 
     def init(self) -> bool:
         def _init():
-            result = self._run(["init"])
+            # -input=false: a backend change (e.g. first init against the new
+            # S3 remote state) prompts interactively to migrate existing
+            # state; under a non-interactive docker exec that has no stdin,
+            # the prompt just hangs forever instead of erroring -- confirmed
+            # live during feature/destroy-deployment testing.
+            # -migrate-state -force-copy: a workspace with pre-existing local
+            # state (DEPLOYOPS_WORKSPACE_ROOT is a persisted volume, so state
+            # from before TF_STATE_BUCKET was configured can still be on disk)
+            # triggers a migration prompt when the backend changes to S3;
+            # -force-copy auto-approves it non-interactively instead of
+            # hanging or erroring under -input=false.
+            result = self._run(["init", "-input=false", "-migrate-state", "-force-copy"])
             if result.returncode != 0:
                 raise RuntimeError(f"init failed: {result.stderr}")
             return True
@@ -110,7 +130,12 @@ def _destroy():
             if result.returncode != 0:
                 raise RuntimeError(f"destroy failed: {result.stderr}")
             return True
-        return self._retry_with_backoff(_destroy)
+        # Destroy failures are often AWS eventual-consistency issues (e.g. a
+        # lingering ENI blocking security-group deletion with
+        # DependencyViolation) that clear on the order of tens of seconds,
+        # not the 2s/4s/8s used for apply/init/plan -- so destroy gets a
+        # slower, explicit backoff (feature/destroy-deployment decision).
+        return self._retry_with_backoff(_destroy, max_attempts=4, delays=[10.0, 30.0, 60.0])
 
     def output(self) -> Dict:
         result = self._run(["output", "-json"])