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 51aa1ff0..bf21cd87 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 @@ -109,9 +109,10 @@ import java.nio.file.Paths private val log = LoggerFactory.getLogger("com.correx.apps.server.Main") +private val loopbackHosts = setOf("localhost", "127.0.0.1", "::1") + fun main() { log.info("=== correx server starting ===") - log.info(" port : 8080") val artifactStore = InfrastructureModule.createArtifactStore() val eventStore = LoggingEventStore(InfrastructureModule.createEventStore(artifactStore)) @@ -218,16 +219,19 @@ fun main() { val workingDir = explicitWorkingDir ?: workspaceRoot // One shared HTTP client backs both the default and per-workspace registries' research tools // (web_search/web_fetch). Built only when research is enabled, so the static path stays offline. + // Lives for the process lifetime (shared across requests), so it's closed via shutdown hook + // below rather than `.use { }`. + val researchHttpClient = if (toolsConfig.research.enabled) { + io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO) + } else { + null + } val researchToolConfig = com.correx.infrastructure.tools.ResearchToolConfig( enabled = toolsConfig.research.enabled, searxngUrl = toolsConfig.research.searxngUrl, maxResults = toolsConfig.research.maxResults, maxFetchBytes = toolsConfig.research.maxFetchBytes, - httpClient = if (toolsConfig.research.enabled) { - io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO) - } else { - null - }, + httpClient = researchHttpClient, ) // Agents create/update/delete tasks through the tool system (tier-gated like any tool); // the service appends to the same event log. Project is derived from the task id prefix. @@ -534,6 +538,8 @@ fun main() { ) // observability-spec §4: continuous health watch. Seed the monitor's last-status from the // recorded system-session events so a restart doesn't re-emit a degraded already in the log. + // healthProbeHttpClient lives for the process lifetime; closed via shutdown hook below. + var healthProbeHttpClient: io.ktor.client.HttpClient? = null val healthMonitor = correxConfig.health.let { hc -> if (!hc.enabled) { null @@ -542,6 +548,7 @@ fun main() { val seeded = DefaultEventReplayer(eventStore, com.correx.apps.server.health.HealthProjection()) .rebuild(com.correx.apps.server.health.SYSTEM_SESSION) .subjects.mapValues { it.value.status } + healthProbeHttpClient = io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO) com.correx.apps.server.health.HealthMonitor( eventStore = eventStore, probes = listOfNotNull( @@ -552,7 +559,7 @@ fun main() { ), com.correx.apps.server.health.LlamaServerHealthProbe( llamaBaseUrl = llamaBaseUrl, - httpClient = io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO), + httpClient = healthProbeHttpClient!!, eventStore = eventStore, livenessTimeoutMs = hc.llamaLivenessTimeoutMs, tpsWarnBelow = hc.llamaTpsWarnBelow, @@ -567,6 +574,12 @@ fun main() { ) } } + Runtime.getRuntime().addShutdownHook( + Thread { + researchHttpClient?.close() + healthProbeHttpClient?.close() + }, + ) val module = ServerModule( orchestrator = orchestrator, eventStore = eventStore, @@ -609,7 +622,16 @@ fun main() { module.start() log.info("==============================") - embeddedServer(Netty, port = correxConfig.server.port, host = correxConfig.server.host) { + val serverConfig = correxConfig.server + log.info(" host:port : {}:{}", serverConfig.host, serverConfig.port) + if (serverConfig.host !in loopbackHosts) { + log.warn( + "Server host '{}' is not loopback — the unauthenticated HTTP/WS surface is exposed to the network", + serverConfig.host, + ) + } + + embeddedServer(Netty, host = serverConfig.host, port = serverConfig.port) { configureServer(module) }.start(wait = true) } @@ -674,9 +696,7 @@ private fun loadConfigArtifactKinds(config: CorrexConfig): List { val schema = runCatching { Json.decodeFromString(JsonSchema.serializer(), Files.readString(schemaPath)) }.getOrElse { e -> - System.err.println( - "Warning: artifact kind '${decl.id}' schema '$schemaPath' failed to load: ${e.message}", - ) + log.warn("artifact kind '{}' schema '{}' failed to load: {}", decl.id, schemaPath, e.message) return@mapNotNull null } ConfigArtifactKind(id = decl.id, schema = schema, llmEmitted = decl.llmEmitted) 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 916d139a..32dbb027 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 @@ -624,13 +624,25 @@ class ServerModule( private val sessionSummaryProjector = SessionSummaryProjector() - fun listSessionSummaries(): List = - eventStore.allSessionIds() + // ponytail: memoized on lastGlobalSequence — GET /sessions was re-reading + re-projecting every + // session's full log on every call (S5). Any new event anywhere bumps the global sequence, so a + // stale cache is impossible: a mismatch always forces a full recompute. Volatile pair-swap is + // enough here (no lock) — worst case under a race is one redundant recompute, never a stale read. + @Volatile + private var sessionSummaryCache: Pair>? = null + + suspend fun listSessionSummaries(): List { + val currentSeq = eventStore.lastGlobalSequence() + sessionSummaryCache?.let { (seq, summaries) -> if (seq == currentSeq) return summaries } + val summaries = eventStore.allSessionIds() // The system session carries global health events, not a user workflow — hide it. .filter { it != com.correx.apps.server.health.SYSTEM_SESSION } .map { sessionId -> sessionSummaryProjector.project(sessionId, eventStore.readFrom(sessionId, fromSequence = 0L)) } + sessionSummaryCache = currentSeq to summaries + return summaries + } private fun preRegisterPendingApprovals() { val projector = ApprovalProjector(DefaultApprovalReducer())