Optimize runtime event processing assembly and projection paths

src/main/java/at/procon/eventhub/processing/api/UnifiedRuntimeProcessingController.java

@@ -2,7 +2,6 @@ package at.procon.eventhub.processing.api;
 
 import at.procon.eventhub.processing.dto.UnifiedRuntimeDerivedProjectionResultDto;
 import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
-import at.procon.eventhub.processing.dto.UnifiedRuntimeTachographEsperScopeResultDto;
 import at.procon.eventhub.processing.eventprocessing.RuntimeEventProcessingService;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
@@ -19,10 +18,6 @@ import at.procon.eventhub.processing.model.UnifiedRuntimeEventBundle;
 import at.procon.eventhub.processing.service.UnifiedRuntimeDerivedProjectionService;
 import at.procon.eventhub.processing.service.UnifiedRuntimeDriverTimelineService;
 import at.procon.eventhub.processing.service.UnifiedRuntimeEventAssemblyService;
-import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationApiRequest;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -39,10 +34,8 @@ public class UnifiedRuntimeProcessingController {
     private final UnifiedRuntimeEventAssemblyService eventAssemblyService;
     private final UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService;
     private final UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService;
-    private final RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService;
     private final RuntimeEventProcessingService runtimeEventProcessingService;
     private final RuntimeProcessingExecutionService runtimeProcessingExecutionService;
-    private final RuntimeTachographParityValidationService tachographParityValidationService;
     private final RuntimeMixedSourceEvidenceValidationService mixedSourceEvidenceValidationService;
 
     public UnifiedRuntimeProcessingController(
@@ -50,18 +43,16 @@ public class UnifiedRuntimeProcessingController {
             UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
             UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService
     ) {
-        this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, null, null, null, null, null);
+        this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, null, null, null);
     }
 
     public UnifiedRuntimeProcessingController(
             UnifiedRuntimeEventAssemblyService eventAssemblyService,
             UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
             UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
-            RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
             RuntimeEventProcessingService runtimeEventProcessingService
     ) {
-        this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService,
+        this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService, runtimeEventProcessingService, null, null);
-                tachographEsperScopeProcessingService, runtimeEventProcessingService, null, null, null);
     }
 
     @Autowired
@@ -69,35 +60,18 @@ public class UnifiedRuntimeProcessingController {
             UnifiedRuntimeEventAssemblyService eventAssemblyService,
             UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
             UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
-            RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
             RuntimeEventProcessingService runtimeEventProcessingService,
-            RuntimeTachographParityValidationService tachographParityValidationService,
             RuntimeMixedSourceEvidenceValidationService mixedSourceEvidenceValidationService,
             RuntimeProcessingExecutionService runtimeProcessingExecutionService
     ) {
         this.eventAssemblyService = eventAssemblyService;
         this.runtimeDriverTimelineService = runtimeDriverTimelineService;
         this.runtimeDerivedProjectionService = runtimeDerivedProjectionService;
-        this.tachographEsperScopeProcessingService = tachographEsperScopeProcessingService;
         this.runtimeEventProcessingService = runtimeEventProcessingService;
         this.runtimeProcessingExecutionService = runtimeProcessingExecutionService;
-        this.tachographParityValidationService = tachographParityValidationService;
         this.mixedSourceEvidenceValidationService = mixedSourceEvidenceValidationService;
     }
 
-    public UnifiedRuntimeProcessingController(
-            UnifiedRuntimeEventAssemblyService eventAssemblyService,
-            UnifiedRuntimeDriverTimelineService runtimeDriverTimelineService,
-            UnifiedRuntimeDerivedProjectionService runtimeDerivedProjectionService,
-            RuntimeDriverWorkingTimeScopeProcessingService tachographEsperScopeProcessingService,
-            RuntimeEventProcessingService runtimeEventProcessingService,
-            RuntimeTachographParityValidationService tachographParityValidationService
-    ) {
-        this(eventAssemblyService, runtimeDriverTimelineService, runtimeDerivedProjectionService,
-                tachographEsperScopeProcessingService, runtimeEventProcessingService,
-                tachographParityValidationService, null, null);
-    }
-
     @PostMapping("/driver-events")
     public ResponseEntity<UnifiedRuntimeEventBundle> loadDriverEvents(
             @RequestBody UnifiedRuntimeProcessingApiRequest request
@@ -157,17 +131,6 @@ public class UnifiedRuntimeProcessingController {
     }
 
 
-    @PostMapping("/event-processing/validation/tachograph-parity")
-    public ResponseEntity<RuntimeTachographParityValidationResultDto> validateTachographParity(
-            @RequestBody RuntimeTachographParityValidationApiRequest request
-    ) {
-        if (tachographParityValidationService == null) {
-            throw new IllegalStateException("Runtime tachograph parity validation service is not configured.");
-        }
-        return ResponseEntity.ok(tachographParityValidationService.validate(request));
-    }
-
-
     @PostMapping("/event-processing/validation/mixed-source-evidence")
     public ResponseEntity<RuntimeMixedSourceEvidenceValidationResultDto> validateMixedSourceEvidence(
             @RequestBody RuntimeMixedSourceEvidenceValidationApiRequest request
@@ -177,22 +140,4 @@ public class UnifiedRuntimeProcessingController {
         }
         return ResponseEntity.ok(mixedSourceEvidenceValidationService.validate(request));
     }
-
-    @PostMapping("/tachograph/esper-processing")
-    public ResponseEntity<UnifiedRuntimeTachographEsperScopeResultDto> runTachographEsperProcessing(
-            @RequestBody UnifiedRuntimeProcessingApiRequest request
-    ) {
-        if (runtimeEventProcessingService != null) {
-            RuntimeEventProcessingResultDto genericResult = runtimeEventProcessingService.process(
-                    RuntimeEventProcessingApiRequest.tachographDriverEsper(request)
-            );
-            return ResponseEntity.ok(UnifiedRuntimeTachographEsperScopeResultDto.fromGenericRuntimeEventProcessingResult(genericResult));
-        }
-        if (tachographEsperScopeProcessingService == null) {
-            throw new IllegalStateException("Tachograph Esper scope processing service is not configured.");
-        }
-        return ResponseEntity.ok(UnifiedRuntimeTachographEsperScopeResultDto.fromDriverWorkingTime(
-                tachographEsperScopeProcessingService.processScope(request)
-        ));
-    }
 }

src/main/java/at/procon/eventhub/processing/driverworkingtime/service/DriverWorkingTimeProcessingCore.java

