feat: approval gates block indefinitely instead of auto-rejecting on timeout

Remove the wall-clock approval timeout that auto-rejected pending approvals. The timeout modeled machine latency, but approvals are human latency (unbounded); it also created an intent/outcome divergence where a human approving at the same instant the timeout fired was silently overridden by the auto-reject. Approvals now block until the operator decides; the resolved-request dedup still guards against double-submit. Add ServerMessage.ApprovalResolved and map ApprovalDecisionResolvedEvent to it in DomainEventMapper so clients are notified when any approval is resolved (by anyone) — letting a second connected client clear its prompt. This is the event-sourced replacement for the removed timeout notification path. Drop the now-dead ApprovalConfig (timeout_ms) and its loader/test/sample-config references. ApprovalStatus.TIMED_OUT is retained for replay of historical events.
2026-05-30 21:48:23 +04:00
parent 780a00229e
commit d3ce310100
11 changed files with 106 additions and 98 deletions
@@ -160,7 +160,6 @@ fun main() {
        orchestrationRepository = repositories.orchestrationRepository,
        approvalRepository = repositories.approvalRepository,
        toolRegistry = toolRegistry,
-        approvalConfig = correxConfig.approval,
    )
    module.start()
    log.info("==============================")
@@ -7,7 +7,6 @@ import com.correx.core.approvals.ApprovalProjector
 import com.correx.core.approvals.DefaultApprovalReducer
 import com.correx.core.approvals.DefaultApprovalRepository
 import com.correx.core.artifactstore.ArtifactStore
-import com.correx.core.config.ApprovalConfig
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.EventMetadata
 import com.correx.core.events.events.NewEvent
