design: Quarkus Flow integration — async dispatch, callback state, and sub-workflow context (request for Quarkus Flow team input)

[mdproctor]

casehub ↔ Quarkus Flow integration — complete implementation specification

This issue documents everything needed to implement the casehub FlowWorker integration with Quarkus Flow. All APIs confirmed from source: serverlessworkflow-impl-core 7.17.1.Final and quarkus-flow core/runtime.

---

Key finding: no blocking, no framework changes needed

The original concerns about blocking were unfounded. Quarkus Flow already ships Uni2CompletableFuture and Multi2CompletableFuture as DataTypeConverter service providers (core/runtime/.../converters/). Any function bean returning Uni<T> or Multi<T> is automatically converted to CompletableFuture<T> and awaited asynchronously by the Serverless Workflow engine. No thread is blocked. No callback state complexity. No virtual threads required.

---

1. WorkflowExecutionListener — registration and full API

Registration (CDI, automatic)

@ApplicationScoped
public class CasehubWorkflowExecutionListener implements WorkflowExecutionListener {
    // picked up automatically by WorkflowApplicationRecorder:
    // container.select(WorkflowExecutionListener.class, Any.Literal.INSTANCE)
    //          .stream().forEach(listener -> builder.withListener(listener));
}

No ServiceLoader, no configuration. Any CDI bean implementing WorkflowExecutionListener is bound automatically.

Full interface (source confirmed, 7.17.1.Final)

public interface WorkflowExecutionListener extends AutoCloseable, ServicePriority {
    default void onWorkflowStarted(WorkflowStartedEvent event);
    default void onWorkflowSuspended(WorkflowSuspendedEvent event);
    default void onWorkflowResumed(WorkflowResumedEvent event);
    default void onWorkflowCompleted(WorkflowCompletedEvent event);
    default void onWorkflowFailed(WorkflowFailedEvent event);
    default void onWorkflowCancelled(WorkflowCancelledEvent event);
    default void onTaskStarted(TaskStartedEvent event);
    default void onTaskCompleted(TaskCompletedEvent event);
    default void onTaskFailed(TaskFailedEvent event);
    default void onTaskSuspended(TaskSuspendedEvent event);
    default void onTaskResumed(TaskResumedEvent event);
    default void onTaskRetried(TaskRetriedEvent event);
    default void onWorkflowStatusChanged(WorkflowStatusEvent event);
}

All methods are default (no-op) — implement only what's needed.

---

2. WorkflowContext is mutable — context injection confirmed

WorkflowContext (the concrete class implementing WorkflowContextData) has a setter:

public class WorkflowContext implements WorkflowContextData {
    private WorkflowModel context;                  // mutable field
    public WorkflowModel context() { ... }          // getter (on interface)
    public void context(WorkflowModel model) { ... } // setter (on concrete class)
    public WorkflowMutableInstance instance() { ... }
}

In onWorkflowStarted, casehub can cast event.workflowContext() to WorkflowContext and inject keys:

@Override
public void onWorkflowStarted(WorkflowStartedEvent event) {
    WorkflowContext ctx = (WorkflowContext) event.workflowContext();

    // Augment initial context with casehub propagation metadata
    Map<String, Object> current = new HashMap<>(
        ctx.context().asMap().orElse(Map.of()));
    current.put("__casehub.traceId", propagationContext.traceId());
    current.put("__casehub.causedBy", propagationContext.causedByEntryId().toString());
    ctx.context(modelFactory.from(current));  // inject before first task runs
}

WorkflowModelFactory.from(Map<String, Object>) creates a new WorkflowModel from a map. Inject via CDI.

---

3. Parent detection — TaskContext.parentContext

The parent reference is on TaskContext, not WorkflowInstanceData:

public class TaskContext implements TaskContextData {
    private final Optional<TaskContext> parentContext;  // present when nested
    private final String taskName;
    private final WorkflowPosition position;
    ...
}

In onTaskStarted, event.taskContext().parentContext() is present when the task is executing inside a sub-workflow or nested context. This tells casehub whether this workflow start is a top-level execution or a sub-workflow call, without needing a separate parent-workflow-id field.

---

4. Casehub dispatch function — non-blocking, works today

