kami
68c56b6af6
A plan that failed grounding used to dead-end — every gate rejection in FreestyleDriver.lockAndRun was terminal, so a legitimate grounding catch (e.g. a stage declaring a PROJECT build with no manifest) left the run stuck with no retry. lockAndRun is now a gate loop: on a grounding rejection with retries left it re-runs the planning workflow from the architect stage (rerunArchitect), which emits a corrected plan, then re-gates. Other gate failures — and grounding once maxGroundingRetries is spent — stay terminal. - FreestyleDriver: gate loop + rerunArchitect/maxGroundingRetries seams; groundPlan returns findings (String?) instead of Boolean; post-grounding tail extracted to lockAndRunGrounded. - DefaultSessionOrchestrator.runFrom(startStage) + emitWorkflowStarted(startStage); run() delegates to it. Lets the re-run enter directly at architect. - buildGroundingFeedbackEntry (ContextFeedback) injects the already-recorded PlanGroundingEvaluatedEvent findings into the architect's L1 context on re-run; wired in SessionOrchestratorExecution. - Main: rerunArchitect lambda (rehydrate -> runFrom(architect) -> rehydrate). The architect stage-entry approval gate already reuses a prior APPROVED decision (alreadyApproved), so the re-run does not re-prompt the operator — added a FreestyleApprovalGateTest regression guard proving runFrom(architect) with a seeded approval emits no second request and runs straight through. Tests: FreestyleDriverTest retry-then-lock + exhaustion->reject(source=grounding); FreestyleApprovalGateTest reuse-approval guard. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
402 lines
19 KiB
Kotlin
402 lines
19 KiB
Kotlin
import com.correx.core.approvals.ApprovalOutcome
|
|
import com.correx.core.approvals.ApprovalProjector
|
|
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.DefaultApprovalEngine
|
|
import com.correx.core.approvals.model.ApprovalContext
|
|
import com.correx.core.approvals.model.ApprovalDecision
|
|
import com.correx.core.approvals.model.ApprovalScopeIdentity
|
|
import com.correx.core.artifacts.DefaultArtifactReducer
|
|
import com.correx.core.artifacts.kind.ConfigArtifactKind
|
|
import com.correx.core.artifacts.kind.JsonSchema
|
|
import com.correx.core.artifacts.kind.TypedArtifactSlot
|
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
|
import com.correx.core.events.events.OrchestrationPausedEvent
|
|
import com.correx.core.events.events.WorkflowCompletedEvent
|
|
import com.correx.core.events.execution.RetryPolicy
|
|
import com.correx.core.events.types.ArtifactId
|
|
import com.correx.core.events.types.SessionId
|
|
import com.correx.core.events.types.StageId
|
|
import com.correx.core.events.types.TransitionId
|
|
import com.correx.core.inference.InferenceRepository
|
|
import com.correx.core.inference.InferenceState
|
|
import com.correx.core.inference.ModelCapability
|
|
import com.correx.core.journal.DecisionJournalProjector
|
|
import com.correx.core.journal.DefaultDecisionJournalReducer
|
|
import com.correx.core.journal.DefaultDecisionJournalRepository
|
|
import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
|
|
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
|
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
|
import com.correx.core.kernel.orchestration.OrchestrationProjector
|
|
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
|
import com.correx.core.kernel.orchestration.OrchestratorEngines
|
|
import com.correx.core.kernel.orchestration.OrchestratorRepositories
|
|
import com.correx.core.kernel.retry.DefaultRetryCoordinator
|
|
import com.correx.core.risk.DefaultRiskAssessor
|
|
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.transitions.graph.StageConfig
|
|
import com.correx.core.transitions.graph.TransitionEdge
|
|
import com.correx.core.transitions.graph.WorkflowGraph
|
|
import com.correx.core.validation.pipeline.ValidationPipeline
|
|
import com.correx.infrastructure.persistence.InMemoryEventStore
|
|
import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository
|
|
import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore
|
|
import com.correx.testing.fixtures.cyclePolicyMissingValidator
|
|
import com.correx.testing.fixtures.context.ContextFixtures
|
|
import com.correx.testing.fixtures.inference.MockInferenceProvider
|
|
import com.correx.testing.fixtures.transitions.TransitionFixtures
|
|
import com.correx.testing.kernel.MockSessionEventReplayer
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.runBlocking
|
|
import kotlinx.coroutines.withTimeout
|
|
import kotlinx.coroutines.yield
|
|
import kotlinx.datetime.Clock
|
|
import org.junit.jupiter.api.Assertions.assertNotNull
|
|
import org.junit.jupiter.api.Assertions.assertTrue
|
|
import org.junit.jupiter.api.Test
|
|
|
|
/**
|
|
* Integration test for Task 4.2: the approval gate on stages flagged `requiresApproval`.
|
|
*
|
|
* Drives analyst→architect with a sequenced stub provider. Asserts that:
|
|
* 1. An [OrchestrationPausedEvent] is emitted before the architect runs, with the analyst
|
|
* summary in the paired [ApprovalRequestedEvent.preview].
|
|
* 2. Submitting an approval unblocks the run and the workflow completes with the architect's
|
|
* execution_plan artifact.
|
|
*/
|
|
class FreestyleApprovalGateTest {
|
|
|
|
private val eventStore = InMemoryEventStore()
|
|
private val sessionRepository = DefaultSessionRepository(MockSessionEventReplayer())
|
|
private val orchestrationRepository = OrchestrationRepository(
|
|
DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())),
|
|
)
|
|
private val inferenceRepository = InferenceRepository(
|
|
object : EventReplayer<InferenceState> {
|
|
override fun rebuild(sessionId: SessionId) = InferenceState()
|
|
},
|
|
)
|
|
private val approvalRepository = DefaultApprovalRepository(
|
|
DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())),
|
|
)
|
|
private val decisionJournalRepository = DefaultDecisionJournalRepository(
|
|
DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())),
|
|
)
|
|
|
|
private val repositories = OrchestratorRepositories(
|
|
eventStore = eventStore,
|
|
inferenceRepository = inferenceRepository,
|
|
orchestrationRepository = orchestrationRepository,
|
|
sessionRepository = sessionRepository,
|
|
artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()),
|
|
approvalRepository = approvalRepository,
|
|
)
|
|
|
|
private val analysisKind = ConfigArtifactKind(
|
|
id = "analysis",
|
|
schema = JsonSchema(type = "object", properties = emptyMap()),
|
|
llmEmitted = true,
|
|
)
|
|
|
|
private val executionPlanKind = ConfigArtifactKind(
|
|
id = "execution_plan",
|
|
schema = JsonSchema(type = "object", properties = emptyMap()),
|
|
llmEmitted = true,
|
|
)
|
|
|
|
private val analysisId = ArtifactId("analysis")
|
|
private val executionPlanId = ArtifactId("execution_plan")
|
|
private val analystStage = StageId("analyst")
|
|
private val architectStage = StageId("architect")
|
|
|
|
/** Graph mirrors freestyle_planning.toml: architect stage has requiresApproval metadata. */
|
|
private fun freestyleGraph() = WorkflowGraph(
|
|
id = "freestyle_planning",
|
|
stages = mapOf(
|
|
analystStage to StageConfig(
|
|
produces = listOf(TypedArtifactSlot(name = analysisId, kind = analysisKind)),
|
|
),
|
|
architectStage to StageConfig(
|
|
produces = listOf(TypedArtifactSlot(name = executionPlanId, kind = executionPlanKind)),
|
|
needs = setOf(analysisId),
|
|
metadata = mapOf("requiresApproval" to "true"),
|
|
),
|
|
),
|
|
transitions = setOf(
|
|
TransitionEdge(
|
|
id = TransitionId("analyst-to-architect"),
|
|
from = analystStage,
|
|
to = architectStage,
|
|
condition = { true },
|
|
),
|
|
TransitionEdge(
|
|
id = TransitionId("architect-to-done"),
|
|
from = architectStage,
|
|
to = StageId("done"),
|
|
condition = { true },
|
|
),
|
|
),
|
|
start = analystStage,
|
|
)
|
|
|
|
/** Seeds a prior architect-gate request + resolved decision (the retry/resume situation). */
|
|
private suspend fun seedArchitectDecision(sessionId: SessionId, outcome: ApprovalOutcome) {
|
|
val requestId = com.correx.core.events.types.ApprovalRequestId("seeded-req-${outcome.name}")
|
|
eventStore.append(
|
|
com.correx.testing.fixtures.EventFixtures.newEvent(
|
|
com.correx.core.events.types.EventId("seed-req-${outcome.name}"),
|
|
sessionId,
|
|
ApprovalRequestedEvent(
|
|
requestId = requestId,
|
|
tier = Tier.T2,
|
|
validationReportId = com.correx.core.events.types.ValidationReportId("vr"),
|
|
riskSummaryId = null,
|
|
sessionId = sessionId,
|
|
stageId = architectStage,
|
|
projectId = null,
|
|
toolName = null,
|
|
),
|
|
),
|
|
)
|
|
eventStore.append(
|
|
com.correx.testing.fixtures.EventFixtures.newEvent(
|
|
com.correx.core.events.types.EventId("seed-dec-${outcome.name}"),
|
|
sessionId,
|
|
com.correx.core.events.events.ApprovalDecisionResolvedEvent(
|
|
decisionId = com.correx.core.events.types.ApprovalDecisionId("seeded-dec-${outcome.name}"),
|
|
requestId = requestId,
|
|
outcome = outcome,
|
|
status = ApprovalStatus.COMPLETED,
|
|
tier = Tier.T2,
|
|
resolutionTimestamp = Clock.System.now(),
|
|
reason = null,
|
|
),
|
|
),
|
|
)
|
|
}
|
|
|
|
private val analysisSummary = """{"summary":"open questions: feasibility of X, timeline for Y"}"""
|
|
private val executionPlanJson = """{"plan":"step 1: do A, step 2: do B"}"""
|
|
|
|
private fun buildOrchestrator(): DefaultSessionOrchestrator {
|
|
// Sequenced provider: first call (analyst) returns the analysis JSON,
|
|
// second call (architect) returns the execution plan JSON.
|
|
var callCount = 0
|
|
val router = object : com.correx.core.inference.InferenceRouter {
|
|
override suspend fun route(
|
|
stageId: StageId,
|
|
requiredCapabilities: Set<ModelCapability>,
|
|
) = MockInferenceProvider(
|
|
fixedResponse = if (callCount++ == 0) analysisSummary else executionPlanJson,
|
|
)
|
|
}
|
|
|
|
return DefaultSessionOrchestrator(
|
|
repositories = repositories,
|
|
engines = OrchestratorEngines(
|
|
transitionResolver = TransitionFixtures.simpleResolver(),
|
|
contextPackBuilder = ContextFixtures.simpleBuilder(),
|
|
inferenceRouter = router,
|
|
validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())),
|
|
approvalEngine = DefaultApprovalEngine(),
|
|
riskAssessor = DefaultRiskAssessor(),
|
|
),
|
|
retryCoordinator = DefaultRetryCoordinator(eventStore),
|
|
artifactStore = NoopArtifactStore(),
|
|
decisionJournalRepository = decisionJournalRepository,
|
|
)
|
|
}
|
|
|
|
@Test
|
|
fun `pause event is emitted with analyst summary before architect runs`(): Unit = runBlocking {
|
|
val sessionId = SessionId("freestyle-gate-1")
|
|
val orchestrator = buildOrchestrator()
|
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
|
|
|
|
val runJob = launch { orchestrator.run(sessionId, freestyleGraph(), config) }
|
|
|
|
// Wait for the pause event before architect
|
|
withTimeout(5_000) {
|
|
while (eventStore.read(sessionId).none { it.payload is OrchestrationPausedEvent }) {
|
|
yield()
|
|
}
|
|
}
|
|
|
|
val events = eventStore.read(sessionId)
|
|
val pause = events.firstNotNullOfOrNull { it.payload as? OrchestrationPausedEvent }
|
|
assertNotNull(pause, "Expected OrchestrationPausedEvent")
|
|
assertTrue(pause!!.stageId == architectStage, "Pause should be for architect stage, got ${pause.stageId}")
|
|
assertTrue(pause.reason == "APPROVAL_PENDING", "Expected reason APPROVAL_PENDING, got ${pause.reason}")
|
|
|
|
val approvalRequest = events.firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent }
|
|
assertNotNull(approvalRequest, "Expected ApprovalRequestedEvent")
|
|
assertTrue(
|
|
approvalRequest!!.preview?.contains("open questions") == true ||
|
|
approvalRequest.preview?.contains("summary") == true,
|
|
"Approval preview should carry analyst summary, got: ${approvalRequest.preview}",
|
|
)
|
|
|
|
runJob.cancel()
|
|
runJob.join()
|
|
}
|
|
|
|
@Test
|
|
fun `a prior REJECTED decision does not satisfy the gate — it re-prompts, not runs unapproved`(): Unit =
|
|
runBlocking {
|
|
val sessionId = SessionId("freestyle-gate-reject")
|
|
// Pre-seed the log as if the architect gate had already been prompted and REJECTED (the
|
|
// retry/resume situation): a matching request + a REJECTED decision for it. The pre-fix
|
|
// predicate matched any decision for the request and skipped the gate, running unapproved.
|
|
val requestId = com.correx.core.events.types.ApprovalRequestId("seeded-req")
|
|
eventStore.append(
|
|
com.correx.testing.fixtures.EventFixtures.newEvent(
|
|
com.correx.core.events.types.EventId("seed-req"),
|
|
sessionId,
|
|
ApprovalRequestedEvent(
|
|
requestId = requestId,
|
|
tier = Tier.T2,
|
|
validationReportId = com.correx.core.events.types.ValidationReportId("vr"),
|
|
riskSummaryId = null,
|
|
sessionId = sessionId,
|
|
stageId = architectStage,
|
|
projectId = null,
|
|
toolName = null,
|
|
),
|
|
),
|
|
)
|
|
eventStore.append(
|
|
com.correx.testing.fixtures.EventFixtures.newEvent(
|
|
com.correx.core.events.types.EventId("seed-dec"),
|
|
sessionId,
|
|
com.correx.core.events.events.ApprovalDecisionResolvedEvent(
|
|
decisionId = com.correx.core.events.types.ApprovalDecisionId("seeded-dec"),
|
|
requestId = requestId,
|
|
outcome = ApprovalOutcome.REJECTED,
|
|
status = ApprovalStatus.COMPLETED,
|
|
tier = Tier.T2,
|
|
resolutionTimestamp = Clock.System.now(),
|
|
reason = "operator said no",
|
|
),
|
|
),
|
|
)
|
|
|
|
val orchestrator = buildOrchestrator()
|
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
|
|
val runJob = launch { orchestrator.run(sessionId, freestyleGraph(), config) }
|
|
|
|
// The gate must re-prompt: a SECOND architect approval request appears (the seeded one is
|
|
// the first). With the bug the seeded REJECTED decision satisfied the gate and no new
|
|
// request was emitted — the architect ran and the workflow completed unapproved.
|
|
withTimeout(5_000) {
|
|
while (
|
|
eventStore.read(sessionId)
|
|
.mapNotNull { it.payload as? ApprovalRequestedEvent }
|
|
.count { it.stageId == architectStage && it.toolName == null } < 2
|
|
) {
|
|
yield()
|
|
}
|
|
}
|
|
|
|
assertTrue(
|
|
eventStore.read(sessionId).none { it.payload is WorkflowCompletedEvent },
|
|
"Workflow must not complete off a REJECTED decision",
|
|
)
|
|
runJob.cancel()
|
|
runJob.join()
|
|
}
|
|
|
|
@Test
|
|
fun `runFrom architect reuses a prior APPROVED decision and does not re-prompt on grounding retry`(): Unit =
|
|
runBlocking {
|
|
// Return-to-architect loop (FreestyleDriver.rerunArchitect): a grounding-rejected plan re-runs
|
|
// the architect via runFrom(architect). The operator already approved the analyst→architect
|
|
// gate on the first pass, so the re-run must NOT re-park — the seeded APPROVED decision below
|
|
// stands in for that first approval; the gate must reuse it.
|
|
val sessionId = SessionId("freestyle-gate-rerun")
|
|
seedArchitectDecision(sessionId, ApprovalOutcome.APPROVED)
|
|
|
|
// Provider always yields the execution plan — we enter at architect, not analyst.
|
|
val router = object : com.correx.core.inference.InferenceRouter {
|
|
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>) =
|
|
MockInferenceProvider(fixedResponse = executionPlanJson)
|
|
}
|
|
val orchestrator = DefaultSessionOrchestrator(
|
|
repositories = repositories,
|
|
engines = OrchestratorEngines(
|
|
transitionResolver = TransitionFixtures.simpleResolver(),
|
|
contextPackBuilder = ContextFixtures.simpleBuilder(),
|
|
inferenceRouter = router,
|
|
validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())),
|
|
approvalEngine = DefaultApprovalEngine(),
|
|
riskAssessor = DefaultRiskAssessor(),
|
|
),
|
|
retryCoordinator = DefaultRetryCoordinator(eventStore),
|
|
artifactStore = NoopArtifactStore(),
|
|
decisionJournalRepository = decisionJournalRepository,
|
|
)
|
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
|
|
|
|
orchestrator.runFrom(sessionId, freestyleGraph(), config, architectStage)
|
|
|
|
val architectRequests = eventStore.read(sessionId)
|
|
.mapNotNull { it.payload as? ApprovalRequestedEvent }
|
|
.count { it.stageId == architectStage && it.toolName == null }
|
|
assertTrue(
|
|
architectRequests == 1,
|
|
"gate must reuse the prior approval, not re-prompt: got $architectRequests",
|
|
)
|
|
assertNotNull(
|
|
eventStore.read(sessionId).find { it.payload is WorkflowCompletedEvent },
|
|
"architect should run straight through on the reused approval",
|
|
)
|
|
}
|
|
|
|
@Test
|
|
fun `approval resumes workflow and architect produces execution plan`(): Unit = runBlocking {
|
|
val sessionId = SessionId("freestyle-gate-2")
|
|
val orchestrator = buildOrchestrator()
|
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
|
|
|
|
val runJob = launch { orchestrator.run(sessionId, freestyleGraph(), config) }
|
|
|
|
// Wait for the approval request
|
|
val requestId = withTimeout(5_000) {
|
|
var id: com.correx.core.events.types.ApprovalRequestId? = null
|
|
while (id == null) {
|
|
id = eventStore.read(sessionId)
|
|
.firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent }
|
|
?.requestId
|
|
if (id == null) yield()
|
|
}
|
|
id
|
|
}
|
|
|
|
val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = architectStage, projectId = null)
|
|
val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT)
|
|
val decision = ApprovalDecision(
|
|
id = null,
|
|
requestId = requestId,
|
|
outcome = ApprovalOutcome.APPROVED,
|
|
state = ApprovalStatus.COMPLETED,
|
|
tier = Tier.T2,
|
|
contextSnapshot = context,
|
|
resolutionTimestamp = Clock.System.now(),
|
|
reason = null,
|
|
)
|
|
orchestrator.submitApprovalDecision(requestId, decision)
|
|
|
|
runJob.join()
|
|
|
|
val events = eventStore.read(sessionId)
|
|
assertNotNull(
|
|
events.find { it.payload is WorkflowCompletedEvent },
|
|
"Workflow should complete after approval",
|
|
)
|
|
}
|
|
}
|