@@ -49,17 +48,14 @@ class ServerModule(
    val orchestrationRepository: OrchestrationRepository,
    val approvalRepository: DefaultApprovalRepository,
    val toolRegistry: ToolRegistry,
-    approvalConfig: ApprovalConfig = ApprovalConfig(),
-    // Long-lived scope owned by the module — backs the ApprovalCoordinator timers
-    // and event-store subscription. SupervisorJob so one failure doesn't kill the
-    // whole module; Dispatchers.Default since the work is non-blocking and CPU-light.
+    // Long-lived scope owned by the module — backs the event-store subscription.
+    // SupervisorJob so one failure doesn't kill the whole module;
+    // Dispatchers.Default since the work is non-blocking and CPU-light.
    val moduleScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
    approvalCoordinator: ApprovalCoordinator? = null,
 ) {
    val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator(
        orchestrator = orchestrator,
-        config = approvalConfig,
-        scope = moduleScope,
    )

    private var subscriptionJob: Job? = null
@@ -5,39 +5,23 @@ import com.correx.apps.server.protocol.ProtocolSerializer
 import com.correx.apps.server.protocol.RiskSummaryDto
 import com.correx.apps.server.protocol.ServerMessage
 import com.correx.apps.server.protocol.toDto
-import com.correx.core.approvals.ApprovalOutcome
-import com.correx.core.approvals.ApprovalStatus
 import com.correx.core.approvals.Tier
-import com.correx.core.approvals.model.ApprovalContext
-import com.correx.core.approvals.model.ApprovalScopeIdentity
-import com.correx.core.config.ApprovalConfig
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.risk.RiskAction
 import com.correx.core.events.types.ApprovalRequestId
 import com.correx.core.events.types.SessionId
-import com.correx.core.events.types.StageId
 import com.correx.core.kernel.orchestration.ApprovalGateway
-import com.correx.core.sessions.ApprovalMode
 import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.websocket.Frame
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
-import kotlinx.datetime.Clock
 import java.util.concurrent.ConcurrentHashMap
-import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision

 open class ApprovalCoordinator(
    private val orchestrator: ApprovalGateway,
-    private val config: ApprovalConfig,
-    val scope: CoroutineScope,
 ) {
    private val sessionClients: ConcurrentHashMap<SessionId, MutableSet<DefaultWebSocketServerSession>> =
        ConcurrentHashMap()
    private val globalClients: MutableSet<DefaultWebSocketServerSession> = ConcurrentHashMap.newKeySet()
    private val resolved: ConcurrentHashMap<ApprovalRequestId, Boolean> = ConcurrentHashMap()
-    private val timeoutJobs: ConcurrentHashMap<ApprovalRequestId, Job> = ConcurrentHashMap()
    private val requestSessions: ConcurrentHashMap<ApprovalRequestId, SessionId> = ConcurrentHashMap()

    fun registerClient(sessionId: SessionId, session: DefaultWebSocketServerSession) {
@@ -74,7 +58,6 @@ open class ApprovalCoordinator(
            sessionSequence = 0L,
        )
        broadcast(event.sessionId, msg)
-        scheduleTimeout(event.requestId, event.sessionId, event.stageId, event.tier)
    }

    fun lookupSession(requestId: ApprovalRequestId): SessionId? = requestSessions[requestId]
@@ -91,12 +74,11 @@ open class ApprovalCoordinator(
    open suspend fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? {
        if (resolved.putIfAbsent(msg.requestId, true) != null) {
            return ServerMessage.ProtocolError(
-                message = "Approval request ${msg.requestId.value} already resolved",
+                message = "Approval request ${msg.requestId.value} was already resolved",
                sequence = null,
                sessionSequence = null,
            )
        }
-        timeoutJobs.remove(msg.requestId)?.cancel()
        requestSessions.remove(msg.requestId)
        val domain = msg.toDomain(sessionId, null, Tier.T2)
        return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) }
@@ -120,30 +102,4 @@ open class ApprovalCoordinator(
            runCatching { client.send(Frame.Text(encoded)) }
        }
    }
-
-    private fun scheduleTimeout(
-        requestId: ApprovalRequestId,
-        sessionId: SessionId,
-        stageId: StageId?,
-        tier: Tier,
-    ) {
-        val job = scope.launch {
-            delay(config.timeoutMs)
-            if (resolved.putIfAbsent(requestId, true) != null) return@launch
-            val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = stageId, projectId = null)
-            val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT)
-            val decision = DomainApprovalDecision(
-                id = null,
-                requestId = requestId,
-                outcome = ApprovalOutcome.REJECTED,
-                state = ApprovalStatus.COMPLETED,
-                tier = tier,
-                contextSnapshot = context,
-                resolutionTimestamp = Clock.System.now(),
-                reason = "Approval timed out after ${config.timeoutMs}ms",
-            )
-            runCatching { orchestrator.submitApprovalDecision(requestId, decision) }
-        }
-        timeoutJobs[requestId] = job
-    }
 }
@@ -5,6 +5,7 @@ import com.correx.apps.server.protocol.RiskSummaryDto
 import com.correx.apps.server.protocol.ServerMessage
 import com.correx.apps.server.protocol.toDto
 import com.correx.core.artifactstore.ArtifactStore
+import com.correx.core.events.events.ApprovalDecisionResolvedEvent
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.ArtifactCreatedEvent
 import com.correx.core.events.events.ChatSessionStartedEvent
@@ -160,6 +161,15 @@ suspend fun domainEventToServerMessage(
        )

        is ApprovalRequestedEvent -> mapApprovalRequested(p, seq, sessionSequence)
