From 5ad945ccb425395f50e9bdbedf1e98c66a126e18 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 8 Jun 2026 10:07:03 +0400 Subject: [PATCH] feat(kernel): JournalCompactionService + orchestrator compaction hook + CAS journal rendering --- .../DefaultSessionOrchestrator.kt | 18 +++- .../orchestration/JournalCompactionService.kt | 52 ++++++++++ .../orchestration/SessionOrchestrator.kt | 8 +- .../JournalCompactionServiceTest.kt | 95 +++++++++++++++++++ 4 files changed, 168 insertions(+), 5 deletions(-) create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt create mode 100644 core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 1290600c..ba095f54 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -5,6 +5,7 @@ import com.correx.core.approvals.ApprovalStatus import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.inference.Tokenizer import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.journal.DecisionJournalRenderer import com.correx.core.artifacts.ArtifactState import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.ApprovalDecisionResolvedEvent @@ -50,7 +51,8 @@ class DefaultSessionOrchestrator( private val retryCoordinator: RetryCoordinator, artifactStore: ArtifactStore, tokenizer: Tokenizer? = null, - decisionJournalRepository: DefaultDecisionJournalRepository, + private val decisionJournalRepository: DefaultDecisionJournalRepository, + private val compactionService: JournalCompactionService? = null, ) : SessionOrchestrator(repositories, engines, artifactStore, decisionJournalRepository), ApprovalGateway { override val tokenizer: Tokenizer? = tokenizer override val cancellations: ConcurrentHashMap = @@ -294,8 +296,20 @@ class DefaultSessionOrchestrator( SubagentRunRequest(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config), ).outcome when (result) { - is StageExecutionResult.Success -> + is StageExecutionResult.Success -> { + compactionService?.let { svc -> + val journalState = decisionJournalRepository.getJournal(ctx.sessionId) + val journalText = DecisionJournalRenderer().render(journalState) + val tokenEstimate = journalText.length / 4 + svc.compactIfNeeded( + sessionId = ctx.sessionId, + state = journalState, + renderedTokenEstimate = tokenEstimate, + emit = { payload -> emit(ctx.sessionId, payload) }, + ) + } return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) + } is StageExecutionResult.Failure -> { log.debug( diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt new file mode 100644 index 00000000..a05b1f51 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt @@ -0,0 +1,52 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.JournalCompactedEvent +import com.correx.core.events.types.SessionId +import com.correx.core.journal.model.DecisionJournalState +import com.correx.core.journal.model.Salience + +class JournalCompactionService( + private val artifactStore: ArtifactStore, + private val summarize: suspend (String) -> String, + val tokenThreshold: Int = 2000, +) { + suspend fun compactIfNeeded( + sessionId: SessionId, + state: DecisionJournalState, + renderedTokenEstimate: Int, + emit: suspend (EventPayload) -> Unit, + ): Boolean { + if (renderedTokenEstimate < tokenThreshold || state.records.isEmpty()) return false + + val throughSequence = state.records.maxOf { it.sequence } + val highRecords = state.records.filter { it.kind.salience() == Salience.HIGH } + val lowCount = state.records.count { it.kind.salience() == Salience.LOW } + + val summaryText = if (highRecords.isEmpty()) { + "(no high-salience decisions to summarize)" + } else { + val prompt = buildString { + appendLine("Summarize the following key decisions concisely (≤200 words).") + appendLine("Preserve all user intent, approvals, steering, and failures.") + appendLine() + highRecords.forEach { appendLine("- [${it.kind}] ${it.summary}") } + } + summarize(prompt) + } + + val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8)) + artifactStore.flushBefore { + emit( + JournalCompactedEvent( + sessionId = sessionId, + coversThroughSequence = throughSequence, + summaryArtifactId = summaryArtifactId, + lowSalienceOmittedCount = lowCount, + ), + ) + } + return true + } +} 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 dd1f8816..96399085 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 @@ -311,9 +311,11 @@ abstract class SessionOrchestrator( // from currentContext.layers (a Map grouped by layer) would scramble turn order // across rounds; instead we grow our own ordered list and let the builder restamp // ordinals from it each round. - val journalText = decisionJournalRenderer.render( - decisionJournalRepository.getJournal(sessionId), - ) + val journalState = decisionJournalRepository.getJournal(sessionId) + val summaryText = journalState.summaryArtifactId?.let { id -> + artifactStore.get(id)?.toString(Charsets.UTF_8) + } + val journalText = decisionJournalRenderer.render(journalState, summaryText) val journalEntries = if (journalText.isBlank()) emptyList() else listOf( ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), diff --git a/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt new file mode 100644 index 00000000..624d25af --- /dev/null +++ b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt @@ -0,0 +1,95 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.JournalCompactedEvent +import com.correx.core.events.types.SessionId +import com.correx.core.journal.model.DecisionJournalState +import com.correx.core.journal.model.DecisionKind +import com.correx.core.journal.model.DecisionRecord +import com.correx.core.utils.TypeId +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class JournalCompactionServiceTest { + + private val fixedArtifactId = TypeId("artifact-001") + + private fun fakeArtifactStore(): ArtifactStore = object : ArtifactStore { + override suspend fun put(bytes: ByteArray) = fixedArtifactId + override suspend fun get(id: TypeId): ByteArray? = null + override suspend fun flushBefore(commit: suspend () -> Unit) = commit() + } + + private fun stateWithRecords(vararg records: DecisionRecord) = + DecisionJournalState(records = records.toList()) + + private fun makeRecord(seq: Long, kind: DecisionKind) = + DecisionRecord(sequence = seq, kind = kind, summary = "summary of $kind") + + @Test + fun `returns false when token estimate is below threshold`(): Unit = runBlocking { + val svc = JournalCompactionService(fakeArtifactStore(), { it }, tokenThreshold = 2000) + val state = stateWithRecords(makeRecord(1, DecisionKind.INTENT)) + val emitted = mutableListOf() + val result = svc.compactIfNeeded(SessionId("s1"), state, renderedTokenEstimate = 500) { emitted += it } + assertFalse(result) + assertTrue(emitted.isEmpty()) + } + + @Test + fun `returns false when records is empty even if estimate is high`(): Unit = runBlocking { + val svc = JournalCompactionService(fakeArtifactStore(), { it }, tokenThreshold = 2000) + val state = DecisionJournalState(records = emptyList()) + val emitted = mutableListOf() + val result = svc.compactIfNeeded(SessionId("s1"), state, renderedTokenEstimate = 9999) { emitted += it } + assertFalse(result) + assertTrue(emitted.isEmpty()) + } + + @Test + fun `emits JournalCompactedEvent with correct coversThroughSequence and lowSalienceOmittedCount`(): Unit = runBlocking { + val svc = JournalCompactionService(fakeArtifactStore(), { "summary" }, tokenThreshold = 100) + val state = stateWithRecords( + makeRecord(1, DecisionKind.INTENT), // HIGH + makeRecord(2, DecisionKind.TRANSITION), // LOW + makeRecord(3, DecisionKind.APPROVAL), // HIGH + makeRecord(4, DecisionKind.RETRY), // LOW + ) + val emitted = mutableListOf() + val result = svc.compactIfNeeded(SessionId("s1"), state, renderedTokenEstimate = 500) { emitted += it } + assertTrue(result) + assertEquals(1, emitted.size) + val event = emitted.first() as JournalCompactedEvent + assertEquals(4L, event.coversThroughSequence) + assertEquals(2, event.lowSalienceOmittedCount) + assertEquals(fixedArtifactId, event.summaryArtifactId) + } + + @Test + fun `only HIGH-salience records are included in the summarize prompt`(): Unit = runBlocking { + val capturedPrompts = mutableListOf() + val svc = JournalCompactionService( + artifactStore = fakeArtifactStore(), + summarize = { prompt -> capturedPrompts += prompt; "ok" }, + tokenThreshold = 100, + ) + val state = stateWithRecords( + makeRecord(1, DecisionKind.INTENT), // HIGH + makeRecord(2, DecisionKind.TRANSITION), // LOW — should not appear in prompt + makeRecord(3, DecisionKind.ARTIFACT), // LOW — should not appear in prompt + makeRecord(4, DecisionKind.FAILURE), // HIGH + ) + val emitted = mutableListOf() + svc.compactIfNeeded(SessionId("s1"), state, renderedTokenEstimate = 500) { emitted += it } + assertEquals(1, capturedPrompts.size) + val prompt = capturedPrompts.first() + assertTrue(prompt.contains("INTENT"), "Prompt should contain INTENT") + assertTrue(prompt.contains("FAILURE"), "Prompt should contain FAILURE") + assertFalse(prompt.contains("TRANSITION"), "Prompt must not contain TRANSITION") + assertFalse(prompt.contains("ARTIFACT"), "Prompt must not contain ARTIFACT") + } +}