fix: async @EventHandler detached unit publishes lifecycle events under aggregation (#433)

docs/events.md

@@ -264,10 +264,12 @@ gets the same shape). Concretely:
   thread (`CALLER_RUNS`), the `timeout` budget cannot bound it — the time-box arms only
   after the inline run completes. The unit itself (fresh beans, teardown) still applies.
 
-Multi-module note: in aggregated (multi-module) containers, async units get correct frames,
-per-module bean isolation, and teardown — but publish no lifecycle events (module
-containers are constructed silent; the aggregator is not in the async dispatch path —
-tracked by [#433](https://github.com/tomas-samek/tiko-di/issues/433)).
+Multi-module note: in aggregated (multi-module) containers, async units behave exactly as
+single-module ones — correct frames, per-module bean isolation, teardown, **and** one
+`EventStartedEvent`/`EventEndingEvent` pair per dispatch (per retry attempt) on the shared
+bus. The aggregator is not on the async dispatch path, so the per-module container publishes
+its detached unit's lifecycle itself, even though it stays silent for the *sync* units the
+aggregator brackets (resolved in [#433](https://github.com/tomas-samek/tiko-di/issues/433)).
 
 ## Graceful shutdown drain
 

tiko-examples/05_multi_module/app/src/test/java/io/tiko/examples/multimodule/app/MultiModuleLifecycleEventsTest.java

@@ -8,10 +8,13 @@
 import io.tiko.events.ApplicationStartedEvent;
 import io.tiko.events.EventEndingEvent;
 import io.tiko.events.EventStartedEvent;
+import io.tiko.examples.multimodule.modulea.UserCreatedEvent;
 import io.tiko.runtime.AggregatingContainer;
 import io.tiko.runtime.LocalEventBus;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.Test;
 
@@ -109,6 +112,40 @@ void supplyInEventScopePublishesExactlyOneLifecyclePairAcrossModules() {
         assertThat(ending.get(0).eventId()).isEqualTo(started.get(0).eventId());
     }
 
+    @Test
+    void asyncHandlerDispatchPublishesExactlyOneLifecyclePairAcrossModules() throws InterruptedException {
+        // #433: async @EventHandler dispatch under an aggregator opens its own EVENT unit on a
+        // per-module container built with publishLifecycleEvents=false. The aggregator is not on
+        // the async dispatch path (generated dispatch calls the module container directly), so the
+        // module container is the sole publisher of the detached unit's pair — exactly one
+        // EventStarted/EventEnding pair, one eventId, must reach the shared bus.
+        EventBus eventBus = newLocalEventBus();
+        List<EventStartedEvent> started = new CopyOnWriteArrayList<>();
+        List<EventEndingEvent> ending = new CopyOnWriteArrayList<>();
+        CountDownLatch pairComplete = new CountDownLatch(1);
+        eventBus.subscribe(EventStartedEvent.class, started::add);
+        eventBus.subscribe(EventEndingEvent.class, e -> {
+            ending.add(e);
+            pairComplete.countDown();
+        });
+
+        Container container = new AggregatingContainer(eventBus, ctx -> {}, null);
+        try {
+            container.start();
+            // AsyncUserAuditor (module-a) handles this async → detached EVENT unit on the executor.
+            eventBus.publish(new UserCreatedEvent(1L, "Alice", "alice@example.com"));
+            assertThat(pairComplete.await(5, TimeUnit.SECONDS))
+                    .as("async detached unit must complete and publish its EventEndingEvent")
+                    .isTrue();
+        } finally {
+            container.shutdown();
+        }
+
+        assertThat(started).hasSize(1);
+        assertThat(ending).hasSize(1);
+        assertThat(ending.get(0).eventId()).isEqualTo(started.get(0).eventId());
+    }
+
     private EventBus newLocalEventBus() {
         return new LocalEventBus();
     }

tiko-examples/05_multi_module/module-a/src/main/java/io/tiko/examples/multimodule/modulea/AsyncUserAuditor.java

@@ -0,0 +1,26 @@
+package io.tiko.examples.multimodule.modulea;
+
+import io.tiko.Scope;
+import io.tiko.annotations.Component;
+import io.tiko.annotations.EventHandler;
+
+/**
+ * Async cross-frame handler used to prove that async {@code @EventHandler} dispatch under an
+ * {@code AggregatingContainer} opens its own EVENT unit and publishes that unit's lifecycle
+ * events (#433).
+ *
+ * <p>Async dispatch runs the handler in a detached EVENT scope on the shared executor. The
+ * per-module container is constructed by the aggregator with {@code publishLifecycleEvents=false}
+ * (so sync units, which the aggregator brackets, are not double-published) — yet the aggregator
+ * is <em>not</em> on the async dispatch path, so the module container must be the sole publisher
+ * of the detached unit's {@code EventStartedEvent}/{@code EventEndingEvent} pair.
+ */
+@Component(scope = Scope.SINGLETON)
+public class AsyncUserAuditor {
+
+    @EventHandler(async = true)
+    public void onUserCreated(UserCreatedEvent event) {
+        // Body intentionally trivial — the test observes the unit's lifecycle bracket, not a
+        // side effect. The detached EVENT scope opens and tears down around this invocation.
+    }
+}

tiko-processor/src/main/java/io/tiko/processor/generator/ContainerGenerator.java

@@ -196,6 +196,7 @@ private void generateOne(
 
         // Add scope management methods
         containerBuilder.addMethod(createRunInEventScopeMethod());
+        containerBuilder.addMethod(createRunInEventScopeCoreMethod());
         containerBuilder.addMethod(createRunInDetachedEventScopeMethod());
         containerBuilder.addMethod(createSupplyInEventScopeMethod());
         containerBuilder.addMethod(createPublishUnitLifecycleMethod());
@@ -473,7 +474,7 @@ private FieldSpec createStartInvokedField() {
      * the aggregator can publish exactly once on the shared bus (#45).
      */
     private FieldSpec createPublishLifecycleEventsField() {
-        return FieldSpec.builder(TypeName.BOOLEAN, "publishLifecycleEvents", Modifier.PRIVATE, Modifier.FINAL)
+        return FieldSpec.builder(TypeName.BOOLEAN, PUBLISH_LIFECYCLE_FIELD, Modifier.PRIVATE, Modifier.FINAL)
                 .build();
     }
 
@@ -559,7 +560,7 @@ private MethodSpec createConstructor() {
                 .addParameter(EventBus.class, "eventBus")
                 .addParameter(ClassName.get("io.tiko", "ErrorHandler"), "errorHandler")
                 .addParameter(ClassName.get("java.util.concurrent", "ExecutorService"), "userEventExecutor")
-                .addParameter(TypeName.BOOLEAN, "publishLifecycleEvents")
+                .addParameter(TypeName.BOOLEAN, PUBLISH_LIFECYCLE_FIELD)
                 .addParameter(Duration.class, "shutdownTimeout")
                 .addParameter(ClassName.get("io.tiko.runtime", "TikoOptions"), "options")
                 .addStatement("this.eventBus = eventBus")
@@ -970,6 +971,12 @@ private static void emitUnitFrameGuard(MethodSpec.Builder method, String storage
     /** Lifecycle-publish gate shared by start/shutdown and both scope brackets (#45, #339). */
     private static final String IF_PUBLISH_LIFECYCLE = "if (publishLifecycleEvents)";
 
+    /** Standing per-container lifecycle-publish flag: field and constructor-parameter name (S1192). */
+    private static final String PUBLISH_LIFECYCLE_FIELD = "publishLifecycleEvents";
+
+    /** Per-call gate parameter of the {@code __runInEventScope} core (#433, S1192). */
+    private static final String PUBLISH_LIFECYCLE_PARAM = "__publishLifecycle";
+
     private static final String RETURN_EXISTING = "return __existing";
 
     private static final ClassName NO_SUCH_COMPONENT = ClassName.get("io.tiko", "NoSuchComponentException");
@@ -985,13 +992,37 @@ private static void emitUnitFrameGuard(MethodSpec.Builder method, String storage
             .build();
 
     /**
-     * Creates runInEventScope method.
+     * Creates the public {@code runInEventScope} method. Delegates to the
+     * {@code __runInEventScope(task, publishLifecycle)} core with the container's standing
+     * {@code publishLifecycleEvents} flag — so a per-module container under an
+     * AggregatingContainer (flag off) stays silent and the aggregator publishes the sole pair.
      */
     private MethodSpec createRunInEventScopeMethod() {
-        MethodSpec.Builder method = MethodSpec.methodBuilder("runInEventScope")
+        return MethodSpec.methodBuilder("runInEventScope")
                 .addModifiers(Modifier.PUBLIC)
                 .addAnnotation(Override.class)
-                .addParameter(Runnable.class, "task");
+                .addParameter(Runnable.class, "task")
+                .addStatement("__runInEventScope(task, publishLifecycleEvents)")
+                .build();
+    }
+
+    /**
+     * Creates the private {@code __runInEventScope(Runnable, boolean)} core (#433). Holds the
+     * single-frame guard, frame bookkeeping, and the unit-lifecycle bracket; the
+     * {@code __publishLifecycle} parameter — not the standing field — decides whether the
+     * {@code EventStartedEvent}/{@code EventEndingEvent} pair is published.
+     *
+     * <p>Public {@link #createRunInEventScopeMethod()} passes the standing
+     * {@code publishLifecycleEvents} flag (sync path: aggregator brackets the unit, module
+     * stays silent). {@link #createRunInDetachedEventScopeMethod()} forces {@code true}: an
+     * async detached unit is bracketed by no one else — the aggregator is not on the async
+     * dispatch path — so the module container is its sole lifecycle publisher.
+     */
+    private MethodSpec createRunInEventScopeCoreMethod() {
+        MethodSpec.Builder method = MethodSpec.methodBuilder("__runInEventScope")
+                .addModifiers(Modifier.PRIVATE)
+                .addParameter(Runnable.class, "task")
+                .addParameter(TypeName.BOOLEAN, PUBLISH_LIFECYCLE_PARAM);
         method.beginControlFlow("if ($T.TRUE.equals(__unitFrameOpen.get()))", Boolean.class);
         method.addStatement(
                 THROW_TYPE_WITH_MESSAGE,
@@ -1002,12 +1033,12 @@ private MethodSpec createRunInEventScopeMethod() {
         method.addStatement("__unitFrameOpen.set($T.TRUE)", Boolean.class)
                 .addStatement("$T __eventId = $T.randomUUID().toString()", String.class, UUID.class)
                 .addStatement("$T __eventStart = $T.now()", Instant.class, Instant.class);
-        emitGatedUnitStartedPublish(method);
+        emitGatedUnitStartedPublish(method, PUBLISH_LIFECYCLE_PARAM);
         method.beginControlFlow("try")
                 .addStatement("task.run()")
                 .nextControlFlow("finally")
                 .addStatement("$T __eventEnd = $T.now()", Instant.class, Instant.class);
-        emitGatedUnitEndingPublish(method);
+        emitGatedUnitEndingPublish(method, PUBLISH_LIFECYCLE_PARAM);
         method.addStatement("__closeEventScope()").endControlFlow();
         return method.build();
     }
@@ -1030,7 +1061,7 @@ private MethodSpec createRunInDetachedEventScopeMethod() {
                 .addStatement("eventScoped.set(new $T<>())", LinkedHashMap.class)
                 .addStatement("__unitFrameOpen.set($T.FALSE)", Boolean.class);
         method.beginControlFlow("try")
-                .addStatement("runInEventScope(task)")
+                .addStatement("__runInEventScope(task, true)")
                 .nextControlFlow("finally")
                 .addStatement("eventScoped.set(__savedScope)")
                 .addStatement("__unitFrameOpen.set(__savedFrameOpen)")
@@ -1062,12 +1093,12 @@ private MethodSpec createSupplyInEventScopeMethod() {
         method.addStatement("__unitFrameOpen.set($T.TRUE)", Boolean.class)
                 .addStatement("$T __eventId = $T.randomUUID().toString()", String.class, UUID.class)
                 .addStatement("$T __eventStart = $T.now()", Instant.class, Instant.class);
-        emitGatedUnitStartedPublish(method);
+        emitGatedUnitStartedPublish(method, PUBLISH_LIFECYCLE_FIELD);
         method.beginControlFlow("try")
                 .addStatement("return supplier.get()")
                 .nextControlFlow("finally")
                 .addStatement("$T __eventEnd = $T.now()", Instant.class, Instant.class);
-        emitGatedUnitEndingPublish(method);
+        emitGatedUnitEndingPublish(method, PUBLISH_LIFECYCLE_FIELD);
         method.addStatement("__closeEventScope()").endControlFlow();
         return method.build();
     }
@@ -1079,15 +1110,15 @@ private MethodSpec createSupplyInEventScopeMethod() {
      * published by the aggregator inside the innermost frame — instead of one pair per
      * nested module frame.
      */
-    private void emitGatedUnitStartedPublish(MethodSpec.Builder method) {
-        method.beginControlFlow(IF_PUBLISH_LIFECYCLE);
+    private void emitGatedUnitStartedPublish(MethodSpec.Builder method, String gateExpression) {
+        method.beginControlFlow("if ($L)", gateExpression);
         method.addStatement("__publishUnitLifecycle(new $T(__eventId, __eventStart))", EVENT_STARTED);
         method.endControlFlow();
     }
 
     /** EventEndingEvent counterpart of {@link #emitGatedUnitStartedPublish} (#339). */
-    private void emitGatedUnitEndingPublish(MethodSpec.Builder method) {
-        method.beginControlFlow(IF_PUBLISH_LIFECYCLE);
+    private void emitGatedUnitEndingPublish(MethodSpec.Builder method, String gateExpression) {
+        method.beginControlFlow("if ($L)", gateExpression);
         method.addStatement(
                 "__publishUnitLifecycle(new $T(__eventId, __eventEnd, $T.between(__eventStart, __eventEnd)))",
                 EVENT_ENDING,

tiko-processor/src/test/java/io/tiko/processor/ContainerGeneratorLifecycleEventGateTest.java

@@ -59,32 +59,54 @@ void generated_shutdown_gates_application_ending_event_publish() throws IOExcept
     }
 
     @Test
-    void runInEventScopeGatesUnitLifecyclePublishes() throws IOException {
-        assertUnitLifecyclePublishesGated("public void runInEventScope");
+    void runInEventScopeDelegatesToCoreWithStandingFlag() throws IOException {
+        // #433: the public sync bracket carries no publish of its own — it passes the container's
+        // standing publishLifecycleEvents flag to the __runInEventScope core, so a per-module
+        // container under an AggregatingContainer (flag off) stays silent for sync units and the
+        // aggregator remains the sole publisher.
+        String content = generateContainerSource();
+        assertThat(content).contains("__runInEventScope(task, publishLifecycleEvents)");
+    }
+
+    @Test
+    void detachedScopeForcesUnitLifecyclePublish() throws IOException {
+        // #433: an async detached unit is bracketed by no aggregator (async dispatch does not
+        // traverse the aggregator's scope path), so the module container is its sole lifecycle
+        // publisher — detached forces the core's publish flag true regardless of the standing flag.
+        String content = generateContainerSource();
+        assertThat(content).contains("__runInEventScope(task, true)");
+    }
+
+    @Test
+    void coreScopeGatesUnitLifecyclePublishesOnParameter() throws IOException {
+        assertUnitLifecyclePublishesGated("private void __runInEventScope", "__publishLifecycle");
     }
 
     @Test
     void supplyInEventScopeGatesUnitLifecyclePublishes() throws IOException {
-        assertUnitLifecyclePublishesGated("public <T> T supplyInEventScope");
+        assertUnitLifecyclePublishesGated("public <T> T supplyInEventScope", "publishLifecycleEvents");
     }
 
     /**
-     * #339: per-module containers under an AggregatingContainer are constructed with
-     * {@code publishLifecycleEvents=false}; their unit-of-work brackets must not publish
-     * their own EventStarted/EventEnding pair — the aggregator publishes exactly one.
+     * #339 / #433: unit-of-work brackets publish their EventStarted/EventEnding pair only when
+     * their gate is set. The sync core ({@code __runInEventScope}) gates on its
+     * {@code __publishLifecycle} parameter; {@code supplyInEventScope} gates on the standing
+     * {@code publishLifecycleEvents} field — so a module container under an AggregatingContainer
+     * (field off) does not double-publish the aggregator's single pair.
      */
-    private void assertUnitLifecyclePublishesGated(String methodSignature) throws IOException {
+    private void assertUnitLifecyclePublishesGated(String methodSignature, String gateExpression) throws IOException {
         String content = generateContainerSource();
+        String gate = "if (" + gateExpression + ")";
 
         int methodPos = content.indexOf(methodSignature);
         assertThat(methodPos).as("scope method present").isPositive();
 
-        int startedGate = content.indexOf("if (publishLifecycleEvents)", methodPos);
+        int startedGate = content.indexOf(gate, methodPos);
         int startedPublish = content.indexOf("__publishUnitLifecycle(new EventStartedEvent", methodPos);
         assertThat(startedGate).as("gate before EventStartedEvent publish").isPositive();
         assertThat(startedPublish).isGreaterThan(startedGate);
 
-        int endingGate = content.indexOf("if (publishLifecycleEvents)", startedPublish);
+        int endingGate = content.indexOf(gate, startedPublish);
         int endingPublish = content.indexOf("__publishUnitLifecycle(new EventEndingEvent", startedPublish);
         assertThat(endingGate).as("gate before EventEndingEvent publish").isPositive();
         assertThat(endingPublish).isGreaterThan(endingGate);

tiko-processor/src/test/java/io/tiko/processor/DetachedEventScopeTest.java

@@ -34,14 +34,16 @@ void generatedContainerCarriesDetachedEventScopeMethod() throws IOException {
         String content = new String(container.openInputStream().readAllBytes(), StandardCharsets.UTF_8);
 
         // One chain: method present; detachment saves both ThreadLocals (the __savedScope
-        // prefix pins the SAVE, not an incidental getter), clears to a fresh map, delegates,
-        // restores in finally; and the signature stays package-private.
+        // prefix pins the SAVE, not an incidental getter), clears to a fresh map, delegates to
+        // the scope core with the publish flag forced true (#433 — a detached async unit is its
+        // own sole lifecycle publisher), restores in finally; and the signature stays
+        // package-private.
         assertThat(content)
                 .contains("void runInDetachedEventScope(Runnable task)")
                 .contains("__savedScope = eventScoped.get()")
                 .contains("__savedFrameOpen = __unitFrameOpen.get()")
                 .contains("eventScoped.set(new LinkedHashMap<>())")
-                .contains("runInEventScope(task)")
+                .contains("__runInEventScope(task, true)")
                 .contains("eventScoped.set(__savedScope)")
                 .contains("__unitFrameOpen.set(__savedFrameOpen)")
                 .doesNotContain("public void runInDetachedEventScope");