feat(storage): delta storage for target_data — 20x reduction

agent_actions/llm/batch/services/processing.py

@@ -379,6 +379,8 @@ def _merge_carry_forward(
                 len(carry_records),
                 action_name,
             )
+            for record in carry_records:
+                record["_delta_mode"] = "full"
 
         try:
             carry_path.unlink()

agent_actions/output/writer.py

@@ -132,8 +132,16 @@ def do_write() -> int:
 
             self.storage_backend.write_target(self.action_name, relative_path, data)
 
+            # Strip _delta_mode before disk write — it's a storage-layer
+            # internal that must not leak to filesystem files.
+            clean_data = [
+                {k: v for k, v in record.items() if k != "_delta_mode"}
+                if isinstance(record, dict) and "_delta_mode" in record
+                else record
+                for record in data
+            ]
             ensure_directory_exists(file_path, is_file=True)
-            atomic_json_write(file_path, data)
+            atomic_json_write(file_path, clean_data)
 
             return file_path.stat().st_size
 

agent_actions/processing/enrichment.py

@@ -72,6 +72,8 @@ def enrich(self, result: ProcessingResult, context: ProcessingContext) -> Proces
                 item["source_guid"] = IDGenerator.generate_source_guid()
                 if old_source_guid:
                     item["parent_source_guid"] = old_source_guid
+                # New GUIDs have no upstream deltas — store as full
+                item["_delta_mode"] = "full"
 
             if (
                 result.source_mapping is not None

agent_actions/record/envelope.py

@@ -37,6 +37,7 @@
     {
         "_state_history",
         "_state_schema_version",
+        "_delta_mode",
     }
 )