fix(persist): complete the storage boundary; respect binary content

src/_index.yaml

@@ -113,33 +113,6 @@ entries:
       agent_ref: userspace.dataflow:agent_ref
     method: run_tests
 
-  - name: execution_identity
-    kind: contract.definition
-    meta:
-      title: Dataflow Execution Identity
-      comment: Optional app-provided bridge for durable workflow identity capture and orchestrator revival.
-    methods:
-      - name: capture
-        description: Capture the current execution identity as an opaque actor_id plus actor_context row. Input { reason? }.
-        input_schemas:
-          - definition: |
-              { "type": "object", "properties": { "reason": { "type": "string" } } }
-            format: application/schema+json
-        output_schemas:
-          - definition: |
-              { "type": "object", "properties": { "success": { "type": "boolean" }, "actor_id": { "type": "string" }, "actor_context": {}, "error": { "type": "string" } }, "required": ["success"] }
-            format: application/schema+json
-      - name: spawn_orchestrator
-        description: Spawn the dataflow orchestrator for a persisted dataflow under that dataflow's captured identity. Input { dataflow_id, process_id, host_id, args? }.
-        input_schemas:
-          - definition: |
-              { "type": "object", "properties": { "dataflow_id": { "type": "string" }, "process_id": { "type": "string" }, "host_id": { "type": "string" }, "args": { "type": "object" } }, "required": ["dataflow_id", "process_id", "host_id"] }
-            format: application/schema+json
-        output_schemas:
-          - definition: |
-              { "type": "object", "properties": { "success": { "type": "boolean" }, "pid": { "type": "string" }, "error": { "type": "string" } }, "required": ["success"] }
-            format: application/schema+json
-
   - name: execution_frame
     kind: library.lua
     meta:

src/persist/commit_repo.lua

@@ -92,6 +92,7 @@ function commit_repo.create(commit_id, dataflow_id, payload, metadata)
             metadata_json = metadata
         end
     end
+    metadata_json = encoding.ensure_utf8(metadata_json)
 
     -- Create timestamp
     local created_at = time.now():format(time.RFC3339)

src/persist/encoding.lua

@@ -7,6 +7,17 @@ local M = {}
 
 local REPLACEMENT = "\239\191\189" -- U+FFFD
 
+local BINARY_CONTENT_TYPE = "application/octet-stream"
+
+-- ensure_storable(value, content_type) -> value. The boundary is content-type
+-- aware: declared binary passes through byte-identical — it lives in binary
+-- columns and sanitizing it would be corruption. Everything else is text and
+-- must be valid UTF-8.
+function M.ensure_storable(value: any, content_type: any): any
+    if content_type == BINARY_CONTENT_TYPE then return value end
+    return M.ensure_utf8(value)
+end
+
 function M.ensure_utf8(s: any): any
     if type(s) ~= "string" then return s end
     local n = #s

src/persist/ops.lua

