feat: shell execution correctness + artifact_field_equals condition

- Add field to ProcessResultArtifact ("success"/"failed") - Materialize ToolResult.Failure as a failed ProcessResult artifact in the orchestrator, with artifact lifecycle events emitted - Non-recoverable ToolResult.Failure now fails the stage ("FATAL:") instead of being silently fed back as LLM context - Fix enterStage to return Continue on non-retryable failures, enabling back-edge transitions (fix/retry loops) - Add ArtifactFieldEquals transition condition with flat and nested dot-notation field access + FieldOperator enum (eq, neq, gt, gte, lt, lte) - Wire artifact_field_equals through ConditionFactory and TomlWorkflowLoader (condition_field, condition_operator) - Cache ProcessResult artifact content for evaluation context - Add 16 tests covering all condition operators and error cases - DefaultTransitionResolver catches condition evaluation errors and returns Blocked instead of crashing
2026-05-26 17:28:51 +04:00
parent 9734eec63c
commit 3b24c7e91a
11 changed files with 304 additions and 10 deletions
@@ -53,6 +53,7 @@ class DefaultSessionOrchestrator(
        }
    }

+    @Suppress("LongMethod")
    private tailrec suspend fun step(ctx: EnrichedExecutionContext): WorkflowResult {
        log.debug(
            "[Orchestrator] step session={} stage={} stageCount={}",
@@ -82,7 +83,14 @@ class DefaultSessionOrchestrator(
            validated.associateWith { ArtifactState(phase = ArtifactLifecyclePhase.VALIDATED) }
        }

-        val decision = resolveTransition(enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts)
+        val artifactContent: Map<ArtifactId, String> = targetIds.mapNotNull { id ->
+            val cacheKey = "${enriched.sessionId.value}:${id.value}"
+            artifactContentCache[cacheKey]?.let { content -> id to content }
+        }.toMap()
+
+        val decision = resolveTransition(
+            enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent,
+        )
        log.debug(
            "[Orchestrator] transition session={} stage={} decision={}",
            enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName,
@@ -174,6 +182,11 @@ class DefaultSessionOrchestrator(
                        "[Orchestrator] stage failed session=${ctx.sessionId.value} " +
                                "stage=${stageId.value} reason=${result.reason} retryable=${result.retryable}",
                    )
+                    if (!result.retryable) {
+                        // Non-retryable: let the transition resolver handle the outcome
+                        // (e.g., back-edge via artifact_field_equals on failure)
+                        return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1))
+                    }
                    val refreshedState = orchestrationRepository.getState(ctx.sessionId)
                    val shouldRetry = retryCoordinator.shouldRetry(
                        sessionId = ctx.sessionId,
@@ -188,7 +201,7 @@ class DefaultSessionOrchestrator(
                                ctx.sessionId,
                                stageId,
                                result.reason,
-                                retryExhausted = result.retryable,
+                                retryExhausted = true,
                            ),
                        )
                    }
@@ -7,6 +7,7 @@ import com.correx.core.approvals.model.ApprovalDecision
 import com.correx.core.approvals.model.DomainApprovalRequest
 import com.correx.core.artifacts.ArtifactState
 import com.correx.core.artifacts.kind.JsonSchema
+import com.correx.core.artifacts.kind.ProcessResultArtifact
 import com.correx.core.artifactstore.ArtifactStore
 import com.correx.core.context.builder.ContextPackBuilder
 import com.correx.core.context.model.ContextEntry
