cowork: fix terraform rollback generator correctness and safety

src/deploydiff/cli.py

@@ -8,7 +8,7 @@
 from .cloudformation_parser import parse_cloudformation_changeset
 from .cost_estimator import estimate_costs
 from .diff_renderer import render_plan
-from .models import CostEstimate, DeployPlan
+from .models import CostEstimate, DeployPlan, PlanFormatError
 from .pulumi_parser import parse_pulumi_preview
 from .rollback import generate_rollback_commands
 from .terraform_parser import parse_terraform_plan
@@ -208,12 +208,16 @@ def _load_plan(
         )
         raise SystemExit(1)
 
-    if terraform_file:
-        return parse_terraform_plan(terraform_file)
-    elif cloudformation_file:
-        return parse_cloudformation_changeset(cloudformation_file)
-    elif pulumi_file:
-        return parse_pulumi_preview(pulumi_file)
+    try:
+        if terraform_file:
+            return parse_terraform_plan(terraform_file)
+        elif cloudformation_file:
+            return parse_cloudformation_changeset(cloudformation_file)
+        elif pulumi_file:
+            return parse_pulumi_preview(pulumi_file)
+    except PlanFormatError as exc:
+        console.print(f"[red]Error: {exc}[/red]")
+        raise SystemExit(1) from exc
 
     return None
 

src/deploydiff/cloudformation_parser.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Any
 
