From 3559ea67ef595cb6ac8ad5ce0765822e58447799 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 12 Jul 2026 17:05:34 +0400 Subject: [PATCH] fix(kernel,server): evict per-session caches on workflow termination (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two unbounded per-session leaks that never released after a workflow ended: - SessionOrchestrator.artifactContentCache (full file contents, keyed ":") grew for the process lifetime. Added evictArtifactContentCache(sessionId) — session-prefix key removal, rehydrate-safe — called on both completeWorkflow and failWorkflow. cancellations was already evicted on both terminal paths. - NarrationSubscriber leaked a Channel + worker coroutine + lanes map entry per session forever. closeLane() now closes the lane channel on WorkflowCompleted/WorkflowFailed (draining the terminal narration first); the worker self-removes from lanes when its for-loop exits on close. Deferred: SqliteEventStore.subscriptions eviction — removing the SharedFlow mid-life would strand LiveArtifactRepository's downstream collector (a suspended-coroutine leak worse than the tiny empty-SharedFlow entry). Green: :core:kernel compile+detekt, :apps:server *NarrationSubscriber* (8). --- .../server/narration/NarrationSubscriber.kt | 48 ++++++++++++------- .../orchestration/SessionOrchestrator.kt | 10 ++++ 2 files changed, 42 insertions(+), 16 deletions(-) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt index 4b3a9fe3..9fecf431 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt @@ -104,22 +104,28 @@ class NarrationSubscriber( stageId = p.stageId.value, ), ) - is WorkflowCompletedEvent -> enqueue( - sid, - NarrationTrigger( - kind = "workflow_completed", - instruction = "The workflow finished successfully. Provide a brief summary.", - stageId = p.terminalStageId.value, - ), - ) - is WorkflowFailedEvent -> enqueue( - sid, - NarrationTrigger( - kind = "workflow_failed", - instruction = "The workflow failed in stage ${p.stageId.value}: ${p.reason}. Explain the failure to the user.", - stageId = p.stageId.value, - ), - ) + is WorkflowCompletedEvent -> { + enqueue( + sid, + NarrationTrigger( + kind = "workflow_completed", + instruction = "The workflow finished successfully. Provide a brief summary.", + stageId = p.terminalStageId.value, + ), + ) + closeLane(sid) + } + is WorkflowFailedEvent -> { + enqueue( + sid, + NarrationTrigger( + kind = "workflow_failed", + instruction = "The workflow failed in stage ${p.stageId.value}: ${p.reason}. Explain the failure to the user.", + stageId = p.stageId.value, + ), + ) + closeLane(sid) + } // Surface the semantic reviewer's verdict conversationally instead of a raw findings // dump: the narrator turns "FAIL, 2 findings" into a plain-language explanation the // operator can act on. Only narrate when there's something to say (non-PASS or findings). @@ -221,6 +227,13 @@ class NarrationSubscriber( lanes[sessionId.value]?.pendingPauses?.clear() } + /** Closes a terminated session's lane so its worker drains any buffered narration (the terminal + * one enqueued just before this) and then exits, self-removing the lane from [lanes]. Without + * this, every session leaks its channel + worker coroutine + map entry forever. */ + private fun closeLane(sessionId: SessionId) { + lanes[sessionId.value]?.channel?.close() + } + private fun startLane(sessionId: SessionId): SessionLane { val channel = Channel(capacity = Channel.UNLIMITED) val lane = SessionLane(channel, used = 0) @@ -249,6 +262,9 @@ class NarrationSubscriber( ) } } + // Channel closed on workflow termination: drop the lane so it stops leaking. A later event + // for the same session (none expected post-terminal) would lazily start a fresh lane. + lanes.remove(sessionId.value) } return lane } 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 e87ef556..e03d6909 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 @@ -338,6 +338,14 @@ abstract class SessionOrchestrator( * Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */ protected val artifactContentCache: ConcurrentHashMap = ConcurrentHashMap() + /** Drops a terminated session's cached artifact contents (the heaviest per-session state — full + * file/JSON payloads). Safe: rehydrateArtifactContentCache rebuilds it from durable events if the + * session is ever resumed. Called on WorkflowCompleted/WorkflowFailed. */ + private fun evictArtifactContentCache(sessionId: SessionId) { + val prefix = "${sessionId.value}:" + artifactContentCache.keys.removeAll { it.startsWith(prefix) } + } + /** Deterministic extraction/repair ladder for near-miss LLM artifact text (prose-wrapped JSON, * code fences, trailing commas). Pure — recomputes on replay, records no events. */ private val artifactExtractionPipeline = ArtifactExtractionPipeline() @@ -3091,6 +3099,7 @@ abstract class SessionOrchestrator( correlateCritiqueOutcomes(sessionId) emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId)) cancellations.remove(sessionId) + evictArtifactContentCache(sessionId) return WorkflowResult.Completed(sessionId, terminalStageId) } @@ -3134,6 +3143,7 @@ abstract class SessionOrchestrator( ) } cancellations.remove(sessionId) + evictArtifactContentCache(sessionId) return WorkflowResult.Failed(sessionId, reason, retryExhausted) }