@@ -117,6 +118,11 @@ abstract class SessionOrchestrator(
    internal val pendingApprovals: ConcurrentHashMap<ApprovalRequestId, CompletableDeferred<ApprovalDecision>> =
        ConcurrentHashMap()

+    /** Cache mapping "sessionId:artifactId" to JSON-serialized artifact payload content.
+     *  Populated when ProcessResult artifacts are stored on tool failure.
+     *  Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */
+    protected val artifactContentCache: ConcurrentHashMap<String, String> = ConcurrentHashMap()
+
    abstract suspend fun run(
        sessionId: SessionId,
        graph: WorkflowGraph,
@@ -234,9 +240,21 @@ abstract class SessionOrchestrator(
            toolRounds < MAX_TOOL_ROUNDS
        ) {
            val toolEntries = dispatchToolCalls(sessionId, stageId, inferenceResult.response.toolCalls)
-            val firstError = toolEntries.firstOrNull { it.content.startsWith("ERROR:") }
-            if (firstError != null) {
-                return StageExecutionResult.Failure(firstError.content.removePrefix("ERROR: "), retryable = true)
+            val fatalEntry = toolEntries.firstOrNull { it.content.startsWith("FATAL:") }
+            if (fatalEntry != null) {
+                storeFailedProcessResult(sessionId, stageId, stageConfig,
+                    inferenceResult.response.toolCalls, fatalEntry)
+                return StageExecutionResult.Failure(
+                    fatalEntry.content.removePrefix("FATAL: "),
+                    retryable = false,
+                )
+            }
+            val errorEntry = toolEntries.firstOrNull { it.content.startsWith("ERROR:") }
+            if (errorEntry != null) {
+                return StageExecutionResult.Failure(
+                    errorEntry.content.removePrefix("ERROR: "),
+                    retryable = true,
+                )
            }
            val allEntries = currentContext.layers.values.flatten() + toolEntries
            currentContext = contextPackBuilder.build(
@@ -276,6 +294,7 @@ abstract class SessionOrchestrator(
    ): String = "[SessionOrchestrator] stage=${stageId.value}: " +
            "failed to load prompt '$path': ${throwable.message}"

+    @Suppress("CyclomaticComplexMethod")
    private suspend fun dispatchToolCalls(
        sessionId: SessionId,
        stageId: StageId,
@@ -383,7 +402,10 @@ abstract class SessionOrchestrator(
            )
            val resultContent = when (result) {
                is ToolResult.Success -> result.output
-                is ToolResult.Failure -> "ERROR: ${result.reason}"
+                is ToolResult.Failure -> {
+                    if (!result.recoverable) "FATAL: ${result.reason}"
+                    else "ERROR: ${result.reason}"
+                }
            }
            val resultEntry = ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
@@ -455,6 +477,34 @@ abstract class SessionOrchestrator(
        )
    }

+    private suspend fun storeFailedProcessResult(
+        sessionId: SessionId,
+        stageId: StageId,
+        stageConfig: StageConfig,
+        toolCalls: List<ToolCallRequest>,
+        fatalEntry: ContextEntry,
+    ) {
+        stageConfig.produces.filter { it.kind.id == "process_result" }.forEach { slot ->
+            val cmd = toolCalls.firstOrNull()?.function?.arguments ?: ""
+            val reason = fatalEntry.content.removePrefix("FATAL: ")
+            val processResult = ProcessResultArtifact(
+                status = "failed",
+                command = cmd,
+                exitCode = -1,
+                stdout = "",
+                stderr = reason,
+                durationMs = 0,
+            )
+            val jsonContent = Json.encodeToString(ProcessResultArtifact.serializer(), processResult)
+            artifactStore.put(jsonContent.toByteArray())
+            val cacheKey = "${sessionId.value}:${slot.name.value}"
+            artifactContentCache[cacheKey] = jsonContent
+            emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1))
+            emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId))
+            emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId))
+        }
+    }
+
    internal open suspend fun mapValidationOutcome(
        sessionId: SessionId,
        stageId: StageId,
@@ -568,8 +618,14 @@ abstract class SessionOrchestrator(
        sessionId: SessionId,
        currentStageId: StageId,
        artifacts: Map<ArtifactId, ArtifactState> = emptyMap(),
+        artifactContent: Map<ArtifactId, String> = emptyMap(),
    ): TransitionDecision {
-        val ctx = EvaluationContext(sessionId, currentStageId, artifacts = artifacts)
+        val ctx = EvaluationContext(
+            sessionId = sessionId,
+            currentStage = currentStageId,
+            artifacts = artifacts,
+            artifactContent = artifactContent,
+        )
        return transitionResolver.resolve(graph, ctx)
    }