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:
+10
@@ -338,6 +338,14 @@ abstract class SessionOrchestrator(
|
||||
* Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */
|
||||
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,
|
||||
* 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)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user