@@ -19,6 +19,8 @@ import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 /**
@@ -32,6 +34,8 @@ import org.springframework.stereotype.Service;
 @Service
 public class DriverWorkingTimeProcessingCore {
 
+    private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeProcessingCore.class);
+
     private final DriverWorkingTimeDerivedProjectionEngine derivedProjectionEngine;
 
     public DriverWorkingTimeProcessingCore(
@@ -42,6 +46,7 @@ public class DriverWorkingTimeProcessingCore {
 
     public DriverWorkingTimeProcessingResultDto process(DriverWorkingTimeProcessingInput input) {
         Objects.requireNonNull(input, "input must not be null");
+        long startedAtNanos = System.nanoTime();
         OffsetDateTime loadedFrom = input.loadedFrom();
         OffsetDateTime loadedTo = input.loadedTo();
         String driverKey = input.driverKey();
@@ -51,6 +56,7 @@ public class DriverWorkingTimeProcessingCore {
             throw new IllegalArgumentException("occurredTo must not be before occurredFrom.");
         }
 
+        long activityClipStartedAtNanos = System.nanoTime();
         List<DriverWorkingTimeActivityInterval> activityIntervals = clipActivityIntervals(
                 input.activityIntervals(),
                 requestedFrom,
@@ -63,9 +69,13 @@ public class DriverWorkingTimeProcessingCore {
                 requestedFrom,
                 requestedTo
         );
+        long activityClipMs = elapsedMillis(activityClipStartedAtNanos);
 
+        long derivedProjectionBuildStartedAtNanos = System.nanoTime();
         DriverWorkingTimeDerivedProjectionBundle derivedProjectionBundle = derivedProjectionEngine.build(input);
+        long derivedProjectionBuildMs = elapsedMillis(derivedProjectionBuildStartedAtNanos);
 
+        long derivedIntervalClipStartedAtNanos = System.nanoTime();
         List<DriverWorkingTimeDrivingInterruptionInterval> drivingInterruptionIntervals =
                 clipDrivingInterruptionIntervals(
                         derivedProjectionBundle.drivingInterruptionIntervals(),
@@ -131,8 +141,9 @@ public class DriverWorkingTimeProcessingCore {
                 requestedFrom,
                 requestedTo
         );
+        long derivedIntervalClipMs = elapsedMillis(derivedIntervalClipStartedAtNanos);
 
-        return new DriverWorkingTimeProcessingResultDto(
+        DriverWorkingTimeProcessingResultDto result = new DriverWorkingTimeProcessingResultDto(
                 input.sessionId(),
                 driverKey,
                 input.sourceKind(),
@@ -168,6 +179,19 @@ public class DriverWorkingTimeProcessingCore {
                 supportGeoEvents,
                 combinedNotes(input.notes())
         );
+        LOG.info(
+                "Driver working-time core processed driver {} in {} ms (activityClipMs: {}, derivedProjectionBuildMs: {}, derivedIntervalClipMs: {}, activityIntervals: {}, drivingIntervals: {}, vehicleUsageIntervals: {}, supportGeoEvents: {})",
+                driverKey,
+                elapsedMillis(startedAtNanos),
+                activityClipMs,
+                derivedProjectionBuildMs,
+                derivedIntervalClipMs,
+                result.activityIntervalCount(),
+                result.drivingIntervalCount(),
+                result.vehicleUsageIntervalCount(),
+                result.supportGeoEventCount()
+        );
+        return result;
     }
 
     public DriverWorkingTimeProcessingResultDto processDriverWorkingTime(DriverWorkingTimeProcessingInput input) {
@@ -649,4 +673,8 @@ public class DriverWorkingTimeProcessingCore {
         }
         return left.isBefore(right) ? left : right;
     }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/driverworkingtime/service/DriverWorkingTimeReusableProjectionBuilder.java

@@ -26,15 +26,22 @@ import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.Collections;
+import java.util.Deque;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.stereotype.Service;
@@ -43,11 +50,41 @@ import org.springframework.util.StreamUtils;
 @Service
 public class DriverWorkingTimeReusableProjectionBuilder {
 
+    private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeReusableProjectionBuilder.class);
     private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
     private static final String DRIVING_DERIVED_PROJECTION_BUNDLE_EPL_TEMPLATE =
             loadResource("esper/driver-working-time-derived-projections.epl");
+    private static final Map<String, Object> ACTIVITY_INTERVAL_INPUT_DEFINITION = activityIntervalInputDefinitionStatic();
+    private static final Map<String, Object> VEHICLE_USAGE_INTERVAL_INPUT_DEFINITION = vehicleUsageIntervalInputDefinitionStatic();
+    private static final Map<String, Object> SUPPORT_GEO_EVIDENCE_INPUT_DEFINITION = supportGeoEvidenceInputDefinitionStatic();
+    private static final int MAX_IDLE_RUNTIMES_PER_DEFINITION = 2;
+    private static final List<String> REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES = List.of(
+            "delete from PreviousRestCandidateCoverageInterval",
+            "delete from OpenPotentialInVehicleTripState",
+            "delete from SupportGeoEvidenceWindow",
+            "delete from DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow",
+            "delete from DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow",
+            "delete from DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow",
+            "delete from DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow",
+            "delete from DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow",
+            "delete from DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow",
+            "delete from DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow",
+            "delete from DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow",
+            "delete from DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow",
+            "delete from DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow",
+            "delete from DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow",
+            "delete from DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow",
+            "delete from DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow",
+            "delete from DailyWeeklyRestCandidateCoverageEmittedKeyWindow",
+            "delete from PreviousSignificantDrivingInterval",
+            "context PerDriver delete from PreviousVehicleUsageInterval"
+    );
 
     private final EventHubProperties properties;
+    private final ConcurrentMap<PreparedProjectionDefinitionCacheKey, PreparedProjectionDefinition> preparedDefinitions =
+            new ConcurrentHashMap<>();
+    private final ConcurrentMap<PreparedProjectionDefinitionCacheKey, ReusableProjectionRuntimePool> reusableRuntimePools =
+            new ConcurrentHashMap<>();
 
     @Autowired
     public DriverWorkingTimeReusableProjectionBuilder(
@@ -78,6 +115,7 @@ public class DriverWorkingTimeReusableProjectionBuilder {
             int significantDrivingMinutes,
             int minimumRestPeriodMinutes
     ) {
+        long startedAtNanos = System.nanoTime();
         if ((activityInputEvents == null || activityInputEvents.isEmpty())
                 && (vehicleUsageInputEvents == null || vehicleUsageInputEvents.isEmpty())) {
             return emptyDerivedProjectionBundle();
@@ -93,22 +131,9 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         List<DriverWorkingTimeRestCoverageInterval> potentialInVehicleOvernightStayIntervals = new ArrayList<>();
         List<DriverWorkingTimePotentialInVehicleTripInterval> potentialInVehicleTripIntervals = new ArrayList<>();
 
-        executeWithRuntime(
+        DerivedProjectionRuntimeExecutionMetrics runtimeMetrics = executeWithRuntime(
-                configuration -> {
+                significantDrivingMinutes,
-                    configuration.getCommon().addEventType(
+                minimumRestPeriodMinutes,
-                            "TachographActivityIntervalInputEvent",
-                            activityIntervalInputDefinition()
-                    );
-                    configuration.getCommon().addEventType(
-                            "TachographVehicleUsageIntervalInputEvent",
-                            vehicleUsageIntervalInputDefinition()
-                    );
-                    configuration.getCommon().addEventType(
-                            "TachographSupportGeoEvidenceInputEvent",
-                            supportGeoEvidenceInputDefinition()
-                    );
-                },
-                renderDrivingDerivedProjectionBundleEpl(significantDrivingMinutes, minimumRestPeriodMinutes),
                 Map.of(
                         "drivingInterruptionIntervals",
                         newData -> collectDrivingInterruptionIntervalModels(newData, drivingInterruptionIntervals),
@@ -129,34 +154,26 @@ public class DriverWorkingTimeReusableProjectionBuilder {
                         "potentialInVehicleTripIntervals",
                         newData -> collectPotentialInVehicleTripIntervalModels(newData, potentialInVehicleTripIntervals)
                 ),
-                runtime -> {
+                sender -> {
                     if (supportGeoInputEvents != null) {
                         for (Map<String, Object> supportGeoEvidence : supportGeoInputEvents) {
-                            runtime.getEventService().sendEventMap(
+                            sender.sendSupportGeoEvent(supportGeoEvidence);
-                                    supportGeoEvidence,
-                                    "TachographSupportGeoEvidenceInputEvent"
-                            );
                         }
                     }
                     if (vehicleUsageInputEvents != null) {
                         for (Map<String, Object> interval : vehicleUsageInputEvents) {
-                            runtime.getEventService().sendEventMap(
+                            sender.sendVehicleUsageEvent(interval);
-                                    interval,
-                                    "TachographVehicleUsageIntervalInputEvent"
-                            );
                         }
                     }
                     if (activityInputEvents != null) {
                         for (Map<String, Object> interval : activityInputEvents) {
-                            runtime.getEventService().sendEventMap(
+                            sender.sendActivityEvent(interval);
-                                    interval,
-                                    "TachographActivityIntervalInputEvent"
-                            );
                         }
                     }
                 }
         );
 
+        long sortOutputStartedAtNanos = System.nanoTime();
         List<DriverWorkingTimeRestCoverageInterval> sortedDailyWeeklyRestCandidateCoverageIntervals =
                 sortDailyWeeklyRestCandidateCoverageIntervalsCommon(dailyWeeklyRestCandidateCoverageIntervals);
         List<DriverWorkingTimeRestCoverageInterval> sortedUnclassifiedDailyWeeklyRestCandidateCoverageIntervals =
@@ -165,6 +182,27 @@ public class DriverWorkingTimeReusableProjectionBuilder {
                 sortPotentialHomeOvernightStayIntervalsCommon(potentialHomeOvernightStayIntervals);
         List<DriverWorkingTimeRestCoverageInterval> sortedPotentialInVehicleOvernightStayIntervals =
                 sortPotentialInVehicleOvernightStayIntervalsCommon(potentialInVehicleOvernightStayIntervals);
+        long sortOutputMs = elapsedMillis(sortOutputStartedAtNanos);
+
+        LOG.info(
+                "Driver working-time derived projection bundle built in {} ms (definitionCacheHit: {}, definitionPreparationMs: {}, runtimePoolHit: {}, runtimeInitMs: {}, deployMs: {}, listenerRegistrationMs: {}, runtimeResetMs: {}, sendSupportGeoMs: {}, sendVehicleUsageMs: {}, sendActivityMs: {}, destroyMs: {}, sortOutputMs: {}, activityInputEvents: {}, vehicleUsageInputEvents: {}, supportGeoInputEvents: {})",
+                elapsedMillis(startedAtNanos),
+                runtimeMetrics.definitionCacheHit(),
+                runtimeMetrics.definitionPreparationMs(),
+                runtimeMetrics.runtimePoolHit(),
+                runtimeMetrics.runtimeInitMs(),
+                runtimeMetrics.deployMs(),
+                runtimeMetrics.listenerRegistrationMs(),
+                runtimeMetrics.runtimeResetMs(),
+                runtimeMetrics.sendSupportGeoMs(),
+                runtimeMetrics.sendVehicleUsageMs(),
+                runtimeMetrics.sendActivityMs(),
+                runtimeMetrics.destroyMs(),
+                sortOutputMs,
+                safeList(activityInputEvents).size(),
+                safeList(vehicleUsageInputEvents).size(),
+                safeList(supportGeoInputEvents).size()
+        );
 
         return new DriverWorkingTimeDerivedProjectionBundle(
                 sortDrivingInterruptionIntervalsCommon(drivingInterruptionIntervals),
@@ -237,39 +275,180 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         );
     }
 
-    private void executeWithRuntime(
+    private DerivedProjectionRuntimeExecutionMetrics executeWithRuntime(
-            Consumer<Configuration> configurationSetup,
+            int significantDrivingMinutes,
-            String epl,
+            int minimumRestPeriodMinutes,
             Map<String, Consumer<EventBean[]>> listeners,
-            Consumer<EPRuntime> sender
+            Consumer<DerivedProjectionEventSender> sender
     ) {
+        long definitionPreparationMs = 0L;
+        boolean definitionCacheHit = false;
+        boolean runtimePoolHit = false;
+        long runtimeInitMs = 0L;
+        long deployMs = 0L;
+        long listenerRegistrationMs = 0L;
+        long runtimeResetMs = 0L;
+        long sendSupportGeoMs = 0L;
+        long sendVehicleUsageMs = 0L;
+        long sendActivityMs = 0L;
+        long destroyMs = 0L;
+        ReusableProjectionRuntimePool runtimePool = null;
+        ReusableProjectionRuntime reusableRuntime = null;
+        boolean discardRuntime = false;
+        try {
+            PreparedProjectionDefinitionCacheKey cacheKey = new PreparedProjectionDefinitionCacheKey(
+                    significantDrivingMinutes,
+                    minimumRestPeriodMinutes,
+                    Math.max(1, properties.getRuntimeProcessing().getRestCandidateGeoLookbackMinutes()),
+                    Math.max(1, properties.getRuntimeProcessing().getRestCandidateGeoLookaheadMinutes()),
+                    Math.max(0, properties.getRuntimeProcessing().getRestCandidateGeoStationaryMaxMeters()),
+                    Math.max(
+                            properties.getRuntimeProcessing().getRestCandidateGeoStationaryMaxMeters(),
+                            properties.getRuntimeProcessing().getRestCandidateGeoMinorMovementMaxMeters()
+                    )
+            );
+            PreparedProjectionDefinition preparedDefinition = preparedDefinitions.get(cacheKey);
+            definitionCacheHit = preparedDefinition != null;
+            if (!definitionCacheHit) {
+                long definitionPreparationStartedAtNanos = System.nanoTime();
+                PreparedProjectionDefinition candidate = prepareProjectionDefinition(
+                        significantDrivingMinutes,
+                        minimumRestPeriodMinutes
+                );
+                PreparedProjectionDefinition existing = preparedDefinitions.putIfAbsent(cacheKey, candidate);
+                preparedDefinition = existing == null ? candidate : existing;
+                definitionCacheHit = existing != null;
+                definitionPreparationMs = definitionCacheHit ? 0L : elapsedMillis(definitionPreparationStartedAtNanos);
+            }
+
+            runtimePool = reusableRuntimePools.computeIfAbsent(cacheKey, ignored -> new ReusableProjectionRuntimePool());
+            reusableRuntime = runtimePool.acquire(preparedDefinition, listeners);
+            runtimePoolHit = reusableRuntime.poolHit();
+            runtimeInitMs = reusableRuntime.runtimeInitMs();
+            deployMs = reusableRuntime.deployMs();
+            listenerRegistrationMs = reusableRuntime.listenerRegistrationMs();
+
+            ReusableProjectionRuntimeExecution execution = reusableRuntime.execute(listeners, sender);
+            runtimeResetMs = execution.runtimeResetMs();
+            sendSupportGeoMs = execution.sendSupportGeoMs();
+            sendVehicleUsageMs = execution.sendVehicleUsageMs();
+            sendActivityMs = execution.sendActivityMs();
+        } catch (EPCompileException | EPDeployException e) {
+            discardRuntime = true;
+            throw new IllegalStateException("Cannot compile/deploy reusable driver working-time projection EPL bundle", e);
+        } catch (RuntimeException e) {
+            discardRuntime = true;
+            throw e;
+        } finally {
+            if (reusableRuntime != null) {
+                if (discardRuntime || runtimePool == null) {
+                    destroyMs = reusableRuntime.destroy();
+                } else {
+                    destroyMs = runtimePool.release(reusableRuntime);
+                }
+            }
+        }
+        return new DerivedProjectionRuntimeExecutionMetrics(
+                definitionCacheHit,
+                definitionPreparationMs,
+                runtimePoolHit,
+                runtimeInitMs,
+                deployMs,
+                listenerRegistrationMs,
+                runtimeResetMs,
+                sendSupportGeoMs,
+                sendVehicleUsageMs,
+                sendActivityMs,
+                destroyMs
+        );
+    }
+
+    private ReusableProjectionRuntime createReusableRuntime(
+            PreparedProjectionDefinition preparedDefinition,
+            Map<String, Consumer<EventBean[]>> listeners
+    ) throws EPDeployException, EPCompileException {
         EPRuntime runtime = null;
         try {
-            Configuration configuration = new Configuration();
+            long runtimeInitStartedAtNanos = System.nanoTime();
-            configurationSetup.accept(configuration);
+            Configuration configuration = createRuntimeConfiguration();
             String runtimeUri = "eventhub-driver-working-time-reusable-projection-" + RUNTIME_COUNTER.incrementAndGet();
             runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);
+            long runtimeInitMs = elapsedMillis(runtimeInitStartedAtNanos);
 
-            CompilerArguments arguments = new CompilerArguments(configuration);
+            long deployStartedAtNanos = System.nanoTime();
-            EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, arguments);
+            EPDeployment deployment = runtime.getDeploymentService().deploy(preparedDefinition.compiled());
-            EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);
+            long deployMs = elapsedMillis(deployStartedAtNanos);
-            for (Map.Entry<String, Consumer<EventBean[]>> entry : listeners.entrySet()) {
+
+            ReusableProjectionRuntime reusableRuntime = new ReusableProjectionRuntime(runtime, runtimeInitMs, deployMs, false);
+
+            long listenerRegistrationStartedAtNanos = System.nanoTime();
+            for (String statementName : listeners.keySet()) {
                 runtime.getDeploymentService()
-                        .getStatement(deployment.getDeploymentId(), entry.getKey())
+                        .getStatement(deployment.getDeploymentId(), statementName)
-                        .addListener((newData, oldData, statement, rt) -> entry.getValue().accept(newData));
+                        .addListener((newData, oldData, statement, rt) -> reusableRuntime.onStatementEvents(statement.getName(), newData));
             }
-
+            reusableRuntime.listenerRegistrationMs(elapsedMillis(listenerRegistrationStartedAtNanos));
-            sender.accept(runtime);
+            reusableRuntime.cleanupQueries(compileCleanupQueries(runtime));
-        } catch (EPCompileException | EPDeployException e) {
+            return reusableRuntime;
-            throw new IllegalStateException("Cannot compile/deploy reusable driver working-time projection EPL bundle", e);
+        } catch (EPDeployException | EPCompileException | RuntimeException ex) {
-        } finally {
+            if (runtime != null && !runtime.isDestroyed()) {
-            if (runtime != null) {
                 runtime.destroy();
             }
+            throw ex;
         }
     }
 
+    private List<EPCompiled> compileCleanupQueries(EPRuntime runtime) throws EPCompileException {
+        CompilerArguments arguments = new CompilerArguments(runtime.getConfigurationDeepCopy());
+        arguments.getPath().add(runtime.getRuntimePath());
+        List<EPCompiled> compiledQueries = new ArrayList<>(REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES.size());
+        for (String cleanupQuery : REUSABLE_RUNTIME_STATE_CLEANUP_QUERIES) {
+            compiledQueries.add(EPCompilerProvider.getCompiler().compileQuery(cleanupQuery, arguments));
+        }
+        return List.copyOf(compiledQueries);
+    }
+
     private Map<String, Object> activityIntervalInputDefinition() {
+        return ACTIVITY_INTERVAL_INPUT_DEFINITION;
+    }
+
+    private Map<String, Object> vehicleUsageIntervalInputDefinition() {
+        return VEHICLE_USAGE_INTERVAL_INPUT_DEFINITION;
+    }
+
+    private Map<String, Object> supportGeoEvidenceInputDefinition() {
+        return SUPPORT_GEO_EVIDENCE_INPUT_DEFINITION;
+    }
+
+    private PreparedProjectionDefinition prepareProjectionDefinition(
+            int significantDrivingMinutes,
+            int minimumRestPeriodMinutes
+    ) throws EPCompileException {
+        Configuration configuration = createRuntimeConfiguration();
+        String epl = renderDrivingDerivedProjectionBundleEpl(significantDrivingMinutes, minimumRestPeriodMinutes);
+        CompilerArguments arguments = new CompilerArguments(configuration);
+        EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, arguments);
+        return new PreparedProjectionDefinition(compiled);
+    }
+
+    private Configuration createRuntimeConfiguration() {
+        Configuration configuration = new Configuration();
+        configuration.getCommon().addEventType(
+                "TachographActivityIntervalInputEvent",
+                activityIntervalInputDefinition()
+        );
+        configuration.getCommon().addEventType(
+                "TachographVehicleUsageIntervalInputEvent",
+                vehicleUsageIntervalInputDefinition()
+        );
+        configuration.getCommon().addEventType(
+                "TachographSupportGeoEvidenceInputEvent",
+                supportGeoEvidenceInputDefinition()
+        );
+        return configuration;
+    }
+
+    private static Map<String, Object> activityIntervalInputDefinitionStatic() {
         Map<String, Object> definition = new LinkedHashMap<>();
         definition.put("sessionId", UUID.class);
         definition.put("driverKey", String.class);
@@ -292,10 +471,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         definition.put("synthetic", boolean.class);
         definition.put("clippedToRequestedPeriod", boolean.class);
         definition.put("level", String.class);
-        return definition;
+        return Collections.unmodifiableMap(definition);
     }
 
-    private Map<String, Object> vehicleUsageIntervalInputDefinition() {
+    private static Map<String, Object> vehicleUsageIntervalInputDefinitionStatic() {
         Map<String, Object> definition = new LinkedHashMap<>();
         definition.put("sessionId", UUID.class);
         definition.put("driverKey", String.class);
@@ -313,10 +492,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         definition.put("vehicleKey", String.class);
         definition.put("sourceKind", String.class);
         definition.put("sourceIntervalIds", java.util.List.class);
-        return definition;
+        return Collections.unmodifiableMap(definition);
     }
 
-    private Map<String, Object> supportGeoEvidenceInputDefinition() {
+    private static Map<String, Object> supportGeoEvidenceInputDefinitionStatic() {
         Map<String, Object> definition = new LinkedHashMap<>();
         definition.put("sessionId", UUID.class);
         definition.put("driverKey", String.class);
@@ -330,7 +509,7 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         definition.put("longitude", Double.class);
         definition.put("odometerKm", Long.class);
         definition.put("priority", int.class);
-        return definition;
+        return Collections.unmodifiableMap(definition);
     }
 
     private Map<String, Object> toActivityIntervalInputMap(
@@ -758,6 +937,10 @@ public class DriverWorkingTimeReusableProjectionBuilder {
         return value instanceof Long number ? number : null;
     }
 
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
+
     private static String loadResource(String path) {
         try {
             ClassPathResource resource = new ClassPathResource(path);
@@ -766,4 +949,230 @@ public class DriverWorkingTimeReusableProjectionBuilder {
             throw new IllegalStateException("Cannot load EPL resource: " + path, e);
         }
     }
+
+    private record PreparedProjectionDefinition(
+            EPCompiled compiled
+    ) {
+    }
+
+    private record PreparedProjectionDefinitionCacheKey(
+            int significantDrivingMinutes,
+            int minimumRestPeriodMinutes,
+            int restGeoLookbackMinutes,
+            int restGeoLookaheadMinutes,
+            int restGeoStationaryMaxMeters,
+            int restGeoMinorMovementMaxMeters
+    ) {
+    }
+
+    private record DerivedProjectionRuntimeExecutionMetrics(
+            boolean definitionCacheHit,
+            long definitionPreparationMs,
+            boolean runtimePoolHit,
+            long runtimeInitMs,
+            long deployMs,
+            long listenerRegistrationMs,
+            long runtimeResetMs,
+            long sendSupportGeoMs,
+            long sendVehicleUsageMs,
+            long sendActivityMs,
+            long destroyMs
+    ) {
+    }
+
+    private final class ReusableProjectionRuntimePool {
+
+        private final Deque<ReusableProjectionRuntime> idleRuntimes = new ArrayDeque<>();
+
+        private synchronized ReusableProjectionRuntime acquire(
+                PreparedProjectionDefinition preparedDefinition,
+                Map<String, Consumer<EventBean[]>> listeners
+        ) throws EPDeployException, EPCompileException {
+            ReusableProjectionRuntime runtime = idleRuntimes.pollFirst();
+            if (runtime != null) {
+                return runtime.reused();
+            }
+            return createReusableRuntime(preparedDefinition, listeners);
+        }
+
+        private long release(ReusableProjectionRuntime runtime) {
+            if (runtime == null) {
+                return 0L;
+            }
+            synchronized (this) {
+                if (idleRuntimes.size() < MAX_IDLE_RUNTIMES_PER_DEFINITION) {
+                    runtime.poolHit(false);
+                    idleRuntimes.addLast(runtime);
+                    return 0L;
+                }
+            }
+            return runtime.destroy();
+        }
+    }
+
+    private static final class ReusableProjectionRuntime {
+
+        private final EPRuntime runtime;
+        private final long runtimeInitMs;
+        private final long deployMs;
+        private volatile long listenerRegistrationMs;
+        private volatile List<EPCompiled> cleanupQueries = List.of();
+        private volatile boolean poolHit;
+        private ExecutionListeners currentExecutionListeners;
+
+        private ReusableProjectionRuntime(
+                EPRuntime runtime,
+                long runtimeInitMs,
+                long deployMs,
+                boolean poolHit
+        ) {
+            this.runtime = runtime;
+            this.runtimeInitMs = runtimeInitMs;
+            this.deployMs = deployMs;
+            this.poolHit = poolHit;
+        }
+
+        private synchronized ReusableProjectionRuntimeExecution execute(
+                Map<String, Consumer<EventBean[]>> listeners,
+                Consumer<DerivedProjectionEventSender> sender
+        ) {
+            currentExecutionListeners = new ExecutionListeners(listeners);
+            long runtimeResetStartedAtNanos = System.nanoTime();
+            for (EPCompiled cleanupQuery : cleanupQueries) {
+                runtime.getFireAndForgetService().executeQuery(cleanupQuery);
+            }
+            long runtimeResetMs = elapsedMillisStatic(runtimeResetStartedAtNanos);
+            try {
+                DerivedProjectionEventSender timedSender = new DerivedProjectionEventSender(runtime);
+                sender.accept(timedSender);
+                return new ReusableProjectionRuntimeExecution(
+                        runtimeResetMs,
+                        timedSender.sendSupportGeoMs(),
+                        timedSender.sendVehicleUsageMs(),
+                        timedSender.sendActivityMs()
+                );
+            } finally {
+                currentExecutionListeners = null;
+            }
+        }
+
+        private void onStatementEvents(String statementName, EventBean[] newData) {
+            ExecutionListeners listeners = currentExecutionListeners;
+            if (listeners == null || newData == null || statementName == null) {
+                return;
+            }
+            Consumer<EventBean[]> consumer = listeners.listeners().get(statementName);
+            if (consumer != null) {
+                consumer.accept(newData);
+            }
+        }
+
+        private long destroy() {
+            if (runtime == null || runtime.isDestroyed()) {
+                return 0L;
+            }
+            long destroyStartedAtNanos = System.nanoTime();
+            runtime.destroy();
+            return elapsedMillisStatic(destroyStartedAtNanos);
+        }
+
+        private ReusableProjectionRuntime reused() {
+            poolHit = true;
+            return this;
+        }
+
+        private void cleanupQueries(List<EPCompiled> value) {
+            cleanupQueries = value == null ? List.of() : value;
+        }
+
+        private void listenerRegistrationMs(long value) {
+            listenerRegistrationMs = value;
+        }
+
+        private void poolHit(boolean value) {
+            poolHit = value;
+        }
+
+        private boolean poolHit() {
+            return poolHit;
+        }
+
+        private long runtimeInitMs() {
+            return poolHit ? 0L : runtimeInitMs;
+        }
+
+        private long deployMs() {
+            return poolHit ? 0L : deployMs;
+        }
+
+        private long listenerRegistrationMs() {
+            return poolHit ? 0L : listenerRegistrationMs;
+        }
+    }
+
+    private record ExecutionListeners(
+            Map<String, Consumer<EventBean[]>> listeners
+    ) {
+        private ExecutionListeners {
+            listeners = listeners == null ? Map.of() : Map.copyOf(listeners);
+        }
+    }
+
+    private record ReusableProjectionRuntimeExecution(
+            long runtimeResetMs,
+            long sendSupportGeoMs,
+            long sendVehicleUsageMs,
+            long sendActivityMs
+    ) {
+    }
+
+    private static final class DerivedProjectionEventSender {
+
+        private final EPRuntime delegate;
+        private long sendSupportGeoMs;
+        private long sendVehicleUsageMs;
+        private long sendActivityMs;
+
+        private DerivedProjectionEventSender(EPRuntime delegate) {
+            this.delegate = delegate;
+        }
+
+        private void sendSupportGeoEvent(Map<String, Object> event) {
+            long startedAtNanos = System.nanoTime();
+            delegate.getEventService().sendEventMap(event, "TachographSupportGeoEvidenceInputEvent");
+            sendSupportGeoMs += elapsedMillisStatic(startedAtNanos);
+        }
+
+        private void sendVehicleUsageEvent(Map<String, Object> event) {
+            long startedAtNanos = System.nanoTime();
+            delegate.getEventService().sendEventMap(event, "TachographVehicleUsageIntervalInputEvent");
+            sendVehicleUsageMs += elapsedMillisStatic(startedAtNanos);
+        }
+
+        private void sendActivityEvent(Map<String, Object> event) {
+            long startedAtNanos = System.nanoTime();
+            delegate.getEventService().sendEventMap(event, "TachographActivityIntervalInputEvent");
+            sendActivityMs += elapsedMillisStatic(startedAtNanos);
+        }
+
+        private long sendSupportGeoMs() {
+            return sendSupportGeoMs;
+        }
+
+        private long sendVehicleUsageMs() {
+            return sendVehicleUsageMs;
+        }
+
+        private long sendActivityMs() {
+            return sendActivityMs;
+        }
+
+        private static long elapsedMillisStatic(long startedAtNanos) {
+            return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+        }
+    }
+
+    private static long elapsedMillisStatic(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/dto/UnifiedRuntimeTachographEsperScopeResultDto.java

@@ -1,93 +0,0 @@
-package at.procon.eventhub.processing.dto;
-
-import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
-import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
-import at.procon.eventhub.processing.model.UnifiedDiscoveredVehicleRef;
-import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-public record UnifiedRuntimeTachographEsperScopeResultDto(
-        UnifiedRuntimeProcessingRequest request,
-        int inputEventCount,
-        int selectedDriverCount,
-        int discoveredVehicleCount,
-        List<UnifiedDiscoveredVehicleRef> discoveredVehicles,
-        Map<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults,
-        Map<String, RuntimeDriverPartitionDebugDto> partitionDebugByDriver,
-        List<String> notes,
-        List<String> warnings
-) {
-    public UnifiedRuntimeTachographEsperScopeResultDto {
-        discoveredVehicles = discoveredVehicles == null ? List.of() : List.copyOf(discoveredVehicles);
-        driverResults = driverResults == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(driverResults));
-        partitionDebugByDriver = partitionDebugByDriver == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(partitionDebugByDriver));
-        notes = notes == null ? List.of() : List.copyOf(notes);
-        warnings = warnings == null ? List.of() : List.copyOf(warnings);
-    }
-
-    public static UnifiedRuntimeTachographEsperScopeResultDto fromDriverWorkingTime(
-            UnifiedRuntimeDriverWorkingTimeScopeResultDto result
-    ) {
-        if (result == null) {
-            return null;
-        }
-        return new UnifiedRuntimeTachographEsperScopeResultDto(
-                result.request(),
-                result.inputEventCount(),
-                result.selectedDriverCount(),
-                result.discoveredVehicleCount(),
-                result.discoveredVehicles(),
-                result.driverResults(),
-                result.partitionDebugByDriver(),
-                result.notes(),
-                result.warnings()
-        );
-    }
-
-    public static UnifiedRuntimeTachographEsperScopeResultDto fromGenericRuntimeEventProcessingResult(
-            RuntimeEventProcessingResultDto genericResult
-    ) {
-        if (genericResult == null) {
-            throw new IllegalArgumentException("genericResult must not be null");
-        }
-        LinkedHashMap<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults = new LinkedHashMap<>();
-        for (Map.Entry<String, RuntimeEventProcessingPartitionResultDto> entry : genericResult.partitionResults().entrySet()) {
-            Object value = entry.getValue().result();
-            if (value instanceof UnifiedRuntimeDerivedProjectionResultDto projectionResult) {
-                driverResults.put(entry.getKey(), projectionResult);
-            } else {
-                throw new IllegalArgumentException("Cannot convert generic partition result for key "
-                        + entry.getKey() + " to UnifiedRuntimeDerivedProjectionResultDto: "
-                        + (value == null ? "null" : value.getClass().getName()));
-            }
-        }
-        return new UnifiedRuntimeTachographEsperScopeResultDto(
-                genericResult.request(),
-                genericResult.inputEventCount(),
-                genericResult.selectedPartitionCount(),
-                genericResult.discoveredVehicleCount(),
-                genericResult.discoveredVehicles(),
-                driverResults,
-                extractPartitionDebug(genericResult),
-                genericResult.notes(),
-                genericResult.warnings()
-        );
-    }
-
-    private static Map<String, RuntimeDriverPartitionDebugDto> extractPartitionDebug(
-            RuntimeEventProcessingResultDto genericResult
-    ) {
-        LinkedHashMap<String, RuntimeDriverPartitionDebugDto> debugByDriver = new LinkedHashMap<>();
-        for (Map.Entry<String, RuntimeEventProcessingPartitionResultDto> entry : genericResult.partitionResults().entrySet()) {
-            Object debug = entry.getValue().metadata().get("partitionDebug");
-            if (debug instanceof RuntimeDriverPartitionDebugDto partitionDebug) {
-                debugByDriver.put(entry.getKey(), partitionDebug);
-            }
-        }
-        return debugByDriver;
-    }
-
-}

src/main/java/at/procon/eventhub/processing/eventprocessing/RuntimeEventProcessingService.java

@@ -3,13 +3,17 @@ package at.procon.eventhub.processing.eventprocessing;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingProfileDescriptorDto;
-import java.util.List;
 import at.procon.eventhub.processing.eventprocessing.profile.RuntimeEventProcessingProfileRegistry;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 @Service
 public class RuntimeEventProcessingService {
 
+    private static final Logger LOG = LoggerFactory.getLogger(RuntimeEventProcessingService.class);
+
     private final RuntimeEventProcessingProfileRegistry profileRegistry;
 
     public RuntimeEventProcessingService(RuntimeEventProcessingProfileRegistry profileRegistry) {
@@ -17,10 +21,34 @@ public class RuntimeEventProcessingService {
     }
 
     public RuntimeEventProcessingResultDto process(RuntimeEventProcessingApiRequest request) {
-        return profileRegistry.require(request.profileKey()).process(request);
+        long startedAtNanos = System.nanoTime();
+        try {
+            RuntimeEventProcessingResultDto result = profileRegistry.require(request.profileKey()).process(request);
+            LOG.info(
+                    "Runtime event processing profile {} completed in {} ms (partitions: {}, inputEvents: {}, warnings: {})",
+                    request.profileKey(),
+                    elapsedMillis(startedAtNanos),
+                    result.selectedPartitionCount(),
+                    result.inputEventCount(),
+                    result.warnings().size()
+            );
+            return result;
+        } catch (RuntimeException ex) {
+            LOG.error(
+                    "Runtime event processing profile {} failed after {} ms",
+                    request == null ? null : request.profileKey(),
+                    elapsedMillis(startedAtNanos),
+                    ex
+            );
+            throw ex;
+        }
     }
 
     public List<RuntimeEventProcessingProfileDescriptorDto> listProfiles() {
         return profileRegistry.profileDescriptors();
     }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/dto/RuntimeEventProcessingApiRequest.java

@@ -1,6 +1,7 @@
 package at.procon.eventhub.processing.eventprocessing.dto;
 
 import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -27,9 +28,9 @@ public record RuntimeEventProcessingApiRequest(
                 : Collections.unmodifiableMap(new LinkedHashMap<>(parameters));
     }
 
-    public static RuntimeEventProcessingApiRequest tachographDriverEsper(UnifiedRuntimeProcessingApiRequest scope) {
+    public static RuntimeEventProcessingApiRequest driverWorkingTime(UnifiedRuntimeProcessingApiRequest scope) {
         return new RuntimeEventProcessingApiRequest(
-                "tachograph-driver-esper-v1",
+                DriverWorkingTimeRuntimeProcessingPlan.PLAN_KEY,
                 scope,
                 new RuntimeEventPartitioningApiRequest(
                         at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy.DRIVER,
@@ -39,7 +40,7 @@ public record RuntimeEventProcessingApiRequest(
                         scope != null ? scope.includeAllDrivers() : null,
                         scope != null ? scope.vehicleKeys() : null,
                         scope != null ? scope.includeAllVehicles() : null,
-                        null,
+                        scope != null ? scope.expandVehicleEvents() : null,
                         scope != null ? scope.vehicleExpansionPaddingMinutes() : null,
                         null
                 ),

src/main/java/at/procon/eventhub/processing/eventprocessing/module/DriverWorkingTimeDerivedProjectionsModule.java

@@ -15,12 +15,19 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
 public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcessingModule {
 
+    private static final Logger LOG = LoggerFactory.getLogger(DriverWorkingTimeDerivedProjectionsModule.class);
+    private static final String DRIVER_PROCESSING_TOTAL_MS_METADATA_KEY = "driverProcessingTotalMs";
+    private static final String SLOWEST_DRIVER_KEY_METADATA_KEY = "slowestDriverKey";
+    private static final String SLOWEST_DRIVER_PROCESSING_MS_METADATA_KEY = "slowestDriverProcessingMs";
+
     private final DriverWorkingTimeProcessingCore workingTimeProcessingCore;
     private final RuntimeDriverWorkingTimeScopeProcessingService legacyScopeProcessingService;
 
@@ -60,20 +67,31 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
         if (legacyScopeProcessingService != null) {
             return executeLegacy(context);
         }
+        long startedAtNanos = System.nanoTime();
         UnifiedRuntimeEventBundle broadBundle = runtimeEventBundle(context);
         UnifiedRuntimeProcessingApiRequest scopeRequest = scopeRequest(context);
         Map<String, DriverWorkingTimePreparedInput> preparedInputs = preparedInputs(context);
 
         LinkedHashMap<String, UnifiedRuntimeDerivedProjectionResultDto> driverResults = new LinkedHashMap<>();
         List<String> warnings = new ArrayList<>();
+        long driverProcessingTotalMs = 0L;
+        String slowestDriverKey = null;
+        long slowestDriverProcessingMs = -1L;
         for (Map.Entry<String, DriverWorkingTimePreparedInput> entry : preparedInputs.entrySet()) {
             DriverWorkingTimePreparedInput preparedInput = entry.getValue();
+            long driverStartedAtNanos = System.nanoTime();
             DriverWorkingTimeProcessingResultDto projection =
                     workingTimeProcessingCore.process(preparedInput.processingInput())
                             .withIncludedIntervals(
                                     scopeRequest.includeActivityIntervalsOrDefault(),
                                     scopeRequest.includeDrivingIntervalsOrDefault()
                             );
+            long driverProcessingMs = elapsedMillis(driverStartedAtNanos);
+            driverProcessingTotalMs += driverProcessingMs;
+            if (driverProcessingMs > slowestDriverProcessingMs) {
+                slowestDriverProcessingMs = driverProcessingMs;
+                slowestDriverKey = preparedInput.driverKey();
+            }
             warnings.addAll(preparedInput.partition().warnings());
             UnifiedRuntimeProcessingRequest driverRequest = broadBundle.request().withDriverKey(preparedInput.driverKey());
             driverResults.put(preparedInput.driverKey(), new UnifiedRuntimeDerivedProjectionResultDto(
@@ -110,6 +128,18 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
         metadata.put("selectedDriverCount", result.selectedDriverCount());
         metadata.put("discoveredVehicleCount", result.discoveredVehicleCount());
         metadata.put("driverResultCount", result.driverResults().size());
+        metadata.put(DRIVER_PROCESSING_TOTAL_MS_METADATA_KEY, driverProcessingTotalMs);
+        metadata.put(SLOWEST_DRIVER_KEY_METADATA_KEY, slowestDriverKey);
+        metadata.put(SLOWEST_DRIVER_PROCESSING_MS_METADATA_KEY, Math.max(0L, slowestDriverProcessingMs));
+        LOG.info(
+                "Driving-derived projections processed {} drivers in {} ms (driverProcessingTotalMs: {}, slowestDriverKey: {}, slowestDriverProcessingMs: {}, warnings: {})",
+                result.driverResults().size(),
+                elapsedMillis(startedAtNanos),
+                driverProcessingTotalMs,
+                slowestDriverKey,
+                Math.max(0L, slowestDriverProcessingMs),
+                result.warnings().size()
+        );
         return new RuntimeProcessingModuleResult(
                 moduleKey(),
                 RuntimeProcessingModuleStatus.SUCCESS,
@@ -187,4 +217,8 @@ public class DriverWorkingTimeDerivedProjectionsModule implements RuntimeProcess
         }
         return fallback;
     }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/module/RuntimeEventAssemblyModule.java

@@ -39,6 +39,8 @@ public class RuntimeEventAssemblyModule implements RuntimeProcessingModule {
         UnifiedRuntimeProcessingApiRequest scopeRequest = scopeRequest(context);
         UnifiedRuntimeEventBundle bundle = eventAssemblyService.assembleDriverScopedEvents(scopeRequest.toRuntimeRequest());
         Map<String, Object> metadata = new LinkedHashMap<>();
+        metadata.put("expandVehicleEvents", scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents());
+        metadata.put("vehicleExpansionPaddingMinutes", scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes());
         metadata.put("driverSeedEventCount", bundle.driverSeedEvents().size());
         metadata.put("expandedVehicleEventCount", bundle.expandedVehicleEvents().size());
         metadata.put("mergedEventCount", bundle.mergedEvents().size());

src/main/java/at/procon/eventhub/processing/eventprocessing/module/RuntimeProcessingPipelineExecutor.java

@@ -4,11 +4,17 @@ import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecu
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 @Service
 public class RuntimeProcessingPipelineExecutor {
 
+    public static final String EXECUTION_DURATION_MS_METADATA_KEY = "executionDurationMs";
+
+    private static final Logger LOG = LoggerFactory.getLogger(RuntimeProcessingPipelineExecutor.class);
+
     private final RuntimeProcessingModuleRegistry moduleRegistry;
 
     public RuntimeProcessingPipelineExecutor(RuntimeProcessingModuleRegistry moduleRegistry) {
@@ -29,8 +35,17 @@ public class RuntimeProcessingPipelineExecutor {
                 continue;
             }
             RuntimeProcessingModule module = moduleRegistry.require(moduleKey.trim());
-            RuntimeProcessingModuleResult result = module.execute(context);
+            long startedAtNanos = System.nanoTime();
+            RuntimeProcessingModuleResult result;
+            try {
+                result = withExecutionDuration(module.execute(context), startedAtNanos);
+            } catch (RuntimeException ex) {
+                long durationMs = elapsedMillis(startedAtNanos);
+                LOG.error("Runtime processing module {} failed after {} ms", module.moduleKey(), durationMs, ex);
+                throw ex;
+            }
             results.put(module.moduleKey(), result);
+            logModuleCompletion(result);
             context = new RuntimeProcessingModuleContext(
                     request,
                     context.events(),
@@ -43,4 +58,45 @@ public class RuntimeProcessingPipelineExecutor {
         }
         return Map.copyOf(results);
     }
+
+    private RuntimeProcessingModuleResult withExecutionDuration(
+            RuntimeProcessingModuleResult result,
+            long startedAtNanos
+    ) {
+        long durationMs = elapsedMillis(startedAtNanos);
+        RuntimeProcessingModuleResult safeResult = result == null
+                ? new RuntimeProcessingModuleResult(
+                "UNKNOWN",
+                RuntimeProcessingModuleStatus.FAILED,
+                null,
+                Map.of(),
+                List.of("Runtime processing module returned null result.")
+        )
+                : result;
+        LinkedHashMap<String, Object> metadata = new LinkedHashMap<>(safeResult.metadata());
+        metadata.put(EXECUTION_DURATION_MS_METADATA_KEY, durationMs);
+        return new RuntimeProcessingModuleResult(
+                safeResult.moduleKey(),
+                safeResult.status(),
+                safeResult.output(),
+                metadata,
+                safeResult.warnings()
+        );
+    }
+
+    private void logModuleCompletion(RuntimeProcessingModuleResult result) {
+        Object durationValue = result.metadata().get(EXECUTION_DURATION_MS_METADATA_KEY);
+        long durationMs = durationValue instanceof Number number ? number.longValue() : -1L;
+        LOG.info(
+                "Runtime processing module {} completed with status {} in {} ms (warnings: {})",
+                result.moduleKey(),
+                result.status(),
+                durationMs,
+                result.warnings().size()
+        );
+    }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/module/VehicleEvidenceAttachmentModule.java

@@ -85,8 +85,16 @@ public class VehicleEvidenceAttachmentModule implements RuntimeProcessingModule
                     directDriverEvents,
                     broadBundle.mergedEvents(),
                     driverVehicleUsageIntervals,
-                    scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents(),
+                    booleanAttribute(
-                    scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes(),
+                            context,
+                            at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan.ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE,
+                            scopeRequest.expandVehicleEvents() == null || scopeRequest.expandVehicleEvents()
+                    ),
+                    integerAttribute(
+                            context,
+                            at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan.VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE,
+                            scopeRequest.vehicleExpansionPaddingMinutes() == null ? 0 : scopeRequest.vehicleExpansionPaddingMinutes()
+                    ),
                     includePartitionDebug
             );
             for (EventHubEventDto attachedEvent : attachmentResult.attachedVehicleEvidenceEvents()) {
@@ -291,4 +299,23 @@ public class VehicleEvidenceAttachmentModule implements RuntimeProcessingModule
         }
         return fallback;
     }
+
+    private int integerAttribute(RuntimeProcessingModuleContext context, String key, int fallback) {
+        Object value = context.attributes().get(key);
+        if (value instanceof Number number) {
+            return Math.max(0, number.intValue());
+        }
+        if (value == null) {
+            return Math.max(0, fallback);
+        }
+        String text = value.toString();
+        if (text.isBlank()) {
+            return Math.max(0, fallback);
+        }
+        try {
+            return Math.max(0, Integer.parseInt(text.trim()));
+        } catch (NumberFormatException ignored) {
+            return Math.max(0, fallback);
+        }
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/module/epl/RuntimeEplModuleExecutor.java

@@ -13,10 +13,15 @@ import com.espertech.esper.runtime.client.EPRuntimeProvider;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StreamUtils;
@@ -24,24 +29,63 @@ import org.springframework.util.StreamUtils;
 @Service
 public class RuntimeEplModuleExecutor {
 
+    public static final String DEFINITION_CACHE_HIT_METADATA_KEY = "eplDefinitionCacheHit";
+    public static final String DEFINITION_PREPARATION_MS_METADATA_KEY = "eplDefinitionPreparationMs";
+    public static final String RUNTIME_INIT_MS_METADATA_KEY = "eplRuntimeInitMs";
+    public static final String DEPLOY_MS_METADATA_KEY = "eplDeployMs";
+    public static final String LISTENER_REGISTRATION_MS_METADATA_KEY = "eplListenerRegistrationMs";
+    public static final String EVENT_SEND_MS_METADATA_KEY = "eplEventSendMs";
+    public static final String DESTROY_MS_METADATA_KEY = "eplDestroyMs";
+
+    private static final Logger LOG = LoggerFactory.getLogger(RuntimeEplModuleExecutor.class);
     private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
 
+    private final ConcurrentMap<PreparedDefinitionCacheKey, PreparedDefinition> preparedDefinitions = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, String> eplResourceContents = new ConcurrentHashMap<>();
+
     public RuntimeEplModuleExecutionResult execute(RuntimeEplModuleDefinition definition) {
         EPRuntime runtime = null;
+        Map<String, List<Map<String, Object>>> outputs = null;
+        LinkedHashMap<String, Object> metadata = null;
+        PreparedDefinition preparedDefinition = null;
+        boolean definitionCacheHit = false;
+        long definitionPreparationMs = 0L;
+        long runtimeInitMs = 0L;
+        long deployMs = 0L;
+        long listenerRegistrationMs = 0L;
+        long eventSendMs = 0L;
+        long destroyMs = 0L;
+        long executionStartedAtNanos = System.nanoTime();
         try {
-            Configuration configuration = new Configuration();
+            PreparedDefinitionCacheKey cacheKey = PreparedDefinitionCacheKey.from(definition);
-            definition.eventTypes().forEach((eventTypeName, eventTypeDefinition) ->
+            preparedDefinition = preparedDefinitions.get(cacheKey);
-                    configuration.getCommon().addEventType(eventTypeName, eventTypeDefinition));
+            definitionCacheHit = preparedDefinition != null;
+            if (!definitionCacheHit) {
+                long definitionPreparationStartedAtNanos = System.nanoTime();
+                PreparedDefinition candidate = prepareDefinition(definition);
+                PreparedDefinition existing = preparedDefinitions.putIfAbsent(cacheKey, candidate);
+                preparedDefinition = existing == null ? candidate : existing;
+                definitionCacheHit = existing != null;
+                definitionPreparationMs = definitionCacheHit
+                        ? 0L
+                        : elapsedMillis(definitionPreparationStartedAtNanos);
+            }
+
+            long runtimeInitStartedAtNanos = System.nanoTime();
+            Configuration configuration = createConfiguration(preparedDefinition.eventTypes());
 
             String runtimeUri = "eventhub-runtime-epl-module-"
                     + definition.moduleKey().replaceAll("[^A-Za-z0-9_-]", "-")
                     + "-" + RUNTIME_COUNTER.incrementAndGet();
             runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);
+            runtimeInitMs = elapsedMillis(runtimeInitStartedAtNanos);
 
-            EPCompiled compiled = EPCompilerProvider.getCompiler().compile(renderEpl(definition.eplResources()), new CompilerArguments(configuration));
+            long deployStartedAtNanos = System.nanoTime();
-            EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);
+            EPDeployment deployment = runtime.getDeploymentService().deploy(preparedDefinition.compiled());
+            deployMs = elapsedMillis(deployStartedAtNanos);
 
-            Map<String, List<Map<String, Object>>> outputs = new LinkedHashMap<>();
+            long listenerRegistrationStartedAtNanos = System.nanoTime();
+            outputs = new LinkedHashMap<>();
             for (String statementName : definition.outputStatementNames()) {
                 outputs.put(statementName, new ArrayList<>());
                 var statement = runtime.getDeploymentService().getStatement(deployment.getDeploymentId(), statementName);
@@ -49,18 +93,28 @@ public class RuntimeEplModuleExecutor {
                     throw new IllegalStateException("EPL module " + definition.moduleKey()
                             + " did not deploy expected statement '" + statementName + "'.");
                 }
-                statement.addListener((newData, oldData, stmt, rt) -> collect(newData, outputs.get(statementName)));
+                List<Map<String, Object>> statementOutputs = outputs.get(statementName);
+                statement.addListener((newData, oldData, stmt, rt) -> collect(newData, statementOutputs));
             }
+            listenerRegistrationMs = elapsedMillis(listenerRegistrationStartedAtNanos);
 
+            long eventSendStartedAtNanos = System.nanoTime();
             for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
                 for (Map<String, Object> event : inputStream.events()) {
                     runtime.getEventService().sendEventMap(event, inputStream.eventTypeName());
                 }
             }
+            eventSendMs = elapsedMillis(eventSendStartedAtNanos);
 
-            Map<String, Object> metadata = new LinkedHashMap<>();
+            metadata = new LinkedHashMap<>();
             metadata.put("engine", "EPL");
             metadata.put("eplResources", definition.eplResources());
+            metadata.put(DEFINITION_CACHE_HIT_METADATA_KEY, definitionCacheHit);
+            metadata.put(DEFINITION_PREPARATION_MS_METADATA_KEY, definitionPreparationMs);
+            metadata.put(RUNTIME_INIT_MS_METADATA_KEY, runtimeInitMs);
+            metadata.put(DEPLOY_MS_METADATA_KEY, deployMs);
+            metadata.put(LISTENER_REGISTRATION_MS_METADATA_KEY, listenerRegistrationMs);
+            metadata.put(EVENT_SEND_MS_METADATA_KEY, eventSendMs);
             Map<String, Integer> inputCounts = new LinkedHashMap<>();
             for (RuntimeEplInputEventStream inputStream : definition.inputStreams()) {
                 inputCounts.merge(inputStream.eventTypeName(), inputStream.events().size(), Integer::sum);
@@ -69,14 +123,48 @@ public class RuntimeEplModuleExecutor {
             Map<String, Integer> outputCounts = new LinkedHashMap<>();
             outputs.forEach((statement, events) -> outputCounts.put(statement, events.size()));
             metadata.put("outputCounts", outputCounts);
-            return new RuntimeEplModuleExecutionResult(outputs, metadata);
         } catch (EPCompileException | EPDeployException e) {
             throw new IllegalStateException("Cannot compile/deploy runtime EPL module " + definition.moduleKey(), e);
         } finally {
             if (runtime != null) {
+                long destroyStartedAtNanos = System.nanoTime();
                 runtime.destroy();
+                destroyMs = elapsedMillis(destroyStartedAtNanos);
+            }
+            if (metadata != null) {
+                metadata.put(DESTROY_MS_METADATA_KEY, destroyMs);
+                LOG.info(
+                        "Runtime EPL module {} completed in {} ms (definitionCacheHit: {}, definitionPreparationMs: {}, runtimeInitMs: {}, deployMs: {}, listenerRegistrationMs: {}, eventSendMs: {}, destroyMs: {})",
+                        definition.moduleKey(),
+                        elapsedMillis(executionStartedAtNanos),
+                        definitionCacheHit,
+                        definitionPreparationMs,
+                        runtimeInitMs,
+                        deployMs,
+                        listenerRegistrationMs,
+                        eventSendMs,
+                        destroyMs
+                );
             }
         }
+        return new RuntimeEplModuleExecutionResult(
+                outputs == null ? Map.of() : outputs,
+                metadata == null ? Map.of() : metadata
+        );
+    }
+
+    private PreparedDefinition prepareDefinition(RuntimeEplModuleDefinition definition) throws EPCompileException {
+        Configuration configuration = createConfiguration(definition.eventTypes());
+        String epl = renderEpl(definition.eplResources());
+        EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, new CompilerArguments(configuration));
+        return new PreparedDefinition(definition.eventTypes(), compiled);
+    }
+
+    private Configuration createConfiguration(Map<String, Map<String, Object>> eventTypes) {
+        Configuration configuration = new Configuration();
+        eventTypes.forEach((eventTypeName, eventTypeDefinition) ->
+                configuration.getCommon().addEventType(eventTypeName, eventTypeDefinition));
+        return configuration;
     }
 
     private String renderEpl(List<String> resources) {
@@ -88,7 +176,7 @@ public class RuntimeEplModuleExecutor {
             if (!builder.isEmpty()) {
                 builder.append("\n\n");
             }
-            builder.append(loadResource(resource.trim()));
+            builder.append(eplResourceContents.computeIfAbsent(resource.trim(), RuntimeEplModuleExecutor::loadResource));
         }
         return builder.toString();
     }
@@ -117,4 +205,40 @@ public class RuntimeEplModuleExecutor {
             target.add(values);
         }
     }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
+
+    private record PreparedDefinition(
+            Map<String, Map<String, Object>> eventTypes,
+            EPCompiled compiled
+    ) {
+        private PreparedDefinition {
+            eventTypes = eventTypes == null ? Map.of() : immutableNestedMap(eventTypes);
+        }
+    }
+
+    private record PreparedDefinitionCacheKey(
+            String moduleKey,
+            Map<String, Map<String, Object>> eventTypes,
+            List<String> eplResources
+    ) {
+        private static PreparedDefinitionCacheKey from(RuntimeEplModuleDefinition definition) {
+            return new PreparedDefinitionCacheKey(
+                    definition.moduleKey(),
+                    immutableNestedMap(definition.eventTypes()),
+                    List.copyOf(definition.eplResources())
+            );
+        }
+    }
+
+    private static Map<String, Map<String, Object>> immutableNestedMap(Map<String, Map<String, Object>> source) {
+        LinkedHashMap<String, Map<String, Object>> copy = new LinkedHashMap<>();
+        source.forEach((key, value) -> copy.put(
+                key,
+                value == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(value))
+        ));
+        return Collections.unmodifiableMap(copy);
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/plan/DriverWorkingTimeRuntimeProcessingPlan.java

@@ -30,7 +30,8 @@ import org.springframework.stereotype.Component;
 public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessingPlan {
 
     public static final String PLAN_KEY = "driver-working-time-v1";
-    public static final String LEGACY_PROFILE_ALIAS = "tachograph-driver-esper-v1";
+    public static final String ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE = "attachVehicleOnlyEvents";
+    public static final String VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE = "vehicleEvidencePaddingMinutes";
 
     private final RuntimeProcessingPipelineExecutor pipelineExecutor;
     private final boolean includeRuntimeEventAssemblyModule;
@@ -76,7 +77,7 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
 
     @Override
     public Set<String> aliases() {
-        return Set.of(LEGACY_PROFILE_ALIAS);
+        return Set.of();
     }
 
     @Override
@@ -181,6 +182,16 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
                 "includePartitionDebug",
                 request.partitioning() != null && request.partitioning().includeDebugOrDefault()
         );
+        boolean attachVehicleOnlyEvents = resolveAttachVehicleOnlyEvents(
+                request.sourceSelection(),
+                request.partitioning(),
+                request.parameters()
+        );
+        int vehicleEvidencePaddingMinutes = resolveVehicleEvidencePaddingMinutes(
+                request.sourceSelection(),
+                request.partitioning(),
+                request.parameters()
+        );
         UnifiedRuntimeProcessingApiRequest scopeRequest = applyExecutionRequest(
                 request.sourceSelection(),
                 request.partitioning(),
@@ -190,6 +201,8 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
         Map<String, Object> attributes = new LinkedHashMap<>();
         attributes.put("runtimeScopeApiRequest", scopeRequest);
         attributes.put("includePartitionDebug", includePartitionDebug);
+        attributes.put(ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE, attachVehicleOnlyEvents);
+        attributes.put(VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE, vehicleEvidencePaddingMinutes);
         RuntimeProcessingModuleContext initialContext = new RuntimeProcessingModuleContext(
                 request,
                 List.of(),
@@ -374,12 +387,6 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
                 sourceSelection.includeActivityIntervalsOrDefault());
         boolean includeDrivingIntervals = booleanParameter(parameters, "includeDrivingIntervals",
                 sourceSelection.includeDrivingIntervalsOrDefault());
-        boolean attachVehicleOnlyEvents = booleanParameter(parameters, "attachVehicleOnlyEvents",
-                partitioning == null ? sourceSelection.expandVehicleEvents() == null || sourceSelection.expandVehicleEvents() : partitioning.attachVehicleEvidenceOrDefault());
-        Integer vehicleEvidencePaddingMinutes = nonNegativeIntegerParameter(parameters, "vehicleEvidencePaddingMinutes",
-                partitioning == null
-                        ? sourceSelection.vehicleExpansionPaddingMinutes()
-                        : partitioning.vehicleEvidencePaddingMinutesOrDefault(sourceSelection.vehicleExpansionPaddingMinutes() == null ? 0 : sourceSelection.vehicleExpansionPaddingMinutes()));
 
         return new UnifiedRuntimeProcessingApiRequest(
                 sourceSelection.sessionId(),
@@ -399,8 +406,8 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
                 sourceSelection.driverCardNumber(),
                 sourceSelection.occurredFrom(),
                 sourceSelection.occurredTo(),
-                attachVehicleOnlyEvents,
+                sourceSelection.expandVehicleEvents(),
-                vehicleEvidencePaddingMinutes,
+                sourceSelection.vehicleExpansionPaddingMinutes(),
                 sourceSelection.includeIntersectingIntervals(),
                 significantDrivingMinutes,
                 minimumRestPeriodMinutes,
@@ -410,6 +417,37 @@ public class DriverWorkingTimeRuntimeProcessingPlan implements RuntimeProcessing
         );
     }
 
+    private boolean resolveAttachVehicleOnlyEvents(
+            UnifiedRuntimeProcessingApiRequest sourceSelection,
+            RuntimeEventPartitioningApiRequest partitioning,
+            Map<String, Object> parameters
+    ) {
+        return booleanParameter(
+                parameters,
+                "attachVehicleOnlyEvents",
+                partitioning == null
+                        ? sourceSelection.expandVehicleEvents() == null || sourceSelection.expandVehicleEvents()
+                        : partitioning.attachVehicleEvidenceOrDefault()
+        );
+    }
+
+    private int resolveVehicleEvidencePaddingMinutes(
+            UnifiedRuntimeProcessingApiRequest sourceSelection,
+            RuntimeEventPartitioningApiRequest partitioning,
+            Map<String, Object> parameters
+    ) {
+        Integer resolved = nonNegativeIntegerParameter(
+                parameters,
+                "vehicleEvidencePaddingMinutes",
+                partitioning == null
+                        ? sourceSelection.vehicleExpansionPaddingMinutes()
+                        : partitioning.vehicleEvidencePaddingMinutesOrDefault(
+                                sourceSelection.vehicleExpansionPaddingMinutes() == null ? 0 : sourceSelection.vehicleExpansionPaddingMinutes()
+                        )
+        );
+        return resolved == null ? 0 : resolved;
+    }
+
     private List<String> requestedOrDefaultModules(List<String> requestedModules) {
         if (requestedModules != null && !requestedModules.isEmpty()) {
             LinkedHashMap<String, String> requested = new LinkedHashMap<>();

src/main/java/at/procon/eventhub/processing/eventprocessing/plan/RuntimeProcessingExecutionApiRequest.java

@@ -48,7 +48,7 @@ public record RuntimeProcessingExecutionApiRequest(
                         sourceSelection != null ? sourceSelection.includeAllDrivers() : null,
                         sourceSelection != null ? sourceSelection.vehicleKeys() : null,
                         sourceSelection != null ? sourceSelection.includeAllVehicles() : null,
-                        null,
+                        sourceSelection != null ? sourceSelection.expandVehicleEvents() : null,
                         sourceSelection != null ? sourceSelection.vehicleExpansionPaddingMinutes() : null,
                         null
                 ),

src/main/java/at/procon/eventhub/processing/eventprocessing/plan/RuntimeProcessingExecutionService.java

@@ -1,11 +1,17 @@
 package at.procon.eventhub.processing.eventprocessing.plan;
 
 import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import at.procon.eventhub.processing.eventprocessing.module.RuntimeProcessingModuleResult;
+import at.procon.eventhub.processing.eventprocessing.module.RuntimeProcessingPipelineExecutor;
 import org.springframework.stereotype.Service;
 
 @Service
 public class RuntimeProcessingExecutionService {
 
+    private static final Logger LOG = LoggerFactory.getLogger(RuntimeProcessingExecutionService.class);
+
     private final RuntimeProcessingPlanRegistry planRegistry;
 
     public RuntimeProcessingExecutionService(RuntimeProcessingPlanRegistry planRegistry) {
@@ -13,12 +19,42 @@ public class RuntimeProcessingExecutionService {
     }
 
     public RuntimeProcessingExecutionResultDto execute(RuntimeProcessingExecutionApiRequest request) {
+        long startedAtNanos = System.nanoTime();
         RuntimeProcessingPlan plan = planRegistry.require(request.processingPlanKey());
-        plan.validatePartitioning(request.partitioning());
+        try {
-        return plan.execute(request);
+            plan.validatePartitioning(request.partitioning());
+            RuntimeProcessingExecutionResultDto result = plan.execute(request);
+            RuntimeProcessingModuleResult slowestModule = result.moduleResults().values().stream()
+                    .filter(moduleResult -> moduleResult.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY) instanceof Number)
+                    .max(java.util.Comparator.comparingLong(moduleResult ->
+                            ((Number) moduleResult.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue()))
+                    .orElse(null);
+            LOG.info(
+                    "Runtime processing plan {} completed in {} ms (modules: {}, slowestModule: {}, slowestModuleMs: {}, warnings: {})",
+                    request.processingPlanKey(),
+                    elapsedMillis(startedAtNanos),
+                    result.executedModules().size(),
+                    slowestModule == null ? null : slowestModule.moduleKey(),
+                    slowestModule == null ? null : slowestModule.metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY),
+                    result.warnings().size()
+            );
+            return result;
+        } catch (RuntimeException ex) {
+            LOG.error(
+                    "Runtime processing plan {} failed after {} ms",
+                    request == null ? null : request.processingPlanKey(),
+                    elapsedMillis(startedAtNanos),
+                    ex
+            );
+            throw ex;
+        }
     }
 
     public List<RuntimeProcessingPlanDescriptorDto> listPlans() {
         return planRegistry.planDescriptors();
     }
+
+    private long elapsedMillis(long startedAtNanos) {
+        return Math.max(0L, (System.nanoTime() - startedAtNanos) / 1_000_000L);
+    }
 }

src/main/java/at/procon/eventhub/processing/eventprocessing/profile/DriverWorkingTimeRuntimeEventProcessingProfile.java

@@ -1,38 +1,28 @@
-package at.procon.eventhub.tachographfilesession.processing.profile;
+package at.procon.eventhub.processing.eventprocessing.profile;
 
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
 import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
+import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
 import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
 import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionResultDto;
-import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
-import at.procon.eventhub.processing.eventprocessing.profile.RuntimeEventProcessingProfile;
-import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
 import java.util.List;
 import java.util.Set;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
-public class TachographDriverEsperRuntimeEventProcessingProfile implements RuntimeEventProcessingProfile {
+public class DriverWorkingTimeRuntimeEventProcessingProfile implements RuntimeEventProcessingProfile {
 
-    /**
+    public static final String PROFILE_KEY = DriverWorkingTimeRuntimeProcessingPlan.PLAN_KEY;
-     * Legacy compatibility key. New clients should use processingPlanKey=driver-working-time-v1
-     * via /api/eventhub/runtime-processing/executions.
-     */
-    public static final String PROFILE_KEY = DriverWorkingTimeRuntimeProcessingPlan.LEGACY_PROFILE_ALIAS;
 
     private final DriverWorkingTimeRuntimeProcessingPlan plan;
 
     @Autowired
