fix(persist): the storage boundary guarantees valid UTF-8

src/persist/_index.yaml

@@ -51,6 +51,7 @@ entries:
       activation_repo: userspace.dataflow.persist:activation_repo
       commit_repo: userspace.dataflow.persist:commit_repo
       dataflow_consts: userspace.dataflow:consts
+      encoding: userspace.dataflow.persist:encoding
       ops: userspace.dataflow.persist:ops
 
   # userspace.dataflow.persist:commit_repo
@@ -68,6 +69,8 @@ entries:
       - sql
       - json
       - time
+    imports:
+      encoding: userspace.dataflow.persist:encoding
 
   # userspace.dataflow.persist:commit_repo_test
   - name: commit_repo_test
@@ -281,6 +284,7 @@ entries:
     imports:
       activation_repo: userspace.dataflow.persist:activation_repo
       dataflow_consts: userspace.dataflow:consts
+      encoding: userspace.dataflow.persist:encoding
 
   # userspace.dataflow.persist:ops_test
   - name: ops_test
@@ -308,3 +312,23 @@ entries:
       ops: userspace.dataflow.persist:ops
       test: wippy.test:test
     method: run_tests
+
+  # userspace.dataflow.persist:encoding
+  - name: encoding
+    kind: library.lua
+    meta:
+      comment: The storage boundary — every persisted string is valid UTF-8; invalid sequences become U+FFFD in place.
+    source: file://encoding.lua
+
+  # userspace.dataflow.persist:encoding_test
+  - name: encoding_test
+    kind: function.lua
+    meta:
+      type: test
+      group: Workflow
+      comment: Invalid external bytes persist as replacement characters instead of failing the commit.
+    source: file://encoding_test.lua
+    method: run_tests
+    imports:
+      test: wippy.test:test
+      encoding: userspace.dataflow.persist:encoding

src/persist/commit.lua

@@ -6,6 +6,7 @@ local security = require("security")
 local ops = require("ops")
 local commit_repo = require("commit_repo")
 local consts = require("dataflow_consts")
+local encoding = require("encoding")
 local activation_repo = require("activation_repo")
 
 local commit = {}
@@ -626,6 +627,9 @@ function commit._create_commit_only(commit_id, dataflow_id, payload, metadata)
     else
         payload_json = tostring(payload)
     end
+    -- Storage boundary: the commit payload carries node content verbatim and
+    -- external bytes must not fail the write.
+    payload_json = encoding.ensure_utf8(payload_json)
 
     -- Process metadata - encode tables as JSON or use empty object
     local metadata_json = "{}"
@@ -641,6 +645,7 @@ function commit._create_commit_only(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.RFC3339NANO)

src/persist/commit_repo.lua

@@ -1,6 +1,7 @@
 local sql = require("sql")
 local json = require("json")
 local time = require("time")
+local encoding = require("encoding")
 
 local DB_RESOURCE = "app:db"
 
@@ -75,6 +76,7 @@ function commit_repo.create(commit_id, dataflow_id, payload, metadata)
     else
         payload_json = tostring(payload)
     end
+    payload_json = encoding.ensure_utf8(payload_json)
 
     -- Process metadata - encode tables as JSON or use empty object
     local metadata_json = "{}"

src/persist/encoding.lua