-from .models import ChangeAction, ChangeSource, DeployPlan, ResourceChange
+from .models import ChangeAction, ChangeSource, DeployPlan, PlanFormatError, ResourceChange
 
 # CloudFormation action mapping
 CFN_ACTION_MAP: dict[str, ChangeAction] = {
@@ -51,10 +51,13 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
                 data = json.load(f)
     else:
         data = changeset_json
+    if not isinstance(data, dict) or ("Changes" not in data and "changes" not in data):
+        raise PlanFormatError("Input does not look like a CloudFormation change set JSON (expected 'Changes' or 'changes' key). Did you pass the right --cfn file?")
 
     changes: list[ResourceChange] = []
     changes_list = data.get("Changes", data.get("changes", []))
 
+
     for change_entry in changes_list:
         resource_change_data = change_entry.get(
             "ResourceChange", change_entry.get("resource_change", {})

src/deploydiff/models.py

@@ -7,6 +7,10 @@
 from typing import Any
 
 
+class PlanFormatError(ValueError):
+    """Raised when an input document does not match the expected plan format."""
+
+
 class ChangeAction(Enum):
     CREATE = "create"
     READ = "read"

src/deploydiff/pulumi_parser.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Any
 
-from .models import ChangeAction, ChangeSource, DeployPlan, ResourceChange
+from .models import ChangeAction, ChangeSource, DeployPlan, PlanFormatError, ResourceChange
 
 # Pulumi step mapping
 PULUMI_STEP_MAP: dict[str, ChangeAction] = {
@@ -49,12 +49,19 @@ def parse_pulumi_preview(preview_json: str | dict[str, Any]) -> DeployPlan:
                 data = json.load(f)
     else:
         data = preview_json
+    if not isinstance(data, dict) or (
+        "steps" not in data
+        and "resourceChanges" not in data
+        and "resources" not in data
+    ):
+        raise PlanFormatError("Input does not look like a Pulumi preview JSON (expected 'steps', 'resourceChanges', or 'resources' keys). Did you pass the right --pulumi file?")
 
     changes: list[ResourceChange] = []
 
     # Pulumi preview JSON has a "steps" array
     steps = data.get("steps", [])
 
+
     # Also support the resource-oriented format
     resources = data.get("resourceChanges", data.get("resources", {}))
 

src/deploydiff/rollback.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from .models import ChangeSource, DeployPlan
+from .models import ChangeAction, ChangeSource, DeployPlan
 
 
 def generate_rollback_commands(plan: DeployPlan) -> list[str]:
@@ -28,17 +28,47 @@ def _terraform_rollback(plan: DeployPlan) -> list[str]:
 
     Strategy: target the reverse of each destructive/create change.
     """
+    if not plan.changes:
+        return ["# No changes to roll back"]
+
     commands: list[str] = []
     commands.append("# Terraform Rollback Commands")
     commands.append("# Run these in reverse order to undo the deployment")
     commands.append("")
 
-    # For each create, we need to destroy it
-    for change in plan.creates:
+    # Replacements (create-before-delete / delete-before-create): revert by
+    # re-applying the PREVIOUS config. Do NOT also emit destroy + apply for
+    # these -- they used to appear in both the creates and destructive lists,
+    # producing contradictory commands for the same resource.
+    replacements = [
+        c
+        for c in plan.destructive_changes
+        if c.action
+        in (
+            ChangeAction.CREATE_BEFORE_DELETE,
+            ChangeAction.DELETE_BEFORE_CREATE,
+            ChangeAction.REPLACE,
+        )
+    ]
+
+    # For each pure create, we need to destroy it
+    pure_creates = [c for c in plan.creates if c.action == ChangeAction.CREATE]
+    for change in pure_creates:
         commands.append(f"terraform destroy -target={change.address} -auto-approve")
 
-    # For each destructive change (delete/replace), we need to re-apply it
-    for change in plan.destructive_changes:
+    # For each pure delete, we need to re-create it from the previous config
+    pure_deletes = [c for c in plan.destructive_changes if c.action == ChangeAction.DELETE]
+    for change in pure_deletes:
+        commands.append(
+            f"# To restore {change.address}, restore previous config and run:"
+        )
+        commands.append(f"terraform apply -target={change.address} -auto-approve")
+
+    # For each replacement, revert with the previous config
+    for change in replacements:
+        commands.append(
+            f"# To revert replaced {change.address}, restore previous config and run:"
+        )
         commands.append(f"terraform apply -target={change.address} -auto-approve")
 
     # For updates, we can try to revert with the previous state
@@ -48,9 +78,6 @@ def _terraform_rollback(plan: DeployPlan) -> list[str]:
         )
         commands.append(f"terraform apply -target={change.address} -auto-approve")
 
-    if not plan.changes:
-        commands.append("# No changes to roll back")
-
     # Add a full rollback option
     commands.append("")
     commands.append("# Or rollback the entire stack:")

src/deploydiff/terraform_parser.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Any
 
-from .models import ChangeAction, ChangeSource, DeployPlan, ResourceChange
+from .models import ChangeAction, ChangeSource, DeployPlan, PlanFormatError, ResourceChange
 
 # Terraform plan action mapping
 TF_ACTION_MAP: dict[str, ChangeAction] = {
@@ -43,8 +43,11 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
                 data = json.load(f)
     else:
         data = plan_json
+    if not isinstance(data, dict) or ("resource_changes" not in data and "format_version" not in data):
+        raise PlanFormatError("Input does not look like a Terraform plan JSON (expected 'resource_changes' or 'format_version' keys). Did you pass the right --tf file?")
 
     format_version = data.get("format_version", "")
+
     changes: list[ResourceChange] = []
 
     # Parse planned changes

tests/test_edge_cases.py

@@ -143,8 +143,9 @@ def test_pulumi_rollback_unsupported_source_fallback(self):
         # all produce meaningful output.
         plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[])
         cmds = generate_rollback_commands(plan)
-        assert len(cmds) > 1
-        assert "Terraform" in cmds[0]
+        # Empty plans short-circuit: no header, no blanket destroy-everything
+        # suggestion for a plan with nothing to roll back.
+        assert cmds == ["# No changes to roll back"]
 
     def test_cloudformation_rollback_no_raw_data(self):
         """_cloudformation_rollback with no raw_data uses STACK_NAME."""

tests/test_plan_format_validation.py

@@ -0,0 +1,68 @@
+"""Regression tests: wrong/non-plan JSON must fail loudly, not report 'no changes'."""
+
+import json
+
+import pytest
+from click.testing import CliRunner
+
+from deploydiff.cli import main
+from deploydiff.cloudformation_parser import parse_cloudformation_changeset
+from deploydiff.models import PlanFormatError
+from deploydiff.pulumi_parser import parse_pulumi_preview
+from deploydiff.terraform_parser import parse_terraform_plan
+
+WRONG_DOCS = [
+    {"name": "not-a-plan", "version": "1.0"},
+    {"foo": []},
+    [1, 2, 3],
+]
+
+TF_EMPTY_PLAN = {"format_version": "1.2", "resource_changes": []}
+CFN_EMPTY_CHANGESET = {"ChangeSetName": "cs", "Changes": []}
+PULUMI_EMPTY = {"steps": []}
+
+
+@pytest.mark.parametrize("doc", WRONG_DOCS)
+def test_terraform_parser_rejects_non_plan(doc):
+    with pytest.raises(PlanFormatError):
+        parse_terraform_plan(doc)
+
+
+@pytest.mark.parametrize("doc", WRONG_DOCS)
+def test_cfn_parser_rejects_non_plan(doc):
+    with pytest.raises(PlanFormatError):
+        parse_cloudformation_changeset(doc)
+
+
+@pytest.mark.parametrize("doc", WRONG_DOCS)
+def test_pulumi_parser_rejects_non_plan(doc):
+    with pytest.raises(PlanFormatError):
+        parse_pulumi_preview(doc)
+
+
+def test_valid_empty_plans_still_parse():
+    assert parse_terraform_plan(TF_EMPTY_PLAN).changes == []
+    assert parse_cloudformation_changeset(CFN_EMPTY_CHANGESET).changes == []
+    assert parse_pulumi_preview(PULUMI_EMPTY).changes == []
+
+
+def _write(tmp_path, doc):
+    f = tmp_path / "plan.json"
+    f.write_text(json.dumps(doc))
+    return str(f)
+
+
+@pytest.mark.parametrize("flag,doc", [
+    ("--tf", {"random": True}),
+    ("--cfn", {"random": True}),
+    ("--pulumi", {"random": True}),
+])
+def test_cli_exits_1_on_wrong_format(tmp_path, flag, doc):
+    result = CliRunner().invoke(main, ["preview", flag, _write(tmp_path, doc)])
+    assert result.exit_code == 1
+    assert "does not look like" in result.output
+
+
+def test_cli_still_accepts_valid_empty_plan(tmp_path):
+    result = CliRunner().invoke(main, ["preview", "--tf", _write(tmp_path, TF_EMPTY_PLAN)])
+    assert result.exit_code == 0, result.output

tests/test_rollback_safety.py

@@ -0,0 +1,50 @@
+"""Regression tests for rollback command generation safety/correctness."""
+
+from deploydiff.models import (
+    ChangeAction,
+    ChangeSource,
+    DeployPlan,
+    ResourceChange,
+)
+from deploydiff.rollback import generate_rollback_commands
+
+
+def _tf_change(address, action):
+    return ResourceChange(
+        address=address,
+        action=action,
+        resource_type="aws_instance",
+        resource_name=address.split(".")[-1],
+        source=ChangeSource.TERRAFORM,
+    )
+
+
+def test_empty_plan_returns_only_noop_message():
+    plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[])
+    commands = generate_rollback_commands(plan)
+    assert commands == ["# No changes to roll back"]
+    # The dangerous blanket destroy-everything suggestion must not appear.
+    assert not any("destroy -auto-approve &&" in c for c in commands)
+
+
+def test_create_before_delete_not_double_commanded():
+    """A create-first replacement must not produce both destroy and apply
+    for the same resource (contradictory rollback commands)."""
+    change = _tf_change("aws_instance.web", ChangeAction.CREATE_BEFORE_DELETE)
+    plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[change])
+    commands = generate_rollback_commands(plan)
+    destroys = [c for c in commands if c.startswith("terraform destroy -target=aws_instance.web")]
+    applies = [c for c in commands if c.startswith("terraform apply -target=aws_instance.web")]
+    assert destroys == [], "replacement should not be destroyed on rollback"
+    assert len(applies) == 1
+
+
+def test_pure_create_gets_destroy_pure_delete_gets_apply():
+    created = _tf_change("aws_instance.new", ChangeAction.CREATE)
+    deleted = _tf_change("aws_instance.old", ChangeAction.DELETE)
+    plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[created, deleted])
+    commands = generate_rollback_commands(plan)
+    assert "terraform destroy -target=aws_instance.new -auto-approve" in commands
+    assert any(
+        c.startswith("terraform apply -target=aws_instance.old") for c in commands
+    )