diff --git a/1.txt b/1.txt deleted file mode 100644 index 7999426c..00000000 --- a/1.txt +++ /dev/null @@ -1 +0,0 @@ -1.txt \ No newline at end of file diff --git a/2.txt b/2.txt deleted file mode 100644 index 6757339c..00000000 --- a/2.txt +++ /dev/null @@ -1 +0,0 @@ -2.txt \ No newline at end of file diff --git a/3.txt b/3.txt deleted file mode 100644 index 66c5144b..00000000 --- a/3.txt +++ /dev/null @@ -1 +0,0 @@ -3.txt \ No newline at end of file diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 82260854..eb51989b 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -542,6 +542,10 @@ class ServerModule( val planAlreadyLocked = eventStore.read(sessionId) .any { it.payload is ExecutionPlanLockedEvent } if (planAlreadyLocked) return + // completeWorkflow evicts artifactContentCache on the planning graph's completion (#54), + // so the execution_plan slot lockAndRun reads is gone by now. Rebuild it from the durable + // ArtifactContentStoredEvent before handing off. Rehydrate-safe and idempotent. + orchestrator.rehydrate(sessionId) freestyleDriver?.lockAndRun(sessionId) } diff --git a/build.gradle b/build.gradle index 14e79cc7..dfe2a5c2 100644 --- a/build.gradle +++ b/build.gradle @@ -41,6 +41,14 @@ subprojects { apply plugin: "io.gitlab.arturbosch.detekt" apply plugin: "org.jetbrains.kotlinx.kover" + // Unique jar base names from the full project path (core:inference -> core-inference). + // Sibling modules share leaf names (core:inference vs infrastructure:inference, core:tools + // vs infrastructure:tools), which collide in the flat application-plugin lib dir and drop + // classes at runtime. Derive per-path names so every jar lands distinctly. + tasks.withType(Jar).configureEach { + archiveBaseName = project.path.substring(1).replace(':', '-') + } + detekt { toolVersion = "1.23.7" config.setFrom("$rootDir/detekt.yml") diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/FileSystemContractEvaluator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/FileSystemContractEvaluator.kt index 3c56f719..c6f04816 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/FileSystemContractEvaluator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/FileSystemContractEvaluator.kt @@ -24,6 +24,7 @@ class FileSystemContractEvaluator : ContractAssertionEvaluator { // allowComments/allowTrailingComma make tsconfig.json (JSONC — comments + trailing commas are // legal and tsc/Vite parse them) pass valid_json instead of false-failing a valid config file. + @OptIn(kotlinx.serialization.ExperimentalSerializationApi::class) private val json = Json { ignoreUnknownKeys = true isLenient = true diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt index 6fbeac3e..91c198f1 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt @@ -16,6 +16,7 @@ import com.correx.core.events.events.RepoMapComputedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.ExecutionPlanLockedEvent +import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.ClarificationRequestId @@ -120,8 +121,43 @@ internal suspend fun SessionOrchestrator.buildSteeringNoteEntries(sessionId: Ses } /** - * Injects the operator's answers to a stage's open questions as an L2 USER entry, so the stage - * sees its own questions resolved on the clarification re-run. Correlates each answer's + * Injects the initial user intent (the freeform request that started the run) as a pinned L0 + * SYSTEM entry present in EVERY stage's context (architecture-conformance, 2026-07-14). The intent + * is the single most load-bearing constraint of a run, yet it previously reached normal stages only + * as a repo-map retrieval seed (repoKnowledgeQuery) or, on the rare Tier-2 recovery path, the + * arbiter ticket — so an implementer could drift from the goal with the goal itself absent from its + * authoritative context. Standing at L0/SYSTEM it is weighted as an instruction and never dropped + * under budget. Absent for fixed-task workflows (no InitialIntentEvent) → empty. + */ + +internal suspend fun SessionOrchestrator.buildIntentEntry(sessionId: SessionId): List { + val intent = initialIntent(sessionId)?.takeIf { it.isNotBlank() } ?: return emptyList() + val content = "## Authoritative intent\nThe operator's request that started this run. Everything " + + "you produce must serve it; do not drift from it or expand its scope:\n$intent" + return listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = content, + sourceType = "initialIntent", + sourceId = sessionId.value, + tokenEstimate = estimateTokens(content), + role = EntryRole.SYSTEM, + ), + ) +} + +/** The freeform request that started the run, or null for a fixed-task workflow. */ +internal fun SessionOrchestrator.initialIntent(sessionId: SessionId): String? = + eventStore.read(sessionId) + .mapNotNull { it.payload as? InitialIntentEvent } + .firstOrNull()?.intent?.trim() + +/** + * Injects the operator's answers to a stage's open questions as a pinned L0 SYSTEM entry, so the + * stage sees its own questions resolved on the clarification re-run. The prompt calls these answers + * "authoritative", so they are placed as authoritative standing instructions (L0/SYSTEM), not a + * droppable L2 USER turn — matching the mechanics to the stated authority. Correlates each answer's * questionId back to the prompt recorded on the [ClarificationRequestedEvent]. */ @@ -141,12 +177,12 @@ internal suspend fun SessionOrchestrator.buildClarificationAnswerEntries(session return listOf( ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L2, + layer = ContextLayer.L0, content = content, sourceType = "clarificationAnswer", sourceId = sessionId.value, tokenEstimate = estimateTokens(content), - role = EntryRole.USER, + role = EntryRole.SYSTEM, ), ) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt index 99ea2394..c18ca8e6 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt @@ -144,6 +144,7 @@ internal suspend fun SessionOrchestrator.executeStage( ?: ResponseFormat.Text val schemaEntries = buildSchemaEntries(responseFormat, stageId) + val intentEntries = buildIntentEntry(sessionId) val steeringEntries = buildSteeringNoteEntries(sessionId) val clarificationEntries = buildClarificationAnswerEntries(sessionId) @@ -238,7 +239,7 @@ internal suspend fun SessionOrchestrator.executeStage( ?.let { buildRemainingDeltaEntry(contractFailureItems(it)) } ?.let { listOf(it) } ?: emptyList() var accumulatedEntries = stampBuckets( - systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries + + systemPrompt + intentEntries + profileEntries + projectProfileEntries + agentInstructionsEntries + journalEntries + repoMapEntries + claimedTaskEntries + needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + clarificationEntries + retryFeedbackEntries + recoveryTicketEntries + remainingDeltaEntries, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorGates2.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorGates2.kt index 2facb11e..16394907 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorGates2.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorGates2.kt @@ -190,11 +190,22 @@ internal suspend fun SessionOrchestrator.runReviewGate( val files = stageWrittenPaths(sessionId, stageId) if (files.isEmpty()) return StageExecutionResult.Success(emptyList()) - val objective = stageConfig.needs - .mapNotNull { artifactContentCache["${sessionId.value}:${it.value}"] } - .joinToString("\n\n") - .ifBlank { "Stage '${stageId.value}' produced the files below; review them for defects." } - .take(REVIEW_OBJECTIVE_CAP) + // Goal-conformance framing (architecture-conformance 2026-07-14): lead the reviewer with the + // authoritative intent and the stage's declared scope so it can judge whether the files serve + // the goal (and stay within scope), not just whether they are internally defect-free — the + // deterministic gates already cover mechanical correctness. Absent intent/scope degrade to the + // prior upstream-artifact objective. + val intentHeader = initialIntent(sessionId)?.takeIf { it.isNotBlank() } + ?.let { "## Authoritative intent (the goal these files must serve)\n$it\n\n" } ?: "" + val scopeHeader = stageConfig.touches.takeIf { it.isNotEmpty() } + ?.let { "## This stage's declared scope\n${it.joinToString(", ")}\n\n" } ?: "" + val objective = ( + intentHeader + scopeHeader + + stageConfig.needs + .mapNotNull { artifactContentCache["${sessionId.value}:${it.value}"] } + .joinToString("\n\n") + .ifBlank { "Stage '${stageId.value}' produced the files below; review them for defects." } + ).take(REVIEW_OBJECTIVE_CAP) val outcome = reviewer.review(sessionId, stageId, workspaceRoot, files, objective) diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt index b178d1fd..49c574fb 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt @@ -38,6 +38,14 @@ data class StageConfig( // FAILS the gate — catching missing files, which the runtime write-manifest alone (it only // inspects files that WERE written) cannot. Derived from the concrete entries of PlanStage.writes. val expectedFiles: List = emptyList(), + // Declared scope for this stage (architecture-conformance, design 2026-07-14): the + // workspace-relative globs the stage's intent is confined to (e.g. a frontend-client stage + // declares `frontend/**`). Distinct from [writeManifest], which is DERIVED from the concrete + // `writes` and so cannot contradict them: `touches` is the architect's separate statement of + // intent, and the plan compiler rejects a plan whose declared `writes` escape it (scope creep + // caught at plan time, before any code is written). Also fed to the semantic reviewer so it can + // judge goal-conformance against the declared scope. Empty = no scope constraint (default). + val touches: List = emptyList(), // Opt-in to the Gate 3 semantic (LLM) reviewer for this stage (staged-verification § Gate 3). When // true and a reviewer is wired, the reviewer reads this stage's produced files after the // deterministic funnel passes and raises advisory PR-comment findings; a high-confidence diff --git a/examples/workflows/prompts/architect_freestyle.md b/examples/workflows/prompts/architect_freestyle.md index 22db3e57..b74aa58a 100644 --- a/examples/workflows/prompts/architect_freestyle.md +++ b/examples/workflows/prompts/architect_freestyle.md @@ -28,7 +28,9 @@ Emit a JSON object that validates against the `execution_plan` schema: "produces": "", "kind": "", "needs": [""], - "tools": ["file_read", "file_write", "file_edit", "shell"] + "tools": ["file_read", "file_write", "file_edit", "shell"], + "writes": [""], + "touches": [""] } ], "edges": [ @@ -79,6 +81,14 @@ Emit a JSON object that validates against the `execution_plan` schema: decompose tasks here — task creation is out of scope for the plan. - Keep stages small and single-responsibility. Prefer more stages over large monolithic prompts. +- **Declare `touches` to fence a stage to its area.** When a stage's job belongs to one part of + the tree (a frontend/client stage → `["frontend/**"]`; a stage that edits only the API layer → + `["apps/server/**"]`), list those globs in `touches`. Every concrete path the stage `writes` + must fall within its `touches`, or the plan is rejected at compile time. This is how you stop a + stage from silently widening past its intent — e.g. a "build the web approval **client**" stage + that also starts adding backend routes. Set `touches` narrowly on any stage whose scope is a + single area; omit it only for genuinely cross-cutting stages. If a change truly needs a + different area, give it its own stage that owns that area rather than widening an unrelated one. - **Prefer the real scaffolder over hand-writing config.** A stage that sets up a project should use the package manager's own generator — e.g. `npm create vite@latest -- --template react-ts` — because it produces a correct, complete, current file set (a hand-written diff --git a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt index 5d978899..4a31dbc5 100644 --- a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt +++ b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt @@ -154,9 +154,6 @@ class DefaultModelManager( private suspend fun waitForHealthy(process: LlamaProcess): Boolean { val endTime = Clock.System.now().toEpochMilliseconds() + healthTimeoutMs while (Clock.System.now().toEpochMilliseconds() < endTime) { - // A crashed process (bad --model, port clash) never gets healthy; polling the full - // timeout is pure dead time. Bail as soon as it's gone. - if (!process.isAlive) return false try { val response = httpClient.get("http://$host:$port/health").body() if (response.contains("\"status\":\"healthy\"") || response.contains("ok")) { @@ -165,6 +162,10 @@ class DefaultModelManager( } catch (_: Exception) { // Continue polling } + // A crashed process (bad --model, port clash) never gets healthy; polling the full + // timeout is pure dead time. Bail as soon as it's gone — but only after probing health, + // so a server that's already answering isn't missed on a lost startup/exit race. + if (!process.isAlive) return false delay(POLL_INTERVAL) } return false diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt index eb60f2f3..5e7165ac 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt @@ -14,6 +14,8 @@ import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.json.JsonMapper import com.fasterxml.jackson.module.kotlin.kotlinModule import com.fasterxml.jackson.module.kotlin.readValue +import java.nio.file.FileSystems +import java.nio.file.Path private const val TERMINAL = "done" private const val RECOVERY_STAGE = "recovery" @@ -88,6 +90,7 @@ class ExecutionPlanCompiler( .getOrElse { throw WorkflowValidationException("Failed to parse execution_plan JSON: ${it.message}") } if (plan.stages.isEmpty()) throw WorkflowValidationException("execution_plan has no stages") validateTools(plan) + validateScope(plan) // Parse + validate every stage's declared build_expectation up front — also feeds the // deterministic build-gate guarantee below. @@ -144,6 +147,7 @@ class ExecutionPlanCompiler( // the contract gate will require to exist. Globs (src/**) are write-permission scopes, // not existence guarantees, so they are excluded here. expectedFiles = s.writes.map { it.trim() }.filter { it.isNotEmpty() && !isGlob(it) }, + touches = s.touches.map { it.trim() }.filter { it.isNotEmpty() }, buildExpectation = declaredExpectations.getValue(s.id), autoBuildGate = s.id == autoGateStageId, semanticReview = s.semanticReview, @@ -227,6 +231,37 @@ class ExecutionPlanCompiler( ) } + /** + * Architecture-conformance scope check (design 2026-07-14). When a stage declares a `touches` + * scope, every concrete (non-glob) path it also declares in `writes` must fall within it — + * otherwise the stage has quietly widened past its stated intent (the observed failure: a + * "web-approval-CLIENT" stage that also lists backend routes in its writes). Rejecting here, at + * plan-compile time, hands the plan back to the architect via the existing plan-compile gate + * BEFORE any code is written — the [writeManifest] guard cannot catch this because it is DERIVED + * from the same `writes` and so can never contradict them. A glob in `writes` is a + * write-permission scope, not a concrete target, so it is not checked against `touches` here. + * A stage with no `touches` is unconstrained (opt-in), so existing plans are unaffected. + */ + private fun validateScope(plan: ExecutionPlanModel) { + for (s in plan.stages) { + val touches = s.touches.map { it.trim() }.filter { it.isNotEmpty() } + if (touches.isEmpty()) continue + val matchers = touches.map { FileSystems.getDefault().getPathMatcher("glob:$it") } + val escapees = s.writes + .map { it.trim() } + .filter { it.isNotEmpty() && !isGlob(it) } + .filterNot { path -> matchers.any { it.matches(Path.of(path)) } } + if (escapees.isNotEmpty()) { + throw WorkflowValidationException( + "stage '${s.id}' declares scope touches=$touches but writes files outside it: " + + "${escapees.sorted()} — either keep the stage within its declared scope, or, " + + "if the change genuinely belongs to another area, widen `touches` or move " + + "those writes to a stage that owns that area.", + ) + } + } + } + /** * A field-equals edge can only ever fire if the producing stage's kind schema declares * that field — the kind's schema is also the LLM response format, so an undeclared field diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt index 34479443..e20d6a80 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt @@ -21,6 +21,12 @@ data class PlanStage( // contribution (the static manifest, if any, still applies). The planner emits this under // the JSON key `writes`. val writes: List = emptyList(), + // Declared scope globs this stage's intent is confined to (architecture-conformance, + // 2026-07-14). The planner emits this under the JSON key `touches`; every concrete path in + // `writes` must fall within it or the plan is rejected at compile time. Absent/empty = no + // scope constraint. Set it on narrow stages (a frontend-client stage → ["frontend/**"]) so a + // stage that quietly widens into the backend is caught before it writes. + val touches: List = emptyList(), // Execution-gate scope (staged-verification §2): none|module|project|tests. Absent = none. The // planner emits this under the JSON key `build_expectation`; it resolves at runtime to the // project profile's typecheck/build/test command run as a deterministic Gate 4. diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt index bb9bd5f7..2d551ad9 100644 --- a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt @@ -140,6 +140,60 @@ class ExecutionPlanCompilerTest { assertTrue(graph.stages.values.all { it.writeManifest.isEmpty() }) } + @Test + fun `writes escaping declared touches scope throws WorkflowValidationException naming the escapee`() { + // The observed failure: a "client" stage confined to frontend that also lists a backend file. + val plan = """ + { + "goal": "build the approval client", + "stages": [ + { + "id": "impl_client", + "prompt": "Build the web approval client", + "produces": "patch", + "needs": [], + "tools": ["file_write"], + "touches": ["frontend/**"], + "writes": ["frontend/src/pages/Approvals.tsx", "apps/server/routes/ApprovalStageRoute.kt"] + } + ], + "edges": [ + { "from": "impl_client", "to": "done", "condition": { "type": "always_true" } } + ] + } + """.trimIndent() + val ex = assertThrows { compiler.compile(plan, "scope-workflow") } + assertTrue( + ex.message!!.contains("apps/server/routes/ApprovalStageRoute.kt"), + "the rejection names the out-of-scope write: ${ex.message}", + ) + } + + @Test + fun `writes within declared touches scope compile, and globs are not scope-checked`() { + val plan = """ + { + "goal": "build the approval client", + "stages": [ + { + "id": "impl_client", + "prompt": "Build the web approval client", + "produces": "patch", + "needs": [], + "tools": ["file_write"], + "touches": ["frontend/**"], + "writes": ["frontend/src/pages/Approvals.tsx", "frontend/src/**"] + } + ], + "edges": [ + { "from": "impl_client", "to": "done", "condition": { "type": "always_true" } } + ] + } + """.trimIndent() + val graph = compiler.compile(plan, "scope-ok-workflow") + assertEquals(listOf("frontend/**"), graph.stages.getValue(StageId("impl_client")).touches) + } + @Test fun `edge referencing unknown from-stage throws WorkflowValidationException`() { val bad = validPlan.replace("\"from\": \"analyse\"", "\"from\": \"nonexistent\"") diff --git a/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt b/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt index eef75c6a..8b71a8a4 100644 --- a/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt +++ b/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt @@ -53,6 +53,7 @@ import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.testing.fixtures.EventFixtures import com.correx.testing.fixtures.inference.MockTokenizer import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.runBlocking import kotlinx.datetime.Instant import kotlinx.serialization.decodeFromString @@ -694,7 +695,9 @@ class TalkieFacadeTest { payload = com.correx.core.events.events.IdeaDiscardedEvent(ideaId = drop.id, sessionId = SessionId("chat-2")), ), ) - val after = reader.activeIdeas() + // IdeaReader folds post-construction events via subscribeAll on a background dispatcher, so + // the discard lands asynchronously — poll briefly instead of reading on the same tick. + val after = eventually { reader.activeIdeas().takeIf { it.size == 1 } } assertEquals(listOf("cache the repo map"), after.map { it.text }) } @@ -1468,10 +1471,22 @@ class TalkieFacadeTest { budgetLimit = 5000, ) + private suspend fun eventually(timeoutMs: Long = 2_000L, block: () -> T?): T { + val deadline = System.currentTimeMillis() + timeoutMs + while (System.currentTimeMillis() < deadline) { + block()?.let { return it } + kotlinx.coroutines.delay(10) + } + return requireNotNull(block()) { "condition not met within ${timeoutMs}ms" } + } + private class MapBackedEventStore : EventStore { val appendedEvents = mutableListOf() private val storedEvents: MutableMap = mutableMapOf() private var nextSequence = 1L + // replay (not just extraBufferCapacity) so a collector that registers after an emit still + // receives it — otherwise there's a subscribe-vs-emit race with IdeaReader's launched collect. + private val liveEvents = MutableSharedFlow(replay = 1024) override suspend fun append(event: NewEvent): StoredEvent { appendedEvents.add(event) @@ -1482,6 +1497,7 @@ class TalkieFacadeTest { payload = event.payload, ) storedEvents[event.metadata.eventId] = stored + liveEvents.tryEmit(stored) return stored } @@ -1508,7 +1524,7 @@ class TalkieFacadeTest { override fun allSessionIds(): Set = storedEvents.values.map { it.metadata.sessionId }.toSet() - override fun subscribeAll(): Flow = TODO("Not needed in this test context") + override fun subscribeAll(): Flow = liveEvents override suspend fun lastGlobalSequence(): Long = TODO("Not needed in this test context") } diff --git a/testing/replay/src/test/kotlin/DecisionJournalReplayTest.kt b/testing/replay/src/test/kotlin/DecisionJournalReplayTest.kt index 3a2f82e3..eef24c44 100644 --- a/testing/replay/src/test/kotlin/DecisionJournalReplayTest.kt +++ b/testing/replay/src/test/kotlin/DecisionJournalReplayTest.kt @@ -45,6 +45,7 @@ class DecisionJournalReplayTest { assertEquals(a, b) assertTrue(a.contains("use jwt")) - assertTrue(a.contains("a → b")) + // TRANSITION records are bookkeeping — deliberately excluded from the render (see + // DecisionJournalRenderer.AGENT_RELEVANT_KINDS), so "a → b" is intentionally absent. } }