From 04e931c860b146c260cb664655b161b70a17c7fa Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 14 Jun 2026 14:33:11 +0400 Subject: [PATCH] feat(server): idea board list + discard ListIdeas -> idea.list (StreamQueries folds the board from the log, replay-neutral) and DiscardIdea -> append IdeaDiscardedEvent landed in the capturing session, then reply with the refreshed board. IdeaDto on the wire. --- .../apps/server/protocol/ClientMessage.kt | 8 +++++ .../com/correx/apps/server/protocol/Dtos.kt | 8 +++++ .../apps/server/protocol/ServerMessage.kt | 12 +++++++ .../apps/server/ws/GlobalStreamHandler.kt | 32 +++++++++++++++++++ .../correx/apps/server/ws/StreamQueries.kt | 13 ++++++++ .../ServerMessageSerializationTest.kt | 25 +++++++++++++++ 6 files changed, 98 insertions(+) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt index f4b9c001..1c796f69 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt @@ -48,6 +48,14 @@ sealed class ClientMessage { @Serializable data class GetSessionStats(val sessionId: SessionId) : ClientMessage() + /** Operator request for the cross-session idea board (replied to with an IdeaList). */ + @Serializable + data object ListIdeas : ClientMessage() + + /** Operator request to remove an idea from the board (tombstones it; reply is a fresh IdeaList). */ + @Serializable + data class DiscardIdea(val ideaId: String) : ClientMessage() + /** Operator request for the current editable config (replied to with a ConfigSnapshot). */ @Serializable data object GetConfig : ClientMessage() diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt index 1811a83e..bbd16985 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt @@ -113,6 +113,14 @@ data class ArtifactSummaryDto( val content: String?, ) +/** One idea on the cross-session board (a captured, not-yet-discarded note). */ +@Serializable +data class IdeaDto( + val ideaId: String, + val text: String, + val capturedAtMs: Long, +) + internal fun RiskSummary.toDto(): RiskSummaryDto = RiskSummaryDto( level = level.name, factors = signals.map { signal -> 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 15bbba8f..31625a57 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 @@ -464,6 +464,18 @@ sealed interface ServerMessage { override val sessionSequence: Long? = null, ) : ServerMessage, NonEventMessage + /** + * The cross-session idea board (reply to ListIdeas / DiscardIdea). Not event-derived (a + * snapshot read folded from the whole event log), so cursors are null. + */ + @Serializable + @SerialName("idea.list") + data class IdeaList( + val ideas: List, + override val sequence: Long? = null, + override val sessionSequence: Long? = null, + ) : ServerMessage, NonEventMessage + /** * A session's derived metrics, returned in response to a GetSessionStats request. Not * event-derived (it is a projection-replay snapshot), so cursors are null. [stats] is the same diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index f29ed987..41d3c9e5 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -18,6 +18,7 @@ import com.correx.core.approvals.Tier import com.correx.core.events.events.ApprovalGrantCreatedEvent import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.IdeaDiscardedEvent import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.NewEvent import com.correx.core.events.events.SessionWorkspaceBoundEvent @@ -29,6 +30,7 @@ import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.inference.ProviderHealth import com.correx.core.kernel.orchestration.WorkspaceContext +import com.correx.core.router.IdeaReader import com.correx.core.utils.TypeId import com.correx.infrastructure.inference.commons.ResourceProbe import com.correx.infrastructure.inference.commons.ResourceSnapshot @@ -229,6 +231,8 @@ class GlobalStreamHandler(private val module: ServerModule) { .onFailure { log.error("cancel failed for session={}: {}", msg.sessionId.value, it.message, it) } } is ClientMessage.ListArtifacts -> sendFrame(queries.listArtifacts(msg.sessionId)) + is ClientMessage.ListIdeas -> sendFrame(queries.listIdeas()) + is ClientMessage.DiscardIdea -> handleDiscardIdea(msg, sendFrame) is ClientMessage.GetSessionStats -> sendFrame(queries.sessionStats(msg.sessionId)) is ClientMessage.GetConfig -> sendFrame(queries.configSnapshot(error = null, restartRequired = emptyList())) is ClientMessage.UpdateConfig -> sendFrame(queries.updateConfig(msg.patch)) @@ -304,6 +308,34 @@ class GlobalStreamHandler(private val module: ServerModule) { } } + // Tombstone an idea, landing the discard in the session that captured it (so all of one idea's + // events stay co-located), then reply with the refreshed board. A missing idea is a no-op. + private suspend fun handleDiscardIdea( + msg: ClientMessage.DiscardIdea, + sendFrame: suspend (ServerMessage) -> Unit, + ) { + runCatching { + IdeaReader(module.eventStore).sessionOf(msg.ideaId)?.let { sessionId -> + module.eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = IdeaDiscardedEvent(ideaId = msg.ideaId, sessionId = sessionId), + ), + ) + } + }.onFailure { + log.error("discardIdea failed for ideaId={}: {}", msg.ideaId, it.message, it) + } + sendFrame(queries.listIdeas()) + } + private fun errorResponse(message: String) = ServerMessage.ProtocolError(message) private fun encodeError(message: String): Frame.Text = diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt index bc2bd1c2..0c1331ff 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt @@ -5,7 +5,9 @@ import com.correx.apps.server.config.ConfigUpdateResult import com.correx.apps.server.metrics.MetricsInspectionService import com.correx.apps.server.protocol.ArtifactSummaryDto import com.correx.apps.server.protocol.ConfigFieldDto +import com.correx.apps.server.protocol.IdeaDto import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.router.IdeaReader import com.correx.core.config.EditableConfig import com.correx.core.events.events.ArtifactContentStoredEvent import com.correx.core.events.events.ArtifactCreatedEvent @@ -91,6 +93,17 @@ class StreamQueries(private val module: ServerModule) { } ?: rawString } + /** + * The cross-session idea board as a snapshot frame, folded from the whole event log via + * [IdeaReader]. Pure read — no events appended, so it is replay-neutral (invariant #1/#8). + */ + suspend fun listIdeas(): ServerMessage.IdeaList { + val ideas = withContext(Dispatchers.IO) { IdeaReader(module.eventStore).activeIdeas() } + return ServerMessage.IdeaList( + ideas = ideas.map { IdeaDto(ideaId = it.id, text = it.text, capturedAtMs = it.capturedAtMs) }, + ) + } + /** Current editable config as a snapshot frame. [error] / [restartRequired] are set by updates. */ fun configSnapshot(error: String?, restartRequired: List): ServerMessage.ConfigSnapshot { val fields = module.configService.snapshot().map { (field, value) -> diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt index 56b3ec9f..8be8fa5c 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt @@ -242,6 +242,31 @@ class ServerMessageSerializationTest { assertEquals("", decoded.candidates[1].reason) } + @Test + fun `IdeaList encodes type and ideas`() { + val msg = ServerMessage.IdeaList( + ideas = listOf( + IdeaDto(ideaId = "i1", text = "cache the repo map", capturedAtMs = 1000), + IdeaDto(ideaId = "i2", text = "add --dry-run", capturedAtMs = 2000), + ), + ) + val jsonStr = ProtocolSerializer.encodeServerMessage(msg) + assert(jsonStr.contains("\"type\":\"idea.list\"")) { "expected type=idea.list" } + + val decoded = json.decodeFromString(jsonStr) + assertEquals(2, decoded.ideas.size) + assertEquals("cache the repo map", decoded.ideas.first().text) + assertEquals(2000, decoded.ideas[1].capturedAtMs) + } + + @Test + fun `DiscardIdea decodes from the client wire format`() { + val wire = """{"type":"com.correx.apps.server.protocol.ClientMessage.DiscardIdea","ideaId":"i1"}""" + val decoded = ProtocolSerializer.decodeClientMessage(wire) + assert(decoded is ClientMessage.DiscardIdea) { "expected DiscardIdea, got $decoded" } + assertEquals("i1", (decoded as ClientMessage.DiscardIdea).ideaId) + } + @Test fun `ClarificationResponse decodes from the client wire format`() { val wire = """