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 cdd294c6..8d6193e6 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 @@ -705,7 +705,21 @@ abstract class SessionOrchestrator( stageId: StageId, stageConfig: StageConfig, ): StageExecutionResult { - if (stageConfig.produces.isEmpty()) return StageExecutionResult.Success(emptyList()) + if (stageConfig.produces.isEmpty()) { + if (stageConfig.allowedTools.isEmpty()) return StageExecutionResult.Success(emptyList()) + val ranTool = eventStore.read(sessionId).any { + (it.payload as? ToolInvocationRequestedEvent)?.stageId == stageId + } + if (ranTool) return StageExecutionResult.Success(emptyList()) + log.error( + "[Orchestrator] stage declared no artifacts and ran no tools " + + "session=${sessionId.value} stage=${stageId.value}", + ) + return StageExecutionResult.Failure( + "stage ${stageId.value} declared no artifacts and ran no tools", + retryable = true, + ) + } val producedIds = stageConfig.produces.map { it.name }.toSet() val present = eventStore.read(sessionId) .mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId } diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index 014d7048..7dcc2289 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -4,10 +4,13 @@ import com.correx.core.approvals.ApprovalStatus import com.correx.core.approvals.DefaultApprovalReducer import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.approvals.Tier +import com.correx.core.approvals.domain.ApprovalEngine import com.correx.core.approvals.domain.DefaultApprovalEngine import com.correx.core.approvals.model.ApprovalContext import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.approvals.model.ApprovalGrant import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.approvals.model.DomainApprovalRequest import com.correx.core.artifacts.DefaultArtifactReducer import com.correx.core.artifactstore.ArtifactStore import com.correx.core.context.builder.DefaultContextPackBuilder @@ -21,18 +24,33 @@ import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.events.ToolRequest import com.correx.core.events.execution.RetryPolicy import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ProviderId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +import com.correx.core.events.types.ToolInvocationId import com.correx.core.events.types.TransitionId +import com.correx.core.inference.CapabilityScore +import com.correx.core.inference.FinishReason +import com.correx.core.inference.InferenceProvider import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceRequest +import com.correx.core.inference.InferenceResponse import com.correx.core.inference.InferenceRouter import com.correx.core.inference.InferenceState import com.correx.core.inference.ModelCapability +import com.correx.core.inference.ProviderHealth +import com.correx.core.inference.Token +import com.correx.core.inference.TokenUsage +import com.correx.core.inference.Tokenizer +import com.correx.core.inference.ToolCallFunction +import com.correx.core.inference.ToolCallRequest import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator import com.correx.core.kernel.orchestration.OrchestrationConfig @@ -46,6 +64,8 @@ import com.correx.core.sessions.ApprovalMode import com.correx.core.sessions.DefaultSessionRepository import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.core.tools.contract.ToolExecutor +import com.correx.core.tools.contract.ToolResult import com.correx.core.transitions.evaluation.PromptResolver import com.correx.core.transitions.graph.StageConfig import com.correx.core.transitions.graph.TransitionEdge @@ -68,10 +88,12 @@ import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout import kotlinx.coroutines.yield import kotlinx.datetime.Clock +import kotlinx.datetime.Instant import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test +import java.util.concurrent.atomic.AtomicInteger @Suppress("UnusedPrivateProperty") class SessionOrchestratorIntegrationTest { @@ -436,6 +458,159 @@ class SessionOrchestratorIntegrationTest { assertEquals(7, state.retryPolicy?.maxAttempts) } + @Test + fun `stage with allowedTools but no tool invocation fails liveness check`(): Unit = runBlocking { + val sessionId = SessionId("s-liveness-neg") + // Stage offers a tool but the stubbed LLM immediately returns Stop (stage_complete + // shortcut) without ever calling it. The new gate must reject this vacuous pass. + val livenessGraph = WorkflowGraph( + id = "liveness-test", + stages = mapOf( + StageId("A") to StageConfig(allowedTools = setOf("my_tool")), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 2, backoffMs = 0)) + // Default MockInferenceProvider returns FinishReason.Stop with no tool calls. + val livenessOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + ) + + livenessOrchestrator.run(sessionId, livenessGraph, config) + + val events = eventStore.read(sessionId) + val failed = events.find { it.payload is WorkflowFailedEvent } + assertNotNull(failed, "expected WorkflowFailedEvent when tool was offered but never called") + val payload = failed?.payload as WorkflowFailedEvent + assertTrue(payload.retryExhausted, "expected retryExhausted=true") + assertTrue( + payload.reason.contains("declared no artifacts and ran no tools"), + "unexpected failure reason: ${payload.reason}", + ) + } + + @Test + fun `stage with allowedTools passes liveness check after tool is invoked`(): Unit = runBlocking { + val sessionId = SessionId("s-liveness-pos") + val livenessGraph = WorkflowGraph( + id = "liveness-pos-test", + stages = mapOf( + StageId("A") to StageConfig(allowedTools = setOf("my_tool")), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + // Round 0: return a tool call for "my_tool"; round 1: call stage_complete to finish. + val callCount = AtomicInteger(0) + val sequencedProvider = object : InferenceProvider { + override val id: ProviderId = ProviderId("seq-mock") + override val name: String = "Sequenced Mock" + override val tokenizer: Tokenizer = object : Tokenizer { + override suspend fun tokenize(text: String): List = emptyList() + override suspend fun countTokens(text: String) = text.length / 4 + } + override suspend fun infer(request: InferenceRequest): InferenceResponse { + val round = callCount.getAndIncrement() + return if (round == 0) { + InferenceResponse( + requestId = request.requestId, + text = "", + finishReason = FinishReason.ToolCall, + tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), + latencyMs = 0, + toolCalls = listOf( + ToolCallRequest( + id = "call-1", + function = ToolCallFunction(name = "my_tool", arguments = "{}"), + ), + ), + ) + } else { + InferenceResponse( + requestId = request.requestId, + text = "", + finishReason = FinishReason.ToolCall, + tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), + latencyMs = 0, + toolCalls = listOf( + ToolCallRequest( + id = "call-2", + function = ToolCallFunction(name = "stage_complete", arguments = "{}"), + ), + ), + ) + } + } + override suspend fun healthCheck() = ProviderHealth.Healthy + override fun capabilities() = setOf(CapabilityScore(ModelCapability.General, 1.0)) + } + + // Auto-approving engine so T2 tool calls are not blocked in tests. + val autoApproveEngine = object : ApprovalEngine { + override fun evaluate( + request: DomainApprovalRequest, + context: ApprovalContext, + grants: List, + now: Instant, + ) = ApprovalDecision( + id = null, + requestId = request.id, + outcome = ApprovalOutcome.AUTO_APPROVED, + state = ApprovalStatus.COMPLETED, + tier = request.tier, + contextSnapshot = context, + resolutionTimestamp = now, + reason = "test-auto-approve", + ) + } + + // Tool executor that succeeds for any tool call. + val succeedingExecutor = object : ToolExecutor { + override suspend fun execute(request: ToolRequest) = ToolResult.Success( + invocationId = request.invocationId, + output = "ok", + ) + } + + val livenessOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines.copy( + inferenceRouter = object : InferenceRouter { + override suspend fun route( + stageId: StageId, + requiredCapabilities: Set, + ): InferenceProvider = sequencedProvider + }, + approvalEngine = autoApproveEngine, + toolExecutor = succeedingExecutor, + ), + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + ) + + livenessOrchestrator.run(sessionId, livenessGraph, config) + + val events = eventStore.read(sessionId) + assertNotNull( + events.find { it.payload is WorkflowCompletedEvent }, + "expected WorkflowCompletedEvent when tool was invoked before stage_complete", + ) + assertTrue( + events.any { (it.payload as? ToolInvocationRequestedEvent)?.stageId == StageId("A") }, + "expected ToolInvocationRequestedEvent for stage A", + ) + } + } private class RecordingArtifactStore : ArtifactStore {