622b331de3
Wires the workspace handshake end to end so a session's workspace is bound
from the client's cwd at connect time and is event-sourced for replay.
- Go TUI sends Hello{workingDir=os.Getwd()} as the first WS frame on connect
(protocol.go encoder + client.go connect path; golden test pins the wire
format against the Kotlin discriminator).
- Server adds ClientMessage.Hello, stashes the per-connection workingDir, and
on session start resolves it through WorkspaceResolver's trust pipeline,
emits SessionWorkspaceBoundEvent (invariant #9), and threads the resolved
workspace into OrchestrationConfig for the live run. A Hello after the first
StartSession is ignored (warn); a rejected path binds the resolver fallback.
- Replay derives the workspace from the recorded event: SessionState gains
boundWorkspace, DefaultSessionReducer fills it from SessionWorkspaceBoundEvent,
and ReplayOrchestrator uses it (Path.of only, no filesystem re-query —
invariant #8) with graceful fallback to config for pre-Phase-B logs.
Absent Hello / null resolver degrades to the prior config-workspace behavior.
296 lines
8.8 KiB
Kotlin
296 lines
8.8 KiB
Kotlin
import com.correx.core.events.events.OrchestrationPausedEvent
|
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
|
import com.correx.core.events.events.SessionWorkspaceBoundEvent
|
|
import com.correx.core.events.events.StageCompletedEvent
|
|
import com.correx.core.events.events.StageFailedEvent
|
|
import com.correx.core.events.events.TransitionExecutedEvent
|
|
import com.correx.core.events.events.WorkflowCompletedEvent
|
|
import com.correx.core.events.events.WorkflowFailedEvent
|
|
import com.correx.core.events.events.WorkflowStartedEvent
|
|
import com.correx.core.events.types.SessionId
|
|
import com.correx.core.events.types.StageId
|
|
import com.correx.core.events.types.TransitionId
|
|
import com.correx.core.sessions.BoundWorkspace
|
|
import com.correx.core.sessions.DefaultSessionReducer
|
|
import com.correx.core.sessions.SessionState
|
|
import com.correx.core.sessions.SessionStatus
|
|
import com.correx.testing.fixtures.EventFixtures.stored
|
|
import kotlinx.datetime.Instant
|
|
import org.junit.jupiter.api.Assertions.assertEquals
|
|
import org.junit.jupiter.api.Test
|
|
|
|
class DefaultSessionReducerTest {
|
|
|
|
private val reducer = DefaultSessionReducer()
|
|
|
|
private val sessionId = SessionId("session-1")
|
|
|
|
@Test
|
|
fun `WorkflowStartedEvent leaves CREATED status`() {
|
|
val state = initialState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = WorkflowStartedEvent(sessionId, workflowId = "test-wf", startStageId = StageId("st-1"))
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.CREATED, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `OrchestrationPausedEvent leaves ACTIVE status`() {
|
|
val state = activeState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = OrchestrationPausedEvent(sessionId, stageId = StageId("st-1"), reason = "APPROVAL_PENDING")
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `OrchestrationResumedEvent leaves PAUSED status`() {
|
|
val state = pausedState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = OrchestrationResumedEvent(sessionId, stageId = StageId("st-1"))
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.PAUSED, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `WorkflowCompletedEvent leaves ACTIVE status`() {
|
|
val state = activeState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = WorkflowCompletedEvent(sessionId, terminalStageId = StageId("st-1"), totalStages = 0)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `WorkflowFailedEvent leaves ACTIVE status`() {
|
|
val state = activeState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = WorkflowFailedEvent(sessionId, stageId = StageId("st-1"), reason = "error", retryExhausted = false)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `TransitionExecutedEvent from StageStartedEvent sets ACTIVE status`() {
|
|
val state = pausedState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = TransitionExecutedEvent(
|
|
sessionId,
|
|
from = StageId("st-1"),
|
|
to = StageId("st-1"),
|
|
transitionId = TransitionId("transition-a")
|
|
)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `StageCompletedEvent sets ACTIVE status`() {
|
|
val state = pausedState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = StageCompletedEvent(
|
|
sessionId,
|
|
stageId = StageId("stage-a"),
|
|
transitionId = TransitionId("transition-a")
|
|
)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `StageFailedEvent sets FAILED status`() {
|
|
val state = activeState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = StageFailedEvent(
|
|
sessionId,
|
|
stageId = StageId("stage-a"),
|
|
transitionId = TransitionId("transition-a"),
|
|
reason = "boom"
|
|
)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.FAILED, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `TransitionExecutedEvent sets ACTIVE status`() {
|
|
val state = pausedState()
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = TransitionExecutedEvent(
|
|
sessionId,
|
|
from = StageId("stage-a"),
|
|
to = StageId("stage-b"),
|
|
transitionId = TransitionId("transition-a"),
|
|
)
|
|
)
|
|
)
|
|
|
|
assertEquals(SessionStatus.ACTIVE, result.status)
|
|
}
|
|
|
|
@Test
|
|
fun `createdAt is initialized once`() {
|
|
val timestamp = Instant.parse("2026-01-01T00:00:00Z")
|
|
|
|
val result = reducer.reduce(
|
|
state = initialState(),
|
|
event = stored(
|
|
payload = WorkflowStartedEvent(sessionId, workflowId = "test-wf", startStageId = StageId("st-1")),
|
|
timestamp = timestamp
|
|
)
|
|
)
|
|
|
|
assertEquals(timestamp, result.createdAt)
|
|
}
|
|
|
|
@Test
|
|
fun `createdAt is preserved after initialization`() {
|
|
val createdAt = Instant.parse("2026-01-01T00:00:00Z")
|
|
val updatedAt = Instant.parse("2026-01-02T00:00:00Z")
|
|
|
|
val state = activeState().copy(
|
|
createdAt = createdAt
|
|
)
|
|
|
|
val result = reducer.reduce(
|
|
state = state,
|
|
event = stored(
|
|
payload = OrchestrationPausedEvent(sessionId, stageId = StageId("st-1"), reason = "APPROVAL_PENDING"),
|
|
timestamp = updatedAt
|
|
)
|
|
)
|
|
|
|
assertEquals(createdAt, result.createdAt)
|
|
}
|
|
|
|
@Test
|
|
fun `updatedAt always reflects latest event timestamp`() {
|
|
val timestamp = Instant.parse("2026-01-02T00:00:00Z")
|
|
|
|
val result = reducer.reduce(
|
|
state = activeState(),
|
|
event = stored(
|
|
payload = OrchestrationPausedEvent(sessionId, stageId = StageId("st-1"), reason = "APPROVAL_PENDING"),
|
|
timestamp = timestamp
|
|
)
|
|
)
|
|
|
|
assertEquals(timestamp, result.updatedAt)
|
|
}
|
|
|
|
@Test
|
|
fun `SessionWorkspaceBoundEvent sets boundWorkspace in state`() {
|
|
val result = reducer.reduce(
|
|
state = initialState(),
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = SessionWorkspaceBoundEvent(
|
|
sessionId = sessionId,
|
|
workspaceRoot = "/home/user/project",
|
|
allowedPaths = listOf("/home/user/project"),
|
|
)
|
|
)
|
|
)
|
|
|
|
assertEquals(
|
|
BoundWorkspace(
|
|
workspaceRoot = "/home/user/project",
|
|
allowedPaths = listOf("/home/user/project"),
|
|
),
|
|
result.boundWorkspace,
|
|
)
|
|
}
|
|
|
|
@Test
|
|
fun `boundWorkspace is preserved by subsequent unrelated events`() {
|
|
val withWorkspace = initialState().copy(
|
|
boundWorkspace = BoundWorkspace(
|
|
workspaceRoot = "/home/user/project",
|
|
allowedPaths = listOf("/home/user/project"),
|
|
)
|
|
)
|
|
|
|
val result = reducer.reduce(
|
|
state = withWorkspace,
|
|
event = stored(
|
|
sessionId = sessionId,
|
|
payload = WorkflowStartedEvent(sessionId, workflowId = "wf", startStageId = StageId("s1"))
|
|
)
|
|
)
|
|
|
|
assertEquals(
|
|
BoundWorkspace(
|
|
workspaceRoot = "/home/user/project",
|
|
allowedPaths = listOf("/home/user/project"),
|
|
),
|
|
result.boundWorkspace,
|
|
)
|
|
}
|
|
|
|
private fun initialState() =
|
|
SessionState(
|
|
status = SessionStatus.CREATED
|
|
)
|
|
|
|
private fun activeState() =
|
|
SessionState(
|
|
status = SessionStatus.ACTIVE
|
|
)
|
|
|
|
private fun pausedState() =
|
|
SessionState(
|
|
status = SessionStatus.PAUSED
|
|
)
|
|
}
|