diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 47ede1aa..e075fb29 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -160,7 +160,6 @@ fun main() { orchestrationRepository = repositories.orchestrationRepository, approvalRepository = repositories.approvalRepository, toolRegistry = toolRegistry, - approvalConfig = correxConfig.approval, ) module.start() log.info("==============================") diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index eecc4ad3..c625bd67 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -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 diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt index 2588408e..1a9c774a 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -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> = ConcurrentHashMap() private val globalClients: MutableSet = ConcurrentHashMap.newKeySet() private val resolved: ConcurrentHashMap = ConcurrentHashMap() - private val timeoutJobs: ConcurrentHashMap = ConcurrentHashMap() private val requestSessions: ConcurrentHashMap = 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 - } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt index 65766f38..b1e1c752 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt @@ -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, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 6372a75b..577baa9c 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -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 diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt index 6ca3ce0b..5e83c312 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt @@ -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)) + } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt index e8773680..c969ea2d 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt @@ -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")) diff --git a/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt index 400fb5be..5446b534 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt @@ -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 diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt index 5912b757..e1d83287 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt @@ -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 = 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 = "", diff --git a/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt b/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt index 0ac550e7..7f360c22 100644 --- a/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt +++ b/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt @@ -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 diff --git a/docs/sample-config.toml b/docs/sample-config.toml index 734a5bae..de9b093c 100644 --- a/docs/sample-config.toml +++ b/docs/sample-config.toml @@ -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"