feat(journal): extend DecisionJournalState + reducer arm for JournalCompactedEvent
This commit is contained in:
@@ -3,6 +3,7 @@ package com.correx.core.journal
|
|||||||
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
||||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
import com.correx.core.events.events.ArtifactValidatedEvent
|
||||||
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
||||||
|
import com.correx.core.events.events.JournalCompactedEvent
|
||||||
import com.correx.core.events.events.InitialIntentEvent
|
import com.correx.core.events.events.InitialIntentEvent
|
||||||
import com.correx.core.events.events.RetryAttemptedEvent
|
import com.correx.core.events.events.RetryAttemptedEvent
|
||||||
import com.correx.core.events.events.StageFailedEvent
|
import com.correx.core.events.events.StageFailedEvent
|
||||||
@@ -16,6 +17,15 @@ import com.correx.core.journal.model.DecisionRecord
|
|||||||
class DefaultDecisionJournalReducer : DecisionJournalReducer {
|
class DefaultDecisionJournalReducer : DecisionJournalReducer {
|
||||||
override fun reduce(state: DecisionJournalState, event: StoredEvent): DecisionJournalState {
|
override fun reduce(state: DecisionJournalState, event: StoredEvent): DecisionJournalState {
|
||||||
if (event.payload is ExecutionPlanLockedEvent) return state.copy(planLocked = true)
|
if (event.payload is ExecutionPlanLockedEvent) return state.copy(planLocked = true)
|
||||||
|
if (event.payload is JournalCompactedEvent) {
|
||||||
|
val p = event.payload as JournalCompactedEvent
|
||||||
|
return state.copy(
|
||||||
|
compactedThroughSequence = p.coversThroughSequence,
|
||||||
|
summaryArtifactId = p.summaryArtifactId,
|
||||||
|
lowSalienceOmittedCount = state.lowSalienceOmittedCount + p.lowSalienceOmittedCount,
|
||||||
|
records = state.records.filter { it.sequence > p.coversThroughSequence },
|
||||||
|
)
|
||||||
|
}
|
||||||
val record = when (val p = event.payload) {
|
val record = when (val p = event.payload) {
|
||||||
is InitialIntentEvent -> DecisionRecord(
|
is InitialIntentEvent -> DecisionRecord(
|
||||||
sequence = event.sessionSequence,
|
sequence = event.sessionSequence,
|
||||||
|
|||||||
@@ -1,9 +1,13 @@
|
|||||||
package com.correx.core.journal.model
|
package com.correx.core.journal.model
|
||||||
|
|
||||||
|
import com.correx.core.events.types.ArtifactId
|
||||||
import kotlinx.serialization.Serializable
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
@Serializable
|
@Serializable
|
||||||
data class DecisionJournalState(
|
data class DecisionJournalState(
|
||||||
val records: List<DecisionRecord> = emptyList(),
|
val records: List<DecisionRecord> = emptyList(),
|
||||||
val planLocked: Boolean = false,
|
val planLocked: Boolean = false,
|
||||||
|
val compactedThroughSequence: Long = -1L,
|
||||||
|
val summaryArtifactId: ArtifactId? = null,
|
||||||
|
val lowSalienceOmittedCount: Int = 0,
|
||||||
)
|
)
|
||||||
|
|||||||
+107
@@ -0,0 +1,107 @@
|
|||||||
|
package com.correx.core.journal
|
||||||
|
|
||||||
|
import com.correx.core.events.events.EventMetadata
|
||||||
|
import com.correx.core.events.events.JournalCompactedEvent
|
||||||
|
import com.correx.core.events.events.StoredEvent
|
||||||
|
import com.correx.core.events.types.ArtifactId
|
||||||
|
import com.correx.core.events.types.EventId
|
||||||
|
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 kotlinx.datetime.Instant
|
||||||
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
|
||||||
|
class DecisionJournalCompactionTest {
|
||||||
|
|
||||||
|
private val reducer = DefaultDecisionJournalReducer()
|
||||||
|
|
||||||
|
private fun compactionEvent(
|
||||||
|
coversThroughSequence: Long,
|
||||||
|
summaryArtifactId: ArtifactId = ArtifactId("artifact-1"),
|
||||||
|
lowSalienceOmittedCount: Int = 0,
|
||||||
|
seq: Long = coversThroughSequence + 1,
|
||||||
|
): StoredEvent = StoredEvent(
|
||||||
|
metadata = EventMetadata(
|
||||||
|
eventId = EventId("e-compact"),
|
||||||
|
sessionId = SessionId("s1"),
|
||||||
|
timestamp = Instant.parse("2026-01-01T00:00:00Z"),
|
||||||
|
schemaVersion = 1,
|
||||||
|
causationId = null,
|
||||||
|
correlationId = null,
|
||||||
|
),
|
||||||
|
sequence = seq,
|
||||||
|
sessionSequence = seq,
|
||||||
|
payload = JournalCompactedEvent(
|
||||||
|
sessionId = SessionId("s1"),
|
||||||
|
coversThroughSequence = coversThroughSequence,
|
||||||
|
summaryArtifactId = summaryArtifactId,
|
||||||
|
lowSalienceOmittedCount = lowSalienceOmittedCount,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
private fun record(seq: Long) = DecisionRecord(
|
||||||
|
sequence = seq,
|
||||||
|
kind = DecisionKind.TRANSITION,
|
||||||
|
summary = "record-$seq",
|
||||||
|
)
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `records at or below coversThroughSequence are purged`() {
|
||||||
|
val state = DecisionJournalState(
|
||||||
|
records = listOf(record(1), record(2), record(3), record(5)),
|
||||||
|
)
|
||||||
|
val result = reducer.reduce(state, compactionEvent(coversThroughSequence = 3, seq = 4))
|
||||||
|
assertEquals(listOf(record(5)), result.records)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `records above coversThroughSequence survive`() {
|
||||||
|
val state = DecisionJournalState(
|
||||||
|
records = listOf(record(10), record(20)),
|
||||||
|
)
|
||||||
|
val result = reducer.reduce(state, compactionEvent(coversThroughSequence = 5, seq = 6))
|
||||||
|
assertEquals(listOf(record(10), record(20)), result.records)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `compactedThroughSequence summaryArtifactId and lowSalienceOmittedCount update correctly`() {
|
||||||
|
val artifactId = ArtifactId("artifact-42")
|
||||||
|
val state = DecisionJournalState()
|
||||||
|
val result = reducer.reduce(
|
||||||
|
state,
|
||||||
|
compactionEvent(
|
||||||
|
coversThroughSequence = 7L,
|
||||||
|
summaryArtifactId = artifactId,
|
||||||
|
lowSalienceOmittedCount = 3,
|
||||||
|
seq = 8,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
assertEquals(7L, result.compactedThroughSequence)
|
||||||
|
assertEquals(artifactId, result.summaryArtifactId)
|
||||||
|
assertEquals(3, result.lowSalienceOmittedCount)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `lowSalienceOmittedCount accumulates across two compaction events`() {
|
||||||
|
val state = DecisionJournalState()
|
||||||
|
val afterFirst = reducer.reduce(
|
||||||
|
state,
|
||||||
|
compactionEvent(coversThroughSequence = 5, lowSalienceOmittedCount = 4, seq = 6),
|
||||||
|
)
|
||||||
|
val afterSecond = reducer.reduce(
|
||||||
|
afterFirst,
|
||||||
|
compactionEvent(coversThroughSequence = 10, lowSalienceOmittedCount = 6, seq = 11),
|
||||||
|
)
|
||||||
|
assertEquals(10, afterSecond.lowSalienceOmittedCount)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `empty records list survives a compaction event`() {
|
||||||
|
val state = DecisionJournalState()
|
||||||
|
val result = reducer.reduce(state, compactionEvent(coversThroughSequence = 100, seq = 101))
|
||||||
|
assertEquals(emptyList<DecisionRecord>(), result.records)
|
||||||
|
assertEquals(100L, result.compactedThroughSequence)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user