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
 
@@ -240,15 +244,34 @@ def _render_costs(
         else:
             delta_str = "$0.00"
 
+        address_cell = est.resource_address
+        if est.used_default_pricing:
+            address_cell += " [yellow]⚠ generic est.[/yellow]"
         table.add_row(
-            est.resource_address,
+            address_cell,
             f"${est.monthly_cost_before:.2f}",
             f"${est.monthly_cost_after:.2f}",
             delta_str,
         )
 
     console.print(table)
 
+    # Surface unpriced resource types: their numbers are generic defaults,
+    # not real estimates. Never let a confident-looking total hide this.
+    unpriced_types = sorted(
+        {
+            change.resource_type
+            for change, est in zip(plan.changes, estimates, strict=False)
+            if est.used_default_pricing
+        }
+    )
+    if unpriced_types:
+        console.print(
+            f"\n[yellow]\u26a0 {len(unpriced_types)} resource type(s) have no "
+            f"pricing data ({', '.join(unpriced_types)}); their rows use a "
+            f"generic default estimate and totals may be inaccurate.[/yellow]"
+        )
+
     total = plan.total_monthly_delta
     if total > 0:
         console.print(f"\n[bold red]Total monthly increase: +${total:.2f}[/bold red]")

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/cost_estimator.py

@@ -173,11 +173,18 @@ def estimate_costs(
         before_cost = _estimate_resource_cost(change, pricing, before=True)
         after_cost = _estimate_resource_cost(change, pricing, before=False)
 
+        # Flag resources with no pricing entry at all: they silently fall back
+        # to the generic default below, which can badly understate real cost.
+        unpriced = change.resource_type not in pricing
+
         estimate = CostEstimate(
             resource_address=change.address,
             monthly_cost_before=before_cost,
             monthly_cost_after=after_cost,
-            description=_build_cost_description(change, before_cost, after_cost),
+            description=_build_cost_description(
+                change, before_cost, after_cost, unpriced=unpriced
+            ),
+            used_default_pricing=unpriced,
         )
         estimates.append(estimate)
 
@@ -224,26 +231,40 @@ def _estimate_resource_cost(
     return type_pricing.get("default", 5.00)
 
 
-def _build_cost_description(change: ResourceChange, before: float, after: float) -> str:
-    """Build a human-readable cost description."""
+def _build_cost_description(
+    change: ResourceChange,
+    before: float,
+    after: float,
+    unpriced: bool = False,
+) -> str:
+    """Build a human-readable cost description.
+
+    When *unpriced* is True the resource type has no entry in the pricing
+    table, so both figures come from the generic default; say so explicitly
+    instead of presenting an invented number as if it were priced data.
+    """
     delta = after - before
     if delta > 0:
-        return f"+${delta:.2f}/mo"
+        desc = f"+${delta:.2f}/mo"
     elif delta < 0:
-        return f"-${abs(delta):.2f}/mo"
-    return "no change"
+        desc = f"-${abs(delta):.2f}/mo"
+    else:
+        desc = "no change"
+    if unpriced:
+        desc += f" [no pricing data for {change.resource_type}; generic default applied]"
+    return desc
 
 
 def _load_pricing(
     pricing_file: str | Path | None = None,
 ) -> dict[str, dict[str, float]]:
     """Load pricing data from a custom file, falling back to defaults."""
     if pricing_file is None:
-        return DEFAULT_PRICING.copy()
+        return copy.deepcopy(DEFAULT_PRICING)
 
     path = Path(pricing_file)
     if not path.exists():
-        return DEFAULT_PRICING.copy()
+        return copy.deepcopy(DEFAULT_PRICING)
 
     with path.open() as f:
         custom = json.load(f)

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"
@@ -84,6 +88,9 @@ class CostEstimate:
     monthly_cost_after: float = 0.0
     currency: str = "USD"
     description: str = ""
+    # True when deploydiff had no pricing entry for this resource type and fell
+    # back to the generic default estimate. Surfaces silent under-estimation.
+    used_default_pricing: bool = False
 
     @property
     def monthly_delta(self) -> float:

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