DBOS Instance support

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

@@ -29,6 +29,33 @@
 import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
 
+class ClientWorkflowHandle<T, E extends Exception> implements WorkflowHandle<T, E> {
+
+  private final SystemDatabase systemDatabase;
+  private final String workflowId;
+
+  public ClientWorkflowHandle(SystemDatabase systemDatabase, String workflowId) {
+    this.systemDatabase = systemDatabase;
+    this.workflowId = workflowId;
+  }
+
+  @Override
+  public String workflowId() {
+    return workflowId;
+  }
+
+  @Override
+  public T getResult() throws E {
+    var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
+    return Result.<T, E>process(result);
+  }
+
+  @Override
+  public WorkflowStatus getStatus() {
+    return systemDatabase.getWorkflowStatus(workflowId);
+  }
+}
+
 /**
  * DBOSClient allows external programs to interact with DBOS apps via direct system database access.
  * Example interactions: Start/enqueue a workflow, and get the result Get events and send messages
@@ -470,31 +497,36 @@ public EnqueueOptions(
     String serializationFormat =
         options.serialization() != null ? options.serialization().formatName() : null;
 
-    return DBOSExecutor.enqueueWorkflow(
-        Objects.requireNonNull(
-            options.workflowName(), "EnqueueOptions workflowName must not be null"),
-        Objects.requireNonNull(options.className(), "EnqueueOptions className must not be null"),
-        Objects.requireNonNullElse(options.instanceName(), ""),
-        null,
-        args,
-        new DBOSExecutor.ExecutionOptions(
-            Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString()),
-            Timeout.of(options.timeout()),
-            options.deadline,
+    var workflowId =
+        DBOSExecutor.enqueueWorkflow(
+            Objects.requireNonNull(
+                options.workflowName(), "EnqueueOptions workflowName must not be null"),
             Objects.requireNonNull(
-                options.queueName(), "EnqueueOptions queueName must not be null"),
-            options.deduplicationId,
-            options.priority,
-            options.queuePartitionKey,
-            false,
-            false,
-            serializationFormat),
-        null,
-        null,
-        null,
-        options.appVersion,
-        systemDatabase,
-        this.serializer);
+                options.className(), "EnqueueOptions className must not be null"),
+            Objects.requireNonNullElse(options.instanceName(), ""),
+            null,
+            args,
+            new DBOSExecutor.ExecutionOptions(
+                Objects.requireNonNullElseGet(
+                    options.workflowId(), () -> UUID.randomUUID().toString()),
+                Timeout.of(options.timeout()),
+                options.deadline,
+                Objects.requireNonNull(
+                    options.queueName(), "EnqueueOptions queueName must not be null"),
+                options.deduplicationId,
+                options.priority,
+                options.queuePartitionKey,
+                false,
+                false,
+                serializationFormat),
+            null,
+            null,
+            null,
+            options.appVersion,
+            systemDatabase,
+            this.serializer);
+
+    return new ClientWorkflowHandle<>(systemDatabase, workflowId);
   }
 
   /**

transact/src/main/java/dev/dbos/transact/context/DBOSContext.java

@@ -163,4 +163,9 @@ public static boolean inStep() {
     var ctx = DBOSContextHolder.get();
     return ctx == null ? false : ctx.isInStep();
   }
+
+  public static SerializationStrategy serializationStrategy() {
+    var ctx = DBOSContextHolder.get();
+    return ctx != null ? ctx.getSerialization() : null;
+  }
 }

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

@@ -812,7 +812,7 @@ void resumeWorkflow(String workflowId) throws SQLException {
   String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions options)
       throws SQLException {
 
-    Objects.requireNonNull(options);
+    options = Objects.requireNonNullElseGet(options, ForkOptions::new);
 
     var status = getWorkflowStatus(originalWorkflowId);
     if (status == null) {
@@ -897,7 +897,11 @@ private static void insertForkedWorkflowStatus(
       stmt.setString(6, applicationVersion);
       stmt.setString(7, originalStatus.appId());
       stmt.setString(8, originalStatus.authenticatedUser());
-      stmt.setString(9, JSONUtil.serializeArray(originalStatus.authenticatedRoles()));
+      stmt.setString(
+          9,
+          originalStatus.authenticatedRoles() == null
+              ? null
+              : JSONUtil.serializeArray(originalStatus.authenticatedRoles()));
       stmt.setString(10, originalStatus.assumedRole());
       stmt.setString(11, Constants.DBOS_INTERNAL_QUEUE);
       stmt.setString(

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

@@ -196,7 +196,7 @@ public ExecutorService get() {
       listeners.add(schedulerService);
 
       for (var listener : listeners) {
-        listener.dbosLaunched();
+        listener.dbosLaunched(dbos);
       }
 
       var recoveryTask =
@@ -650,7 +650,7 @@ public <T, E extends Exception> T runStepInternal(
   /** Retrieve the workflowHandle for the workflowId */
   public <R, E extends Exception> WorkflowHandle<R, E> retrieveWorkflow(String workflowId) {
     logger.debug("retrieveWorkflow {}", workflowId);
-    return new WorkflowHandleDBPoll<R, E>(workflowId);
+    return new WorkflowHandleDBPoll<>(this, workflowId);
   }
 
   public void sleep(Duration duration) {
@@ -1089,6 +1089,10 @@ public <T, E extends Exception> WorkflowHandle<T, E> startWorkflow(
     logger.debug("startWorkflow {}", options);
 
     var invocation = captureInvocation(supplier);
+    if (invocation.executor() != this) {
+      throw new IllegalStateException(
+          "The @Workflow method must be called on the DBOS instance passed to the startWorkflow lambda");
+    }
     var workflow = getWorkflow(invocation);
 
     var ctx = DBOSContextHolder.get();
@@ -1278,19 +1282,21 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
             "queue %s does not exist".formatted(options.queueName()));
       }
 
