perf(kernel): batch the artifact-lifecycle event triple via appendAll (#59)

The Created->Validating->Validated triple fires unconditionally back-to-back at
three artifact-emission sites — three single-event append transactions (each its
own commit + flow-publish pass) for one logical unit. Added emitAll(sessionId,
payloads) which routes 2+ events through EventStore.appendAll (one transaction,
one publish pass) and falls back to emit() for 0/1. Same events, same order.

Adjacent-but-conditional emits (clarification/approval pause+resume) are left on
emit(): they straddle suspension points, so batching would change observable
ordering.
This commit is contained in:
2026-07-12 13:22:05 +04:00
parent 65d5e9de83
commit 530b118b67
@@ -2074,9 +2074,14 @@ abstract class SessionOrchestrator(
) { ) {
return@forEach return@forEach
} }
emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) emitAll(
emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) sessionId,
emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) 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 return@forEach
} }
emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) emitAll(
emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) sessionId,
emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) 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. // Emit artifact lifecycle events for process_result slots on failure branches.
// Content was already stored in CAS + cache by dispatchToolCalls for every outcome. // Content was already stored in CAS + cache by dispatchToolCalls for every outcome.
stageConfig.produces.filter { it.kind.id == "process_result" }.forEach { slot -> stageConfig.produces.filter { it.kind.id == "process_result" }.forEach { slot ->
emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1)) emitAll(
emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId)) sessionId,
emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId)) 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<EventPayload>) {
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 --- // --- token estimation ---
protected open suspend fun estimateTokens(content: String): Int { protected open suspend fun estimateTokens(content: String): Int {