+        is ApprovalDecisionResolvedEvent -> ServerMessage.ApprovalResolved(
+            // ApprovalDecisionResolvedEvent has no sessionId on its payload — read it from the event envelope
+            sessionId = event.metadata.sessionId,
+            requestId = p.requestId,
+            outcome = p.outcome.name,
+            reason = p.reason,
+            sequence = seq,
+            sessionSequence = sessionSequence,
+        )
        is ArtifactCreatedEvent -> ServerMessage.ArtifactCreated(
            sessionId = p.sessionId,
            stageId = p.stageId,
@@ -223,6 +223,17 @@ sealed interface ServerMessage {
        override val sessionSequence: Long,
    ) : ServerMessage, SessionMessage

+    @Serializable
+    @SerialName("approval.resolved")
+    data class ApprovalResolved(
+        val sessionId: SessionId,
+        val requestId: ApprovalRequestId,
+        val outcome: String,
+        val reason: String?,
+        override val sequence: Long,
+        override val sessionSequence: Long,
+    ) : ServerMessage, SessionMessage
+
    // -- System / infra --

    @Serializable
@@ -4,7 +4,6 @@ import com.correx.apps.server.protocol.ApprovalDecision
 import com.correx.apps.server.protocol.ClientMessage
 import com.correx.apps.server.protocol.ServerMessage
 import com.correx.core.approvals.Tier
-import com.correx.core.config.ApprovalConfig
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.EventMetadata
 import com.correx.core.events.events.EventPayload
@@ -29,6 +28,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.filter
@@ -43,6 +43,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertInstanceOf
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import java.util.concurrent.CopyOnWriteArrayList

@@ -96,7 +97,7 @@ class ApprovalCoordinatorWiringTest {
    @Test
    fun `handleResponse forwards to gateway and returns null on success`(): Unit = runBlocking {
        val gateway = RecordingGateway()
-        val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
+        val coord = ApprovalCoordinator(gateway)
        val msg = ClientMessage.ApprovalResponse(
            requestId = requestId,
            decision = ApprovalDecision.APPROVE,
@@ -113,7 +114,7 @@ class ApprovalCoordinatorWiringTest {
    @Test
    fun `duplicate ApprovalResponse for same requestId returns ProtocolError`(): Unit = runBlocking {
        val gateway = RecordingGateway()
-        val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
+        val coord = ApprovalCoordinator(gateway)
        val msg = ClientMessage.ApprovalResponse(
            requestId = requestId,
            decision = ApprovalDecision.APPROVE,
@@ -130,16 +131,6 @@ class ApprovalCoordinatorWiringTest {
        assertEquals(1, gateway.submissions.size)
    }

-    /**
-     * Proves the ServerModule.start() subscription wiring: an [ApprovalRequestedEvent] flowing
-     * through the event store reaches the coordinator's onApprovalRequested. We verify this by
-     * subscribing the same wiring the module uses and checking the coordinator's behavior
-     * (it schedules a timeout job and dedupes the requestId).
-     *
-     * After the event is observed, a subsequent handleResponse with a matching requestId should
-     * succeed once (the coordinator's internal `resolved` map was NOT pre-populated by the event).
-     * The behavioral assertion: the gateway sees the decision exactly once.
-     */
    /**
     * Proves the ServerModule.start() subscription wiring shape: an [ApprovalRequestedEvent]
     * flowing through the event store reaches the wired-up callback, and non-approval events
@@ -183,7 +174,7 @@ class ApprovalCoordinatorWiringTest {
    @Test
    fun `lookupSession returns mapping registered by onApprovalRequested and cleared by handleResponse`(): Unit = runBlocking {
        val gateway = RecordingGateway()
-        val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
+        val coord = ApprovalCoordinator(gateway)
        val otherSessionId = SessionId("other-session")

        // Initially no mapping
@@ -215,7 +206,7 @@ class ApprovalCoordinatorWiringTest {
    @Test
    fun `onApprovalRequested with non-null riskSummary registers session without error`(): Unit = runBlocking {
        val gateway = RecordingGateway()
-        val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
+        val coord = ApprovalCoordinator(gateway)
        val riskSummary = RiskSummary(
            level = RiskLevel.HIGH,
            signals = listOf(RiskSignal.ValidationErrors(errorCount = 2)),
@@ -235,4 +226,31 @@ class ApprovalCoordinatorWiringTest {
        coord.onApprovalRequested(event)
        assertEquals(sessionId, coord.lookupSession(requestId))
    }
+
+    /**
+     * Approval gates block indefinitely — no auto-reject after a delay.
+     * An unanswered request must not cause any decision to be submitted to the gateway.
+     */
+    @Test
+    fun `unanswered approval request does not auto-reject after a delay`(): Unit = runBlocking {
+        val gateway = RecordingGateway()
+        val coord = ApprovalCoordinator(gateway)
+        val event = ApprovalRequestedEvent(
+            requestId = requestId,
+            tier = Tier.T2,
+            validationReportId = ValidationReportId("vr-1"),
+            riskSummaryId = null,
+            sessionId = sessionId,
+            stageId = null,
+            projectId = null,
+        )
+
+        coord.onApprovalRequested(event)
+        // Give any hypothetical background job time to fire.
+        delay(200L)
+
+        assertTrue(gateway.submissions.isEmpty(), "No decision should be submitted for an unanswered approval")
+        // The session mapping must still be present — the request is still pending.
+        assertEquals(sessionId, coord.lookupSession(requestId))
+    }
 }
@@ -2,8 +2,11 @@ package com.correx.apps.server.bridge

 import com.correx.apps.server.protocol.PauseReason
 import com.correx.apps.server.protocol.ServerMessage
+import com.correx.core.approvals.ApprovalOutcome
+import com.correx.core.approvals.ApprovalStatus
 import com.correx.core.approvals.Tier
 import com.correx.core.artifactstore.ArtifactStore
+import com.correx.core.events.events.ApprovalDecisionResolvedEvent
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.EventMetadata
 import com.correx.core.events.events.InferenceCompletedEvent
@@ -28,6 +31,7 @@ import com.correx.core.events.risk.RiskAction
 import com.correx.core.events.risk.RiskLevel
 import com.correx.core.events.risk.RiskSignal
 import com.correx.core.events.risk.RiskSummary
+import com.correx.core.events.types.ApprovalDecisionId
 import com.correx.core.events.types.ApprovalRequestId
 import com.correx.core.events.types.ArtifactId
 import com.correx.core.events.types.EventId
@@ -410,6 +414,50 @@ class DomainEventMapperTest {
        assertTrue(result.riskSummary.factors[1].contains("Repeated failure"))
    }

+    @Test
+    fun `ApprovalDecisionResolvedEvent maps to ApprovalResolved with correct fields`(): Unit = runTest {
+        val requestId = ApprovalRequestId("req-3")
+        val event = storedEvent(
+            ApprovalDecisionResolvedEvent(
+                decisionId = ApprovalDecisionId("dec-1"),
+                requestId = requestId,
+                outcome = ApprovalOutcome.APPROVED,
+                status = ApprovalStatus.COMPLETED,
+                tier = Tier.T2,
+                resolutionTimestamp = timestamp,
+                reason = "user approved",
+                userSteering = null,
+            ),
+        )
+        val result = domainEventToServerMessage(event, noopStore, sessionSequence = 5L) as ServerMessage.ApprovalResolved
+        assertEquals(sessionId, result.sessionId)
+        assertEquals(requestId, result.requestId)
+        assertEquals("APPROVED", result.outcome)
+        assertEquals("user approved", result.reason)
+        assertEquals(event.sequence, result.sequence)
+        assertEquals(5L, result.sessionSequence)
+    }
+
+    @Test
+    fun `ApprovalDecisionResolvedEvent maps to ApprovalResolved with null reason`(): Unit = runTest {
+        val requestId = ApprovalRequestId("req-4")
+        val event = storedEvent(
+            ApprovalDecisionResolvedEvent(
+                decisionId = ApprovalDecisionId("dec-2"),
+                requestId = requestId,
+                outcome = ApprovalOutcome.REJECTED,
+                status = ApprovalStatus.COMPLETED,
+                tier = Tier.T3,
+                resolutionTimestamp = timestamp,
+                reason = null,
+                userSteering = null,
+            ),
+        )
+        val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) as ServerMessage.ApprovalResolved
+        assertEquals("REJECTED", result.outcome)
+        assertNull(result.reason)
+    }
+
    @Test
    fun `unmapped event returns null`(): Unit = runTest {
        val event = storedEvent(SteeringNoteAddedEvent(sessionId = sessionId, content = "test"))
@@ -204,13 +204,11 @@ object ConfigLoader {
        val serverSection = sections["server"] ?: emptyMap()
        val tuiSection = sections["tui"] ?: emptyMap()
        val cliSection = sections["cli"] ?: emptyMap()
-        val approvalSection = sections["approval"] ?: emptyMap()
        val toolsSection = sections["tools"] ?: emptyMap()
        val toolsShellSection = sections["tools.shell"] ?: emptyMap()
        val toolsFileReadSection = sections["tools.file_read"] ?: emptyMap()
        val toolsFileWriteSection = sections["tools.file_write"] ?: emptyMap()
        val toolsFileEditSection = sections["tools.file_edit"] ?: emptyMap()
-        val routerSection = sections["router"] ?: emptyMap()
        val routerEmbedderSection = sections["router.embedder"] ?: emptyMap()
        val routerL3Section = sections["router.l3"] ?: emptyMap()

@@ -228,10 +226,6 @@ object ConfigLoader {
            defaultOutput = asString(cliSection["default_output"], "human"),
        )

-        val approval = ApprovalConfig(
-            timeoutMs = asLong(approvalSection["timeout_ms"], 300_000L),
-        )
-
        // Resolve tool enable flags: prefer nested [tools.shell], [tools.file_read], etc.
        // Fall back to flat [tools] section for backward compat
        val shellEnabled = when {
@@ -350,7 +344,6 @@ object ConfigLoader {
            server = server,
            tui = tui,
            cli = cli,
-            approval = approval,
            tools = tools,
            providers = providers,
            router = router,
@@ -392,15 +385,6 @@ object ConfigLoader {
        }
    }

-    private fun asLong(value: Any?, default: Long = 0L): Long {
-        return when (value) {
-            is Long -> value
-            is Int -> value.toLong()
-            is String -> value.toLongOrNull() ?: default
-            else -> default
-        }
-    }
-
    private fun asBoolean(value: Any?, default: Boolean = false): Boolean {
        return when (value) {
            is Boolean -> value
@@ -7,7 +7,6 @@ data class CorrexConfig(
    val server: ServerConfig = ServerConfig(),
    val tui: TuiConfig = TuiConfig(),
    val cli: CliConfig = CliConfig(),
-    val approval: ApprovalConfig = ApprovalConfig(),
    val tools: ToolsConfig = ToolsConfig(),
    val providers: List<ProviderConfig> = emptyList(),
    val router: RouterConfig = RouterConfig(),
@@ -30,11 +29,6 @@ data class CliConfig(
    val defaultOutput: String = "human",
 )

-@Serializable
-data class ApprovalConfig(
-    val timeoutMs: Long = 300_000L,
-)
-
@Serializable
 data class ToolsConfig(
    val sandboxRoot: String = "",
@@ -12,7 +12,6 @@ class ConfigLoaderTest {
        assertEquals("dark", config.tui.theme)
        assertEquals(5, config.tui.sessionListLimit)
        assertEquals("human", config.cli.defaultOutput)
-        assertEquals(300_000L, config.approval.timeoutMs)
    }

    @Test
@@ -28,9 +27,6 @@ class ConfigLoaderTest {

            [cli]
            default_output = "json"
-
-            [approval]
-            timeout_ms = 600000
        """.trimIndent()

        val loader = ConfigLoader::class.java
@@ -43,7 +39,6 @@ class ConfigLoaderTest {
        assertEquals("light", result.tui.theme)
        assertEquals(10, result.tui.sessionListLimit)
        assertEquals("json", result.cli.defaultOutput)
-        assertEquals(600_000L, result.approval.timeoutMs)
    }

    @Test
@@ -12,9 +12,6 @@ session_list_limit = 5
 [cli]
 default_output = "human"

-[approval]
-timeout_ms = 300000
-
 [tools]
 sandbox_root = "~/.config/correx/sandbox"
 working_dir = "/tmp"