-      return enqueueWorkflow(
-          workflow.name(),
-          workflow.className(),
-          workflow.instanceName(),
-          maxRetries,
-          args,
-          options,
-          parent,
-          executorId(),
-          appVersion(),
-          appId(),
-          systemDatabase,
-          this.serializer);
+      var workflowId =
+          enqueueWorkflow(
+              workflow.name(),
+              workflow.className(),
+              workflow.instanceName(),
+              maxRetries,
+              args,
+              options,
+              parent,
+              executorId(),
+              appVersion(),
+              appId(),
+              systemDatabase,
+              this.serializer);
+      return new WorkflowHandleDBPoll<>(this, workflowId);
     }
 
     logger.debug("executeWorkflow {}({}) {}", workflow.fullyQualifiedName(), args, options);
@@ -1413,10 +1419,10 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
           TimeUnit.MILLISECONDS);
     }
 
-    return new WorkflowHandleFuture<T, E>(workflowId, future, this);
+    return new WorkflowHandleFuture<T, E>(this, workflowId, future);
   }
 
-  public static <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
+  public static String enqueueWorkflow(
       String name,
       String className,
       String instanceName,
@@ -1468,10 +1474,10 @@ public static <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
           options.isDequeuedRequest,
           options.serialization(),
           serializer);
-      return new WorkflowHandleDBPoll<T, E>(workflowId);
+      return workflowId;
     } catch (DBOSWorkflowExecutionConflictException e) {
       logger.debug("Workflow execution conflict for workflowId {}", workflowId);
-      return new WorkflowHandleDBPoll<T, E>(workflowId);
+      return workflowId;
     } catch (DBOSQueueDuplicatedException e) {
       logger.debug(
           "Workflow queue {} reused deduplicationId {}", e.queueName(), e.deduplicationId());

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

@@ -1,12 +1,14 @@
 package dev.dbos.transact.execution;
 
+import dev.dbos.transact.DBOS;
+
 /**
  * For registering callbacks that hear about `DBOS.launch()` and `DBOS.shutdown()`. At this point,
  * DBOS is ready to run workflows, and no additional registrations are allowed.
  */
 public interface DBOSLifecycleListener {
   /** Called from within DBOS.launch, after workflow processing is allowed */
-  void dbosLaunched();
+  void dbosLaunched(DBOS.Instance dbos);
 
   /** Called from within DBOS.shutdown, before workflow processing is stopped */
   void dbosShutDown();