From 30321e08a580de6ef3c58c909cd63c0eb94a8017 Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 24 Jun 2026 20:35:40 +0000 Subject: [PATCH] feat(toolintent): record the session's active task as a session-local fact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claiming a task now emits SessionWorkingTaskEvent{taskId, affectedPaths} into the session's own stream (via a SessionFactRecorder port + event-store adapter), and SessionContextProjection folds it into SessionContext.activeTask. So a gate learns "which task is this session working, and its scope" by reading the session stream — no cross-stream scan, no kernel->tasks dependency. Re-emitted on an affected_paths edit by the claimant, so the snapshot scope stays current; latest wins on replay. This is the data source the write-scope gate needs (next). Co-Authored-By: Claude Opus 4.8 --- .../kotlin/com/correx/apps/server/Main.kt | 1 + .../tasks/EventStoreSessionFactRecorder.kt | 36 +++++++++++++++++++ .../core/events/events/SessionEvents.kt | 15 ++++++++ .../events/serialization/Serialization.kt | 2 ++ .../correx/core/toolintent/SessionContext.kt | 7 ++++ .../SessionContextProjectionTest.kt | 14 ++++++++ .../tools/task/SessionFactRecorder.kt | 14 ++++++++ .../infrastructure/tools/task/TaskTools.kt | 3 +- .../tools/task/TaskUpdateTool.kt | 23 +++++++++++- .../tools/task/TaskToolsTest.kt | 18 ++++++++++ 10 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/tasks/EventStoreSessionFactRecorder.kt create mode 100644 infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/SessionFactRecorder.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 93621ec2..8bf9a944 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -238,6 +238,7 @@ fun main() { taskDocumentResolver, taskArtifactResolver, taskSessionResolver, + com.correx.apps.server.tasks.EventStoreSessionFactRecorder(eventStore), ) val toolRegistry = InfrastructureModule.createToolRegistry( buildToolConfig( diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/tasks/EventStoreSessionFactRecorder.kt b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/EventStoreSessionFactRecorder.kt new file mode 100644 index 00000000..6a16a92e --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/EventStoreSessionFactRecorder.kt @@ -0,0 +1,36 @@ +package com.correx.apps.server.tasks + +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.SessionWorkingTaskEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.TaskId +import com.correx.infrastructure.tools.task.SessionFactRecorder +import kotlinx.datetime.Clock +import java.util.UUID + +/** + * Adapter for [SessionFactRecorder]: appends a [SessionWorkingTaskEvent] to the session's own stream + * when it claims (or re-scopes) a task, so [com.correx.core.toolintent.SessionContextProjection] + * folds it into the active task without scanning task streams. + */ +class EventStoreSessionFactRecorder(private val eventStore: EventStore) : SessionFactRecorder { + + override suspend fun recordWorkingTask(sessionId: SessionId, taskId: TaskId, affectedPaths: List) { + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = SessionWorkingTaskEvent(sessionId, taskId, affectedPaths), + ), + ) + } +} diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt index 5dbefe81..86087dcd 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt @@ -1,6 +1,7 @@ package com.correx.core.events.events import com.correx.core.events.types.SessionId +import com.correx.core.events.types.TaskId import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable @@ -10,6 +11,20 @@ data class ChatSessionStartedEvent( val sessionId: SessionId, ) : EventPayload +/** + * Records, in the session's own stream, that this run claimed [taskId] and is working it with the + * declared [affectedPaths] write scope — a session-local fact so the gates know the active task + * without scanning task streams. Re-emitted when the claimant edits affected_paths, so the folded + * scope stays current; SessionContextProjection takes the most recent. + */ +@Serializable +@SerialName("SessionWorkingTask") +data class SessionWorkingTaskEvent( + val sessionId: SessionId, + val taskId: TaskId, + val affectedPaths: List = emptyList(), +) : EventPayload + @Serializable @SerialName("SessionWorkspaceBound") data class SessionWorkspaceBoundEvent( 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 76667f4d..e30f4f27 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 @@ -12,6 +12,7 @@ import com.correx.core.events.events.ArtifactValidatingEvent import com.correx.core.events.events.BriefEchoMismatchEvent import com.correx.core.events.events.BriefGroundingCheckedEvent import com.correx.core.events.events.ChatSessionStartedEvent +import com.correx.core.events.events.SessionWorkingTaskEvent import com.correx.core.events.events.ClarificationAnsweredEvent import com.correx.core.events.events.ClarificationRequestedEvent import com.correx.core.events.events.ChatTurnEvent @@ -139,6 +140,7 @@ val eventModule = SerializersModule { subclass(BriefEchoMismatchEvent::class) subclass(RiskAssessedEvent::class) subclass(ChatSessionStartedEvent::class) + subclass(SessionWorkingTaskEvent::class) subclass(ChatTurnEvent::class) subclass(RouterNarrationEvent::class) subclass(SessionWorkspaceBoundEvent::class) diff --git a/core/toolintent/src/main/kotlin/com/correx/core/toolintent/SessionContext.kt b/core/toolintent/src/main/kotlin/com/correx/core/toolintent/SessionContext.kt index d7127a00..599f2dfd 100644 --- a/core/toolintent/src/main/kotlin/com/correx/core/toolintent/SessionContext.kt +++ b/core/toolintent/src/main/kotlin/com/correx/core/toolintent/SessionContext.kt @@ -1,5 +1,6 @@ package com.correx.core.toolintent +import com.correx.core.events.events.SessionWorkingTaskEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.ToolExecutionCompletedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent @@ -40,6 +41,12 @@ class SessionContextProjection(private val sessionId: SessionId) : Projection recordPendingRead(state, payload) is ToolExecutionCompletedEvent -> recordCompletion(state, payload) + is SessionWorkingTaskEvent -> + if (payload.sessionId == sessionId) { + state.copy(activeTask = ActiveTask(payload.taskId.value, payload.affectedPaths)) + } else { + state + } else -> state } diff --git a/core/toolintent/src/test/kotlin/com/correx/core/toolintent/SessionContextProjectionTest.kt b/core/toolintent/src/test/kotlin/com/correx/core/toolintent/SessionContextProjectionTest.kt index 93e870ed..0f1053a7 100644 --- a/core/toolintent/src/test/kotlin/com/correx/core/toolintent/SessionContextProjectionTest.kt +++ b/core/toolintent/src/test/kotlin/com/correx/core/toolintent/SessionContextProjectionTest.kt @@ -6,9 +6,11 @@ import com.correx.core.events.events.EventPayload import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.ToolExecutionCompletedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.SessionWorkingTaskEvent import com.correx.core.events.events.ToolReceipt import com.correx.core.events.events.ToolRequest import com.correx.core.events.types.EventId +import com.correx.core.events.types.TaskId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.events.types.ToolInvocationId @@ -76,4 +78,16 @@ class SessionContextProjectionTest { assertEquals(setOf("core/A.kt", "core/B.kt"), ctx.writes) assertTrue(ctx.reads.isEmpty()) } + + @Test + fun `a SessionWorkingTask event sets the active task and the latest wins`() { + val ctx = fold( + listOf( + stored(SessionWorkingTaskEvent(session, TaskId("auth-1"), listOf("core/a/**"))), + stored(SessionWorkingTaskEvent(session, TaskId("auth-2"), listOf("core/auth/**"))), + ), + ) + assertEquals("auth-2", ctx.activeTask?.taskId) + assertEquals(listOf("core/auth/**"), ctx.activeTask?.scope) + } } diff --git a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/SessionFactRecorder.kt b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/SessionFactRecorder.kt new file mode 100644 index 00000000..ab64119b --- /dev/null +++ b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/SessionFactRecorder.kt @@ -0,0 +1,14 @@ +package com.correx.infrastructure.tools.task + +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.TaskId + +/** + * Port: record that a session is working a task — emitted on claim (and on an affected_paths edit by + * the claimant) so the gates can fold it into SessionContext.activeTask without scanning task + * streams. The host supplies an adapter that appends a SessionWorkingTaskEvent to the session + * stream. A null binding means no recording (the bare tool still works). + */ +fun interface SessionFactRecorder { + suspend fun recordWorkingTask(sessionId: SessionId, taskId: TaskId, affectedPaths: List) +} diff --git a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskTools.kt b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskTools.kt index ee93d5e7..e7460d96 100644 --- a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskTools.kt +++ b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskTools.kt @@ -21,10 +21,11 @@ object TaskTools { documentResolver: TaskDocumentResolver? = null, artifactResolver: TaskArtifactResolver? = null, sessionResolver: TaskSessionResolver? = null, + sessionFacts: SessionFactRecorder? = null, ): List = listOf( TaskCreateTool(service), - TaskUpdateTool(service), + TaskUpdateTool(service, sessionFacts), TaskDeleteTool(service), TaskSearchTool(service), TaskReadyTool(service), diff --git a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskUpdateTool.kt b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskUpdateTool.kt index bc9a5677..2c6f4c26 100644 --- a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskUpdateTool.kt +++ b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/task/TaskUpdateTool.kt @@ -25,7 +25,10 @@ import kotlinx.serialization.json.putJsonObject * Agent-facing tool: update an existing task — edit fields, transition status, add a work-graph * link, or attach a note. The task's project is derived from its id, so only the id is required. */ -class TaskUpdateTool(private val service: TaskService) : Tool, ToolExecutor { +class TaskUpdateTool( + private val service: TaskService, + private val sessionFacts: SessionFactRecorder? = null, +) : Tool, ToolExecutor { override val name: String = "task_update" override val description: String = @@ -110,6 +113,8 @@ class TaskUpdateTool(private val service: TaskService) : Tool, ToolExecutor { request.stringParam("note")?.let { service.addNote(taskId, TaskNoteAuthor.AGENT, it) } + recordWorkingTask(request, taskId, action) + val status = service.getTask(taskId)?.state?.status?.name ?: "UNKNOWN" val warning = if (action == "claim") claimWarning(taskId) else "" return ToolResult.Success( @@ -128,6 +133,22 @@ class TaskUpdateTool(private val service: TaskService) : Tool, ToolExecutor { return fail(request, "Cannot claim ${taskId.value} — unmet blockers: $listed. Complete them, or pass force=true.") } + /** + * Record the session→task working relationship (for SessionContext.activeTask) on a claim, or on + * an affected_paths edit by the recorded claimant (to refresh the snapshot scope). No-op without + * a recorder bound. + */ + private suspend fun recordWorkingTask(request: ToolRequest, taskId: TaskId, action: String?) { + val recorder = sessionFacts ?: return + val task = service.getTask(taskId) ?: return + val isClaim = action == "claim" + val scopeEditByClaimant = request.parameters.containsKey("affected_paths") && + task.state.claimant == request.sessionId.value + if (isClaim || scopeEditByClaimant) { + recorder.recordWorkingTask(request.sessionId, taskId, task.state.affectedPaths) + } + } + /** When a force-claimed task still has unmet blockers, flag it so the agent knows it jumped a dependency. */ private fun claimWarning(taskId: TaskId): String { val blockers = service.blockers(taskId) diff --git a/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/task/TaskToolsTest.kt b/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/task/TaskToolsTest.kt index 20abf70a..8a5b5123 100644 --- a/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/task/TaskToolsTest.kt +++ b/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/task/TaskToolsTest.kt @@ -210,6 +210,24 @@ class TaskToolsTest { assertEquals(TaskStatus.IN_PROGRESS, service.getTask(TaskId("auth-2"))!!.state.status) } + @Test + fun `claiming a task records the session working it`() = runBlocking { + create.execute( + request( + "task_create", + mapOf("project" to "auth", "title" to "A", "goal" to "g", "affected_paths" to listOf("core/auth/**")), + ), + ) + val captured = mutableListOf>>() + val recorder = SessionFactRecorder { sid, tid, paths -> captured.add(Triple(sid.value, tid.value, paths)) } + + TaskUpdateTool(service, recorder).execute(request("task_update", mapOf("id" to "auth-1", "action" to "claim"))) + + assertEquals(1, captured.size) + assertEquals("auth-1", captured.single().second) + assertEquals(listOf("core/auth/**"), captured.single().third) + } + @Test fun `task_delete tombstones the task`() = runBlocking { val id = createTask()