From fe561ada093f44b595ac94d93ab1b75d170c5b05 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 4 Jun 2026 02:16:23 +0400 Subject: [PATCH] =?UTF-8?q?feat(server):=20intent=20input=20channel=20?= =?UTF-8?q?=E2=80=94=20StartSession.input=20seeds=20decision=20journal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../apps/server/protocol/ClientMessage.kt | 8 +++++++- .../apps/server/ws/GlobalStreamHandler.kt | 17 +++++++++++++++++ .../correx/core/events/events/IntentEvents.kt | 18 ++++++++++++++++++ .../core/events/serialization/Serialization.kt | 2 ++ .../journal/DefaultDecisionJournalReducer.kt | 6 ++++++ .../core/journal/model/DecisionRecord.kt | 2 +- .../kotlin/DecisionJournalProjectorTest.kt | 12 ++++++++++++ 7 files changed, 63 insertions(+), 2 deletions(-) create mode 100644 core/events/src/main/kotlin/com/correx/core/events/events/IntentEvents.kt 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 85db2c0b..02f89727 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 @@ -24,7 +24,13 @@ enum class GrantScopeDto { @Serializable sealed class ClientMessage { @Serializable - data class StartSession(val workflowId: String, val config: SessionConfigDto?) : ClientMessage() + data class StartSession( + val workflowId: String, + val config: SessionConfigDto?, + // Optional freeform request for intent-driven pipelines. Seeded into the decision + // journal so every stage sees it. Fixed-task workflows (e.g. healthcheck) omit it. + val input: String? = null, + ) : ClientMessage() @Serializable data class ResumeSession(val sessionId: SessionId) : ClientMessage() 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 83a50911..e82ccf24 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 @@ -17,6 +17,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.InitialIntentEvent import com.correx.core.events.events.NewEvent import com.correx.core.events.events.SessionWorkspaceBoundEvent import com.correx.core.events.events.StoredEvent @@ -466,6 +467,22 @@ class GlobalStreamHandler(private val module: ServerModule) { // the bound workspace root for tool path containment and per-session tool registry. val sessionConfig = module.defaultOrchestrationConfig.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). + msg.input?.takeIf { it.isNotBlank() }?.let { intent -> + module.eventStore.append(NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = InitialIntentEvent(sessionId = sessionId, intent = intent), + )) + } + // Only launch orchestrator after protocol messages are delivered successfully. // Uses launchSessionRun() so the job is tracked in activeSessionJobs — this prevents // the OrchestrationResumedEvent subscription from spuriously launching a duplicate resume. diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/IntentEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/IntentEvents.kt new file mode 100644 index 00000000..ee5e8552 --- /dev/null +++ b/core/events/src/main/kotlin/com/correx/core/events/events/IntentEvents.kt @@ -0,0 +1,18 @@ +package com.correx.core.events.events + +import com.correx.core.events.types.SessionId +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +/** + * The freeform user request that kicked off a workflow run (e.g. "add an auth layer to the + * account service"). Recorded once at session start so it flows into the decision journal and + * is pinned into every stage's context — the input channel for intent-driven pipelines. + * Optional: fixed-task workflows (e.g. healthcheck) start without one. + */ +@Serializable +@SerialName("InitialIntent") +data class InitialIntentEvent( + val sessionId: SessionId, + val intent: String, +) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index e608d4eb..6b5b4b84 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt @@ -16,6 +16,7 @@ import com.correx.core.events.events.EventPayload import com.correx.core.events.events.FileWrittenEvent import com.correx.core.events.events.L3MemoryRetrievedEvent import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.InferenceFailedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceTimeoutEvent @@ -58,6 +59,7 @@ val eventModule = SerializersModule { subclass(ToolCallAssessedEvent::class) subclass(FileWrittenEvent::class) subclass(SteeringNoteAddedEvent::class) + subclass(InitialIntentEvent::class) subclass(StageFailedEvent::class) subclass(StageCompletedEvent::class) subclass(TransitionExecutedEvent::class) diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt index 56843d55..c48d07ce 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt @@ -2,6 +2,7 @@ package com.correx.core.journal import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.SteeringNoteAddedEvent @@ -14,6 +15,11 @@ import com.correx.core.journal.model.DecisionRecord class DefaultDecisionJournalReducer : DecisionJournalReducer { override fun reduce(state: DecisionJournalState, event: StoredEvent): DecisionJournalState { val record = when (val p = event.payload) { + is InitialIntentEvent -> DecisionRecord( + sequence = event.sessionSequence, + kind = DecisionKind.INTENT, + summary = "Goal: ${p.intent}", + ) is SteeringNoteAddedEvent -> DecisionRecord( sequence = event.sessionSequence, kind = DecisionKind.STEERING, diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt index 60adfb60..58809ae7 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt @@ -2,7 +2,7 @@ package com.correx.core.journal.model import kotlinx.serialization.Serializable -enum class DecisionKind { STEERING, APPROVAL, TRANSITION, RETRY, FAILURE, ARTIFACT } +enum class DecisionKind { INTENT, STEERING, APPROVAL, TRANSITION, RETRY, FAILURE, ARTIFACT } /** * One distilled decision. [sequence] is the source event's sessionSequence, used for diff --git a/testing/projections/src/test/kotlin/DecisionJournalProjectorTest.kt b/testing/projections/src/test/kotlin/DecisionJournalProjectorTest.kt index bd4bfde4..e544a6b1 100644 --- a/testing/projections/src/test/kotlin/DecisionJournalProjectorTest.kt +++ b/testing/projections/src/test/kotlin/DecisionJournalProjectorTest.kt @@ -1,3 +1,4 @@ +import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.types.SessionId @@ -35,6 +36,17 @@ class DecisionJournalProjectorTest { assertEquals(DecisionKind.TRANSITION, s.records[1].kind) } + @Test + fun `initial intent becomes a journal record`() { + val s = projector.apply( + projector.initial(), + stored(payload = InitialIntentEvent(SessionId("x"), "add auth layer"), sequence = 1L), + ) + assertEquals(1, s.records.size) + assertEquals(DecisionKind.INTENT, s.records[0].kind) + assertEquals("Goal: add auth layer", s.records[0].summary) + } + @Test fun `unrelated events are ignored`() { val s0 = projector.initial()