@@ -0,0 +1,66 @@
+-- The storage boundary: every string the engine persists is valid UTF-8.
+-- External content — scraped pages, model output — carries arbitrary bytes,
+-- and Postgres rejects an invalid sequence, which would fail the whole commit
+-- and strand the run. Invalid sequences are replaced with U+FFFD in place;
+-- the record persists and the damage stays visible and local.
+local M = {}
+
+local REPLACEMENT = "\239\191\189" -- U+FFFD
+
+function M.ensure_utf8(s: any): any
+    if type(s) ~= "string" then return s end
+    local n = #s
+    local i = 1
+    local out: { string }? = nil
+    local last = 1
+    while i <= n do
+        local c = string.byte(s, i)
+        local len = 0
+        if c < 0x80 then
+            len = 1
+        elseif c >= 0xC2 and c <= 0xDF then
+            len = 2
+        elseif c >= 0xE0 and c <= 0xEF then
+            len = 3
+        elseif c >= 0xF0 and c <= 0xF4 then
+            len = 4
+        end
+        local ok = len > 0
+        if ok and len > 1 then
+            if i + len - 1 > n then
+                ok = false
+            else
+                for j = 1, len - 1 do
+                    local cc = string.byte(s, i + j)
+                    if cc < 0x80 or cc > 0xBF then
+                        ok = false
+                        break
+                    end
+                end
+                if ok and len == 3 then
+                    local c2 = string.byte(s, i + 1)
+                    if (c == 0xE0 and c2 < 0xA0) or (c == 0xED and c2 > 0x9F) then ok = false end
+                elseif ok and len == 4 then
+                    local c2 = string.byte(s, i + 1)
+                    if (c == 0xF0 and c2 < 0x90) or (c == 0xF4 and c2 > 0x8F) then ok = false end
+                end
+            end
+        end
+        if ok then
+            i = i + len
+        else
+            if out == nil then out = {} end
+            local acc = out :: { string }
+            acc[#acc + 1] = s:sub(last, i - 1)
+            acc[#acc + 1] = REPLACEMENT
+            i = i + 1
+            last = i
+        end
+    end
+    if out == nil then return s end
+    local acc = out :: { string }
+    acc[#acc + 1] = s:sub(last)
+    return table.concat(acc)
+end
+
+return M

src/persist/encoding_test.lua

@@ -0,0 +1,49 @@
+local test = require("test")
+local encoding = require("encoding")
+
+local function define_tests()
+    describe("persist encoding boundary", function()
+        it("passes valid UTF-8 through untouched", function()
+            local samples = {
+                "plain ascii",
+                "кириллица и ümlaut",
+                "emoji \240\159\154\128 and CJK \228\184\173\230\150\135",
+                "",
+            }
+            for _, s in ipairs(samples) do
+                test.eq(encoding.ensure_utf8(s), s)
+            end
+        end)
+
+        it("replaces invalid bytes with U+FFFD instead of failing", function()
+            local dirty = "scraped \255\254 bytes \192\128 inside"
+            local clean = encoding.ensure_utf8(dirty)
+            test.is_true(clean:find("\255", 1, true) == nil)
+            test.is_true(clean:find("\192", 1, true) == nil)
+            test.is_true(clean:find("scraped ", 1, true) ~= nil)
+            test.is_true(clean:find(" inside", 1, true) ~= nil)
+            test.is_true(clean:find("\239\191\189", 1, true) ~= nil)
+        end)
+
+        it("replaces truncated multi-byte sequences at end of string", function()
+            local truncated = "tail \226\130" -- first two bytes of a 3-byte sequence
+            local clean = encoding.ensure_utf8(truncated)
+            test.eq(clean, "tail \239\191\189\239\191\189")
+        end)
+
+        it("rejects UTF-16 surrogates encoded as UTF-8", function()
+            local surrogate = "x\237\160\128y" -- U+D800
+            local clean = encoding.ensure_utf8(surrogate)
+            test.is_true(clean:find("\237", 1, true) == nil)
+            test.eq(clean:sub(1, 1), "x")
+            test.eq(clean:sub(-1), "y")
+        end)
+
+        it("leaves non-strings alone", function()
+            test.eq(encoding.ensure_utf8(nil), nil)
+            test.eq(encoding.ensure_utf8(42), 42)
+        end)
+    end)
+end
+
+return { run_tests = test.run_cases(define_tests) }

src/persist/ops.lua

@@ -4,6 +4,7 @@ local uuid = require("uuid")
 local json = require("json")
 local consts = require("dataflow_consts")
 local activation_repo = require("activation_repo")
+local encoding = require("encoding")
 
 -- Use shared constants from consts
 local constants = {
@@ -295,6 +296,7 @@ handlers[constants.COMMAND_TYPES.CREATE_NODE] = function(tx, dataflow_id, op_id,
         end
         metadata = encoded
     end
+    metadata = encoding.ensure_utf8(metadata)
 
     local config = payload.config or "{}"
     if type(config) == "table" then
@@ -552,6 +554,9 @@ 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"
     local node_id = payload.node_id or sql.as.null()
@@ -795,12 +800,14 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id,
             if encode_err then return nil, "Failed to encode content: " .. encode_err end
             content_value = encoded
         end
+        content_value = encoding.ensure_utf8(content_value)
         local metadata = payload.metadata or "{}"
         if type(metadata) == "table" then
             local encoded, encode_err = json.encode(metadata)
             if encode_err then return nil, "Failed to encode metadata: " .. encode_err end
             metadata = encoded
         end
+        metadata = encoding.ensure_utf8(metadata)
         local inserted, insert_err = sql.builder.insert("dataflow_data")
             :set_map({
                 data_id = payload.data_id,