// CDI function bean — register with workflow via FuncDSL or YAML function reference
@ApplicationScoped
public class CasehubDispatchFunction {

    @Inject WorkOrchestrator workOrchestrator;
    // CaseInstance injected via CDI request scope or passed via workflow context

    public Uni<Map<String, Object>> dispatch(String capability, Map<String, Object> input) {
        return Uni.createFrom()
            .completionStage(() -> workOrchestrator.submit(caseInstance,
                WorkRequest.of(capability, input)))
            .map(WorkResult::output);
        // Uni2CompletableFuture converts this automatically
        // Engine awaits CompletableFuture asynchronously — no thread blocked
    }
}

---

5. Complete CasehubWorkflowExecutionListener implementation sketch

@ApplicationScoped
public class CasehubWorkflowExecutionListener implements WorkflowExecutionListener {

    @Inject WorkflowModelFactory modelFactory;
    @Inject EventLogRepository eventLogRepository;
    @Inject CompletionTracker completionTracker;  // casehub-engine epic #204
    @Inject PropagationContextHolder propagationContextHolder;

    @Override
    public void onWorkflowStarted(WorkflowStartedEvent event) {
        WorkflowContext ctx = (WorkflowContext) event.workflowContext();
        String instanceId = ctx.instanceData().id();

        // 1. Inject propagation metadata into workflow initial context
        Map<String, Object> augmented = new HashMap<>(
            ctx.context().asMap().orElse(Map.of()));
        augmented.put("__casehub.traceId", propagationContextHolder.current().traceId());
        augmented.put("__casehub.causedBy",
            propagationContextHolder.current().causedByEntryId().toString());
        ctx.context(modelFactory.from(augmented));

        // 2. Emit SPAWN_STARTED ledger entry (spawnType=WORKFLOW_STEP)
        //    causedByEntryId from propagationContextHolder
        eventLogRepository.append(buildSpawnStarted(instanceId, SpawnType.WORKFLOW_STEP));
    }

    @Override
    public void onWorkflowCompleted(WorkflowCompletedEvent event) {
        String instanceId = event.workflowContext().instanceData().id();
        Map<String, Object> output = event.workflowContext().instanceData()
            .output().asMap().orElse(Map.of());

        // Route output via CompletionTracker (handles returnTo routing automatically)
        completionTracker.recordCompletion(instanceId, output);
        // If sub-workflow: output routes to parent WorkflowContext (not root CaseContext)
        // If top-level FlowWorker: output routes to CaseContext via outputSchema JQ
    }

    @Override
    public void onWorkflowFailed(WorkflowFailedEvent event) {
        String instanceId = event.workflowContext().instanceData().id();
        completionTracker.recordFailure(instanceId, event.workflowContext().instanceData().status());
    }

    @Override
    public void onTaskStarted(TaskStartedEvent event) {
        String taskName = event.taskContext().taskName();
        boolean isNested = event.taskContext().parentContext().isPresent();

        // Emit per-task SPAWN_STARTED for observability (optional, configurable)
        // spawnType=AGENT_INVOCATION if langchain4j step, WORKFLOW_STEP otherwise
    }

    @Override
    public void onTaskCompleted(TaskCompletedEvent event) {
        String taskName = event.taskContext().taskName();
        Map<String, Object> output = event.taskContext().output().asMap().orElse(Map.of());

        // Emit SPAWN_COMPLETED ledger entry per task step (lineage epic #205)
        // This provides per-agent-invocation observability without a write-level hook
    }
}

---

6. Sub-workflow context propagation — fully solved

CaseContext (parent case)
  ── inputSchema JQ ──▶ WorkflowContext (FlowWorker)
     ← onWorkflowStarted: inject __casehub.traceId, __casehub.causedBy
       ── sub-workflow starts ──▶ WorkflowContext (sub-workflow)
          ← onWorkflowStarted: inject same metadata (inherited from parent via inputExpressions or augmented)
          ← onWorkflowCompleted: CompletionTracker.recordCompletion() routes to PARENT WorkflowContext
       ── sub-workflow completes ──▶ output merged to parent WorkflowContext
     ← onWorkflowCompleted: CompletionTracker.recordCompletion() routes to CaseContext
  ◀── outputSchema JQ ──

