feat(ergo): gzip compression and target destroy improvements

Add gzip compression for all Ergo message paths (Create, Update,
TrackedProgress) to handle resources exceeding the 64KB buffer limit.
CompressJSON/DecompressJSON helpers on json.RawMessage fields; resolve
cache decompresses transparently.

Target destroy behavior:
- Application formae (with resources) only delete targets whose config
  has $ref dependencies on resources being destroyed. Plain targets
  (docker, us-east-1) survive because they exist independently.
- Target-only formae always delete their targets (unchanged behavior).
- Cascade source displayed as human-readable resource label instead of
  raw KSUID in CLI output and simulation tree.
- Simulated target updates now show (cascade) marker matching resources.

Author: JeroenSoeters

docs/docs.go

@@ -690,12 +690,14 @@ const docTemplate = `{
                 "Array",
                 "EntitySet",
                 "Set",
+                "Atomic",
                 ""
             ],
             "x-enum-varnames": [
                 "FieldUpdateMethodArray",
                 "FieldUpdateMethodEntitySet",
                 "FieldUpdateMethodSet",
+                "FieldUpdateMethodAtomic",
                 "FieldUpdateMethodNone"
             ]
         },
@@ -1231,6 +1233,9 @@ const docTemplate = `{
         "model.TargetUpdate": {
             "type": "object",
             "properties": {
+                "CascadeSource": {
+                    "type": "string"
+                },
                 "Discoverable": {
                     "type": "boolean"
                 },
@@ -1241,6 +1246,9 @@ const docTemplate = `{
                 "ErrorMessage": {
                     "type": "string"
                 },
+                "IsCascade": {
+                    "type": "boolean"
+                },
                 "ModifiedTs": {
                     "type": "string"
                 },

docs/swagger.json

@@ -684,12 +684,14 @@
                 "Array",
                 "EntitySet",
                 "Set",
+                "Atomic",
                 ""
             ],
             "x-enum-varnames": [
                 "FieldUpdateMethodArray",
                 "FieldUpdateMethodEntitySet",
                 "FieldUpdateMethodSet",
+                "FieldUpdateMethodAtomic",
                 "FieldUpdateMethodNone"
             ]
         },
@@ -1225,6 +1227,9 @@
         "model.TargetUpdate": {
             "type": "object",
             "properties": {
+                "CascadeSource": {
+                    "type": "string"
+                },
                 "Discoverable": {
                     "type": "boolean"
                 },
@@ -1235,6 +1240,9 @@
                 "ErrorMessage": {
                     "type": "string"
                 },
+                "IsCascade": {
+                    "type": "boolean"
+                },
                 "ModifiedTs": {
                     "type": "string"
                 },

docs/swagger.yaml

@@ -72,12 +72,14 @@ definitions:
     - Array
     - EntitySet
     - Set
+    - Atomic
     - ""
     type: string
     x-enum-varnames:
     - FieldUpdateMethodArray
     - FieldUpdateMethodEntitySet
     - FieldUpdateMethodSet
+    - FieldUpdateMethodAtomic
     - FieldUpdateMethodNone
   model.Forma:
     properties:
@@ -434,13 +436,17 @@ definitions:
     type: object
   model.TargetUpdate:
     properties:
+      CascadeSource:
+        type: string
       Discoverable:
         type: boolean
       Duration:
         description: milliseconds
         type: integer
       ErrorMessage:
         type: string
+      IsCascade:
+        type: boolean
       ModifiedTs:
         type: string
       Operation:

internal/cli/renderer/renderer.go

@@ -553,7 +553,15 @@ func formatSimulatedTargetUpdate(root *gtree.Node, tu apimodel.TargetUpdate) {
 		line = display.Greyf("%s target %s", op, tu.TargetLabel)
 	}
 
-	root.Add(line)
+	if tu.IsCascade {
+		line = line + display.Gold(" (cascade)")
+	}
+
+	node := root.Add(line)
+
+	if tu.IsCascade && tu.CascadeSource != "" {
+		node.Add(display.Grey("because it depends on ") + display.LightBlue(tu.CascadeSource))
+	}
 }
 
 // formatTargetUpdate formats a target update for status view

internal/metastructure/changeset/changeset.go

@@ -12,6 +12,7 @@ import (
 
 	"log/slog"
 
+	"github.com/platform-engineering-labs/formae/internal/metastructure/resolver"
 	"github.com/platform-engineering-labs/formae/internal/metastructure/resource_update"
 	"github.com/platform-engineering-labs/formae/internal/metastructure/target_update"
 	apimodel "github.com/platform-engineering-labs/formae/pkg/api/model"
@@ -341,7 +342,17 @@ func (p *ExecutionDAG) buildTargetResolvableEdges() {
 		if !ok {
 			continue
 		}
-		for _, uri := range tu.RemainingResolvables {
+
+		// For delete operations, extract resolvable URIs from the existing
+		// target config rather than RemainingResolvables. Delete target updates
+		// intentionally don't set RemainingResolvables to avoid the target
+		// updater attempting resolution during destroy.
+		resolvables := tu.RemainingResolvables
+		if tu.Operation == target_update.TargetOperationDelete && tu.ExistingTarget != nil {
+			resolvables = resolver.ExtractResolvableURIsFromJSON(tu.ExistingTarget.Config)
+		}
+
+		for _, uri := range resolvables {
 			ksuid := uri.KSUID()
 			if tu.Operation == target_update.TargetOperationDelete {
 				// REVERSED: resource delete waits for target delete

internal/metastructure/changeset/resolve_cache.go

@@ -254,6 +254,14 @@ func (r *ResolveCache) readViaPlugin(retry resolveRetry) (*plugin.TrackedProgres
 	if !ok {
 		return nil, fmt.Errorf("unexpected result type from plugin operator: %T", progressResult)
 	}
+	// Decompress resource properties if sent compressed over Ergo (64KB limit)
+	if len(progress.CompressedResourceProperties) > 0 && len(progress.ResourceProperties) == 0 {
+		decompressed, err := plugin.DecompressJSON(progress.CompressedResourceProperties)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decompress resource properties: %w", err)
+		}
+		progress.ResourceProperties = decompressed
+	}
 	return &progress, nil
 }
 

internal/metastructure/metastructure.go

@@ -1390,14 +1390,17 @@ func findCascadeTargetDeletes(
 	ds datastore.Datastore,
 ) ([]target_update.TargetUpdate, []resource_update.ResourceUpdate, error) {
 
-	// Collect KSUIDs of all resources being deleted
+	// Collect KSUIDs of all resources being deleted and build a KSUID→label
+	// lookup so cascade sources can be displayed as human-readable labels.
 	deletingKSUIDs := make([]string, 0)
 	deletingSet := make(map[string]bool)
+	ksuidToLabel := make(map[string]string)
 	for _, ru := range resourceUpdates {
 		if ru.Operation == resource_update.OperationDelete && ru.DesiredState.Ksuid != "" {
 			if !deletingSet[ru.DesiredState.Ksuid] {
 				deletingSet[ru.DesiredState.Ksuid] = true
 				deletingKSUIDs = append(deletingKSUIDs, ru.DesiredState.Ksuid)
+				ksuidToLabel[ru.DesiredState.Ksuid] = ru.DesiredState.Label
 			}
 		}
 	}
@@ -1447,12 +1450,16 @@ func findCascadeTargetDeletes(
 				}
 				alreadyDeleting[target.Label] = true
 
+				cascadeSourceLabel := ksuidToLabel[sourceKSUID]
+				if cascadeSourceLabel == "" {
+					cascadeSourceLabel = sourceKSUID
+				}
 				slog.Debug("Cascade target delete detected",
 					"target", target.Label,
-					"cascadeSource", sourceKSUID)
+					"cascadeSource", cascadeSourceLabel)
 
 				cascadeTargetUpdates = append(cascadeTargetUpdates,
-					target_update.NewTargetUpdateForCascadeDelete(target, sourceKSUID))
+					target_update.NewTargetUpdateForCascadeDelete(target, cascadeSourceLabel))
 				newDeletedTargetLabels = append(newDeletedTargetLabels, target.Label)
 			}
 		}
@@ -1644,7 +1651,7 @@ func FormaCommandFromForma(forma *pkgmodel.Forma,
 		}
 	}
 
-	targetUpdates, err := target_update.NewTargetUpdateGenerator(ds).GenerateTargetUpdates(forma.Targets, command)
+	targetUpdates, err := target_update.NewTargetUpdateGenerator(ds).GenerateTargetUpdates(forma.Targets, command, len(forma.Resources) > 0)
 	if err != nil {
 		return nil, err
 	}

internal/metastructure/resource_update/resource_update_generator.go

@@ -208,8 +208,14 @@ func generateResourceUpdatesForDestroy(
 		}
 	}
 
+	// Build KSUID→label lookup for human-readable cascade sources
+	ksuidToLabel := make(map[string]string, len(resourceDestroys))
+	for _, rd := range resourceDestroys {
+		ksuidToLabel[rd.DesiredState.Ksuid] = rd.DesiredState.Label
+	}
+
 	// Find cascade deletes - resources that reference the resources being deleted
-	cascadeDeletes, err := findCascadeDeletes(explicitDeleteKSUIDs, existingTargetMap, source, ds)
+	cascadeDeletes, err := findCascadeDeletes(explicitDeleteKSUIDs, ksuidToLabel, existingTargetMap, source, ds)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find cascade deletes: %w", err)
 	}
@@ -266,6 +272,7 @@ func generateResourceUpdatesForDestroy(
 // resources being deleted. Uses level-by-level BFS with batched queries to find the full cascade chain.
 func findCascadeDeletes(
 	toDelete map[string]bool,
+	ksuidToLabel map[string]string,
 	existingTargetMap map[string]*pkgmodel.Target,
 	source FormaCommandSource,
 	ds ResourceDataLookup) ([]ResourceUpdate, error) {
@@ -313,9 +320,9 @@ func findCascadeDeletes(
 					continue
 				}
 
-				// Use the source KSUID for the cascade message
-				sourceLabel := sourceKSUID
-				if _, isExplicit := toDelete[sourceKSUID]; isExplicit {
+				// Use human-readable label for the cascade source
+				sourceLabel := ksuidToLabel[sourceKSUID]
+				if sourceLabel == "" {
 					sourceLabel = sourceKSUID
 				}
 
@@ -339,8 +346,9 @@ func findCascadeDeletes(
 
 				cascadeDeletes = append(cascadeDeletes, resourceDestroy)
 
-				// Add to next level for further cascade detection
+				// Add to next level for further cascade detection and update label lookup
 				nextLevel = append(nextLevel, dependent.Ksuid)
+				ksuidToLabel[dependent.Ksuid] = dependent.Label
 			}
 		}
 

internal/metastructure/resource_update/resource_updater.go

@@ -470,12 +470,19 @@ func create(state gen.Atom, data ResourceUpdateData, proc gen.Process) (gen.Atom
 		return StateFinishedWithError, data, nil, nil
 	}
 
+	compressedProps, err := plugin.CompressJSON(convertedResource.Properties)
+	if err != nil {
+		proc.Log().Error("failed to compress resource properties: %v", err)
+		data.resourceUpdate.MarkAsFailed()
+		return StateFinishedWithError, data, nil, nil
+	}
+
 	createOperation := plugin.CreateResource{
-		Namespace:    convertedResource.Namespace(),
-		ResourceType: convertedResource.Type,
-		Label:        convertedResource.Label,
-		Properties:   convertedResource.Properties,
-		TargetConfig: data.resourceUpdate.ResourceTarget.Config,
+		Namespace:            convertedResource.Namespace(),
+		ResourceType:         convertedResource.Type,
+		Label:                convertedResource.Label,
+		CompressedProperties: compressedProps,
+		TargetConfig:         data.resourceUpdate.ResourceTarget.Config,
 	}
 
 	// First we check if progress already was made on the create operation. This can happen for example if the node crashed while the
@@ -556,15 +563,28 @@ func update(state gen.Atom, data ResourceUpdateData, proc gen.Process) (gen.Atom
 		return StateFinishedWithError, data, nil, nil
 	}
 
+	compressedDesired, err := plugin.CompressJSON(convertedResource.Properties)
+	if err != nil {
+		proc.Log().Error("failed to compress desired properties: %v", err)
+		data.resourceUpdate.MarkAsFailed()
+		return StateFinishedWithError, data, nil, nil
+	}
+	compressedPrior, err := plugin.CompressJSON(convertedExisting.Properties)
+	if err != nil {
+		proc.Log().Error("failed to compress prior properties: %v", err)
+		data.resourceUpdate.MarkAsFailed()
+		return StateFinishedWithError, data, nil, nil
+	}
+
 	updateOperation := plugin.UpdateResource{
-		Namespace:         convertedResource.Namespace(),
-		NativeID:          convertedResource.NativeID,
-		ResourceType:      convertedResource.Type,
-		Label:             convertedResource.Label,
-		PriorProperties:   convertedExisting.Properties,
-		DesiredProperties: convertedResource.Properties,
-		PatchDocument:     string(data.resourceUpdate.DesiredState.PatchDocument),
-		TargetConfig:      data.resourceUpdate.ResourceTarget.Config,
+		Namespace:                   convertedResource.Namespace(),
+		NativeID:                    convertedResource.NativeID,
+		ResourceType:                convertedResource.Type,
+		Label:                       convertedResource.Label,
+		CompressedPriorProperties:   compressedPrior,
+		CompressedDesiredProperties: compressedDesired,
+		PatchDocument:               string(data.resourceUpdate.DesiredState.PatchDocument),
+		TargetConfig:                data.resourceUpdate.ResourceTarget.Config,
 	}
 
 	// First we check if progress already was made on the update operation. This can happen for example if the node crashed while the
@@ -635,6 +655,17 @@ func resumeWaitingForResource(state gen.Atom, data ResourceUpdateData, progress
 // state machine to the next state when the plugin operation finished successfully. After the last plugin operation, or
 // after the first error, it reports the final state to the stack updater and exits.
 func handleProgressUpdate(from gen.PID, state gen.Atom, data ResourceUpdateData, message plugin.TrackedProgress, proc gen.Process) (gen.Atom, ResourceUpdateData, []statemachine.Action, error) {
+	// Decompress resource properties if sent compressed over Ergo
+	if len(message.CompressedResourceProperties) > 0 && len(message.ResourceProperties) == 0 {
+		decompressed, err := plugin.DecompressJSON(message.CompressedResourceProperties)
+		if err != nil {
+			proc.Log().Error("failed to decompress resource properties", "error", err)
+			data.resourceUpdate.MarkAsFailed()
+			return StateFinishedWithError, data, nil, nil
+		}
+		message.ResourceProperties = decompressed
+	}
+
 	err := data.resourceUpdate.RecordProgress(&message)
 	if err != nil {
 		proc.Log().Error("failed to record progress for resource update", "error", err)
@@ -845,6 +876,15 @@ func doPluginOperation(resourceURI pkgmodel.FormaeURI, operation plugin.PluginOp
 		return nil, fmt.Errorf("expected TrackedProgress, got %T", response)
 	}
 
+	// Decompress resource properties if sent compressed over Ergo (64KB limit)
+	if len(progressResult.CompressedResourceProperties) > 0 && len(progressResult.ResourceProperties) == 0 {
+		decompressed, err := plugin.DecompressJSON(progressResult.CompressedResourceProperties)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decompress resource properties: %w", err)
+		}
+		progressResult.ResourceProperties = decompressed
+	}
+
 	return &progressResult, nil
 }
 

internal/metastructure/target_update/target_update.go

@@ -48,13 +48,13 @@ type TargetUpdate struct {
 }
 
 // NewTargetUpdateForCascadeDelete creates a cascade delete TargetUpdate for a target
-// that depends on a resource being deleted. The cascadeSourceKSUID is the KSUID of
-// the resource that triggered the cascade.
+// that depends on a resource being deleted. cascadeSource is the label of the
+// resource that triggered the cascade.
 //
 // Unlike regular target deletes, cascade deletes do NOT set RemainingResolvables
 // because the referenced resources are themselves being deleted. Attempting to
 // resolve those references during execution would fail.
-func NewTargetUpdateForCascadeDelete(target *pkgmodel.Target, cascadeSourceKSUID string) TargetUpdate {
+func NewTargetUpdateForCascadeDelete(target *pkgmodel.Target, cascadeSource string) TargetUpdate {
 	now := util.TimeNow()
 	return TargetUpdate{
 		Target:         *target,
@@ -64,7 +64,7 @@ func NewTargetUpdateForCascadeDelete(target *pkgmodel.Target, cascadeSourceKSUID
 		StartTs:        now,
 		ModifiedTs:     now,
 		IsCascade:      true,
-		CascadeSource:  cascadeSourceKSUID,
+		CascadeSource:  cascadeSource,
 	}
 }