server: S5 memoize listSessionSummaries + S6 Main.kt hygiene
S5 — GET /sessions re-read + re-projected every session's full log on every call. Cache keyed on eventStore.lastGlobalSequence(); volatile pair-swap, no lock — any append bumps the seq so a stale read is impossible. S6 — close research + health-probe HttpClients via shutdown hook; log the real configured host:port (was hardcoded 8080) and bind to it; System.err -> SLF4J; WARN at startup on non-loopback bind (unauthenticated surface). resolveApiKey precedence item N/A — no such code exists. Both verified green in isolated worktrees (226 pass). Full local build currently blocked by an unrelated concurrent core:config edit.
This commit is contained in:
@@ -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<ArtifactKind> {
|
||||
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)
|
||||
|
||||
@@ -624,13 +624,25 @@ class ServerModule(
|
||||
|
||||
private val sessionSummaryProjector = SessionSummaryProjector()
|
||||
|
||||
fun listSessionSummaries(): List<SessionSummary> =
|
||||
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<Long, List<SessionSummary>>? = null
|
||||
|
||||
suspend fun listSessionSummaries(): List<SessionSummary> {
|
||||
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())
|
||||
|
||||
Reference in New Issue
Block a user