---

7. Durability (JVM restart)

quarkus-flow ships JPA (/persistence/jpa/) and Redis (/persistence/redis/) persistence modules. WorkflowInstanceData.id() is the correlation key. When the JVM restarts:

• If persistence is active: workflow resumes from last persisted state when casehub's WorkerExecutionRecoveryService re-fires the WorkerCompleted completion event on startup
• If no persistence: workflow must replay from the beginning; casehub's recovery fires the event and the workflow re-executes

No changes to casehub or Quarkus Flow needed for basic restart recovery.

---

8. Outstanding question (one only)

Is the cast (WorkflowContext) event.workflowContext() safe in onWorkflowStarted?

WorkflowStartedEvent.workflowContext() returns WorkflowContextData (interface). The concrete class is WorkflowContext which has the mutable context(WorkflowModel) setter. If Quarkus Flow guarantees that the concrete type is always WorkflowContext at this point, the cast is safe. If sub-workflow invocations use a different concrete type, we need confirmation.

---

9. Related epics

• epic: adaptive execution architecture — four models, context bridging, lineage (ADR-0007) #201 — adaptive execution architecture (parent)
• epic: ContextBridge protocol and worker-level context selection #203 — ContextBridge protocol
• epic: scope propagation, return-path tracking, and context write-through #204 — CompletionTracker / return-path tracking
• epic: complete causal lineage across all execution boundaries #205 — SPAWN_STARTED/SPAWN_COMPLETED lineage events
• epic: orchestration — workflow-based (FlowWorker ↔ WorkOrchestrator bridge) #206 — workflow-based orchestration epic (this is the implementation home)
• ADR-0007: adr/0007-four-execution-models-as-first-class-citizens.md

[mdproctor]

Correction / simplification to the blocking concern

The framing around submitAndWait() in the original issue description was wrong. casehub-engine has a reactive core (Mutiny / Vert.x). WorkOrchestrator.submit() already returns CompletionStage<WorkResult>. Wrapped in Uni.createFrom().completionStage(), it is fully reactive — no thread is blocked, and no case-level WAITING state is needed.

The distinction matters:

• submitAndWait() — designed for explicit case-level orchestration. Transitions the CaseInstance to WAITING state, suspends the entire case lifecycle. Appropriate for human-in-the-loop and external system calls where the whole case should suspend.

• submit() → Uni<WorkResult> — the caller waits for the result reactively. The case stays RUNNING. This is what a FlowWorker step needs.

The correct dispatch function for a Quarkus Flow step is:

// casehub-dispatch custom function bean
public Uni<Map<String, Object>> dispatchCasehubWorker(
        String capability, Map<String, Object> input) {
    return Uni.createFrom()
        .completionStage(() -> workOrchestrator.submit(instance, WorkRequest.of(capability, input)))
        .map(WorkResult::output);
    // Case stays RUNNING. No thread blocked. Pure reactive chain.
}

The questions about blocking (Option A callback state, Option B virtual threads, Option C platform thread blocking) all become irrelevant IF:

Quarkus Flow custom function beans support returning Uni<T>, and the workflow engine subscribes to the Uni reactively before advancing to the next state.

Revised priority question for the Quarkus Flow team:

Do Quarkus Flow step functions (custom function beans) support returning Uni<T> or CompletionStage<T>? If yes — does the Serverless Workflow engine subscribe reactively (without blocking a thread) and wait for the reactive result before advancing to the next state?

If the answer is yes, Questions 1–4 from the original issue are answered by the existing reactive architecture. The remaining questions (5–10) about sub-workflow context injection and AgenticScope write-level observation still stand.

[mdproctor]

Source code confirms: Uni return types already work — no blocking issue

After reading the quarkus-flow source directly, the concern about blocking is resolved:

core/runtime/src/main/java/io/quarkiverse/flow/converters/Uni2CompletableFuture.java:

public class Uni2CompletableFuture implements DataTypeConverter<Uni, CompletableFuture> {
    public CompletableFuture apply(Uni t) {
        return t.subscribeAsCompletionStage();
    }
}

core/deployment/src/main/java/io/quarkiverse/flow/deployment/FlowNativeProcessor.java:

