diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 311cdb9e..7553c93b 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -2074,9 +2074,14 @@ abstract class SessionOrchestrator( ) { return@forEach } - emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) - emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) - emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) + emitAll( + sessionId, + listOf( + ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1), + ArtifactValidatingEvent(slot.name, sessionId, stageId), + ArtifactValidatedEvent(slot.name, sessionId, stageId), + ), + ) } } @@ -2198,9 +2203,14 @@ abstract class SessionOrchestrator( ) return@forEach } - emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) - emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) - emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) + emitAll( + sessionId, + listOf( + ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1), + ArtifactValidatingEvent(slot.name, sessionId, stageId), + ArtifactValidatedEvent(slot.name, sessionId, stageId), + ), + ) } } @@ -2810,9 +2820,14 @@ abstract class SessionOrchestrator( // Emit artifact lifecycle events for process_result slots on failure branches. // Content was already stored in CAS + cache by dispatchToolCalls for every outcome. stageConfig.produces.filter { it.kind.id == "process_result" }.forEach { slot -> - emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) - emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) - emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) + emitAll( + sessionId, + listOf( + ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1), + ArtifactValidatingEvent(slot.name, sessionId, stageId), + ArtifactValidatedEvent(slot.name, sessionId, stageId), + ), + ) } } @@ -3192,6 +3207,29 @@ abstract class SessionOrchestrator( ) } + /** Appends [payloads] as one transaction (one flow-publish pass) for events that belong together. */ + internal suspend fun emitAll(sessionId: SessionId, payloads: List) { + if (payloads.size <= 1) { + payloads.firstOrNull()?.let { emit(sessionId, it) } + return + } + eventStore.appendAll( + payloads.map { payload -> + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = payload, + ) + }, + ) + } + // --- token estimation --- protected open suspend fun estimateTokens(content: String): Int {