@@ -142,9 +142,9 @@ local function replace_iteration_terminal(tx, dataflow_id, op_id, payload, data_
         :set("type", payload.data_type)
         :set("discriminator", payload.discriminator)
         :set("key", payload.key)
-        :set("content", content_value)
+        :set("content", encoding.ensure_storable(content_value, content_type))
         :set("content_type", content_type)
-        :set("metadata", metadata)
+        :set("metadata", encoding.ensure_utf8(metadata))
         :run_with(tx)
         :exec()
     if err then return nil, "Failed to replace iteration terminal data: " .. err end
@@ -317,7 +317,7 @@ handlers[constants.COMMAND_TYPES.CREATE_NODE] = function(tx, dataflow_id, op_id,
             parent_node_id = parent_node_id,
             type = payload.node_type,
             status = status,
-            config = config,
+            config = encoding.ensure_utf8(config),
             metadata = metadata,
             created_at = now_ts,
             updated_at = now_ts
@@ -379,7 +379,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_NODE] = function(tx, dataflow_id, op_id,
             end
             config = encoded
         end
-        update_query = update_query:set("config", config)
+        update_query = update_query:set("config", encoding.ensure_utf8(config))
         has_update = true
     end
 
@@ -464,7 +464,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_NODE] = function(tx, dataflow_id, op_id,
             end
         end
 
-        update_query = update_query:set("metadata", meta_val_for_db)
+        update_query = update_query:set("metadata", encoding.ensure_utf8(meta_val_for_db))
         has_update = true
     end
 
@@ -554,11 +554,10 @@ handlers[constants.COMMAND_TYPES.CREATE_DATA] = function(tx, dataflow_id, op_id,
         end
         content_value = encoded
     end
-    -- Storage boundary: external content can carry arbitrary bytes; what is
-    -- written is always valid UTF-8.
-    content_value = encoding.ensure_utf8(content_value)
-
     local content_type = payload.content_type or "application/json"
+    -- Storage boundary: external content can carry arbitrary bytes; every
+    -- textual write is valid UTF-8, declared binary passes byte-identical.
+    content_value = encoding.ensure_storable(content_value, content_type)
     local node_id = payload.node_id or sql.as.null()
     local metadata = payload.metadata or "{}"
 
@@ -569,7 +568,7 @@ handlers[constants.COMMAND_TYPES.CREATE_DATA] = function(tx, dataflow_id, op_id,
         end
         metadata = encoded
     end
-
+    metadata = encoding.ensure_utf8(metadata)
 
     if is_iteration_terminal_unique_slot(payload) then
         local existing_row, existing_err = find_existing_unique_data_row(tx, dataflow_id, payload)
@@ -732,7 +731,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id,
             content_value = encoded
         end
 
-        update_query = update_query:set("content", content_value)
+        update_query = update_query:set("content", encoding.ensure_storable(content_value, payload.content_type))
         has_update = true
     end
 
@@ -751,7 +750,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id,
             metadata = encoded
         end
 
-        update_query = update_query:set("metadata", metadata)
+        update_query = update_query:set("metadata", encoding.ensure_utf8(metadata))
         has_update = true
     end
 
@@ -1084,7 +1083,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op
             end
         end
 
-        update_query_builder = update_query_builder:set("metadata", meta_val_for_db)
+        update_query_builder = update_query_builder:set("metadata", encoding.ensure_utf8(meta_val_for_db))
         has_real_update_field = true
     end
 

src/runner/workflow_state.lua

@@ -1182,6 +1182,14 @@ function methods:handle_process_exit(pid, success, result)
         and content_kind ~= "boolean" and content_kind ~= "table" then
         result_content = result_content ~= nil and tostring(result_content) or nil
     end
+    if content_kind == "table" then
+        -- A table must actually encode (cycles do not); an unserializable
+        -- result persists as its failure text instead of poisoning the batch.
+        local ok, _, encode_err = pcall(function() return json.encode(result_content) end)
+        if not ok or encode_err ~= nil then
+            result_content = "unserializable node result: " .. tostring(result)
+        end
+    end
     if result_content == nil then
         result_content = success and "Completed" or "Failed"
     end

src/runner/workflow_state_test.lua

@@ -860,24 +860,23 @@ local function define_tests()
                 test.eq(type(content), "string")
             end)
 
-            it("persists a table result whose values cannot encode as its string form", function()
+            it("persists a cyclic table result as its failure text", function()
                 local ws = workflow_state.new(test_ctx.dataflow_id) :: any
                 ws.nodes["node-1"] = { status = consts.STATUS.RUNNING, type = "test_node" }
                 ws:track_process("node-1", "pid-123")
 
-                local poison = { message = "boom", raw = coroutine.create(function() end) }
-                local exit_info = ws:handle_process_exit("pid-123", false, poison) :: any
+                local cyclic: any = { message = "boom" }
+                cyclic.self = cyclic
+                local exit_info = ws:handle_process_exit("pid-123", false, cyclic) :: any
                 test.not_nil(exit_info)
 
                 local content = nil
                 for _, cmd in ipairs(ws.queued_commands) do
                     local p = (cmd :: any).payload or {}
                     if p.data_type == consts.DATA_TYPE.NODE_RESULT then content = p.content end
                 end
-                test.not_nil(content)
-                local encoded, encode_err = json.encode(content)
-                test.is_nil(encode_err)
-                test.not_nil(encoded)
+                test.eq(type(content), "string")
+                test.is_true((content :: string):find("unserializable", 1, true) ~= nil)
             end)
         end)
 

test/_index.yaml

@@ -156,42 +156,6 @@ entries:
       test_only: true
       comment: Named policy group for test-only process services.
 
-  - name: execution_identity.capture
-    kind: function.lua
-    meta:
-      environment: test
-      test_only: true
-      comment: Captures the explicit test actor and policy used by the test runner.
-    source: file://execution_identity.lua
-    method: capture
-    modules: &execution_identity_modules [json, process, security]
-    imports: &execution_identity_imports
-      dataflow_repo: userspace.dataflow.persist:dataflow_repo
-
-  - name: execution_identity.spawn_orchestrator
-    kind: function.lua
-    meta:
-      environment: test
-      test_only: true
-      comment: Reconstructs the persisted test identity for durable wake delivery.
-    source: file://execution_identity.lua
-    method: spawn_orchestrator
-    modules: *execution_identity_modules
-    imports: *execution_identity_imports
-
-  - name: execution_identity.binding
-    kind: contract.binding
-    meta:
-      environment: test
-      test_only: true
-      comment: Test-only implementation of Dataflow's application-owned identity bridge.
-    contracts:
-      - contract: userspace.dataflow:execution_identity
-        default: true
-        methods:
-          capture: app:execution_identity.capture
-          spawn_orchestrator: app:execution_identity.spawn_orchestrator
-
   - name: gateway
     kind: http.service
     addr: :19097