fix(kernel): emit ArtifactCreatedEvent for tool-produced slots and fix verifyProduces race
LiveArtifactRepository uses a hot subscription that does not replay history, so its cache was always empty when verifyProduces ran immediately after emit. Fix by reading directly from EventStore (synchronous) instead of the repository. Also emit ArtifactCreatedEvent for non-llmEmitted produces slots after stage validation passes, and fix tool parameter decoding to use JsonPrimitive.content so escape sequences like \n are resolved rather than passed as literal backslash-n.
This commit is contained in:
+31
-3
@@ -10,6 +10,7 @@ import com.correx.core.context.model.ContextLayer
|
|||||||
import com.correx.core.context.model.ContextPack
|
import com.correx.core.context.model.ContextPack
|
||||||
import com.correx.core.context.model.TokenBudget
|
import com.correx.core.context.model.TokenBudget
|
||||||
import com.correx.core.events.types.ContextEntryId
|
import com.correx.core.events.types.ContextEntryId
|
||||||
|
import com.correx.core.events.events.ArtifactCreatedEvent
|
||||||
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
||||||
import com.correx.core.events.events.ApprovalRequestedEvent
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
||||||
import com.correx.core.events.events.EventMetadata
|
import com.correx.core.events.events.EventMetadata
|
||||||
@@ -262,7 +263,10 @@ abstract class SessionOrchestrator(
|
|||||||
if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false)
|
if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false)
|
||||||
val validationCtx = ValidationContext(graph = graph, sessionState = session.state)
|
val validationCtx = ValidationContext(graph = graph, sessionState = session.state)
|
||||||
when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) {
|
when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) {
|
||||||
is StageExecutionResult.Success -> verifyProduces(sessionId, stageId, stageConfig)
|
is StageExecutionResult.Success -> {
|
||||||
|
emitToolArtifacts(sessionId, stageId, stageConfig)
|
||||||
|
verifyProduces(sessionId, stageId, stageConfig)
|
||||||
|
}
|
||||||
is StageExecutionResult.Failure -> outcome
|
is StageExecutionResult.Failure -> outcome
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -280,7 +284,9 @@ abstract class SessionOrchestrator(
|
|||||||
val parameters = runCatching {
|
val parameters = runCatching {
|
||||||
val element = Json.parseToJsonElement(toolCall.function.arguments)
|
val element = Json.parseToJsonElement(toolCall.function.arguments)
|
||||||
val jsonObject = element as? kotlinx.serialization.json.JsonObject ?: return@runCatching emptyMap()
|
val jsonObject = element as? kotlinx.serialization.json.JsonObject ?: return@runCatching emptyMap()
|
||||||
jsonObject.entries.associate { (k, v) -> k to v.toString().removeSurrounding("\"") as Any }
|
jsonObject.entries.associate { (k, v) ->
|
||||||
|
k to ((v as? kotlinx.serialization.json.JsonPrimitive)?.content ?: v.toString()) as Any
|
||||||
|
}
|
||||||
}.getOrElse { emptyMap() }
|
}.getOrElse { emptyMap() }
|
||||||
val request = ToolRequest(
|
val request = ToolRequest(
|
||||||
invocationId = invocationId,
|
invocationId = invocationId,
|
||||||
@@ -405,6 +411,26 @@ abstract class SessionOrchestrator(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private suspend fun emitToolArtifacts(
|
||||||
|
sessionId: SessionId,
|
||||||
|
stageId: StageId,
|
||||||
|
stageConfig: StageConfig,
|
||||||
|
) {
|
||||||
|
stageConfig.produces
|
||||||
|
.filter { !it.kind.llmEmitted }
|
||||||
|
.forEach { slot ->
|
||||||
|
emit(
|
||||||
|
sessionId,
|
||||||
|
ArtifactCreatedEvent(
|
||||||
|
artifactId = slot.name,
|
||||||
|
sessionId = sessionId,
|
||||||
|
stageId = stageId,
|
||||||
|
schemaVersion = 1,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun verifyProduces(
|
private fun verifyProduces(
|
||||||
sessionId: SessionId,
|
sessionId: SessionId,
|
||||||
stageId: StageId,
|
stageId: StageId,
|
||||||
@@ -412,7 +438,9 @@ abstract class SessionOrchestrator(
|
|||||||
): StageExecutionResult {
|
): StageExecutionResult {
|
||||||
if (stageConfig.produces.isEmpty()) return StageExecutionResult.Success(emptyList())
|
if (stageConfig.produces.isEmpty()) return StageExecutionResult.Success(emptyList())
|
||||||
val producedIds = stageConfig.produces.map { it.name }.toSet()
|
val producedIds = stageConfig.produces.map { it.name }.toSet()
|
||||||
val present = artifactRepository.getByIds(sessionId, producedIds).keys
|
val present = eventStore.read(sessionId)
|
||||||
|
.mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId }
|
||||||
|
.toSet()
|
||||||
val missing = producedIds - present
|
val missing = producedIds - present
|
||||||
if (missing.isEmpty()) return StageExecutionResult.Success(emptyList())
|
if (missing.isEmpty()) return StageExecutionResult.Success(emptyList())
|
||||||
val missingIds = missing.joinToString(", ") { it.value }
|
val missingIds = missing.joinToString(", ") { it.value }
|
||||||
|
|||||||
Reference in New Issue
Block a user