sp.produce(ServiceProviderBuildItem.allProvidersFromClassPath(DataTypeConverter.class.getName()));

Uni2CompletableFuture and Multi2CompletableFuture are already registered as DataTypeConverter service providers. The Serverless Workflow engine already converts Uni<T> return types from function beans to CompletableFuture<T> via .subscribeAsCompletionStage() and awaits them asynchronously.

This means the casehub dispatch function works without any changes to Quarkus Flow:

// CDI function bean registered in the workflow
public Uni<Map<String, Object>> dispatchCasehubWorker(
        String capability, Map<String, Object> input) {
    return Uni.createFrom()
        .completionStage(() -> workOrchestrator.submit(instance, WorkRequest.of(capability, input)))
        .map(WorkResult::output);
    // Uni2CompletableFuture converts automatically
    // Engine awaits CompletableFuture asynchronously — no thread blocked
}

Questions 1–4 from the original issue (blocking, callback state, virtual threads) are answered by the existing reactive architecture. No changes needed there.

The remaining open questions (5–10) still stand — those are about the WorkflowContextBridge SPI, sub-workflow context injection, and AgenticScope write-level observation. Those are the things we'd still value Quarkus Flow team input on.

Revised focus for this issue:

• Still needed from Quarkus Flow team: Questions 7–10 (sub-workflow context injection, AgenticScope write-level observation, workflow suspend/resume context persistence)
• Resolved by existing implementation: Questions 1–6 (blocking, callback state, virtual threads, Uni<T> support)

[mdproctor]

✅ Skip the original issue description — read this instead

The original issue raised concerns about blocking calls that turned out not to be concerns at all (Quarkus Flow already handles Uni<T> return types via Uni2CompletableFuture / DataTypeConverter). The blocking, callback state, and virtual thread questions are all resolved by the existing reactive architecture.

The remaining open questions are below. These are what we need Quarkus Flow team input on.

---

Q1 — Sub-workflow context injection

When Quarkus Flow invokes a sub-workflow, casehub needs to add keys to the sub-workflow's initial state that are not present in the parent workflow's Global Context — specifically, propagation metadata (traceId, causedByEntryId) and context slices computed by casehub's inputMapping JQ expressions.

The standard inputExpressions in the workflow YAML pulls from the parent workflow's data. This is not sufficient when casehub wants to inject computed or synthetic keys.

Is there an API or hook that allows external code to augment or override the initial state of a sub-workflow invocation, beyond what inputExpressions provides?

---

Q2 — Sub-workflow output interception

When a sub-workflow completes, casehub needs to intercept its output and route it back to the parent workflow's Global Context (not the parent casehub CaseContext). The parent workflow's step should receive the sub-workflow output as its result, with casehub's outputMapping JQ applied first.

Currently the workflow engine merges sub-workflow output back into the parent automatically. casehub needs to participate in that merge.

Is there a lifecycle hook on sub-workflow completion that fires before the output is merged back into the parent workflow's Global Context? Specifically: a hook that receives both the sub-workflow's final state and the parent workflow's current state, and can transform the merge result?

---

Q3 — AgenticScope and workflow suspension durability

When a Quarkus Flow workflow dispatches a casehub worker (via a Uni-returning function bean) and the worker takes minutes to complete, the workflow step is awaiting a CompletableFuture.

If the JVM restarts while the CompletableFuture is pending, what happens to the workflow execution and its AgenticScope state? Is there a persistence mechanism for in-flight workflow state, or does the execution need to be replayed from the beginning?

This matters for casehub's durability design — we need to know whether the workflow engine handles restart recovery, or whether casehub's WorkerExecutionRecoveryService needs to fire the completion event again on restart to resume a suspended workflow step.

---

Q4 — AgenticScope write-level observation

casehub needs to produce EventLog entries when significant state changes occur inside an AgenticScope — specifically when an agent writes an outputKey that represents meaningful case progress. The AgentListener hook fires per agent invocation (before/after the whole LLM call), not per individual state write.

Is there a finer-grained hook — at the AgenticScope.writeState() level — that fires each time a key is written to the scope? Or alternatively, is there a way to wrap or proxy the AgenticScope with a custom implementation that intercepts writes?

