feat(kernel): JournalCompactionService + orchestrator compaction hook + CAS journal rendering

2026-06-08 10:07:03 +04:00
parent 084820a8a3
commit 5ad945ccb4
4 changed files with 168 additions and 5 deletions
@@ -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<SessionId, AtomicBoolean> =
@@ -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(
@@ -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
+    }
+}
@@ -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()),
@@ -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<EventPayload>()
+        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<EventPayload>()
+        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<EventPayload>()
+        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<String>()
+        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<EventPayload>()
+        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")
+    }
+}