feat(server): intent input channel — StartSession.input seeds decision journal

This commit is contained in:
2026-06-04 02:16:23 +04:00
parent 6393d578fd
commit fe561ada09
7 changed files with 63 additions and 2 deletions
@@ -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()
@@ -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.
@@ -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
@@ -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)
@@ -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,
@@ -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
@@ -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()