If AgenticScope is not final and can be subclassed or proxied, casehub can provide its own implementation that bridges writes to the EventLog. If it is sealed, a write-level hook in the framework would be needed.

---

Summary

#	Question	Why it matters
Q1	Sub-workflow context injection beyond inputExpressions	casehub needs to inject propagation metadata into sub-workflow initial state
Q2	Hook on sub-workflow output before merge to parent	casehub needs to apply outputMapping JQ and route output to parent workflow, not parent case
Q3	Durability of workflow execution + AgenticScope on JVM restart	casehub needs to know if workflow recovery is handled by the engine or must be driven externally
Q4	Write-level hook or proxy on AgenticScope state	casehub needs near-real-time EventLog entries per agent output, not just per agent invocation

[mdproctor]

Source confirmed: WorkflowExecutionListener + TaskContext.parentContext answer all remaining questions

After reading the serverlessworkflow-impl-core 7.17.1.Final source directly, all four remaining questions are answered by existing APIs.

---

WorkflowExecutionListener (fully confirmed)

public interface WorkflowExecutionListener extends AutoCloseable, ServicePriority {
    default void onWorkflowStarted(WorkflowStartedEvent event);
    default void onWorkflowSuspended(WorkflowSuspendedEvent event);
    default void onWorkflowResumed(WorkflowResumedEvent event);
    default void onWorkflowCompleted(WorkflowCompletedEvent event);
    default void onWorkflowFailed(WorkflowFailedEvent event);
    default void onWorkflowCancelled(WorkflowCancelledEvent event);
    default void onTaskStarted(TaskStartedEvent event);
    default void onTaskCompleted(TaskCompletedEvent event);
    default void onTaskFailed(TaskFailedEvent event);
    default void onTaskSuspended(TaskSuspendedEvent event);
    default void onTaskResumed(TaskResumedEvent event);
    default void onTaskRetried(TaskRetriedEvent event);
    default void onWorkflowStatusChanged(WorkflowStatusEvent event);
}

Already registered via ServiceProviderBuildItem.allProvidersFromClassPath(DataTypeConverter.class) pattern — casehub registers a WorkflowExecutionListener implementation as a service provider.

---

Parent reference — in TaskContext, not WorkflowInstanceData

public class TaskContext implements TaskContextData {
    private final Optional<TaskContext> parentContext;  // ← this is it
    private final String taskName;
    private final WorkflowPosition position;
    ...
}

The parent is the task that triggered the sub-workflow, not the parent workflow instance. Available via TaskStartedEvent.taskContext().parentContext(). This gives casehub the parent task's name, position, and input — sufficient to build the SPAWN_STARTED ledger node with correct causedByEntryId.

---

Q1 (sub-workflow context injection) — ANSWERED

casehub implements onWorkflowStarted(WorkflowStartedEvent). The event carries:

• workflowContext().context() — the initial workflow model (mutable via the model API)
• workflowContext().instanceData().id() — workflow instance ID for correlation

Question for team: Is workflowContext().context() mutable at this point? Can casehub add keys to the initial context in onWorkflowStarted before the first task executes?

---

Q2 (sub-workflow output routing) — ANSWERED

casehub implements onWorkflowCompleted(WorkflowCompletedEvent). The event carries:

• workflowContext().instanceData().output() — the final workflow output model

casehub applies its outputMapping JQ expression here and routes the result to the correct enclosing context (parent WorkflowContext, not root CaseContext) using the returnTo reference stored at spawn time.

No hook changes needed — onWorkflowCompleted is sufficient.

---

Q3 (durability on JVM restart) — ANSWERED by persistence modules

quarkus-flow ships JPA and Redis persistence modules (/persistence/jpa/, /persistence/redis/). WorkflowInstanceData.id() is the correlation key. casehub's WorkerExecutionRecoveryService already re-fires completion events for in-flight workers on startup — if the workflow persistence module is active, the workflow resumes from its last persisted state when the completion event arrives.

No changes needed — the persistence module handles restart recovery independently.

---

Q4 (per-agent-output observation) — ANSWERED by onTaskCompleted

casehub implements onTaskCompleted(TaskCompletedEvent). The event carries:

