Compatibility fixes for system database sharing

transact/src/main/java/dev/dbos/transact/DBOSClient.java

@@ -371,6 +371,49 @@ public EnqueueOptions(@NonNull String workflowName, @NonNull String queueName) {
           null);
     }
 
+    /**
+     * Constructs options with no explicit owning application, which enqueues the workflow for the
+     * enqueueing application.
+     */
+    public EnqueueOptions(
+        @NonNull String workflowName,
+        @Nullable String className,
+        @Nullable String instanceName,
+        @NonNull String queueName,
+        @Nullable String workflowId,
+        @Nullable String appVersion,
+        @Nullable Duration timeout,
+        @Nullable Instant deadline,
+        @Nullable String deduplicationId,
+        @Nullable Integer priority,
+        @Nullable String queuePartitionKey,
+        @Nullable Duration delay,
+        @Nullable SerializationStrategy serialization,
+        @Nullable String authenticatedUser,
+        @Nullable String assumedRole,
+        @Nullable List<String> authenticatedRoles,
+        @Nullable Map<String, Object> attributes) {
+      this(
+          workflowName,
+          className,
+          instanceName,
+          queueName,
+          workflowId,
+          appVersion,
+          timeout,
+          deadline,
+          deduplicationId,
+          priority,
+          queuePartitionKey,
+          delay,
+          serialization,
+          authenticatedUser,
+          assumedRole,
+          authenticatedRoles,
+          attributes,
+          null);
+    }
+
     public EnqueueOptions(
         @NonNull String workflowName, @Nullable String className, @NonNull String queueName) {
       this(

transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java

@@ -437,6 +437,16 @@ public static Duration toDuration(Long ms) {
     return ms != null ? Duration.ofMillis(ms) : null;
   }
 
+  /**
+   * The error column of a workflow or a step, or null when it records no error.
+   *
+   * <p>The Go SDK stores the error as a non-nullable string, so a workflow that succeeded leaves ""
+   * behind where this one writes NULL. Empty only: no SDK writes a blank non-empty value.
+   */
+  public static String errorOrNull(String error) {
+    return error != null && error.isEmpty() ? null : error;
+  }
+
   /**
    * Initializes the status of a workflow.
    *

transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java

@@ -1,6 +1,7 @@
 package dev.dbos.transact.database.dao;
 
 import dev.dbos.transact.database.DbContext;
+import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.exceptions.*;
 import dev.dbos.transact.internal.DebugTriggers;
 import dev.dbos.transact.json.DBOSSerializer;
@@ -144,7 +145,7 @@ public static StepResult checkStepResult(
       try (ResultSet rs = pstmt.executeQuery()) {
         if (rs.next()) { // Check if any operation output row exists
           String output = rs.getString("output");
-          String error = rs.getString("error");
+          String error = SystemDatabase.errorOrNull(rs.getString("error"));
           String _stepName = rs.getString("function_name");
           String serialization = rs.getString("serialization");
           result =
@@ -222,7 +223,7 @@ static List<StepInfo> listWorkflowSteps(
           int functionId = rs.getInt("function_id");
           String functionName = rs.getString("function_name");
           String outputData = rs.getString("output");
-          String errorData = rs.getString("error");
+          String errorData = SystemDatabase.errorOrNull(rs.getString("error"));
           String childWorkflowId = rs.getString("child_workflow_id");
           Long startedAt = rs.getObject("started_at_epoch_ms", Long.class);
           Long completedAt = rs.getObject("completed_at_epoch_ms", Long.class);
@@ -231,7 +232,10 @@ static List<StepInfo> listWorkflowSteps(
           Object outputVal = null;
           ErrorResult stepError = null;
 
-          if (Objects.requireNonNullElse(loadOutput, true)) {
+          // As for a workflow's status: the steps are what is wanted, the payloads one field of
+          // each. See SerializationUtil.canDeserialize.
+          if (Objects.requireNonNullElse(loadOutput, true)
+              && SerializationUtil.canDeserialize(serialization, serializer)) {
             if (outputData != null) {
               try {
                 outputVal =

transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java

@@ -25,6 +25,7 @@
 import dev.dbos.transact.workflow.GetWorkflowAggregatesInput;
 import dev.dbos.transact.workflow.ListWorkflowsInput;
 import dev.dbos.transact.workflow.StepAggregateRow;
+import dev.dbos.transact.workflow.StepInfo;
 import dev.dbos.transact.workflow.WorkflowAggregateRow;
 import dev.dbos.transact.workflow.WorkflowDelay;
 import dev.dbos.transact.workflow.WorkflowEvent;
@@ -47,6 +48,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -1233,8 +1235,11 @@ private static WorkflowStatus resultsToWorkflowStatus(
     String attributesJson = rs.getString("attributes");
     String serializedInput = loadInput ? rs.getString("inputs") : null;
     String serializedOutput = loadOutput ? rs.getString("output") : null;
-    String serializedError = loadOutput ? rs.getString("error") : null;
+    String serializedError = loadOutput ? SystemDatabase.errorOrNull(rs.getString("error")) : null;
     String serialization = loadInput || loadOutput ? rs.getString("serialization") : null;
+    // A status read reaches other applications' rows on purpose, and wants their metadata; an
+    // unreadable payload comes back null rather than failing the read.
+    boolean readable = SerializationUtil.canDeserialize(serialization, serializer);
     WorkflowStatus info =
         new WorkflowStatus(
             rs.getString("workflow_uuid"),
@@ -1247,14 +1252,16 @@ private static WorkflowStatus resultsToWorkflowStatus(
             (authenticatedRolesJson != null)
                 ? JsonUtility.fromJson(authenticatedRolesJson, new TypeReference<List<String>>() {})
                 : null,
-            loadInput
+            loadInput && readable
                 ? SerializationUtil.deserializePositionalArgs(
                     serializedInput, serialization, serializer)
                 : null,
-            loadOutput
+            loadOutput && readable
                 ? SerializationUtil.deserializeValue(serializedOutput, serialization, serializer)
                 : null,
-            loadOutput ? ErrorResult.deserialize(serializedError, serialization, serializer) : null,
+            loadOutput && readable
+                ? ErrorResult.deserialize(serializedError, serialization, serializer)
+                : null,
             rs.getString("executor_id"),
             SystemDatabase.toInstant(rs.getObject("created_at", Long.class)),
             SystemDatabase.toInstant(rs.getObject("updated_at", Long.class)),
@@ -1327,7 +1334,7 @@ public static <T> Result<T> awaitWorkflowResult(
               }
 
               case ERROR -> {
-                String error = rs.getString("error");
+                String error = SystemDatabase.errorOrNull(rs.getString("error"));
                 Throwable t = SerializationUtil.deserializeError(error, serialization, serializer);
                 return Result.failure(t);
               }
@@ -2246,6 +2253,35 @@ static List<WorkflowStream> listWorkflowStreams(Connection conn, String schema,
     return streams;
   }
 
+  /**
+   * Refuse to move a workflow whose payloads this runtime cannot handle.
+   *
+   * <p>Export and import round-trip the payloads through this runtime's serializers, so neither can
+   * settle for the null a status read reports: a payload dropped on the way through restores a
+   * workflow that never had it.
+   */
+  private static void requireSerializerFor(
+      String action,
+      String workflowId,
+      String workflowSerialization,
+      List<StepInfo> steps,
+      DBOSSerializer serializer) {
+    var formats = new LinkedHashSet<String>();
+    if (!SerializationUtil.canDeserialize(workflowSerialization, serializer)) {
+      formats.add(workflowSerialization);
+    }
+    for (var step : steps) {
+      if (!SerializationUtil.canDeserialize(step.serialization(), serializer)) {
+        formats.add(step.serialization());
+      }
+    }
+    if (!formats.isEmpty()) {
+      throw new IllegalStateException(
+          "Cannot %s workflow %s: it is serialized as %s, which this application has no serializer for"
+              .formatted(action, workflowId, String.join(", ", formats)));
+    }
+  }
+
   public static List<ExportedWorkflow> exportWorkflow(
       DbContext ctx, String workflowId, boolean exportChildren) throws SQLException {
 
@@ -2263,6 +2299,9 @@ public static List<ExportedWorkflow> exportWorkflow(
         var steps =
             StepsDAO.listWorkflowSteps(
                 conn, ctx.schema(), ctx.serializer(), wfid, true, null, null);
+        if (status != null) {
+          requireSerializerFor("export", wfid, status.serialization(), steps, ctx.serializer());
+        }
         var events = listWorkflowEvents(conn, ctx.schema(), wfid);
         var eventHistory = listWorkflowEventHistory(conn, ctx.schema(), wfid);
         var streams = listWorkflowStreams(conn, ctx.schema(), wfid);
@@ -2276,6 +2315,14 @@ public static void importWorkflow(DbContext ctx, List<ExportedWorkflow> workflow
       throws SQLException {
 
     DBOSSerializer serializer = ctx.serializer();
+    // The whole batch, before anything is written: export and import are a single-SDK affair but
+    // not a single-configuration one, and a payload we cannot re-serialize imports empty.
+    for (var workflow : workflows) {
+      var s = workflow.status();
+      requireSerializerFor(
+          "import", s.workflowId(), s.serialization(), workflow.steps(), serializer);
+    }
+
     var wfSQL =
         """
         INSERT INTO "%s".workflow_status (

transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

@@ -1696,6 +1696,22 @@ public <T, E extends Exception> WorkflowHandle<T, E> executeWorkflowById(
       throw new DBOSNonExistentWorkflowException(workflowId);
     }
 
+    // Reading the row reports an unreadable payload as null, so running the workflow refuses for
+    // itself: a workflow invoked with arguments it never had is worse than one marked ERROR.
+    if (!SerializationUtil.canDeserialize(status.serialization(), serializer)) {
+      var e =
+          new IllegalStateException(
+              "Cannot run workflow %s: its arguments are serialized as %s, which this application has no deserializer for"
+                  .formatted(workflowId, status.serialization()));
+      logger.error("Unreadable serialization for workflow {}", workflowId, e);
+      // Recorded in this runtime's own format, not the row's: a format we cannot read is one
+      // we cannot write either, and serializing the error into it would throw and leave the
+      // workflow PENDING forever — the hang this refusal exists to prevent. The status is the
+      // part that has to land.
+      persistWorkflowError(workflowId, e, null);
+      throw e;
+    }
+
     Object[] inputs = status.input();
     var wfName =
         RegisteredWorkflow.fullyQualifiedName(

transact/src/main/java/dev/dbos/transact/json/SerializationUtil.java

@@ -26,6 +26,19 @@ public final class SerializationUtil {
 
   private SerializationUtil() {}
 
+  /**
+   * Whether this runtime can read the given serialization format.
+   *
+   * <p>The two built-in formats always, and a custom serializer only for the format it names. A
+   * null format is the native one, written before the column existed.
+   */
+  public static boolean canDeserialize(String serialization, DBOSSerializer customSerializer) {
+    if (serialization == null || PORTABLE.equals(serialization) || NATIVE.equals(serialization)) {
+      return true;
+    }
+    return customSerializer != null && customSerializer.name().equals(serialization);
+  }
+
   // ============ Value Serialization ============
 
   /**

transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java

@@ -1,5 +1,6 @@
 package dev.dbos.transact.txstep;
 
+import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.workflow.internal.StepResult;
 
 import java.sql.Connection;
@@ -66,7 +67,7 @@ public static Optional<StepResult> readResult(
             stepId,
             stepName,
             rs.getString("output"),
-            rs.getString("error"),
+            SystemDatabase.errorOrNull(rs.getString("error")),
             null,
             rs.getString("serialization")));
   }

transact/src/main/java/dev/dbos/transact/workflow/GetStepAggregatesInput.java

@@ -41,6 +41,35 @@ public record GetStepAggregatesInput(
     }
   }
 
+  /**
+   * Constructs an input with no application filter, which covers this application's own steps plus
+   * unclaimed ones.
+   */
+  public GetStepAggregatesInput(
+      boolean groupByFunctionName,
+      boolean groupByStatus,
+      boolean selectCount,
+      boolean selectMaxDuration,
+      Duration timeBucketSize,
+      List<String> status,
+      List<String> functionName,
+      List<String> workflowIdPrefix,
+      Instant completedAfter,
+      Instant completedBefore) {
+    this(
+        groupByFunctionName,
+        groupByStatus,
+        selectCount,
+        selectMaxDuration,
+        timeBucketSize,
+        status,
+        functionName,
+        workflowIdPrefix,
+        completedAfter,
+        completedBefore,
+        null);
+  }
+
   public GetStepAggregatesInput() {
     this(false, false, true, false, null, null, null, null, null, null, null);
   }

transact/src/main/java/dev/dbos/transact/workflow/GetWorkflowAggregatesInput.java

@@ -65,6 +65,62 @@ public record GetWorkflowAggregatesInput(
   }
 
   /** Constructs a default input with {@code selectCount=true} and no group-by or filter flags. */
+  /**
+   * Constructs an input with no application filter and no grouping by application, which covers
+   * this application's own workflows plus unclaimed ones.
+   */
+  public GetWorkflowAggregatesInput(
+      boolean groupByStatus,
+      boolean groupByName,
+      boolean groupByQueueName,
+      boolean groupByExecutorId,
+      boolean groupByApplicationVersion,
+      boolean selectCount,
+      boolean selectMinCreatedAt,
+      boolean selectMaxQueueWait,
+      boolean selectMaxTotalLatency,
+      Duration timeBucketSize,
+      List<String> workflowName,
+      List<String> status,
+      List<String> queueName,
+      List<String> executorIds,
+      List<String> applicationVersion,
+      List<String> workflowIdPrefix,
+      Instant startTime,
+      Instant endTime,
+      Instant completedAfter,
+      Instant completedBefore,
+      Instant dequeuedAfter,
+      Instant dequeuedBefore,
+      Map<String, Object> attributes) {
+    this(
+        groupByStatus,
+        groupByName,
+        groupByQueueName,
+        groupByExecutorId,
+        groupByApplicationVersion,
+        false,
+        selectCount,
+        selectMinCreatedAt,
+        selectMaxQueueWait,
+        selectMaxTotalLatency,
+        timeBucketSize,
+        workflowName,
+        status,
+        queueName,
+        executorIds,
+        applicationVersion,
+        null,
+        workflowIdPrefix,
+        startTime,
+        endTime,
+        completedAfter,
+        completedBefore,
+        dequeuedAfter,
+        dequeuedBefore,
+        attributes);
+  }
+
   public GetWorkflowAggregatesInput() {
     this(
         false, false, false, false, false, false, true, false, false, false, null, null, null, null,