fix(kernel): require tool activity for produce-less stage completion

A stage declaring no `produces` passed verifyProduces vacuously, letting an
agent stage_complete with zero work. Gate on allowedTools: a produce-less stage
with tools available must have >=1 ToolInvocationRequestedEvent scoped to its
stageId, else Failure(retryable). Produce-less + no tools is exempt (only
stage_complete is ever offered), so pure-reasoning stages aren't bricked.
This commit is contained in:
2026-06-02 19:56:34 +04:00
parent 1dc7eae8a1
commit d1dc9e2f5c
2 changed files with 190 additions and 1 deletions
@@ -705,7 +705,21 @@ abstract class SessionOrchestrator(
stageId: StageId, stageId: StageId,
stageConfig: StageConfig, stageConfig: StageConfig,
): StageExecutionResult { ): 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 producedIds = stageConfig.produces.map { it.name }.toSet()
val present = eventStore.read(sessionId) val present = eventStore.read(sessionId)
.mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId } .mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId }
@@ -4,10 +4,13 @@ import com.correx.core.approvals.ApprovalStatus
import com.correx.core.approvals.DefaultApprovalReducer import com.correx.core.approvals.DefaultApprovalReducer
import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.approvals.DefaultApprovalRepository
import com.correx.core.approvals.Tier 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.domain.DefaultApprovalEngine
import com.correx.core.approvals.model.ApprovalContext import com.correx.core.approvals.model.ApprovalContext
import com.correx.core.approvals.model.ApprovalDecision 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.ApprovalScopeIdentity
import com.correx.core.approvals.model.DomainApprovalRequest
import com.correx.core.artifacts.DefaultArtifactReducer import com.correx.core.artifacts.DefaultArtifactReducer
import com.correx.core.artifactstore.ArtifactStore import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.context.builder.DefaultContextPackBuilder 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.InferenceStartedEvent
import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent 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.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent 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.execution.RetryPolicy
import com.correx.core.events.types.ArtifactId 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.SessionId
import com.correx.core.events.types.StageId 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.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.InferenceRepository
import com.correx.core.inference.InferenceRequest
import com.correx.core.inference.InferenceResponse
import com.correx.core.inference.InferenceRouter import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.InferenceState import com.correx.core.inference.InferenceState
import com.correx.core.inference.ModelCapability 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.DefaultOrchestrationReducer
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
import com.correx.core.kernel.orchestration.OrchestrationConfig 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.DefaultSessionRepository
import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import com.correx.core.sessions.projections.replay.EventReplayer 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.evaluation.PromptResolver
import com.correx.core.transitions.graph.StageConfig import com.correx.core.transitions.graph.StageConfig
import com.correx.core.transitions.graph.TransitionEdge import com.correx.core.transitions.graph.TransitionEdge
@@ -68,10 +88,12 @@ import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.yield import kotlinx.coroutines.yield
import kotlinx.datetime.Clock import kotlinx.datetime.Clock
import kotlinx.datetime.Instant
import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
import java.util.concurrent.atomic.AtomicInteger
@Suppress("UnusedPrivateProperty") @Suppress("UnusedPrivateProperty")
class SessionOrchestratorIntegrationTest { class SessionOrchestratorIntegrationTest {
@@ -436,6 +458,159 @@ class SessionOrchestratorIntegrationTest {
assertEquals(7, state.retryPolicy?.maxAttempts) 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<Token> = 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<ApprovalGrant>,
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<ModelCapability>,
): 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 { private class RecordingArtifactStore : ArtifactStore {