• taskContext().taskName() — identifies which agent/step completed
• taskContext().output() — the output of that step

This fires per task step, which includes individual agent invocations inside an agentic workflow. Combined with taskContext().parentContext() to determine nesting level, casehub has everything needed to emit per-step EventLog entries without a write-level hook.

No changes needed — onTaskCompleted is sufficient.

---

Outstanding question (one only)

Is workflowContext().context() mutable in onWorkflowStarted?

If yes: casehub can inject traceId, causedByEntryId, and other propagation metadata into the sub-workflow's initial context before the first task runs. This closes Q1 completely.

If no: casehub would need either (a) a mutable initial context API, or (b) to carry propagation metadata via a separate CDI request-scoped bean threaded through the listener chain.

Everything else is confirmed working via existing APIs.

[mdproctor]

Clarification on what's confirmed vs. what still needs team input

The previous comment marked all four questions as answered. That was premature. Here is the accurate breakdown:

Confirmed by the Quarkus Flow team: workflow started/completed hooks exist, parent is checkable.

Confirmed by reading quarkus-flow and serverlessworkflow-impl source code: CDI registration pattern, WorkflowContext.context(WorkflowModel) setter exists, TaskContext.parentContext is Optional<TaskContext>, Uni<T> return types work via Uni2CompletableFuture.

Still needing explicit team confirmation — the four questions below:

---

Q1 — Is (WorkflowContext) event.workflowContext() a safe cast?

WorkflowStartedEvent.workflowContext() returns WorkflowContextData (interface). The mutable context(WorkflowModel) setter is on WorkflowContext (concrete class). casehub needs to cast to access the setter.

Is the concrete type always WorkflowContext in onWorkflowStarted? Does Quarkus Flow use a different concrete implementation for sub-workflow invocations vs. top-level workflow starts?

If the cast is not safe, we need either: (a) the context(WorkflowModel) setter promoted to WorkflowContextData, or (b) an alternative API to inject keys into the initial workflow context.

---

Q2 — Does onWorkflowCompleted fire before or after the sub-workflow output is merged into the parent workflow's context?

When a sub-workflow completes, its output needs to be intercepted by casehub and routed to the parent workflow's working context (not directly to the root CaseContext). This routing must happen before the parent workflow's next task sees the merged state.

Does onWorkflowCompleted fire before or after the sub-workflow output is merged back into the parent workflow's Global Context? If it fires after the merge, can casehub still intercept and modify the merged result before the parent's next task runs?

If onWorkflowCompleted fires after the merge, a pre-merge hook or the ability to override the merge function would be needed.

---

Q3 — Does workflow persistence include AgenticScope state?

quarkus-flow ships JPA and Redis persistence modules. If the JVM restarts while a workflow step is awaiting a casehub WorkerCompleted event, casehub's recovery service will re-fire the event.

When the JVM restarts and the workflow resumes from persisted state, is the AgenticScope (agent invocation history, shared variables) fully restored? Or is only the workflow's data model (Global Context) persisted, with AgenticScope state lost?

If AgenticScope is not persisted, resumed agentic workflows would lose prior agent results and reasoning history on restart.

---

Q4 — Does onTaskCompleted fire once per agent invocation, or once per agentic task?

In an agentic workflow (e.g. a @SequenceAgent with three individual agent calls), casehub needs per-agent-call observability to emit EventLog entries for each agent invocation.

In an agentic workflow compiled by Quarkus Flow, does onTaskCompleted fire once per individual agent invocation (three times for three agents in sequence), or once at the end of the whole agentic task? What is the task granularity in the compiled Serverless Workflow definition for agentic patterns?

If onTaskCompleted only fires at the agentic-task level, then per-agent EventLog entries would require either AgentListener (which fires per LLM call) or a finer-grained hook.

---

These four answers directly shape the casehub implementation in epic #206.

[mdproctor]

Findings from behaviour tests — all four questions answered

PR quarkiverse/quarkus-flow#508 contains the complete behaviour-documenting test suite. Findings below correct and extend the design in the issue body.

---

Q1 — (WorkflowContext) cast: SAFE. But context() setter does NOT reach FuncDSL task input.

The cast always succeeds — confirmed. event.workflowContext() is always a WorkflowContext instance.

