feat(tui,server,config): live config editor + artifact viewer
Two operator features over the existing WebSocket protocol, plus a tui-go nav fix. Config editor (g / palette "config"): - core:config gains a thin registry stack: OrchestrationKnobs ([orchestration] block), CorrexConfigWriter (round-trip TOML serializer; parseToml(write(c))==c), ConfigHolder (AtomicReference live config), and EditableConfig (allowlist of editable fields + patch applier; security fields are absent so they can never be patched). - ConfigService validates -> persists (atomic temp+move) -> swaps the holder -> rebuilds config-derived services. Live-apply is scoped to safe seams: stage timeout, compaction threshold (now a supplier), router knobs (routerFacade rebuild), and personalization/project toggles all take effect on the next session/turn; in-flight sessions keep the config they started with. server.host/port are persisted + badged "restart required", not hot-swapped. - Protocol: GetConfig / UpdateConfig / ConfigSnapshot; TUI overlay with type-aware editing (bool toggle, enum cycle, inline int/string edit) and save. Artifact viewer (v / palette "artifacts"): - Server assembles a session's artifacts from the event log and resolves bytes from the CAS store (StreamQueries.listArtifacts) — a replay-neutral snapshot read. TUI overlay lists artifacts and shows scrollable content. tui-go fix: workflow failure no longer clears selectedID (which yanked the user to the list); listNav lands on the first session when nothing is selected instead of wrapping to a random one. Config is not event-sourced (it lives in TOML); editing stays outside the log.
This commit is contained in:
@@ -9,6 +9,8 @@ import com.correx.core.artifacts.kind.ArtifactKind
|
||||
import com.correx.core.artifacts.kind.ConfigArtifactKind
|
||||
import com.correx.core.artifacts.kind.JsonSchema
|
||||
import com.correx.core.artifactstore.ArtifactStore
|
||||
import com.correx.apps.server.config.ConfigService
|
||||
import com.correx.core.config.ConfigHolder
|
||||
import com.correx.core.config.ConfigLoader
|
||||
import com.correx.core.config.CorrexConfig
|
||||
import com.correx.core.config.OperatorProfile
|
||||
@@ -38,6 +40,7 @@ import com.correx.core.kernel.orchestration.WorkspaceTools
|
||||
import com.correx.core.kernel.orchestration.WorkspaceToolRegistryProvider
|
||||
import com.correx.core.kernel.retry.DefaultRetryCoordinator
|
||||
import com.correx.core.risk.DefaultRiskAssessor
|
||||
import com.correx.core.router.RouterFacade
|
||||
import com.correx.core.router.l3.L3MetadataRehydrator
|
||||
import com.correx.core.router.model.RouterConfig
|
||||
import com.correx.core.sessions.DefaultSessionReducer
|
||||
@@ -100,6 +103,10 @@ fun main() {
|
||||
logStoresInfo(eventStore, artifactStore)
|
||||
|
||||
val correxConfig = ConfigLoader.load()
|
||||
// Live, swappable config. Hot-tunable fields (orchestration timeout, router/compaction knobs,
|
||||
// personalization/project toggles) are read from this holder at use-time; ConfigService swaps
|
||||
// it and rebuilds the config-derived services on a TUI edit.
|
||||
val configHolder = ConfigHolder(correxConfig)
|
||||
val operatorProfile: OperatorProfile? = if (correxConfig.personalization.enabled) {
|
||||
ProfileLoader.load()
|
||||
} else null
|
||||
@@ -247,15 +254,17 @@ fun main() {
|
||||
val defaultOrchestrationConfig = OrchestrationConfig(
|
||||
sandboxRoot = sandboxRoot,
|
||||
defaultSystemPromptPath = toolsConfig.defaultSystemPromptPath,
|
||||
// Local models are slow on large implementer/reviewer turns (plan + design + file
|
||||
// contents); 60s was timing out ~mid-run and burning every retry. 3 min gives the
|
||||
// turn room to complete. TODO: surface this as a config knob instead of a hardcode.
|
||||
stageTimeoutMs = 180_000L,
|
||||
// Stage timeout is now an operator-tunable knob ([orchestration] stage_timeout_ms, default
|
||||
// 3 min — local models are slow on large turns). ServerModule.orchestrationConfig() reads
|
||||
// the live value from the holder per session, so this base value is just the boot default.
|
||||
stageTimeoutMs = correxConfig.orchestration.stageTimeoutMs,
|
||||
journalCompactionTokenThreshold = correxConfig.orchestration.journalCompactionTokenThreshold,
|
||||
)
|
||||
val journalCompactionService = JournalCompactionService(
|
||||
artifactStore = artifactStore,
|
||||
summarize = { prompt -> summarizeWithInference(prompt, SessionId("system"), inferenceRouter) },
|
||||
tokenThreshold = defaultOrchestrationConfig.journalCompactionTokenThreshold,
|
||||
// Supplier reads the live threshold each compaction check (no orchestrator rebuild needed).
|
||||
tokenThreshold = { configHolder.get().orchestration.journalCompactionTokenThreshold },
|
||||
)
|
||||
val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
@@ -266,37 +275,42 @@ fun main() {
|
||||
decisionJournalRepository = decisionJournalRepository,
|
||||
compactionService = journalCompactionService,
|
||||
)
|
||||
val routerConfig = correxConfig.router
|
||||
val embedder = InfrastructureModule.createEmbedderFromConfig(routerConfig.embedder)
|
||||
val l3MemoryStore = InfrastructureModule.createL3MemoryStoreFromConfig(routerConfig.l3)
|
||||
val embedder = InfrastructureModule.createEmbedderFromConfig(correxConfig.router.embedder)
|
||||
val l3MemoryStore = InfrastructureModule.createL3MemoryStoreFromConfig(correxConfig.router.l3)
|
||||
// Rebuild the L3 metadata map from the ChatTurnEvent log before serving queries:
|
||||
// persisted vectors survive restart but the metadata map does not (no-op for
|
||||
// non-rehydratable backends like in_memory).
|
||||
runBlocking { L3MetadataRehydrator(eventStore).rehydrate(l3MemoryStore) }
|
||||
// Map the config-file [router] block onto the domain RouterConfig the facade uses.
|
||||
val domainRouterConfig = RouterConfig(
|
||||
conversationKeepLast = routerConfig.conversationKeepLast,
|
||||
retrievalK = routerConfig.retrievalK,
|
||||
tokenBudget = TokenBudget(limit = routerConfig.tokenBudget),
|
||||
generationConfig = GenerationConfig(
|
||||
temperature = routerConfig.generation.temperature,
|
||||
topP = routerConfig.generation.topP,
|
||||
maxTokens = routerConfig.generation.maxTokens,
|
||||
),
|
||||
narrationGenerationConfig = GenerationConfig(
|
||||
temperature = routerConfig.narration.temperature,
|
||||
topP = routerConfig.narration.topP,
|
||||
maxTokens = routerConfig.narration.maxTokens,
|
||||
),
|
||||
)
|
||||
val routerFacade = InfrastructureModule.createRouterFacade(
|
||||
eventStore = eventStore,
|
||||
inferenceRouter = inferenceRouter,
|
||||
config = domainRouterConfig,
|
||||
tokenizer = firstProvider.tokenizer,
|
||||
embedder = embedder,
|
||||
l3MemoryStore = l3MemoryStore,
|
||||
)
|
||||
// Builds the router facade from a config snapshot, mapping the [router] block onto the domain
|
||||
// RouterConfig. Reused by ConfigService's rebuild hook so router knob edits apply live (the
|
||||
// embedder and L3 store are construction-time and reused; the facade is what gets swapped).
|
||||
fun buildRouterFacade(cfg: CorrexConfig): RouterFacade {
|
||||
val rc = cfg.router
|
||||
val domainRouterConfig = RouterConfig(
|
||||
conversationKeepLast = rc.conversationKeepLast,
|
||||
retrievalK = rc.retrievalK,
|
||||
tokenBudget = TokenBudget(limit = rc.tokenBudget),
|
||||
generationConfig = GenerationConfig(
|
||||
temperature = rc.generation.temperature,
|
||||
topP = rc.generation.topP,
|
||||
maxTokens = rc.generation.maxTokens,
|
||||
),
|
||||
narrationGenerationConfig = GenerationConfig(
|
||||
temperature = rc.narration.temperature,
|
||||
topP = rc.narration.topP,
|
||||
maxTokens = rc.narration.maxTokens,
|
||||
),
|
||||
)
|
||||
return InfrastructureModule.createRouterFacade(
|
||||
eventStore = eventStore,
|
||||
inferenceRouter = inferenceRouter,
|
||||
config = domainRouterConfig,
|
||||
tokenizer = firstProvider.tokenizer,
|
||||
embedder = embedder,
|
||||
l3MemoryStore = l3MemoryStore,
|
||||
)
|
||||
}
|
||||
val routerFacade = buildRouterFacade(correxConfig)
|
||||
val sessionUndoService = com.correx.apps.server.undo.SessionUndoService(
|
||||
eventStore = eventStore,
|
||||
artifactStore = artifactStore,
|
||||
@@ -314,28 +328,35 @@ fun main() {
|
||||
privilegedLocations = privilegedLocations,
|
||||
allowedWorkspaceRoots = allowedWorkspaceRoots,
|
||||
)
|
||||
val projectMemory = if (correxConfig.project.enabled) {
|
||||
com.correx.apps.server.memory.ProjectMemoryService(
|
||||
config = correxConfig.project,
|
||||
embedder = embedder,
|
||||
l3MemoryStore = l3MemoryStore,
|
||||
journalRepository = decisionJournalRepository,
|
||||
eventStore = eventStore,
|
||||
indexer = com.correx.apps.server.memory.RepoMapIndexer(
|
||||
maxDepth = correxConfig.project.maxDepth,
|
||||
ignoreGlobs = correxConfig.project.ignoreGlobs,
|
||||
),
|
||||
)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
val profileAdaptationService: ProfileAdaptationService? =
|
||||
if (correxConfig.personalization.enabled && correxConfig.personalization.learn) {
|
||||
// Built from a config snapshot and reused by ConfigService's rebuild hook so toggling
|
||||
// project.enabled / personalization.* applies live to the next session.
|
||||
fun buildProjectMemory(cfg: CorrexConfig): com.correx.apps.server.memory.ProjectMemoryService? =
|
||||
if (cfg.project.enabled) {
|
||||
com.correx.apps.server.memory.ProjectMemoryService(
|
||||
config = cfg.project,
|
||||
embedder = embedder,
|
||||
l3MemoryStore = l3MemoryStore,
|
||||
journalRepository = decisionJournalRepository,
|
||||
eventStore = eventStore,
|
||||
indexer = com.correx.apps.server.memory.RepoMapIndexer(
|
||||
maxDepth = cfg.project.maxDepth,
|
||||
ignoreGlobs = cfg.project.ignoreGlobs,
|
||||
),
|
||||
)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
fun buildProfileAdaptation(cfg: CorrexConfig): ProfileAdaptationService? =
|
||||
if (cfg.personalization.enabled && cfg.personalization.learn) {
|
||||
ProfileAdaptationService(
|
||||
decisionJournalRepository = decisionJournalRepository,
|
||||
summarize = { prompt -> summarizeWithInference(prompt, SessionId("system"), inferenceRouter) },
|
||||
)
|
||||
} else null
|
||||
} else {
|
||||
null
|
||||
}
|
||||
val projectMemory = buildProjectMemory(correxConfig)
|
||||
val profileAdaptationService: ProfileAdaptationService? = buildProfileAdaptation(correxConfig)
|
||||
|
||||
val configArtifactKinds = loadConfigArtifactKinds(correxConfig)
|
||||
val artifactKindRegistry = DefaultArtifactKindRegistry().also { reg ->
|
||||
@@ -366,12 +387,22 @@ fun main() {
|
||||
modelSwapper = modelSwapper,
|
||||
resourceProbe = resourceProbe,
|
||||
workspaceResolver = workspaceResolver,
|
||||
narrationMaxPerRun = routerConfig.narration.maxPerRun,
|
||||
narrationMaxPerRun = correxConfig.router.narration.maxPerRun,
|
||||
projectMemory = projectMemory,
|
||||
configHolder = configHolder,
|
||||
freestyleDriver = freestyleDriver,
|
||||
operatorProfile = operatorProfile,
|
||||
profileAdaptationService = profileAdaptationService,
|
||||
)
|
||||
// Wire live config editing: persist to TOML, swap the holder, and rebuild config-derived
|
||||
// services. Built after the module so the rebuild hook can swap them in place.
|
||||
module.configService = ConfigService(configHolder, ConfigLoader.configPath()) { cfg ->
|
||||
module.applyRebuiltServices(
|
||||
newRouterFacade = buildRouterFacade(cfg),
|
||||
newProjectMemory = buildProjectMemory(cfg),
|
||||
newProfileAdaptationService = buildProfileAdaptation(cfg),
|
||||
)
|
||||
}
|
||||
module.start()
|
||||
log.info("==============================")
|
||||
|
||||
|
||||
@@ -51,7 +51,9 @@ class ServerModule(
|
||||
val workflowRegistry: WorkflowRegistry,
|
||||
val providerRegistry: ProviderRegistry,
|
||||
val defaultOrchestrationConfig: OrchestrationConfig,
|
||||
val routerFacade: RouterFacade,
|
||||
// Rebuilt by ConfigService on a live config edit (router knobs); read at use-time by
|
||||
// GlobalStreamHandler.onUserInput so the next chat turn picks up the new facade.
|
||||
routerFacade: RouterFacade,
|
||||
val orchestrationRepository: OrchestrationRepository,
|
||||
val approvalRepository: DefaultApprovalRepository,
|
||||
val toolRegistry: ToolRegistry,
|
||||
@@ -73,19 +75,62 @@ class ServerModule(
|
||||
approvalCoordinator: ApprovalCoordinator? = null,
|
||||
val narrationMaxPerRun: Int = 100,
|
||||
// Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path).
|
||||
val projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null,
|
||||
// Rebuilt by ConfigService when project.enabled is toggled live.
|
||||
projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null,
|
||||
// Live, swappable config. Null only in tests that don't exercise config editing; defaults to a
|
||||
// holder seeded from defaults so callers always have a value to read.
|
||||
val configHolder: com.correx.core.config.ConfigHolder =
|
||||
com.correx.core.config.ConfigHolder(com.correx.core.config.CorrexConfig()),
|
||||
// Driver for the two-phase freestyle workflow: locks the plan and runs phase 2.
|
||||
// Null on non-freestyle paths; injected by Main when freestyle is configured.
|
||||
private val freestyleDriver: FreestyleDriver? = null,
|
||||
// Operator profile snapshot loaded at server start. Null when personalization is disabled.
|
||||
private val operatorProfile: OperatorProfile? = null,
|
||||
// Propose learned profile updates at session end. Null when learn=false or no profile.
|
||||
private val profileAdaptationService: com.correx.apps.server.personalization.ProfileAdaptationService? = null,
|
||||
// Rebuilt by ConfigService when personalization.enabled/learn is toggled live.
|
||||
profileAdaptationService: com.correx.apps.server.personalization.ProfileAdaptationService? = null,
|
||||
) {
|
||||
val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator(
|
||||
orchestrator = orchestrator,
|
||||
)
|
||||
|
||||
// Config-derived services that ConfigService swaps on a live edit. Reads happen at use-time
|
||||
// (next session / next chat turn), so in-flight sessions keep the config they started with.
|
||||
@Volatile
|
||||
var routerFacade: RouterFacade = routerFacade
|
||||
private set
|
||||
|
||||
@Volatile
|
||||
var projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = projectMemory
|
||||
private set
|
||||
|
||||
@Volatile
|
||||
var profileAdaptationService: com.correx.apps.server.personalization.ProfileAdaptationService? =
|
||||
profileAdaptationService
|
||||
private set
|
||||
|
||||
// Set by Main after construction (it needs a reference to the module to build the rebuild hook).
|
||||
lateinit var configService: com.correx.apps.server.config.ConfigService
|
||||
|
||||
/**
|
||||
* Swaps the config-derived services after a live config edit. Called by [ConfigService].
|
||||
* In-flight sessions are untouched; the next session/turn reads the new instances.
|
||||
*/
|
||||
fun applyRebuiltServices(
|
||||
newRouterFacade: RouterFacade,
|
||||
newProjectMemory: com.correx.apps.server.memory.ProjectMemoryService?,
|
||||
newProfileAdaptationService: com.correx.apps.server.personalization.ProfileAdaptationService?,
|
||||
) {
|
||||
this.routerFacade = newRouterFacade
|
||||
this.projectMemory = newProjectMemory
|
||||
this.profileAdaptationService = newProfileAdaptationService
|
||||
}
|
||||
|
||||
/** Orchestration config for a new session, with live-tunable knobs read from [configHolder]. */
|
||||
fun orchestrationConfig(): OrchestrationConfig = defaultOrchestrationConfig.copy(
|
||||
stageTimeoutMs = configHolder.get().orchestration.stageTimeoutMs,
|
||||
)
|
||||
|
||||
private var subscriptionJob: Job? = null
|
||||
private val activeSessionJobs = ConcurrentHashMap<SessionId, Job>()
|
||||
|
||||
@@ -137,7 +182,7 @@ class ServerModule(
|
||||
fun launchSessionRun(
|
||||
sessionId: SessionId,
|
||||
graph: com.correx.core.transitions.graph.WorkflowGraph,
|
||||
sessionConfig: OrchestrationConfig = defaultOrchestrationConfig,
|
||||
sessionConfig: OrchestrationConfig = orchestrationConfig(),
|
||||
) {
|
||||
activeSessionJobs.computeIfAbsent(sessionId) {
|
||||
moduleScope.launch {
|
||||
@@ -217,7 +262,7 @@ class ServerModule(
|
||||
activeSessionJobs.computeIfAbsent(sessionId) {
|
||||
moduleScope.launch {
|
||||
runCatching {
|
||||
orchestrator.resume(sessionId, graph, defaultOrchestrationConfig)
|
||||
orchestrator.resume(sessionId, graph, orchestrationConfig())
|
||||
}.onFailure { e ->
|
||||
log.error("resumeSession: session={} failed", sessionId.value, e)
|
||||
}
|
||||
@@ -236,7 +281,7 @@ class ServerModule(
|
||||
moduleScope.launch {
|
||||
runCatching {
|
||||
orchestrator.rehydrate(sessionId)
|
||||
orchestrator.resume(sessionId, graph, defaultOrchestrationConfig)
|
||||
orchestrator.resume(sessionId, graph, orchestrationConfig())
|
||||
}.onFailure { e ->
|
||||
log.error("resumeSession (rehydrate): session={} failed", sessionId.value, e)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package com.correx.apps.server.config
|
||||
|
||||
import com.correx.core.config.ConfigHolder
|
||||
import com.correx.core.config.CorrexConfig
|
||||
import com.correx.core.config.CorrexConfigWriter
|
||||
import com.correx.core.config.EditableConfig
|
||||
import com.correx.core.config.EditableField
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.StandardCopyOption
|
||||
|
||||
/** Outcome of a config update. [Ok.restartRequired] lists changed keys that need a restart. */
|
||||
sealed interface ConfigUpdateResult {
|
||||
data class Ok(val restartRequired: Set<String>) : ConfigUpdateResult
|
||||
data class Error(val message: String) : ConfigUpdateResult
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates, persists, and live-applies config edits. The flow on [apply] is: validate the patch
|
||||
* against [EditableConfig] (security-sensitive keys are rejected because they are absent from the
|
||||
* registry), atomically rewrite the TOML file, swap the [holder], then invoke [rebuild] so
|
||||
* config-derived services are reconstructed. Validation/persist failures are reported without
|
||||
* mutating the holder or calling [rebuild].
|
||||
*
|
||||
* Config is not event-sourced — it lives in TOML — so this service deliberately stays outside the
|
||||
* event log.
|
||||
*/
|
||||
class ConfigService(
|
||||
private val holder: ConfigHolder,
|
||||
private val configPath: Path,
|
||||
private val rebuild: (CorrexConfig) -> Unit,
|
||||
) {
|
||||
|
||||
/** Current editable fields paired with their rendered string values, for the protocol snapshot. */
|
||||
fun snapshot(): List<Pair<EditableField, String>> =
|
||||
EditableConfig.fields.map { it to it.getString(holder.get()) }
|
||||
|
||||
fun apply(patch: Map<String, String>): ConfigUpdateResult {
|
||||
val newCfg = EditableConfig.apply(holder.get(), patch)
|
||||
.getOrElse { return ConfigUpdateResult.Error(it.message ?: "invalid config update") }
|
||||
|
||||
return runCatching { persist(newCfg) }.fold(
|
||||
onSuccess = {
|
||||
holder.set(newCfg)
|
||||
rebuild(newCfg)
|
||||
ConfigUpdateResult.Ok(patch.keys intersect EditableConfig.restartRequiredKeys)
|
||||
},
|
||||
onFailure = { ConfigUpdateResult.Error("failed to persist config: ${it.message}") },
|
||||
)
|
||||
}
|
||||
|
||||
/** Writes via a temp file + atomic move so a crash mid-write cannot leave a truncated config. */
|
||||
private fun persist(cfg: CorrexConfig) {
|
||||
val parent = configPath.toAbsolutePath().parent
|
||||
Files.createDirectories(parent)
|
||||
val tmp = Files.createTempFile(parent, "config", ".toml.tmp")
|
||||
Files.writeString(tmp, CorrexConfigWriter.write(cfg))
|
||||
runCatching {
|
||||
Files.move(tmp, configPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE)
|
||||
}.getOrElse {
|
||||
// Some filesystems reject ATOMIC_MOVE across boundaries; fall back to a plain replace.
|
||||
Files.move(tmp, configPath, StandardCopyOption.REPLACE_EXISTING)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -38,6 +38,18 @@ sealed class ClientMessage {
|
||||
@Serializable
|
||||
data class CancelSession(val sessionId: SessionId) : ClientMessage()
|
||||
|
||||
/** Operator request for the artifacts produced by a session, with their content. */
|
||||
@Serializable
|
||||
data class ListArtifacts(val sessionId: SessionId) : ClientMessage()
|
||||
|
||||
/** Operator request for the current editable config (replied to with a ConfigSnapshot). */
|
||||
@Serializable
|
||||
data object GetConfig : ClientMessage()
|
||||
|
||||
/** Operator request to apply a config patch (key → string value) and persist it. */
|
||||
@Serializable
|
||||
data class UpdateConfig(val patch: Map<String, String>) : ClientMessage()
|
||||
|
||||
@Serializable
|
||||
data class ApprovalResponse(
|
||||
val requestId: ApprovalRequestId,
|
||||
|
||||
@@ -86,6 +86,33 @@ data class WorkflowDto(
|
||||
val description: String,
|
||||
)
|
||||
|
||||
/**
|
||||
* One editable config field for the TUI editor. [type] is an [com.correx.core.config.ConfigFieldType]
|
||||
* name; [value] is the current rendered value; [options] is non-empty only for ENUM fields;
|
||||
* [restartRequired] flags fields that persist now but only take effect after a server restart.
|
||||
*/
|
||||
@Serializable
|
||||
data class ConfigFieldDto(
|
||||
val key: String,
|
||||
val type: String,
|
||||
val value: String,
|
||||
val options: List<String> = emptyList(),
|
||||
val restartRequired: Boolean = false,
|
||||
)
|
||||
|
||||
/**
|
||||
* A single artifact in a session listing. [content] is the UTF-8 decoding of the stored
|
||||
* bytes, or null when no content was stored / the bytes are not valid UTF-8 (binary).
|
||||
*/
|
||||
@Serializable
|
||||
data class ArtifactSummaryDto(
|
||||
val artifactId: String,
|
||||
val stageId: String,
|
||||
val schemaVersion: Int,
|
||||
val phase: String,
|
||||
val content: String?,
|
||||
)
|
||||
|
||||
internal fun RiskSummary.toDto(): RiskSummaryDto = RiskSummaryDto(
|
||||
level = level.name,
|
||||
factors = signals.map { signal ->
|
||||
|
||||
@@ -360,6 +360,34 @@ sealed interface ServerMessage {
|
||||
override val sessionSequence: Long,
|
||||
) : ServerMessage, SessionMessage
|
||||
|
||||
/**
|
||||
* Full artifact listing for a session, returned in response to a ListArtifacts request.
|
||||
* Not event-derived (it is a snapshot read), so cursors are null.
|
||||
*/
|
||||
@Serializable
|
||||
@SerialName("artifact.list")
|
||||
data class ArtifactList(
|
||||
val sessionId: SessionId,
|
||||
val artifacts: List<ArtifactSummaryDto>,
|
||||
override val sequence: Long? = null,
|
||||
override val sessionSequence: Long? = null,
|
||||
) : ServerMessage, NonEventMessage
|
||||
|
||||
/**
|
||||
* The current editable config (reply to GetConfig / UpdateConfig). [restartRequired] lists the
|
||||
* keys in a just-applied patch that need a server restart to take effect; [error] is set (and
|
||||
* fields hold the unchanged current values) when an update was rejected.
|
||||
*/
|
||||
@Serializable
|
||||
@SerialName("config.snapshot")
|
||||
data class ConfigSnapshot(
|
||||
val fields: List<ConfigFieldDto>,
|
||||
val restartRequired: List<String> = emptyList(),
|
||||
val error: String? = null,
|
||||
override val sequence: Long? = null,
|
||||
override val sessionSequence: Long? = null,
|
||||
) : ServerMessage, NonEventMessage
|
||||
|
||||
// -- Router --
|
||||
|
||||
@Serializable
|
||||
|
||||
@@ -65,6 +65,8 @@ private fun ResourceSnapshot.toResourceStatus(): ServerMessage.ResourceStatus =
|
||||
|
||||
class GlobalStreamHandler(private val module: ServerModule) {
|
||||
|
||||
private val queries = StreamQueries(module)
|
||||
|
||||
suspend fun handle(session: DefaultWebSocketServerSession) = coroutineScope {
|
||||
log.info("client connected")
|
||||
|
||||
@@ -191,6 +193,10 @@ class GlobalStreamHandler(private val module: ServerModule) {
|
||||
}
|
||||
}
|
||||
|
||||
// Flat dispatch table: an exhaustive `when` over the sealed ClientMessage. Its cyclomatic
|
||||
// complexity scales with the number of message types, which is the false-positive case the
|
||||
// metric flags — each branch is a one-line delegation, not real branching logic.
|
||||
@Suppress("CyclomaticComplexMethod")
|
||||
private suspend fun handleClientMessage(
|
||||
session: DefaultWebSocketServerSession,
|
||||
msg: ClientMessage,
|
||||
@@ -219,6 +225,9 @@ class GlobalStreamHandler(private val module: ServerModule) {
|
||||
runCatching { module.orchestrator.cancel(msg.sessionId) }
|
||||
.onFailure { log.error("cancel failed for session={}: {}", msg.sessionId.value, it.message, it) }
|
||||
}
|
||||
is ClientMessage.ListArtifacts -> sendFrame(queries.listArtifacts(msg.sessionId))
|
||||
is ClientMessage.GetConfig -> sendFrame(queries.configSnapshot(error = null, restartRequired = emptyList()))
|
||||
is ClientMessage.UpdateConfig -> sendFrame(queries.updateConfig(msg.patch))
|
||||
is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported"))
|
||||
is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg, sendFrame)
|
||||
is ClientMessage.CreateGrant -> handleCreateGrant(msg, sendFrame)
|
||||
@@ -465,7 +474,7 @@ class GlobalStreamHandler(private val module: ServerModule) {
|
||||
|
||||
// Thread the resolved workspace into the OrchestrationConfig so the live run uses
|
||||
// the bound workspace root for tool path containment and per-session tool registry.
|
||||
val sessionConfig = module.defaultOrchestrationConfig.copy(workspace = resolvedWorkspace)
|
||||
val sessionConfig = module.orchestrationConfig().copy(workspace = resolvedWorkspace)
|
||||
|
||||
// Seed the freeform request (if any) before the run so it lands in the decision
|
||||
// journal and is pinned into every stage's context (the intent input channel).
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
package com.correx.apps.server.ws
|
||||
|
||||
import com.correx.apps.server.ServerModule
|
||||
import com.correx.apps.server.config.ConfigUpdateResult
|
||||
import com.correx.apps.server.protocol.ArtifactSummaryDto
|
||||
import com.correx.apps.server.protocol.ConfigFieldDto
|
||||
import com.correx.apps.server.protocol.ServerMessage
|
||||
import com.correx.core.config.EditableConfig
|
||||
import com.correx.core.events.events.ArtifactContentStoredEvent
|
||||
import com.correx.core.events.events.ArtifactCreatedEvent
|
||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
||||
import com.correx.core.events.events.ArtifactValidatingEvent
|
||||
import com.correx.core.events.types.ArtifactId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
|
||||
/**
|
||||
* Read-only / snapshot queries served over the global stream: artifact listings and config
|
||||
* get/update. Kept separate from [GlobalStreamHandler] (which owns the connection lifecycle and
|
||||
* session-mutating commands) so each stays focused. Every method here is replay-neutral — it
|
||||
* reads recorded state or edits TOML config, never appends events.
|
||||
*/
|
||||
class StreamQueries(private val module: ServerModule) {
|
||||
|
||||
/**
|
||||
* Reads a session's events to assemble its artifact listing (creation order preserved),
|
||||
* resolving each artifact's stored bytes via the CAS artifact store. Pure snapshot read —
|
||||
* no events appended, so it is replay-neutral (invariant #1/#8).
|
||||
*/
|
||||
suspend fun listArtifacts(sessionId: SessionId): ServerMessage.ArtifactList {
|
||||
val events = withContext(Dispatchers.IO) { module.eventStore.read(sessionId) }
|
||||
val accs = LinkedHashMap<String, ArtifactAcc>()
|
||||
for (event in events) {
|
||||
when (val p = event.payload) {
|
||||
is ArtifactCreatedEvent -> accs.getOrPut(p.artifactId.value) { ArtifactAcc() }.apply {
|
||||
stageId = p.stageId.value
|
||||
schemaVersion = p.schemaVersion
|
||||
}
|
||||
is ArtifactValidatingEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATING" }
|
||||
is ArtifactValidatedEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATED" }
|
||||
is ArtifactContentStoredEvent -> accs[p.artifactId.value]?.let { it.contentHash = p.contentHash }
|
||||
else -> Unit
|
||||
}
|
||||
}
|
||||
val artifacts = accs.map { (id, acc) ->
|
||||
val content = acc.contentHash?.let { hash ->
|
||||
withContext(Dispatchers.IO) { module.artifactStore.get(hash) }?.toString(Charsets.UTF_8)
|
||||
}
|
||||
ArtifactSummaryDto(
|
||||
artifactId = id,
|
||||
stageId = acc.stageId,
|
||||
schemaVersion = acc.schemaVersion,
|
||||
phase = acc.phase,
|
||||
content = content,
|
||||
)
|
||||
}
|
||||
return ServerMessage.ArtifactList(sessionId = sessionId, artifacts = artifacts)
|
||||
}
|
||||
|
||||
/** Current editable config as a snapshot frame. [error] / [restartRequired] are set by updates. */
|
||||
fun configSnapshot(error: String?, restartRequired: List<String>): ServerMessage.ConfigSnapshot {
|
||||
val fields = module.configService.snapshot().map { (field, value) ->
|
||||
ConfigFieldDto(
|
||||
key = field.key,
|
||||
type = field.type.name,
|
||||
value = value,
|
||||
options = field.options,
|
||||
restartRequired = field.key in EditableConfig.restartRequiredKeys,
|
||||
)
|
||||
}
|
||||
return ServerMessage.ConfigSnapshot(fields = fields, restartRequired = restartRequired, error = error)
|
||||
}
|
||||
|
||||
/** Applies a config patch and returns the resulting snapshot (with an error set on failure). */
|
||||
fun updateConfig(patch: Map<String, String>): ServerMessage.ConfigSnapshot =
|
||||
when (val result = module.configService.apply(patch)) {
|
||||
is ConfigUpdateResult.Ok -> configSnapshot(error = null, restartRequired = result.restartRequired.toList())
|
||||
is ConfigUpdateResult.Error -> configSnapshot(error = result.message, restartRequired = emptyList())
|
||||
}
|
||||
|
||||
/** Mutable accumulator for folding a session's artifact events into one summary per artifact. */
|
||||
private class ArtifactAcc {
|
||||
var stageId: String = ""
|
||||
var schemaVersion: Int = 0
|
||||
var phase: String = "CREATED"
|
||||
var contentHash: ArtifactId? = null
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package com.correx.apps.server.config
|
||||
|
||||
import com.correx.core.config.ConfigHolder
|
||||
import com.correx.core.config.ConfigLoader
|
||||
import com.correx.core.config.CorrexConfig
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.junit.jupiter.api.io.TempDir
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class ConfigServiceTest {
|
||||
|
||||
@Test
|
||||
fun `apply persists, swaps the holder, and rebuilds`(@TempDir dir: Path) {
|
||||
val path = dir.resolve("config.toml")
|
||||
val holder = ConfigHolder(CorrexConfig())
|
||||
var rebuilt: CorrexConfig? = null
|
||||
val svc = ConfigService(holder, path) { rebuilt = it }
|
||||
|
||||
val res = svc.apply(mapOf("project.enabled" to "true", "server.port" to "9000"))
|
||||
|
||||
assertTrue(res is ConfigUpdateResult.Ok)
|
||||
assertTrue(holder.get().project.enabled)
|
||||
assertEquals(9000, holder.get().server.port)
|
||||
assertEquals(holder.get(), rebuilt)
|
||||
// Persisted and reloadable to the same config.
|
||||
assertEquals(holder.get(), ConfigLoader.loadFrom(path))
|
||||
assertTrue("server.port" in (res as ConfigUpdateResult.Ok).restartRequired)
|
||||
assertFalse("project.enabled" in res.restartRequired)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `invalid patch does not persist or rebuild`(@TempDir dir: Path) {
|
||||
val path = dir.resolve("config.toml")
|
||||
val holder = ConfigHolder(CorrexConfig())
|
||||
var rebuilt = false
|
||||
val svc = ConfigService(holder, path) { rebuilt = true }
|
||||
|
||||
val res = svc.apply(mapOf("tui.session_list_limit" to "abc"))
|
||||
|
||||
assertTrue(res is ConfigUpdateResult.Error)
|
||||
assertFalse(rebuilt)
|
||||
assertFalse(Files.exists(path))
|
||||
assertEquals(5, holder.get().tui.sessionListLimit) // unchanged default
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `snapshot reflects the current held config`(@TempDir dir: Path) {
|
||||
val holder = ConfigHolder(CorrexConfig())
|
||||
val svc = ConfigService(holder, dir.resolve("config.toml")) {}
|
||||
svc.apply(mapOf("tui.session_list_limit" to "11"))
|
||||
val snap = svc.snapshot().associate { (f, v) -> f.key to v }
|
||||
assertEquals("11", snap["tui.session_list_limit"])
|
||||
}
|
||||
}
|
||||
+21
@@ -125,6 +125,27 @@ class ServerMessageSerializationTest {
|
||||
assert(jsonStr.contains("\"totalTokens\":120")) { "expected totalTokens=120" }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ConfigSnapshot encodes fields, restartRequired, and error`() {
|
||||
val msg = ServerMessage.ConfigSnapshot(
|
||||
fields = listOf(
|
||||
ConfigFieldDto("tui.theme", "ENUM", "dark", options = listOf("dark", "light"), restartRequired = false),
|
||||
ConfigFieldDto("server.port", "INT", "8080", restartRequired = true),
|
||||
),
|
||||
restartRequired = listOf("server.port"),
|
||||
error = null,
|
||||
)
|
||||
val jsonStr = ProtocolSerializer.encodeServerMessage(msg)
|
||||
assert(jsonStr.contains("\"type\":\"config.snapshot\"")) { "expected type=config.snapshot" }
|
||||
assert(jsonStr.contains("\"key\":\"tui.theme\"")) { "expected tui.theme field" }
|
||||
assert(jsonStr.contains("\"restartRequired\":[\"server.port\"]")) { "expected restartRequired list" }
|
||||
|
||||
val decoded = json.decodeFromString<ServerMessage.ConfigSnapshot>(jsonStr)
|
||||
assertEquals(2, decoded.fields.size)
|
||||
assertEquals("8080", decoded.fields[1].value)
|
||||
assertEquals(listOf("server.port"), decoded.restartRequired)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `SessionAnnounced round-trips through JSON`() {
|
||||
val original = ServerMessage.SessionAnnounced(
|
||||
|
||||
Reference in New Issue
Block a user