-    public TachographDriverEsperRuntimeEventProcessingProfile(DriverWorkingTimeRuntimeProcessingPlan plan) {
+    public DriverWorkingTimeRuntimeEventProcessingProfile(DriverWorkingTimeRuntimeProcessingPlan plan) {
         this.plan = plan;
     }
 
-    public TachographDriverEsperRuntimeEventProcessingProfile(RuntimeDriverWorkingTimeScopeProcessingService scopeProcessingService) {
-        this(new DriverWorkingTimeRuntimeProcessingPlan(scopeProcessingService));
-    }
-
     @Override
     public String profileKey() {
         return PROFILE_KEY;
@@ -45,14 +35,13 @@ public class TachographDriverEsperRuntimeEventProcessingProfile implements Runti
 
     @Override
     public String displayName() {
-        return "Tachograph Driver Esper Processing";
+        return plan.displayName();
     }
 
     @Override
     public String description() {
-        return "Compatibility adapter for legacy profileKey=" + PROFILE_KEY
+        return "Compatibility runtime-event-processing adapter for processingPlanKey="
-                + ". New clients should use processingPlanKey=" + plan.processingPlanKey() + ". "
+                + plan.processingPlanKey() + ". " + plan.description();
-                + plan.description();
     }
 
     @Override

src/main/java/at/procon/eventhub/processing/service/UnifiedRuntimeEventAssemblyService.java

@@ -14,11 +14,15 @@ import java.util.Comparator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 @Service
 public class UnifiedRuntimeEventAssemblyService {
 
+    private static final Logger LOG = LoggerFactory.getLogger(UnifiedRuntimeEventAssemblyService.class);
+
     private final List<RuntimeDriverEventLoader> driverEventLoaders;
     private final List<RuntimeVehicleEventLoader> vehicleEventLoaders;
 
@@ -34,10 +38,13 @@ public class UnifiedRuntimeEventAssemblyService {
         List<UnifiedRuntimeSourceInput> sourceInputs = request.normalizedSourceInputs();
         List<EventHubEventDto> driverSeedEvents = loadDriverSeedEvents(request);
         List<UnifiedDiscoveredVehicleRef> discoveredVehicles = discoverVehicles(driverSeedEvents);
-        List<EventHubEventDto> expandedVehicleEvents = request.expandVehicleEvents()
+        boolean expandVehicleEvents = request.expandVehicleEvents();
+        List<EventHubEventDto> expandedVehicleEvents = expandVehicleEvents
                 ? loadExpandedVehicleEvents(request, discoveredVehicles)
                 : List.of();
-        List<EventHubEventDto> mergedEvents = deduplicateAndSort(driverSeedEvents, expandedVehicleEvents);
+        List<EventHubEventDto> mergedEvents = expandVehicleEvents
+                ? deduplicateAndSort(driverSeedEvents, expandedVehicleEvents)
+                : driverSeedEvents;
 
         List<String> notes = new ArrayList<>();
         boolean includesEventHub = sourceInputs.stream()
@@ -63,12 +70,21 @@ public class UnifiedRuntimeEventAssemblyService {
                 notes.add("Tachograph file-session events were loaded from session " + sourceInput.sessionId() + ".");
             }
         }
-        if (request.expandVehicleEvents()) {
+        if (expandVehicleEvents) {
             notes.add("Vehicle expansion loaded additional events for vehicles discovered in the driver seed set.");
             notes.add("Vehicle expansion padding minutes: " + request.vehicleExpansionPaddingMinutes() + ".");
         } else {
             notes.add("Vehicle expansion was disabled for this runtime request.");
         }
+        LOG.info(
+                "Runtime event assembly completed (expandVehicleEvents: {}, sourceInputs: {}, driverSeedEvents: {}, discoveredVehicles: {}, expandedVehicleEvents: {}, mergedEvents: {})",
+                expandVehicleEvents,
+                sourceInputs.size(),
+                driverSeedEvents.size(),
+                discoveredVehicles.size(),
+                expandedVehicleEvents.size(),
+                mergedEvents.size()
+        );
         return new UnifiedRuntimeEventBundle(
                 request,
                 driverSeedEvents,

src/main/java/at/procon/eventhub/tachographfilesession/processing/plan/TachographDriverWorkingTimeRuntimeProcessingPlan.java

@@ -1,17 +0,0 @@
-package at.procon.eventhub.tachographfilesession.processing.plan;
-
-import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
-import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
-
-/**
- * @deprecated Compatibility adapter. Use {@link DriverWorkingTimeRuntimeProcessingPlan}.
- */
-@Deprecated(forRemoval = false)
-public class TachographDriverWorkingTimeRuntimeProcessingPlan extends DriverWorkingTimeRuntimeProcessingPlan {
-
-    public TachographDriverWorkingTimeRuntimeProcessingPlan(
-            RuntimeDriverWorkingTimeScopeProcessingService driverWorkingTimeScopeProcessingService
-    ) {
-        super(driverWorkingTimeScopeProcessingService);
-    }
-}

src/main/java/at/procon/eventhub/tachographfilesession/processing/service/TachographFileSessionUnifiedDriverEventSource.java

@@ -9,7 +9,7 @@ import at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
 import at.procon.eventhub.tachographfilesession.model.TachographFileSession;
 import at.procon.eventhub.tachographfilesession.model.TachographTimelineEventBundle;
 import at.procon.eventhub.tachographfilesession.service.DriverNotFoundInSessionException;
-import at.procon.eventhub.tachographfilesession.service.DriverTimelineEventBuilder;
+import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionNotFoundException;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
 import java.util.List;
@@ -19,14 +19,14 @@ import org.springframework.stereotype.Component;
 public class TachographFileSessionUnifiedDriverEventSource implements UnifiedDriverEventSource {
 
     private final TachographFileSessionRepository repository;
-    private final DriverTimelineEventBuilder eventBuilder;
+    private final RawSourceDriverTimelineEventBuilder rawSourceEventBuilder;
 
     public TachographFileSessionUnifiedDriverEventSource(
             TachographFileSessionRepository repository,
-            DriverTimelineEventBuilder eventBuilder
+            RawSourceDriverTimelineEventBuilder rawSourceEventBuilder
     ) {
         this.repository = repository;
-        this.eventBuilder = eventBuilder;
+        this.rawSourceEventBuilder = rawSourceEventBuilder;
     }
 
     @Override
@@ -56,7 +56,7 @@ public class TachographFileSessionUnifiedDriverEventSource implements UnifiedDri
             DriverExtractionSession driver,
             UnifiedDriverEventsRequest request
     ) {
-        TachographTimelineEventBundle bundle = eventBuilder.buildEventBundle(session, driver);
+        TachographTimelineEventBundle bundle = rawSourceEventBuilder.buildRawEventBundle(session, driver);
         return TachographTimelineEventBundle.fromRuntimeBundle(
                 RuntimeIntervalEventWindowSelector.filterBundle(
                         bundle == null ? null : bundle.toRuntimeBundle(),

src/main/java/at/procon/eventhub/tachographfilesession/processing/service/TachographFileSessionUnifiedVehicleEventSource.java

@@ -10,7 +10,7 @@ import at.procon.eventhub.reference.TachographNationRegistry;
 import at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
 import at.procon.eventhub.tachographfilesession.model.TachographFileSession;
 import at.procon.eventhub.tachographfilesession.model.TachographTimelineEventBundle;
-import at.procon.eventhub.tachographfilesession.service.DriverTimelineEventBuilder;
+import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionNotFoundException;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
 import java.util.List;
@@ -20,14 +20,14 @@ import org.springframework.stereotype.Component;
 public class TachographFileSessionUnifiedVehicleEventSource implements UnifiedVehicleEventSource {
 
     private final TachographFileSessionRepository repository;
-    private final DriverTimelineEventBuilder eventBuilder;
+    private final RawSourceDriverTimelineEventBuilder rawSourceEventBuilder;
 
     public TachographFileSessionUnifiedVehicleEventSource(
             TachographFileSessionRepository repository,
-            DriverTimelineEventBuilder eventBuilder
+            RawSourceDriverTimelineEventBuilder rawSourceEventBuilder
     ) {
         this.repository = repository;
-        this.eventBuilder = eventBuilder;
+        this.rawSourceEventBuilder = rawSourceEventBuilder;
     }
 
     @Override
@@ -52,7 +52,7 @@ public class TachographFileSessionUnifiedVehicleEventSource implements UnifiedVe
             DriverExtractionSession driver,
             UnifiedVehicleEventsRequest request
     ) {
-        TachographTimelineEventBundle bundle = eventBuilder.buildEventBundle(session, driver);
+        TachographTimelineEventBundle bundle = rawSourceEventBuilder.buildRawEventBundle(session, driver);
         return TachographTimelineEventBundle.fromRuntimeBundle(
                 RuntimeIntervalEventWindowSelector.filterBundle(
                         bundle == null ? null : bundle.toRuntimeBundle(),

src/main/java/at/procon/eventhub/tachographfilesession/processing/service/UnifiedRuntimeTachographEsperScopeProcessingService.java

@@ -1,49 +0,0 @@
-package at.procon.eventhub.tachographfilesession.processing.service;
-
-import at.procon.eventhub.processing.dto.UnifiedRuntimeDriverWorkingTimeScopeResultDto;
-import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
-import at.procon.eventhub.processing.dto.UnifiedRuntimeTachographEsperScopeResultDto;
-import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
-
-/**
- * @deprecated Compatibility adapter. Use {@link RuntimeDriverWorkingTimeScopeProcessingService};
- * tachograph files/databases are only one possible source for the common driver working-time runtime plan.
- */
-@Deprecated(forRemoval = false)
-public class UnifiedRuntimeTachographEsperScopeProcessingService {
-
-    private final RuntimeDriverWorkingTimeScopeProcessingService delegate;
-
-    public UnifiedRuntimeTachographEsperScopeProcessingService(
-            RuntimeDriverWorkingTimeScopeProcessingService delegate
-    ) {
-        this.delegate = delegate;
-    }
-
-    public UnifiedRuntimeTachographEsperScopeResultDto processScope(UnifiedRuntimeProcessingApiRequest apiRequest) {
-        return toLegacy(delegate.processScope(apiRequest));
-    }
-
-    public UnifiedRuntimeTachographEsperScopeResultDto processScope(
-            UnifiedRuntimeProcessingApiRequest apiRequest,
-            boolean includePartitionDebug
-    ) {
-        return toLegacy(delegate.processScope(apiRequest, includePartitionDebug));
-    }
-
-    private UnifiedRuntimeTachographEsperScopeResultDto toLegacy(
-            UnifiedRuntimeDriverWorkingTimeScopeResultDto result
-    ) {
-        return new UnifiedRuntimeTachographEsperScopeResultDto(
-                result.request(),
-                result.inputEventCount(),
-                result.selectedDriverCount(),
-                result.discoveredVehicleCount(),
-                result.discoveredVehicles(),
-                result.driverResults(),
-                result.partitionDebugByDriver(),
-                result.notes(),
-                result.warnings()
-        );
-    }
-}

src/main/java/at/procon/eventhub/tachographfilesession/processing/validation/RuntimeTachographParityValidationService.java

@@ -9,7 +9,7 @@ import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingA
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
 import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
-import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
+import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
 import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
 import at.procon.eventhub.processing.model.UnifiedRuntimeEventBackend;
 import at.procon.eventhub.tachographfilesession.dto.TachographEsperDriverProcessingResultDto;
@@ -85,13 +85,13 @@ public class RuntimeTachographParityValidationService {
                 : "DIFFERENT";
         List<String> notes = new ArrayList<>(runtimeResult.notes());
         notes.add("Validation compares the legacy tachograph file-session esper-events path with the generic runtime event-processing profile '"
-                + TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY + "'.");
+                + DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY + "'.");
         if (sessionIds.size() > 1) {
             notes.add("For multiple sessions, file-session reference counts are summed per driver. Runtime processing may intentionally merge or deduplicate intervals across session boundaries, so category differences need domain review.");
         }
 
         return new RuntimeTachographParityValidationResultDto(
-                TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                 status,
                 sessionIds,
                 driverResults.size(),
@@ -207,7 +207,7 @@ public class RuntimeTachographParityValidationService {
         parameters.put("vehicleEvidencePaddingMinutes", request.vehicleExpansionPaddingMinutesOrDefault());
         parameters.put("includePartitionDebug", request.includeDebugOrDefault());
         return new RuntimeEventProcessingApiRequest(
-                TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                 scope,
                 partitioning,
                 parameters

src/main/resources/esper/driver-working-time-derived-projections.epl

@@ -288,7 +288,7 @@ create schema DailyWeeklyRestCandidateCoverageEmittedKey(
     endedAtEpochSecond long
 );
 
-create context PerDriver partition by driverKey from TachographVehicleUsageIntervalInputEvent;
+@public create context PerDriver partition by driverKey from TachographVehicleUsageIntervalInputEvent;
 
 create schema VuCardAbsentInterval(
     sessionId java.util.UUID,
@@ -438,27 +438,27 @@ create schema PotentialInVehicleTripInterval(
     lastNextDrivingSourceIntervalId string
 );
 
-create window PreviousRestCandidateCoverageInterval#unique(driverKey) as DailyWeeklyRestCandidateCoverageInterval;
+@public create window PreviousRestCandidateCoverageInterval#unique(driverKey) as DailyWeeklyRestCandidateCoverageInterval;
 
-create window OpenPotentialInVehicleTripState#unique(driverKey) as PotentialInVehicleTripState;
+@public create window OpenPotentialInVehicleTripState#unique(driverKey) as PotentialInVehicleTripState;
 
-create window SupportGeoEvidenceWindow#keepall as SupportGeoEvidence;
+@public create window SupportGeoEvidenceWindow#keepall as SupportGeoEvidence;
 
-create window DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateBeginGeoEvidenceCandidate;
+@public create window DailyWeeklyRestCandidateBeginGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateBeginGeoEvidenceCandidate;
-create window DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceBestScore;
+@public create window DailyWeeklyRestCandidateBeginGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceBestScore;
-create window DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceResolved;
+@public create window DailyWeeklyRestCandidateBeginGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginGeoEvidenceResolved;
 
-create window DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateEndGeoEvidenceCandidate;
+@public create window DailyWeeklyRestCandidateEndGeoEvidenceCandidateWindow#keepall as DailyWeeklyRestCandidateEndGeoEvidenceCandidate;
-create window DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceBestScore;
+@public create window DailyWeeklyRestCandidateEndGeoEvidenceBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceBestScore;
-create window DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceResolved;
+@public create window DailyWeeklyRestCandidateEndGeoEvidenceResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndGeoEvidenceResolved;
-create window DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateBeginBoundaryOdometerCandidate;
+@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateBeginBoundaryOdometerCandidate;
-create window DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerBestScore;
+@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerBestScore;
-create window DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerResolved;
+@public create window DailyWeeklyRestCandidateBeginBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateBeginBoundaryOdometerResolved;
-create window DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateEndBoundaryOdometerCandidate;
+@public create window DailyWeeklyRestCandidateEndBoundaryOdometerCandidateWindow#keepall as DailyWeeklyRestCandidateEndBoundaryOdometerCandidate;
-create window DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerBestScore;
+@public create window DailyWeeklyRestCandidateEndBoundaryOdometerBestScoreWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerBestScore;
-create window DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerResolved;
+@public create window DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateEndBoundaryOdometerResolved;
-create window DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageCardResolvedInterval;
+@public create window DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageCardResolvedInterval;
-create window DailyWeeklyRestCandidateCoverageEmittedKeyWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageEmittedKey;
+@public create window DailyWeeklyRestCandidateCoverageEmittedKeyWindow#unique(driverKey, startedAtEpochSecond, endedAtEpochSecond) as DailyWeeklyRestCandidateCoverageEmittedKey;
 
 insert into SupportGeoEvidenceWindow
 select
@@ -488,7 +488,7 @@ select
     vehicleKey
 from TachographActivityIntervalInputEvent(activityType = 'DRIVE', durationSeconds > ${SIGNIFICANT_DRIVING_THRESHOLD_SECONDS});
 
-create window PreviousSignificantDrivingInterval#unique(driverKey) as SignificantDrivingInterval;
+@public create window PreviousSignificantDrivingInterval#unique(driverKey) as SignificantDrivingInterval;
 
 on SignificantDrivingInterval as next
 insert into DrivingInterruptionInterval
@@ -1483,7 +1483,7 @@ select
     endedAtEpochSecond
 from DailyWeeklyRestCandidateCoverageInterval;
 
-context PerDriver
+@public context PerDriver
 create window PreviousVehicleUsageInterval#lastevent as TachographVehicleUsageIntervalInputEvent;
 
 @Priority(30)

src/test/java/at/procon/eventhub/processing/api/UnifiedRuntimeProcessingControllerTest.java

@@ -35,10 +35,6 @@ import at.procon.eventhub.processing.model.RuntimeDriverTimeline;
 import at.procon.eventhub.processing.model.RuntimeVehicleUsageInterval;
 import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
 import at.procon.eventhub.processing.service.UnifiedRuntimeDerivedProjectionService;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographDriverParityResultDto;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityCategoryComparisonDto;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
-import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
 import at.procon.eventhub.processing.service.UnifiedRuntimeDriverTimelineService;
 import at.procon.eventhub.processing.service.UnifiedRuntimeEventAssemblyService;
 import at.procon.eventhub.tachographfilesession.dto.TachographEsperDriverProcessingResultDto;
@@ -328,8 +324,6 @@ class UnifiedRuntimeProcessingControllerTest {
                         derivedProjectionService,
                         null,
                         null,
-                        null,
-                        null,
                         executionService
                 ))
                 .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@@ -339,7 +333,7 @@ class UnifiedRuntimeProcessingControllerTest {
         when(executionService.listPlans())
                 .thenReturn(List.of(new RuntimeProcessingPlanDescriptorDto(
                         "driver-working-time-v1",
-                        Set.of("tachograph-driver-esper-v1"),
+                        Set.of(),
                         "Driver working-time processing",
                         "Runs common runtime event processing modules.",
                         RuntimeEventPartitioningStrategy.DRIVER,
@@ -351,8 +345,7 @@ class UnifiedRuntimeProcessingControllerTest {
 
         mockMvc.perform(get("/api/eventhub/runtime-processing/executions/plans"))
                 .andExpect(status().isOk())
-                .andExpect(jsonPath("$[0].processingPlanKey").value("driver-working-time-v1"))
+                .andExpect(jsonPath("$[0].processingPlanKey").value("driver-working-time-v1"));
-                .andExpect(jsonPath("$[0].aliases[0]").value("tachograph-driver-esper-v1"));
     }
 
     @Test
@@ -367,8 +360,6 @@ class UnifiedRuntimeProcessingControllerTest {
                         derivedProjectionService,
                         null,
                         null,
-                        null,
-                        null,
                         executionService
                 ))
                 .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@@ -434,8 +425,6 @@ class UnifiedRuntimeProcessingControllerTest {
                         derivedProjectionService,
                         null,
                         null,
-                        null,
-                        null,
                         executionService
                 ))
                 .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@@ -517,7 +506,6 @@ class UnifiedRuntimeProcessingControllerTest {
                         eventAssemblyService,
                         timelineService,
                         derivedProjectionService,
-                        null,
                         runtimeEventProcessingService
                 ))
                 .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@@ -526,9 +514,9 @@ class UnifiedRuntimeProcessingControllerTest {
 
         when(runtimeEventProcessingService.listProfiles())
                 .thenReturn(List.of(new RuntimeEventProcessingProfileDescriptorDto(
-                        "tachograph-driver-esper-v1",
+                        "driver-working-time-v1",
-                        "Tachograph Driver Esper Processing",
+                        "Driver working-time processing",
-                        "Runs tachograph driver Esper processing over runtime event scopes.",
+                        "Runs common runtime event processing modules.",
                         RuntimeEventPartitioningStrategy.DRIVER,
                         List.of(RuntimeEventPartitioningStrategy.DRIVER),
                         Set.of(),
@@ -537,8 +525,8 @@ class UnifiedRuntimeProcessingControllerTest {
 
         mockMvc.perform(get("/api/eventhub/runtime-processing/event-processing/profiles"))
                 .andExpect(status().isOk())
-                .andExpect(jsonPath("$[0].profileKey").value("tachograph-driver-esper-v1"))
+                .andExpect(jsonPath("$[0].profileKey").value("driver-working-time-v1"))
-                .andExpect(jsonPath("$[0].displayName").value("Tachograph Driver Esper Processing"))
+                .andExpect(jsonPath("$[0].displayName").value("Driver working-time processing"))
                 .andExpect(jsonPath("$[0].defaultPartitioningStrategy").value("DRIVER"))
                 .andExpect(jsonPath("$[0].supportedPartitioningStrategies[0]").value("DRIVER"))
                 .andExpect(jsonPath("$[0].optionalParameters[0]").exists());
@@ -554,7 +542,6 @@ class UnifiedRuntimeProcessingControllerTest {
                         eventAssemblyService,
                         timelineService,
                         derivedProjectionService,
-                        null,
                         runtimeEventProcessingService
                 ))
                 .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
@@ -572,7 +559,7 @@ class UnifiedRuntimeProcessingControllerTest {
         );
         when(runtimeEventProcessingService.process(any()))
                 .thenReturn(new RuntimeEventProcessingResultDto(
-                        "tachograph-driver-esper-v1",
+                        "driver-working-time-v1",
                         RuntimeEventPartitioningStrategy.DRIVER,
                         request,
                         3,
@@ -588,7 +575,7 @@ class UnifiedRuntimeProcessingControllerTest {
                         .contentType("application/json")
                         .content("""
                                 {
-                                  "profileKey": "tachograph-driver-esper-v1",
+                                  "profileKey": "driver-working-time-v1",
                                   "scope": {
                                     "sessionId": "%s",
                                     "sourceFamilies": ["TACHOGRAPH_FILE_SESSION"],
@@ -607,155 +594,12 @@ class UnifiedRuntimeProcessingControllerTest {
                                 }
                                 """.formatted(sessionId)))
                 .andExpect(status().isOk())
-                .andExpect(jsonPath("$.profileKey").value("tachograph-driver-esper-v1"))
+                .andExpect(jsonPath("$.profileKey").value("driver-working-time-v1"))
                 .andExpect(jsonPath("$.partitioningStrategy").value("DRIVER"))
                 .andExpect(jsonPath("$.inputEventCount").value(3))
                 .andExpect(jsonPath("$.discoveredVehicles[0].vin").value("VIN-1"));
     }
 
-
-    @Test
-    void compatibilityTachographEndpointDelegatesThroughGenericProfileRuntimeApi() throws Exception {
-        UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);
-        UnifiedRuntimeDriverTimelineService timelineService = org.mockito.Mockito.mock(UnifiedRuntimeDriverTimelineService.class);
-        UnifiedRuntimeDerivedProjectionService derivedProjectionService = org.mockito.Mockito.mock(UnifiedRuntimeDerivedProjectionService.class);
-        RuntimeEventProcessingService runtimeEventProcessingService = org.mockito.Mockito.mock(RuntimeEventProcessingService.class);
-        MockMvc mockMvc = MockMvcBuilders.standaloneSetup(new UnifiedRuntimeProcessingController(
-                        eventAssemblyService,
-                        timelineService,
-                        derivedProjectionService,
-                        null,
-                        runtimeEventProcessingService
-                ))
-                .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
-                .setControllerAdvice(new UnifiedRuntimeProcessingExceptionHandler())
-                .build();
-
-        UUID sessionId = UUID.randomUUID();
-        UnifiedRuntimeProcessingRequest request = UnifiedRuntimeProcessingRequest.forTachographFileSession(
-                sessionId,
-                "12:123",
-                OffsetDateTime.parse("2026-05-01T08:00:00Z"),
-                OffsetDateTime.parse("2026-05-01T10:00:00Z"),
-                true,
-                0
-        );
-        UnifiedRuntimeDerivedProjectionResultDto driverResult = new UnifiedRuntimeDerivedProjectionResultDto(
-                request,
-                2,
-                1,
-                3,
-                5,
-                List.of(new UnifiedDiscoveredVehicleRef("VEH-1", "VIN-1", "12", "REG-1")),
-                null,
-                List.of("processed through generic profile")
-        );
-        when(runtimeEventProcessingService.process(any()))
-                .thenReturn(new RuntimeEventProcessingResultDto(
-                        "tachograph-driver-esper-v1",
-                        RuntimeEventPartitioningStrategy.DRIVER,
-                        request,
-                        5,
-                        1,
-                        1,
-                        List.of(new UnifiedDiscoveredVehicleRef("VEH-1", "VIN-1", "12", "REG-1")),
-                        Map.of("12:123", new RuntimeEventProcessingPartitionResultDto(
-                                "DRIVER",
-                                "12:123",
-                                "UnifiedRuntimeDerivedProjectionResultDto",
-                                driverResult,
-                                Map.of("mergedEventCount", 5)
-                        )),
-                        List.of("generic adapter"),
-                        List.of()
-                ));
-
-        mockMvc.perform(post("/api/eventhub/runtime-processing/tachograph/esper-processing")
-                        .contentType("application/json")
-                        .content("""
-                                {
-                                  "sessionId": "%s",
-                                  "sourceFamilies": ["TACHOGRAPH_FILE_SESSION"],
-                                  "driverKey": "12:123",
-                                  "occurredFrom": "2026-05-01T08:00:00Z",
-                                  "occurredTo": "2026-05-01T10:00:00Z",
-                                  "expandVehicleEvents": true,
-                                  "significantDrivingMinutes": 3,
-                                  "minimumRestPeriodMinutes": 720
-                                }
-                                """.formatted(sessionId)))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.inputEventCount").value(5))
-                .andExpect(jsonPath("$.selectedDriverCount").value(1))
-                .andExpect(jsonPath("$.driverResults['12:123'].mergedEventCount").value(5))
-                .andExpect(jsonPath("$.notes[0]").value("generic adapter"));
-    }
-
-
-    @Test
-    void validatesTachographParityViaRuntimeApi() throws Exception {
-        UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);
-        UnifiedRuntimeDriverTimelineService timelineService = org.mockito.Mockito.mock(UnifiedRuntimeDriverTimelineService.class);
-        UnifiedRuntimeDerivedProjectionService derivedProjectionService = org.mockito.Mockito.mock(UnifiedRuntimeDerivedProjectionService.class);
-        RuntimeEventProcessingService runtimeEventProcessingService = org.mockito.Mockito.mock(RuntimeEventProcessingService.class);
-        RuntimeTachographParityValidationService parityValidationService = org.mockito.Mockito.mock(RuntimeTachographParityValidationService.class);
-        MockMvc mockMvc = MockMvcBuilders.standaloneSetup(new UnifiedRuntimeProcessingController(
-                        eventAssemblyService,
-                        timelineService,
-                        derivedProjectionService,
-                        null,
-                        runtimeEventProcessingService,
-                        parityValidationService
-                ))
-                .setMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))
-                .setControllerAdvice(new UnifiedRuntimeProcessingExceptionHandler())
-                .build();
-
-        UUID sessionId = UUID.randomUUID();
-        when(parityValidationService.validate(any()))
-                .thenReturn(new RuntimeTachographParityValidationResultDto(
-                        "tachograph-driver-esper-v1",
-                        "EQUAL",
-                        List.of(sessionId),
-                        1,
-                        Map.of("12:123", new RuntimeTachographDriverParityResultDto(
-                                "12:123",
-                                "EQUAL",
-                                "SINGLE_FILE_SESSION",
-                                1,
-                                true,
-                                List.of(new RuntimeTachographParityCategoryComparisonDto(
-                                        "activityIntervals",
-                                        2,
-                                        2,
-                                        true
-                                )),
-                                List.of("validated"),
-                                List.of()
-                        )),
-                        List.of("validation complete"),
-                        List.of()
-                ));
-
-        mockMvc.perform(post("/api/eventhub/runtime-processing/event-processing/validation/tachograph-parity")
-                        .contentType("application/json")
-                        .content("""
-                                {
-                                  "sessionId": "%s",
-                                  "driverKey": "12:123",
-                                  "occurredFrom": "2026-05-01T08:00:00Z",
-                                  "occurredTo": "2026-05-01T10:00:00Z",
-                                  "includeDebug": true
-                                }
-                                """.formatted(sessionId)))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.profileKey").value("tachograph-driver-esper-v1"))
-                .andExpect(jsonPath("$.status").value("EQUAL"))
-                .andExpect(jsonPath("$.driverResults['12:123'].status").value("EQUAL"))
-                .andExpect(jsonPath("$.driverResults['12:123'].comparisons[0].category").value("activityIntervals"))
-                .andExpect(jsonPath("$.driverResults['12:123'].comparisons[0].equal").value(true));
-    }
-
     @Test
     void returnsBadRequestForInvalidRuntimeRequest() throws Exception {
         UnifiedRuntimeEventAssemblyService eventAssemblyService = org.mockito.Mockito.mock(UnifiedRuntimeEventAssemblyService.class);

src/test/java/at/procon/eventhub/processing/driverworkingtime/service/DriverWorkingTimeReusableProjectionBuilderTest.java

@@ -0,0 +1,115 @@
+package at.procon.eventhub.processing.driverworkingtime.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import at.procon.eventhub.config.EventHubProperties;
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeActivityInterval;
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDerivedProjectionBundle;
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeProcessingInput;
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+class DriverWorkingTimeReusableProjectionBuilderTest {
+
+    @Test
+    void reusesWarmRuntimeWithoutLeakingPreviousState() {
+        DriverWorkingTimeReusableProjectionBuilder builder =
+                new DriverWorkingTimeReusableProjectionBuilder(new EventHubProperties());
+        UUID sessionId = UUID.randomUUID();
+        OffsetDateTime from = OffsetDateTime.parse("2026-05-01T08:00:00Z");
+        OffsetDateTime firstDriveEnd = OffsetDateTime.parse("2026-05-01T09:00:00Z");
+        OffsetDateTime secondDriveStart = OffsetDateTime.parse("2026-05-01T10:00:00Z");
+        OffsetDateTime to = OffsetDateTime.parse("2026-05-01T11:00:00Z");
+
+        DriverWorkingTimeProcessingInput input = new DriverWorkingTimeProcessingInput(
+                sessionId,
+                "12:123",
+                "DRIVER_CARD",
+                from,
+                to,
+                from,
+                to,
+                15,
+                30,
+                List.of(
+                        new DriverWorkingTimeActivityInterval(
+                                sessionId,
+                                "12:123",
+                                "ACT-1",
+                                "DRIVE",
+                                "DRIVER",
+                                "INSERTED",
+                                "SINGLE",
+                                "12:REG-1",
+                                "VIN-1",
+                                "DRIVER_CARD",
+                                "ACT-1",
+                                "ACT-1",
+                                from,
+                                firstDriveEnd,
+                                from.toEpochSecond(),
+                                firstDriveEnd.toEpochSecond(),
+                                firstDriveEnd.toEpochSecond() - from.toEpochSecond(),
+                                List.of("ACT-1"),
+                                false,
+                                false,
+                                "RAW_INTERVAL"
+                        ),
+                        new DriverWorkingTimeActivityInterval(
+                                sessionId,
+                                "12:123",
+                                "ACT-2",
+                                "DRIVE",
+                                "DRIVER",
+                                "INSERTED",
+                                "SINGLE",
+                                "12:REG-1",
+                                "VIN-1",
+                                "DRIVER_CARD",
+                                "ACT-2",
+                                "ACT-2",
+                                secondDriveStart,
+                                to,
+                                secondDriveStart.toEpochSecond(),
+                                to.toEpochSecond(),
+                                to.toEpochSecond() - secondDriveStart.toEpochSecond(),
+                                List.of("ACT-2"),
+                                false,
+                                false,
+                                "RAW_INTERVAL"
+                        )
+                ),
+                List.of(
+                        new DriverWorkingTimeVehicleUsageInterval(
+                                sessionId,
+                                "12:123",
+                                "VU-1",
+                                "VU-1",
+                                "VU-1",
+                                from,
+                                to,
+                                from.toEpochSecond(),
+                                to.toEpochSecond(),
+                                to.toEpochSecond() - from.toEpochSecond(),
+                                100L,
+                                150L,
+                                "12:REG-1",
+                                "VIN-1",
+                                "DRIVER_CARD",
+                                List.of("VU-1")
+                        )
+                ),
+                List.of(),
+                List.of()
+        );
+
+        DriverWorkingTimeDerivedProjectionBundle first = builder.buildDerivedProjectionBundle(input);
+        DriverWorkingTimeDerivedProjectionBundle second = builder.buildDerivedProjectionBundle(input);
+
+        assertThat(second).isEqualTo(first);
+        assertThat(second.drivingInterruptionIntervals()).hasSize(1);
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/dto/RuntimeEventProcessingApiRequestTest.java

@@ -0,0 +1,49 @@
+package at.procon.eventhub.processing.eventprocessing.dto;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+class RuntimeEventProcessingApiRequestTest {
+
+    @Test
+    void driverWorkingTimePreservesDisabledVehicleExpansion() {
+        UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
+                UUID.randomUUID(),
+                List.of(),
+                null,
+                null,
+                Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
+                null,
+                null,
+                "12:123",
+                Set.of(),
+                false,
+                Set.of(),
+                false,
+                null,
+                null,
+                null,
+                OffsetDateTime.parse("2026-05-01T00:00:00Z"),
+                OffsetDateTime.parse("2026-05-02T00:00:00Z"),
+                false,
+                0,
+                true,
+                null,
+                null,
+                null,
+                null,
+                List.of()
+        );
+
+        RuntimeEventProcessingApiRequest request = RuntimeEventProcessingApiRequest.driverWorkingTime(scope);
+
+        assertThat(request.partitioning().attachVehicleEvidence()).isFalse();
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/module/DriverVehicleUsageIntervalsModuleTest.java

@@ -15,6 +15,7 @@ import at.procon.eventhub.dto.VehicleRefDto;
 import at.procon.eventhub.dto.VehicleRegistrationRefDto;
 import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
 import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.eventprocessing.module.epl.RuntimeEplModuleExecutor;
 import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
 import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
@@ -78,6 +79,64 @@ class DriverVehicleUsageIntervalsModuleTest {
         assertThat(intervals.get(0).endedAt()).isEqualTo(OffsetDateTime.parse("2026-05-01T10:10:00Z"));
     }
 
+    @Test
+    void reusesPreparedEplDefinitionAcrossExecutions() {
+        DriverVehicleUsageIntervalsModule module = new DriverVehicleUsageIntervalsModule(new RuntimeEplModuleExecutor());
+        RuntimeProcessingModuleContext context = new RuntimeProcessingModuleContext(
+                new RuntimeProcessingExecutionApiRequest(
+                        "driver-working-time-v1",
+                        runtimeScope(),
+                        null,
+                        List.of(),
+                        Map.of()
+                ),
+                List.of(
+                        vehicleUsageEvent(
+                                "CVU-1",
+                                EventType.CARD_INSERTED,
+                                EventLifecycle.INSERT,
+                                "2026-05-01T07:50:00Z",
+                                "2026-05-01T07:50:00Z",
+                                "2026-05-01T10:10:00Z",
+                                100_000L,
+                                "INSERTED"
+                        ),
+                        vehicleUsageEvent(
+                                "CVU-99",
+                                EventType.CARD_WITHDRAWN,
+                                EventLifecycle.WITHDRAW,
+                                "2026-05-01T10:10:00Z",
+                                "2026-05-01T07:50:00Z",
+                                "2026-05-01T10:10:00Z",
+                                140_000L,
+                                "NOT_INSERTED"
+                        )
+                ),
+                Map.of(),
+                Map.of()
+        );
+
+        RuntimeProcessingModuleResult first = module.execute(context);
+        RuntimeProcessingModuleResult second = module.execute(context);
+
+        assertThat(first.metadata())
+                .containsEntry(RuntimeEplModuleExecutor.DEFINITION_CACHE_HIT_METADATA_KEY, false)
+                .containsKey(RuntimeEplModuleExecutor.DEFINITION_PREPARATION_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.RUNTIME_INIT_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.DEPLOY_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.LISTENER_REGISTRATION_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.EVENT_SEND_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.DESTROY_MS_METADATA_KEY);
+        assertThat(second.metadata())
+                .containsEntry(RuntimeEplModuleExecutor.DEFINITION_CACHE_HIT_METADATA_KEY, true)
+                .containsEntry(RuntimeEplModuleExecutor.DEFINITION_PREPARATION_MS_METADATA_KEY, 0L)
+                .containsKey(RuntimeEplModuleExecutor.RUNTIME_INIT_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.DEPLOY_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.LISTENER_REGISTRATION_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.EVENT_SEND_MS_METADATA_KEY)
+                .containsKey(RuntimeEplModuleExecutor.DESTROY_MS_METADATA_KEY);
+    }
+
     private UnifiedRuntimeProcessingApiRequest runtimeScope() {
         return new UnifiedRuntimeProcessingApiRequest(
                 UUID.randomUUID(),

src/test/java/at/procon/eventhub/processing/eventprocessing/module/DriverWorkingTimeDerivedProjectionsModuleTest.java

@@ -211,6 +211,9 @@ class DriverWorkingTimeDerivedProjectionsModuleTest {
                 (UnifiedRuntimeDriverWorkingTimeScopeResultDto) result.output();
         UnifiedRuntimeDerivedProjectionResultDto driverResult = scopeResult.driverResults().get("12:123");
 
+        assertThat(result.metadata()).containsKey("driverProcessingTotalMs");
+        assertThat(result.metadata()).containsEntry("slowestDriverKey", "12:123");
+        assertThat(result.metadata()).containsKey("slowestDriverProcessingMs");
         assertThat(driverResult.projection().activityIntervalCount()).isEqualTo(1);
         assertThat(driverResult.projection().drivingIntervalCount()).isEqualTo(1);
         assertThat(driverResult.projection().activityIntervals()).isEmpty();

src/test/java/at/procon/eventhub/processing/eventprocessing/module/RuntimeProcessingPipelineExecutorTest.java

@@ -0,0 +1,119 @@
+package at.procon.eventhub.processing.eventprocessing.module;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
+import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class RuntimeProcessingPipelineExecutorTest {
+
+    @Test
+    void addsExecutionDurationMetadataToEachModuleResult() {
+        RuntimeProcessingPipelineExecutor executor = new RuntimeProcessingPipelineExecutor(
+                new RuntimeProcessingModuleRegistry(List.of(
+                        new TestModule("first"),
+                        new TestModule("second")
+                ))
+        );
+
+        Map<String, RuntimeProcessingModuleResult> results = executor.execute(
+                new RuntimeProcessingExecutionApiRequest(
+                        "driver-working-time-v1",
+                        new at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest(
+                                java.util.UUID.randomUUID(),
+                                List.of(),
+                                null,
+                                null,
+                                java.util.Set.of(at.procon.eventhub.processing.model.UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
+                                null,
+                                null,
+                                "12:123",
+                                java.util.Set.of(),
+                                false,
+                                java.util.Set.of(),
+                                false,
+                                null,
+                                null,
+                                null,
+                                java.time.OffsetDateTime.parse("2026-05-01T00:00:00Z"),
+                                java.time.OffsetDateTime.parse("2026-05-01T01:00:00Z"),
+                                true,
+                                0,
+                                null,
+                                null,
+                                null,
+                                null,
+                                null
+                        ),
+                        new at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest(
+                                RuntimeEventPartitioningStrategy.DRIVER,
+                                null,
+                                false,
+                                null,
+                                false,
+                                null,
+                                false,
+                                null,
+                                null,
+                                null
+                        ),
+                        List.of(),
+                        Map.of()
+                ),
+                List.of("first", "second"),
+                null
+        );
+
+        assertThat(results).containsKeys("first", "second");
+        assertThat(results.get("first").metadata()).containsEntry("marker", "first");
+        assertThat(results.get("first").metadata()).containsKey(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY);
+        assertThat(results.get("second").metadata()).containsEntry("marker", "second");
+        assertThat(results.get("second").metadata()).containsKey(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY);
+        assertThat(((Number) results.get("first").metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue())
+                .isGreaterThanOrEqualTo(0L);
+        assertThat(((Number) results.get("second").metadata().get(RuntimeProcessingPipelineExecutor.EXECUTION_DURATION_MS_METADATA_KEY)).longValue())
+                .isGreaterThanOrEqualTo(0L);
+    }
+
+    private static final class TestModule implements RuntimeProcessingModule {
+
+        private final String key;
+
+        private TestModule(String key) {
+            this.key = key;
+        }
+
+        @Override
+        public String moduleKey() {
+            return key;
+        }
+
+        @Override
+        public at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingModuleDescriptorDto descriptor() {
+            return new at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingModuleDescriptorDto(
+                    key,
+                    key,
+                    "test",
+                    "JAVA",
+                    java.util.Set.of()
+            );
+        }
+
+        @Override
+        public RuntimeProcessingModuleResult execute(RuntimeProcessingModuleContext context) {
+            if ("second".equals(key)) {
+                assertThat(context.previousResults()).containsKey("first");
+            }
+            return new RuntimeProcessingModuleResult(
+                    key,
+                    RuntimeProcessingModuleStatus.SUCCESS,
+                    null,
+                    Map.of("marker", key),
+                    List.of()
+            );
+        }
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/module/VehicleEvidenceAttachmentModuleTest.java

@@ -0,0 +1,137 @@
+package at.procon.eventhub.processing.eventprocessing.module;
+
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDriverPartition;
+import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
+import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
+import at.procon.eventhub.processing.eventprocessing.plan.RuntimeProcessingExecutionApiRequest;
+import at.procon.eventhub.processing.model.RuntimeDriverVehicleEvidenceAttachmentResult;
+import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
+import at.procon.eventhub.processing.model.UnifiedRuntimeEventBundle;
+import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
+import at.procon.eventhub.processing.service.RuntimeDriverVehicleEvidenceAttachmentService;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+class VehicleEvidenceAttachmentModuleTest {
+
+    @Test
+    void usesAttachmentAttributeInsteadOfScopeExpansionFlag() {
+        RuntimeDriverVehicleEvidenceAttachmentService service =
+                org.mockito.Mockito.mock(RuntimeDriverVehicleEvidenceAttachmentService.class);
+        VehicleEvidenceAttachmentModule module = new VehicleEvidenceAttachmentModule(service);
+
+        UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
+                UUID.randomUUID(),
+                List.of(),
+                null,
+                null,
+                Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
+                null,
+                null,
+                "12:123",
+                Set.of(),
+                false,
+                Set.of(),
+                false,
+                null,
+                null,
+                null,
+                OffsetDateTime.parse("2026-05-01T00:00:00Z"),
+                OffsetDateTime.parse("2026-05-02T00:00:00Z"),
+                true,
+                15,
+                true,
+                null,
+                null,
+                null,
+                null,
+                List.of()
+        );
+        UnifiedRuntimeProcessingRequest runtimeRequest = scope.toRuntimeRequest();
+        UnifiedRuntimeEventBundle bundle = new UnifiedRuntimeEventBundle(
+                runtimeRequest,
+                List.of(),
+                List.of(),
+                List.of(),
+                List.of(),
+                List.of()
+        );
+
+        when(service.attachVehicleEvidence(
+                eq("12:123"),
+                eq(List.of()),
+                eq(List.of()),
+                eq(List.<DriverWorkingTimeVehicleUsageInterval>of()),
+                eq(false),
+                eq(3),
+                anyBoolean()
+        )).thenReturn(new RuntimeDriverVehicleEvidenceAttachmentResult(
+                "12:123",
+                List.of(),
+                List.of(),
+                List.of(),
+                0,
+                0,
+                0,
+                List.of(),
+                List.of(),
+                List.of(),
+                List.of()
+        ));
+
+        RuntimeProcessingModuleContext context = new RuntimeProcessingModuleContext(
+                new RuntimeProcessingExecutionApiRequest("driver-working-time-v1", scope, null, List.of(), Map.of()),
+                List.of(),
+                Map.of(
+                        "runtimeScopeApiRequest", scope,
+                        DriverWorkingTimeRuntimeProcessingPlan.ATTACH_VEHICLE_ONLY_EVENTS_ATTRIBUTE, false,
+                        DriverWorkingTimeRuntimeProcessingPlan.VEHICLE_EVIDENCE_PADDING_MINUTES_ATTRIBUTE, 3
+                ),
+                Map.of(
+                        DriverWorkingTimeModuleKeys.RUNTIME_EVENT_ASSEMBLY,
+                        new RuntimeProcessingModuleResult(
+                                DriverWorkingTimeModuleKeys.RUNTIME_EVENT_ASSEMBLY,
+                                RuntimeProcessingModuleStatus.SUCCESS,
+                                bundle,
+                                Map.of(),
+                                List.of()
+                        ),
+                        DriverWorkingTimeModuleKeys.VEHICLE_USAGE_MERGE,
+                        new RuntimeProcessingModuleResult(
+                                DriverWorkingTimeModuleKeys.VEHICLE_USAGE_MERGE,
+                                RuntimeProcessingModuleStatus.SUCCESS,
+                                List.of(),
+                                Map.of(),
+                                List.of()
+                        )
+                )
+        );
+
+        RuntimeProcessingModuleResult result = module.execute(context);
+
+        verify(service).attachVehicleEvidence(
+                eq("12:123"),
+                eq(List.of()),
+                eq(List.of()),
+                eq(List.<DriverWorkingTimeVehicleUsageInterval>of()),
+                eq(false),
+                eq(3),
+                eq(false)
+        );
+        @SuppressWarnings("unchecked")
+        Map<String, DriverWorkingTimeDriverPartition> partitions =
+                (Map<String, DriverWorkingTimeDriverPartition>) result.output();
+        org.assertj.core.api.Assertions.assertThat(partitions).containsKey("12:123");
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/plan/DriverWorkingTimeRuntimeProcessingPlanTest.java

@@ -0,0 +1,72 @@
+package at.procon.eventhub.processing.eventprocessing.plan;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest;
+import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
+import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
+import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+class DriverWorkingTimeRuntimeProcessingPlanTest {
+
+    @Test
+    void applyExecutionRequestDoesNotRewriteScopeExpansionFromAttachmentFlag() {
+        DriverWorkingTimeRuntimeProcessingPlan plan = new DriverWorkingTimeRuntimeProcessingPlan(
+                org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class)
+        );
+        UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
+                UUID.randomUUID(),
+                List.of(),
+                null,
+                null,
+                Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
+                null,
+                null,
+                "12:123",
+                Set.of(),
+                false,
+                Set.of(),
+                false,
+                null,
+                null,
+                null,
+                OffsetDateTime.parse("2026-05-01T00:00:00Z"),
+                OffsetDateTime.parse("2026-05-02T00:00:00Z"),
+                true,
+                15,
+                true,
+                null,
+                null,
+                null,
+                null,
+                List.of()
+        );
+
+        UnifiedRuntimeProcessingApiRequest resolved = plan.applyExecutionRequest(
+                scope,
+                new RuntimeEventPartitioningApiRequest(
+                        RuntimeEventPartitioningStrategy.DRIVER,
+                        Set.of(),
+                        false,
+                        Set.of(),
+                        false,
+                        Set.of(),
+                        false,
+                        false,
+                        0,
+                        false
+                ),
+                Map.of("attachVehicleOnlyEvents", false)
+        );
+
+        assertThat(resolved.expandVehicleEvents()).isTrue();
+        assertThat(resolved.vehicleExpansionPaddingMinutes()).isEqualTo(15);
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/plan/RuntimeProcessingExecutionApiRequestTest.java

@@ -0,0 +1,49 @@
+package at.procon.eventhub.processing.eventprocessing.plan;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
+import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+class RuntimeProcessingExecutionApiRequestTest {
+
+    @Test
+    void driverWorkingTimePreservesDisabledVehicleExpansion() {
+        UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
+                UUID.randomUUID(),
+                List.of(),
+                null,
+                null,
+                Set.of(UnifiedEventSourceFamily.TACHOGRAPH_FILE_SESSION),
+                null,
+                null,
+                "12:123",
+                Set.of(),
+                false,
+                Set.of(),
+                false,
+                null,
+                null,
+                null,
+                OffsetDateTime.parse("2026-05-01T00:00:00Z"),
+                OffsetDateTime.parse("2026-05-02T00:00:00Z"),
+                false,
+                0,
+                true,
+                null,
+                null,
+                null,
+                null,
+                List.of()
+        );
+
+        RuntimeProcessingExecutionApiRequest request = RuntimeProcessingExecutionApiRequest.driverWorkingTime(scope);
+
+        assertThat(request.partitioning().attachVehicleEvidence()).isFalse();
+    }
+}

src/test/java/at/procon/eventhub/processing/eventprocessing/profile/DriverWorkingTimeRuntimeEventProcessingProfileTest.java

@@ -1,10 +1,6 @@
 package at.procon.eventhub.processing.eventprocessing.profile;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
 import at.procon.eventhub.processing.dto.UnifiedRuntimeDerivedProjectionResultDto;
 import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
@@ -12,28 +8,29 @@ import at.procon.eventhub.processing.dto.UnifiedRuntimeDriverWorkingTimeScopeRes
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventPartitioningApiRequest;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingApiRequest;
 import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
+import at.procon.eventhub.processing.eventprocessing.plan.DriverWorkingTimeRuntimeProcessingPlan;
 import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
 import at.procon.eventhub.processing.model.UnifiedRuntimeProcessingRequest;
 import at.procon.eventhub.processing.service.RuntimeDriverWorkingTimeScopeProcessingService;
-import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
 import java.time.OffsetDateTime;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import org.junit.jupiter.api.Test;
-import org.mockito.ArgumentCaptor;
 
-class TachographDriverEsperRuntimeEventProcessingProfileTest {
+class DriverWorkingTimeRuntimeEventProcessingProfileTest {
 
     @Test
     void exposesDiscoveryMetadata() {
-        TachographDriverEsperRuntimeEventProcessingProfile profile = new TachographDriverEsperRuntimeEventProcessingProfile(
+        DriverWorkingTimeRuntimeEventProcessingProfile profile = new DriverWorkingTimeRuntimeEventProcessingProfile(
-                org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class)
+                new DriverWorkingTimeRuntimeProcessingPlan(
+                        org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class)
+                )
         );
 
-        assertThat(profile.profileKey()).isEqualTo("tachograph-driver-esper-v1");
+        assertThat(profile.profileKey()).isEqualTo("driver-working-time-v1");
-        assertThat(profile.displayName()).isEqualTo("Tachograph Driver Esper Processing");
+        assertThat(profile.displayName()).isEqualTo("Driver working-time processing");
         assertThat(profile.defaultPartitioningStrategy()).isEqualTo(RuntimeEventPartitioningStrategy.DRIVER);
         assertThat(profile.supportedPartitioningStrategies()).containsExactly(RuntimeEventPartitioningStrategy.DRIVER);
         assertThat(profile.optionalParameters()).containsExactlyInAnyOrder(
@@ -48,9 +45,11 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
     }
 
     @Test
-    void delegatesToTachographScopeServiceAndMapsPartitionResults() {
+    void delegatesToDriverWorkingTimePlanAndMapsPartitionResults() {
         RuntimeDriverWorkingTimeScopeProcessingService scopeService = org.mockito.Mockito.mock(RuntimeDriverWorkingTimeScopeProcessingService.class);
-        TachographDriverEsperRuntimeEventProcessingProfile profile = new TachographDriverEsperRuntimeEventProcessingProfile(scopeService);
+        DriverWorkingTimeRuntimeEventProcessingProfile profile = new DriverWorkingTimeRuntimeEventProcessingProfile(
+                new DriverWorkingTimeRuntimeProcessingPlan(scopeService)
+        );
 
         UUID sessionId = UUID.randomUUID();
         UnifiedRuntimeProcessingApiRequest scope = new UnifiedRuntimeProcessingApiRequest(
@@ -80,7 +79,7 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
                 null
         );
         RuntimeEventProcessingApiRequest request = new RuntimeEventProcessingApiRequest(
-                TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                 scope,
                 new RuntimeEventPartitioningApiRequest(
                         RuntimeEventPartitioningStrategy.DRIVER,
@@ -137,7 +136,7 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
                 null,
                 List.of("driver processed")
         );
-        when(scopeService.processScope(any(), anyBoolean()))
+        org.mockito.Mockito.when(scopeService.processScope(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyBoolean()))
                 .thenReturn(new UnifiedRuntimeDriverWorkingTimeScopeResultDto(
                         processedRequest,
                         5,
@@ -152,23 +151,10 @@ class TachographDriverEsperRuntimeEventProcessingProfileTest {
 
         var result = profile.process(request);
 
-        assertThat(result.profileKey()).isEqualTo(TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY);
+        assertThat(result.profileKey()).isEqualTo(DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY);
         assertThat(result.partitioningStrategy()).isEqualTo(RuntimeEventPartitioningStrategy.DRIVER);
         assertThat(result.partitionResults()).containsOnlyKeys("12:DRIVER-1");
         assertThat(result.partitionResults().get("12:DRIVER-1").partitionType()).isEqualTo("DRIVER");
         assertThat(result.partitionResults().get("12:DRIVER-1").result()).isSameAs(driverResult);
-
-        ArgumentCaptor<UnifiedRuntimeProcessingApiRequest> captor = ArgumentCaptor.forClass(UnifiedRuntimeProcessingApiRequest.class);
-        ArgumentCaptor<Boolean> debugCaptor = ArgumentCaptor.forClass(Boolean.class);
-        verify(scopeService).processScope(captor.capture(), debugCaptor.capture());
-        UnifiedRuntimeProcessingApiRequest delegated = captor.getValue();
-        assertThat(delegated.driverKeys()).containsExactly("12:DRIVER-1");
-        assertThat(delegated.significantDrivingMinutes()).isEqualTo(5);
-        assertThat(delegated.minimumRestPeriodMinutes()).isEqualTo(600);
-        assertThat(delegated.vehicleExpansionPaddingMinutes()).isEqualTo(20);
-        assertThat(delegated.expandVehicleEvents()).isTrue();
-        assertThat(delegated.includeActivityIntervals()).isTrue();
-        assertThat(delegated.includeDrivingIntervals()).isTrue();
-        assertThat(debugCaptor.getValue()).isTrue();
     }
 }

src/test/java/at/procon/eventhub/processing/eventprocessing/validation/RuntimeMixedSourceEvidenceValidationServiceTest.java

@@ -14,7 +14,7 @@ import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingA
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
 import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
-import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
+import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
 import at.procon.eventhub.processing.model.UnifiedEventSourceFamily;
 import at.procon.eventhub.processing.model.UnifiedRuntimeEventBackend;
 import at.procon.eventhub.processing.dto.UnifiedRuntimeProcessingApiRequest;
@@ -58,7 +58,7 @@ class RuntimeMixedSourceEvidenceValidationServiceTest {
                 )
         );
         when(processingService.process(any())).thenReturn(new RuntimeEventProcessingResultDto(
-                TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                 RuntimeEventPartitioningStrategy.DRIVER,
                 null,
                 3,
@@ -78,7 +78,7 @@ class RuntimeMixedSourceEvidenceValidationServiceTest {
 
         RuntimeMixedSourceEvidenceValidationResultDto result = service.validate(new RuntimeMixedSourceEvidenceValidationApiRequest(
                 new RuntimeEventProcessingApiRequest(
-                        TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                        DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                         new UnifiedRuntimeProcessingApiRequest(
                                 null,
                                 List.of(),

src/test/java/at/procon/eventhub/processing/eventprocessing/validation/RuntimeTachographParityValidationServiceTest.java

@@ -10,7 +10,7 @@ import at.procon.eventhub.processing.eventprocessing.RuntimeEventProcessingServi
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingPartitionResultDto;
 import at.procon.eventhub.processing.eventprocessing.dto.RuntimeEventProcessingResultDto;
 import at.procon.eventhub.processing.eventprocessing.partition.RuntimeEventPartitioningStrategy;
-import at.procon.eventhub.tachographfilesession.processing.profile.TachographDriverEsperRuntimeEventProcessingProfile;
+import at.procon.eventhub.processing.eventprocessing.profile.DriverWorkingTimeRuntimeEventProcessingProfile;
 import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationApiRequest;
 import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationResultDto;
 import at.procon.eventhub.tachographfilesession.processing.validation.RuntimeTachographParityValidationService;
@@ -73,7 +73,7 @@ class RuntimeTachographParityValidationServiceTest {
         );
         when(runtimeEventProcessingService.process(any()))
                 .thenReturn(new RuntimeEventProcessingResultDto(
-                        TachographDriverEsperRuntimeEventProcessingProfile.PROFILE_KEY,
+                        DriverWorkingTimeRuntimeEventProcessingProfile.PROFILE_KEY,
                         RuntimeEventPartitioningStrategy.DRIVER,
                         runtimeRequest,
                         4,

src/test/java/at/procon/eventhub/processing/service/TachographFileSessionRuntimeEventLoaderTest.java

@@ -28,6 +28,7 @@ import at.procon.eventhub.tachographfilesession.model.TachographCompositeSession
 import at.procon.eventhub.tachographfilesession.service.InMemoryTachographCompositeSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
+import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -53,9 +54,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
                 new VehicleKeyFactory(),
                 new EventDetailsFactory(new ObjectMapper())
         );
+        RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
         TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
-                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
+                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
-                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
+                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
                 compositeRepository,
                 new EventAcquisitionRecordKeyService(),
                 new EventHubEventSorter()
@@ -90,9 +92,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
                 new VehicleKeyFactory(),
                 new EventDetailsFactory(new ObjectMapper())
         );
+        RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
         TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
-                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
+                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
-                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
+                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
                 compositeRepository,
                 new EventAcquisitionRecordKeyService(),
                 new EventHubEventSorter()
@@ -135,9 +138,10 @@ class TachographFileSessionRuntimeEventLoaderTest {
                 new VehicleKeyFactory(),
                 new EventDetailsFactory(new ObjectMapper())
         );
+        RawSourceDriverTimelineEventBuilder rawSourceEventBuilder = new RawSourceDriverTimelineEventBuilder(eventBuilder);
         TachographFileSessionRuntimeEventLoader loader = new TachographFileSessionRuntimeEventLoader(
-                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, eventBuilder))),
+                new UnifiedDriverEventSourceService(List.of(new TachographFileSessionUnifiedDriverEventSource(repository, rawSourceEventBuilder))),
-                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, eventBuilder))),
+                new UnifiedVehicleEventSourceService(List.of(new TachographFileSessionUnifiedVehicleEventSource(repository, rawSourceEventBuilder))),
                 compositeRepository,
                 new EventAcquisitionRecordKeyService(),
                 new EventHubEventSorter()

src/test/java/at/procon/eventhub/processing/service/UnifiedDriverEventSourceServiceTest.java

@@ -23,6 +23,7 @@ import at.procon.eventhub.tachographfilesession.service.DriverKeyFactory;
 import at.procon.eventhub.tachographfilesession.service.DriverTimelineBuilder;
 import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
+import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -45,12 +46,12 @@ class UnifiedDriverEventSourceServiceTest {
         UnifiedDriverEventSourceService service = new UnifiedDriverEventSourceService(List.of(
                 new TachographFileSessionUnifiedDriverEventSource(
                         repository,
-                        new IntervalBackedDriverTimelineEventBuilder(
+                        new RawSourceDriverTimelineEventBuilder(new IntervalBackedDriverTimelineEventBuilder(
                                 timelineBuilder,
                                 new DriverKeyFactory(),
                                 new VehicleKeyFactory(),
                                 new EventDetailsFactory(new ObjectMapper())
-                        )
+                        ))
                 )
         ));
 

src/test/java/at/procon/eventhub/processing/service/UnifiedVehicleEventSourceServiceTest.java

@@ -23,6 +23,7 @@ import at.procon.eventhub.tachographfilesession.service.DriverKeyFactory;
 import at.procon.eventhub.tachographfilesession.service.DriverTimelineBuilder;
 import at.procon.eventhub.tachographfilesession.service.InMemoryTachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.IntervalBackedDriverTimelineEventBuilder;
+import at.procon.eventhub.tachographfilesession.service.RawSourceDriverTimelineEventBuilder;
 import at.procon.eventhub.tachographfilesession.service.TachographFileSessionRepository;
 import at.procon.eventhub.tachographfilesession.service.VehicleKeyFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -42,15 +43,16 @@ class UnifiedVehicleEventSourceServiceTest {
         EventHubProperties properties = new EventHubProperties();
         TachographFileSessionRepository repository = new InMemoryTachographFileSessionRepository(properties);
         DriverTimelineBuilder timelineBuilder = new DriverTimelineBuilder();
+        IntervalBackedDriverTimelineEventBuilder eventBuilder = new IntervalBackedDriverTimelineEventBuilder(
+                timelineBuilder,
+                new DriverKeyFactory(),
+                new VehicleKeyFactory(),
+                new EventDetailsFactory(new ObjectMapper())
+        );
         UnifiedVehicleEventSourceService service = new UnifiedVehicleEventSourceService(List.of(
                 new TachographFileSessionUnifiedVehicleEventSource(
                         repository,
-                        new IntervalBackedDriverTimelineEventBuilder(
+                        new RawSourceDriverTimelineEventBuilder(eventBuilder)
-                                timelineBuilder,
-                                new DriverKeyFactory(),
-                                new VehicleKeyFactory(),
-                                new EventDetailsFactory(new ObjectMapper())
-                        )
                 )
         ));