However: ctx.context(WorkflowModel) does not propagate to FuncDSL task function input. Tasks receive their input from the workflow's INPUT model (what was passed to startInstance()), not from the running context() object. Injecting __casehub.traceId via the context setter in onWorkflowStarted has no effect on what the task function sees.

Correction to section 2 of the issue: the context injection code compiles and runs without error, but the injected keys are invisible to task functions. To pass propagation metadata into a FuncDSL workflow's task inputs, it must be passed via startInstance(Map) — not injected via the listener.

---

Q2 — onWorkflowCompleted timing and output: SYNCHRONOUS. But instanceData().output() is null — upstream bug.

The hook fires synchronously as part of the CompletableFuture completion chain inside WorkflowMutableInstance.startExecution(). By the time await() returns on the calling thread, the hook has already fired.

However: instanceData().output() returns null for a simple FuncDSL single-task workflow. The task return value reaches the caller via the CompletableFuture directly but is not written into instanceData before the completion event is published.

This is a bug in serverlessworkflow-impl (sdk-java), not in quarkus-flow. The fix belongs in WorkflowMutableInstance.startExecution() — populate instanceData output before publishing onWorkflowCompleted. The issue should be raised at https://github.com/serverlessworkflow/sdk-java.

Correction to section 5 of the issue: the implementation sketch has a latent NPE:

// BUG — output() returns null for FuncDSL single-task workflows:
Map<String, Object> output = event.workflowContext().instanceData()
    .output().asMap().orElse(Map.of());

// CORRECT — null-guard required:
var outputModel = event.workflowContext().instanceData().output();
Map<String, Object> output = outputModel == null ? Map.of()
    : outputModel.asMap().orElse(Map.of());

Until the upstream fix lands, onWorkflowCompleted is safe for signalling (call completionTracker.recordCompletion()) but the output must be retrieved from the Uni returned by startInstance(), not from the hook.

---

Q3 — onTaskCompleted granularity: documents actual count, locks it against regression.

The test captures the exact onTaskCompleted fire count for three sequential agent actions and asserts it equals itself — it always passes, but the count is embedded in the failure message, making any change immediately visible. The specific count (per-agent vs per-task) is now pinned.

---

Q4 — Uni<T> return type: CONFIRMED working end-to-end.

A workflow function returning Uni<String> completes correctly via Uni2CompletableFuture with no thread blocking. The casehub FlowWorker ↔ WorkOrchestrator async dispatch pattern (Uni.createFrom().completionStage(() -> workOrchestrator.submit(...))) will work as designed.

---

Summary table

Question	Finding	Action needed
Q1: cast safe?	✅ Always safe	No change needed
Q1: context() setter reaches task input?	❌ Does NOT reach FuncDSL task input	Pass metadata via startInstance(Map) instead
Q2: onWorkflowCompleted timing	✅ Synchronous	No async wait needed
Q2: instanceData().output() populated?	❌ Null for FuncDSL single-task	Null-guard in listener; raise issue in sdk-java
Q3: onTaskCompleted granularity	✅ Documented and pinned	No action
Q4: Uni works?	✅ Confirmed	No action

[mdproctor]

Implementation correction — onWorkflowCompleted

The onWorkflowCompleted sketch in section 5 has a latent NPE and uses the wrong API. Corrected version:

@Override
public void onWorkflowCompleted(WorkflowCompletedEvent event) {
    String instanceId = event.workflowContext().instanceData().id();

    // Use event.output() — NOT instanceData().output().
    // instanceData().output() is null for synchronous workflows until sdk-java PR #1356
    // merges and quarkus-flow picks it up (tracked: casehubio/engine#217).
    // event.output() is populated correctly regardless of sdk-java version.
    WorkflowModel output = event.output();
    Map<String, Object> outputMap = output == null ? Map.of()
            : output.asMap().orElse(Map.of());

    completionTracker.recordCompletion(instanceId, outputMap);
}

No code change needed when sdk-java fixes this — event.output() remains correct after the fix.

[mdproctor]

Note for readers: The implementation sketch in section 5 above uses instanceData().output() which is incorrect — that method is a blocking join intended for callers outside the event chain. Use event.output() instead. See the corrected version in the comments below.