refactor(toolintent): one SessionContext read-model for the gates
Replace the single-purpose sessionReadPaths plumbing with a SessionContext
{reads, writes, activeTask} folded by SessionContextProjection over the session
stream — the read-model every anti-hallucination gate consumes (plane-2 rules via
ToolCallAssessmentInput.session; tool gates via a port, next).
- reads: completions whose recorded capability includes FILE_READ.
- writes: the resolved paths the orchestrator already records on each completion's
receipt.affectedEntities (FileAffectingTool) — no param-parsing reinvention.
- activeTask: null for now; populated once claim records a session fact.
ReadBeforeWriteRule now reads input.session.reads; ReadFilesProjection is retired.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+7
-8
@@ -69,7 +69,7 @@ import com.correx.core.events.events.ToolReceipt
|
||||
import com.correx.core.events.events.ToolRequest
|
||||
import com.correx.core.events.risk.RiskAction
|
||||
import com.correx.core.events.risk.RiskSummary
|
||||
import com.correx.core.toolintent.ReadFilesProjection
|
||||
import com.correx.core.toolintent.SessionContextProjection
|
||||
import com.correx.core.toolintent.ToolCallAssessmentInput
|
||||
import com.correx.core.toolintent.ToolCallAssessor
|
||||
import com.correx.core.toolintent.WorkspacePolicy
|
||||
@@ -970,7 +970,7 @@ abstract class SessionOrchestrator(
|
||||
paramRoles = tool?.paramRoles ?: emptyMap(),
|
||||
writeManifest = writeManifest,
|
||||
sessionEgressHosts = sessionEgressHosts,
|
||||
sessionReadPaths = resolveSessionReadPaths(sessionId),
|
||||
session = resolveSessionContext(sessionId),
|
||||
),
|
||||
)
|
||||
emit(
|
||||
@@ -1001,15 +1001,14 @@ abstract class SessionOrchestrator(
|
||||
}
|
||||
|
||||
/**
|
||||
* Folds this session's file_read tool events through [ReadFilesProjection] into the set of paths
|
||||
* it has successfully read — the input to [com.correx.core.toolintent.rules.ReadBeforeWriteRule].
|
||||
* Replay-safe (reads the log, never live state); empty before any read completes.
|
||||
* Folds this session's tool events through [SessionContextProjection] into the read-model the
|
||||
* anti-hallucination gates consume (reads, writes, active task). Replay-safe (reads the log,
|
||||
* never live state); empty before anything has happened.
|
||||
*/
|
||||
internal fun resolveSessionReadPaths(sessionId: SessionId): Set<String> {
|
||||
val projection = ReadFilesProjection(sessionId)
|
||||
internal fun resolveSessionContext(sessionId: SessionId): com.correx.core.toolintent.SessionContext {
|
||||
val projection = SessionContextProjection(sessionId)
|
||||
return eventStore.read(sessionId)
|
||||
.fold(projection.initial()) { acc, event -> projection.apply(acc, event) }
|
||||
.readPaths
|
||||
}
|
||||
|
||||
private suspend fun buildSchemaEntries(
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
package com.correx.core.toolintent
|
||||
|
||||
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.types.SessionId
|
||||
import com.correx.core.sessions.projections.Projection
|
||||
import com.correx.core.toolintent.rules.candidatePathStrings
|
||||
import com.correx.core.tools.contract.ToolCapability
|
||||
|
||||
/** Reads still awaiting their completion event, plus the paths confirmed read so far. */
|
||||
data class ReadFilesState(
|
||||
val pending: Map<String, List<String>> = emptyMap(),
|
||||
val readPaths: Set<String> = emptySet(),
|
||||
)
|
||||
|
||||
/**
|
||||
* Folds one session's tool events into the set of file paths it has successfully read. A call counts
|
||||
* as a read by its recorded [ToolCapability.FILE_READ] capability (not its tool name), and only once
|
||||
* its [ToolExecutionCompletedEvent] lands — a read that failed (file not found) never enters the
|
||||
* set, so it can't satisfy [com.correx.core.toolintent.rules.ReadBeforeWriteRule]. Paths are kept as
|
||||
* the raw arguments; the rule resolves both sides through the probe, so `./Foo.kt` read and
|
||||
* `Foo.kt` written still match. Same projection precedent as
|
||||
* [com.correx.core.sessions.projections.EgressAllowlistProjection].
|
||||
*/
|
||||
class ReadFilesProjection(private val sessionId: SessionId) : Projection<ReadFilesState> {
|
||||
|
||||
override fun initial(): ReadFilesState = ReadFilesState()
|
||||
|
||||
override fun apply(state: ReadFilesState, event: StoredEvent): ReadFilesState =
|
||||
when (val payload = event.payload) {
|
||||
is ToolInvocationRequestedEvent -> recordPending(state, payload)
|
||||
is ToolExecutionCompletedEvent -> confirmRead(state, payload)
|
||||
else -> state
|
||||
}
|
||||
|
||||
private fun recordPending(state: ReadFilesState, payload: ToolInvocationRequestedEvent): ReadFilesState {
|
||||
if (payload.sessionId != sessionId || ToolCapability.FILE_READ !in payload.capabilities) return state
|
||||
val paths = candidatePathStrings(emptyMap(), payload.request.parameters)
|
||||
if (paths.isEmpty()) return state
|
||||
return state.copy(pending = state.pending + (payload.invocationId.value to paths))
|
||||
}
|
||||
|
||||
private fun confirmRead(state: ReadFilesState, payload: ToolExecutionCompletedEvent): ReadFilesState {
|
||||
val paths = state.pending[payload.invocationId.value]
|
||||
if (payload.sessionId != sessionId || paths == null) return state
|
||||
return state.copy(
|
||||
pending = state.pending - payload.invocationId.value,
|
||||
readPaths = state.readPaths + paths,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package com.correx.core.toolintent
|
||||
|
||||
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.types.SessionId
|
||||
import com.correx.core.sessions.projections.Projection
|
||||
import com.correx.core.toolintent.rules.candidatePathStrings
|
||||
import com.correx.core.tools.contract.ToolCapability
|
||||
|
||||
/**
|
||||
* What a session has done to the workspace this run, plus the task it is working — the single
|
||||
* read-model every anti-hallucination gate consumes (plane-2 rules via the threaded
|
||||
* [ToolCallAssessmentInput.session]; tool gates via a provider port). [pendingReads] is
|
||||
* fold-internal (reads awaiting their completion event) and not meant for consumers.
|
||||
*/
|
||||
data class SessionContext(
|
||||
val reads: Set<String> = emptySet(),
|
||||
val writes: Set<String> = emptySet(),
|
||||
val activeTask: ActiveTask? = null,
|
||||
val pendingReads: Map<String, List<String>> = emptyMap(),
|
||||
)
|
||||
|
||||
/** The task this session has claimed and is working, with its declared write scope. */
|
||||
data class ActiveTask(val taskId: String, val scope: List<String>)
|
||||
|
||||
/**
|
||||
* Folds one session's tool events into a [SessionContext]:
|
||||
* - **reads** — completed invocations whose recorded capability includes FILE_READ (path from the
|
||||
* request args; a read that failed never completes, so it never counts).
|
||||
* - **writes** — the resolved paths the orchestrator already records on each completion's
|
||||
* `receipt.affectedEntities` (via FileAffectingTool), so writes need no param-parsing here.
|
||||
* - **activeTask** — set when the session claims a task; null until a SessionWorkingTaskEvent lands.
|
||||
*/
|
||||
class SessionContextProjection(private val sessionId: SessionId) : Projection<SessionContext> {
|
||||
|
||||
override fun initial(): SessionContext = SessionContext()
|
||||
|
||||
override fun apply(state: SessionContext, event: StoredEvent): SessionContext =
|
||||
when (val payload = event.payload) {
|
||||
is ToolInvocationRequestedEvent -> recordPendingRead(state, payload)
|
||||
is ToolExecutionCompletedEvent -> recordCompletion(state, payload)
|
||||
else -> state
|
||||
}
|
||||
|
||||
private fun recordPendingRead(state: SessionContext, payload: ToolInvocationRequestedEvent): SessionContext {
|
||||
if (payload.sessionId != sessionId || ToolCapability.FILE_READ !in payload.capabilities) return state
|
||||
val paths = candidatePathStrings(emptyMap(), payload.request.parameters)
|
||||
if (paths.isEmpty()) return state
|
||||
return state.copy(pendingReads = state.pendingReads + (payload.invocationId.value to paths))
|
||||
}
|
||||
|
||||
private fun recordCompletion(state: SessionContext, payload: ToolExecutionCompletedEvent): SessionContext {
|
||||
if (payload.sessionId != sessionId) return state
|
||||
val justRead = state.pendingReads[payload.invocationId.value]
|
||||
return state.copy(
|
||||
reads = if (justRead != null) state.reads + justRead else state.reads,
|
||||
writes = state.writes + payload.receipt.affectedEntities,
|
||||
pendingReads = state.pendingReads - payload.invocationId.value,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -29,10 +29,10 @@ data class ToolCallAssessmentInput(
|
||||
// EgressAllowlistProjection). Unioned with the static allow-list at the network gate; empty
|
||||
// means "no per-session grants", which never narrows the static allow-list.
|
||||
val sessionEgressHosts: Set<String> = emptySet(),
|
||||
// Paths this session has successfully file_read (folded via ReadFilesProjection). The input to
|
||||
// ReadBeforeWriteRule; empty means nothing has been read yet (so any write to an existing file
|
||||
// is unread). Raw argument strings — the rule resolves both sides through the probe.
|
||||
val sessionReadPaths: Set<String> = emptySet(),
|
||||
// What this session has done this run (reads, writes) and the task it's working — folded via
|
||||
// SessionContextProjection. The input to the anti-hallucination rules (read-before-write reads
|
||||
// .reads, write-scope reads .activeTask). Empty by default so unrelated rules ignore it.
|
||||
val session: SessionContext = SessionContext(),
|
||||
)
|
||||
|
||||
data class ToolCallAssessment(
|
||||
|
||||
+1
-1
@@ -28,7 +28,7 @@ class ReadBeforeWriteRule : ToolCallRule {
|
||||
|
||||
override fun assess(input: ToolCallAssessmentInput): ToolCallAssessment {
|
||||
val root = input.workspace.workspaceRoot
|
||||
val readReal = input.sessionReadPaths.map { realOf(input, root, it) }.toSet()
|
||||
val readReal = input.session.reads.map { realOf(input, root, it) }.toSet()
|
||||
|
||||
val issues = mutableListOf<ValidationIssue>()
|
||||
val observations = mutableListOf<ToolCallObservation>()
|
||||
|
||||
+1
-1
@@ -36,7 +36,7 @@ class ReadBeforeWriteRuleTest {
|
||||
capabilities = setOf(ToolCapability.FILE_WRITE),
|
||||
workspace = WorkspacePolicy(workspace, emptyList()),
|
||||
probe = probe,
|
||||
sessionReadPaths = read,
|
||||
session = SessionContext(reads = read),
|
||||
)
|
||||
|
||||
@Test
|
||||
|
||||
+14
-20
@@ -18,14 +18,14 @@ import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class ReadFilesProjectionTest {
|
||||
class SessionContextProjectionTest {
|
||||
|
||||
private val session = SessionId("s")
|
||||
private val projection = ReadFilesProjection(session)
|
||||
private val projection = SessionContextProjection(session)
|
||||
|
||||
private var seq = 0L
|
||||
private fun stored(payload: EventPayload) = StoredEvent(
|
||||
metadata = EventMetadata(EventId("e-${seq}"), session, Instant.parse("2026-01-01T00:00:00Z"), 1, null, null),
|
||||
metadata = EventMetadata(EventId("e-$seq"), session, Instant.parse("2026-01-01T00:00:00Z"), 1, null, null),
|
||||
sequence = ++seq,
|
||||
sessionSequence = seq,
|
||||
payload = payload,
|
||||
@@ -43,16 +43,17 @@ class ReadFilesProjectionTest {
|
||||
),
|
||||
)
|
||||
|
||||
private fun completed(invId: String, tool: String = "file_read") = stored(
|
||||
private fun completed(invId: String, affected: List<String> = emptyList()) = stored(
|
||||
ToolExecutionCompletedEvent(
|
||||
invocationId = ToolInvocationId(invId),
|
||||
sessionId = session,
|
||||
toolName = tool,
|
||||
toolName = "tool",
|
||||
receipt = ToolReceipt(
|
||||
invocationId = ToolInvocationId(invId),
|
||||
toolName = tool,
|
||||
toolName = "tool",
|
||||
exitCode = 0,
|
||||
outputSummary = "ok",
|
||||
affectedEntities = affected,
|
||||
durationMs = 1,
|
||||
tier = Tier.T1,
|
||||
timestamp = Instant.parse("2026-01-01T00:00:00Z"),
|
||||
@@ -64,22 +65,15 @@ class ReadFilesProjectionTest {
|
||||
events.fold(projection.initial()) { acc, e -> projection.apply(acc, e) }
|
||||
|
||||
@Test
|
||||
fun `a completed file_read adds the path to the read set`() {
|
||||
val state = fold(listOf(readRequested("r1", "src/A.kt"), completed("r1")))
|
||||
assertEquals(setOf("src/A.kt"), state.readPaths)
|
||||
fun `a completed read enters reads but a pending one does not`() {
|
||||
assertEquals(setOf("src/A.kt"), fold(listOf(readRequested("r1", "src/A.kt"), completed("r1"))).reads)
|
||||
assertTrue(fold(listOf(readRequested("r1", "src/A.kt"))).reads.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a read still pending its completion does not count`() {
|
||||
val state = fold(listOf(readRequested("r1", "src/A.kt")))
|
||||
assertTrue(state.readPaths.isEmpty())
|
||||
assertTrue(state.pending.containsKey("r1"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a non-read tool completion is ignored`() {
|
||||
// file_write request never enters pending, so its completion adds nothing.
|
||||
val state = fold(listOf(completed("w1", tool = "file_write")))
|
||||
assertTrue(state.readPaths.isEmpty())
|
||||
fun `writes come from the completion's affectedEntities`() {
|
||||
val ctx = fold(listOf(completed("w1", affected = listOf("core/A.kt", "core/B.kt"))))
|
||||
assertEquals(setOf("core/A.kt", "core/B.kt"), ctx.writes)
|
||||
assertTrue(ctx.reads.isEmpty())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user