fix(kernel,server): evict per-session caches on workflow termination (#54)

Two unbounded per-session leaks that never released after a workflow ended:

- SessionOrchestrator.artifactContentCache (full file contents, keyed
  "<sessionId>:<path>") 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).
This commit is contained in:
2026-07-12 17:05:34 +04:00
parent 97e56b9275
commit 3559ea67ef
2 changed files with 42 additions and 16 deletions
@@ -104,22 +104,28 @@ class NarrationSubscriber(
stageId = p.stageId.value, stageId = p.stageId.value,
), ),
) )
is WorkflowCompletedEvent -> enqueue( is WorkflowCompletedEvent -> {
sid, enqueue(
NarrationTrigger( sid,
kind = "workflow_completed", NarrationTrigger(
instruction = "The workflow finished successfully. Provide a brief summary.", kind = "workflow_completed",
stageId = p.terminalStageId.value, instruction = "The workflow finished successfully. Provide a brief summary.",
), stageId = p.terminalStageId.value,
) ),
is WorkflowFailedEvent -> enqueue( )
sid, closeLane(sid)
NarrationTrigger( }
kind = "workflow_failed", is WorkflowFailedEvent -> {
instruction = "The workflow failed in stage ${p.stageId.value}: ${p.reason}. Explain the failure to the user.", enqueue(
stageId = p.stageId.value, 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 // 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 // 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). // 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() 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 { private fun startLane(sessionId: SessionId): SessionLane {
val channel = Channel<QueuedNarration>(capacity = Channel.UNLIMITED) val channel = Channel<QueuedNarration>(capacity = Channel.UNLIMITED)
val lane = SessionLane(channel, used = 0) 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 return lane
} }
@@ -338,6 +338,14 @@ abstract class SessionOrchestrator(
* Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */ * Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */
protected val artifactContentCache: ConcurrentHashMap<String, String> = ConcurrentHashMap() protected val artifactContentCache: ConcurrentHashMap<String, String> = 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, /** Deterministic extraction/repair ladder for near-miss LLM artifact text (prose-wrapped JSON,
* code fences, trailing commas). Pure — recomputes on replay, records no events. */ * code fences, trailing commas). Pure — recomputes on replay, records no events. */
private val artifactExtractionPipeline = ArtifactExtractionPipeline() private val artifactExtractionPipeline = ArtifactExtractionPipeline()
@@ -3091,6 +3099,7 @@ abstract class SessionOrchestrator(
correlateCritiqueOutcomes(sessionId) correlateCritiqueOutcomes(sessionId)
emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId)) emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId))
cancellations.remove(sessionId) cancellations.remove(sessionId)
evictArtifactContentCache(sessionId)
return WorkflowResult.Completed(sessionId, terminalStageId) return WorkflowResult.Completed(sessionId, terminalStageId)
} }
@@ -3134,6 +3143,7 @@ abstract class SessionOrchestrator(
) )
} }
cancellations.remove(sessionId) cancellations.remove(sessionId)
evictArtifactContentCache(sessionId)
return WorkflowResult.Failed(sessionId, reason, retryExhausted) return WorkflowResult.Failed(sessionId, reason, retryExhausted)
} }