From 4be8f292ae44fd638d393205f8eac8c1924a7155 Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 1 Jul 2026 13:41:06 +0400 Subject: [PATCH] fix(kernel,server,workflow): per-stage retry budget, sane freestyle budget, tail /events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects surfaced by a live headless freestyle_planning run (all made a long execution workflow fail or stall in ways invisible to a REST operator): 1. retryCount was session-global (DefaultOrchestrationReducer): TransitionExecuted updated currentStageId but never reset retryCount, while shouldRetry gates on it per-stage. An early stage that burned its 3 retries starved every later stage of its own — the next stage's first *retryable* failure went terminal. Reset retryCount on stage entry; back-edge loops stay bounded by the separate refinementIterations counter. Regression test added. 2. Freestyle stages ran at StageConfig's 4096-token default (ExecutionPlanCompiler never set a budget) vs 16384-32768 for static role_pipeline stages. A tool-heavy stage truncated its context every round, evicting the file_read results it just gathered, so it re-read forever and never accumulated enough to write. Set 16384. 3. GET /events returned the OLDEST `limit` events (readFrom(0) then take): for a session past `limit` events the tail — including a pending ApprovalRequested — was invisible to a headless/REST poller, the exact caller POST /approve serves. Default now tails; explicit fromSeq still paginates forward. --- .../apps/server/routes/SessionRoutes.kt | 11 ++++++++--- .../DefaultOrchestrationReducer.kt | 8 +++++++- .../workflow/ExecutionPlanCompiler.kt | 9 +++++++++ .../test/kotlin/OrchestrationReducerTest.kt | 19 +++++++++++++++++++ 4 files changed, 43 insertions(+), 4 deletions(-) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt index b40ee3cb..ee0e5705 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt @@ -199,10 +199,11 @@ private fun Route.getEventsRoute(module: ServerModule) { get("/events") { val id = call.parameters["id"] ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") - val fromSeq = call.request.queryParameters["fromSeq"]?.toLongOrNull() ?: 0L + val fromSeqParam = call.request.queryParameters["fromSeq"]?.toLongOrNull() + val fromSeq = fromSeqParam ?: 0L val limit = call.request.queryParameters["limit"]?.toIntOrNull() ?: DEFAULT_EVENT_LIMIT val typeFilter = call.request.queryParameters["type"] - val rows = module.eventStore.readFrom(TypeId(id), fromSeq) + val mapped = module.eventStore.readFrom(TypeId(id), fromSeq) .asSequence() .map { stored -> val payloadJson = eventJson.encodeToJsonElement(EventPayload.serializer(), stored.payload) @@ -218,8 +219,12 @@ private fun Route.getEventsRoute(module: ServerModule) { ) } .filter { typeFilter == null || it.type == typeFilter } - .take(limit) .toList() + // No explicit fromSeq = a "what's happening now" tail: return the most recent `limit` events + // (a pending ApprovalRequested lives at the tail, so the oldest-200 default hid it from a + // headless/REST operator — the very caller POST /approve exists to serve). Explicit fromSeq + // keeps forward, ascending pagination from that point. + val rows = if (fromSeqParam == null) mapped.takeLast(limit) else mapped.take(limit) call.respond(rows) } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt index 325ff4be..b246cace 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt @@ -53,7 +53,13 @@ class DefaultOrchestrationReducer : OrchestrationReducer { pauseReason = null, ) - is TransitionExecutedEvent -> state.copy(currentStageId = p.to) + // Entering a new stage resets the retry budget: retryCount is per-stage (RetryAttemptedEvent + // is stage-scoped, maxAttempts is a per-stage cap), so without this reset the budget is shared + // session-wide — an early stage that exhausts its retries starves every later stage of its own, + // turning the next stage's first retryable failure into a terminal WorkflowFailed (live repro: + // implement burned 3/3, then review's retryable JSON-decode error got 0 retries). Back-edge + // loops stay bounded by the separate refinementIterations counter, so this can't loop forever. + is TransitionExecutedEvent -> state.copy(currentStageId = p.to, retryCount = 0) is RetryAttemptedEvent -> state.copy( status = OrchestrationStatus.RUNNING, diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt index 0e1177cd..482303e7 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt @@ -15,6 +15,14 @@ import com.fasterxml.jackson.module.kotlin.readValue private const val TERMINAL = "done" +// Freestyle plans don't specify a per-stage token budget, so without this every compiled stage +// inherits StageConfig's 4096 default — 1/8th of what the static role_pipeline execution stages +// use (16384–32768). At 4096 a tool-heavy stage (scaffold/implement) truncates its context every +// round, evicting the file_read results the model just gathered, so it re-reads forever and never +// accumulates enough to write. Match the static execution baseline. (gemma ctx is 32768, so this +// leaves ample headroom for completion.) +private const val DEFAULT_STAGE_TOKEN_BUDGET = 16384 + class ExecutionPlanCompiler( private val registry: ArtifactKindRegistry, // Names of every registered tool. A stage that references a tool the runtime can't resolve @@ -55,6 +63,7 @@ class ExecutionPlanCompiler( needs = s.needs.map { ArtifactId(it) }.toSet(), allowedTools = s.tools.toSet(), writeManifest = writeManifest, + tokenBudget = DEFAULT_STAGE_TOKEN_BUDGET, metadata = mapOf("promptInline" to s.prompt), ) } diff --git a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt index 4bbc02bc..a59cb78a 100644 --- a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt +++ b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt @@ -1,6 +1,7 @@ import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RetryAttemptedEvent +import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.events.WorkflowStartedEvent @@ -8,6 +9,7 @@ import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.orchestration.OrchestrationStatus 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.kernel.orchestration.DefaultOrchestrationReducer import com.correx.testing.fixtures.EventFixtures.stored import org.junit.jupiter.api.Assertions.assertEquals @@ -133,6 +135,23 @@ class OrchestrationReducerTest { assertNull(retrying1.failureReason) } + @Test + fun `TransitionExecutedEvent resets retryCount so each stage gets its own retry budget`() { + val exhausted = reducer.reduce( + state, + stored(payload = RetryAttemptedEvent(sessionId, stageId, 3, 3, "boom")), + ) + assertEquals(3, exhausted.retryCount) + + val nextStage = StageId("stage-2") + val advanced = reducer.reduce( + exhausted, + stored(payload = TransitionExecutedEvent(sessionId, stageId, nextStage, TransitionId("t1"))), + ) + assertEquals(nextStage, advanced.currentStageId) + assertEquals(0, advanced.retryCount) + } + @Test fun `unrelated event does nothing`() {