feat(kernel): deterministic per-task claim stage + per-task write scope

A stage flagged claimTask gets the kernel (not the LLM) to claim the next ready task — or re-surface the one already claimed on a review bounce — and inject its TaskContextAssembler bundle as L0. While a task is claimed, plane-2's write manifest narrows to the task's affectedPaths. Project identity is resolved from the origin-session link the task tools already record (TaskService.projectForSession), so the loop predicate and claim stage agree regardless of the free-form project key the agent passed. Claiming flows through a kernel-declared TaskClaimCoordinator implemented in apps/server (no cross-core import).
2026-06-29 00:56:55 +04:00
parent c1e4c7b25e
commit 3e44c6d107
8 changed files with 159 additions and 23 deletions
@@ -368,7 +368,17 @@ fun main() {
        compactionService = journalCompactionService,
        artifactKindRegistry = artifactKindRegistry,
        repoKnowledgeRetriever = repoKnowledgeRetriever,
-        readyTaskCounter = com.correx.apps.server.tasks.ProjectReadyTaskCounter(eventStore, taskService),
+        readyTaskCounter = com.correx.apps.server.tasks.ProjectReadyTaskCounter(taskService),
+        taskClaimCoordinator = com.correx.apps.server.tasks.DefaultTaskClaimCoordinator(
+            taskService,
+            com.correx.core.tasks.TaskContextAssembler(
+                taskService,
+                taskKnowledgeRetriever,
+                taskDocumentResolver,
+                taskArtifactResolver,
+                taskSessionResolver,
+            ),
+        ),
    )
    val workflowRegistry = FileSystemWorkflowRegistry(
        InfrastructureModule.createWorkflowLoader(configArtifactKindsEarly),
@@ -0,0 +1,29 @@
+package com.correx.apps.server.tasks
+
+import com.correx.core.events.types.SessionId
+import com.correx.core.kernel.orchestration.TaskClaimCoordinator
+import com.correx.core.tasks.TaskContextAssembler
+import com.correx.core.tasks.TaskService
+
+/**
+ * [TaskClaimCoordinator] over the task projection: claims the session's next ready task (claimant =
+ * session id) and renders its [TaskContextAssembler] bundle as the implementer's L0 context. The
+ * claim emits a `TaskClaimedEvent`, so the task leaves `ready()` and the `tasks_ready` predicate can
+ * eventually terminate the loop — driven by recorded fact, not agent compliance.
+ */
+class DefaultTaskClaimCoordinator(
+    private val taskService: TaskService,
+    private val assembler: TaskContextAssembler,
+) : TaskClaimCoordinator {
+
+    override suspend fun claimNext(sessionId: SessionId): String? {
+        val active = taskService.activeClaim(sessionId.value)
+        val task = active ?: taskService.nextReadyForSession(sessionId.value)
+            ?.also { taskService.claim(it.taskId, sessionId.value) }
+            ?: return null
+        return assembler.assemble(task.taskId)?.render()
+    }
+
+    override fun activeScope(sessionId: SessionId): List<String> =
+        taskService.activeClaim(sessionId.value)?.state?.affectedPaths ?: emptyList()
+}
@@ -1,31 +1,18 @@
 package com.correx.apps.server.tasks

-import com.correx.core.approvals.ProjectIdentity
-import com.correx.core.events.events.SessionWorkspaceBoundEvent
-import com.correx.core.events.stores.EventStore
 import com.correx.core.events.types.SessionId
 import com.correx.core.kernel.orchestration.ReadyTaskCounter
 import com.correx.core.tasks.TaskService

 /**
- * [ReadyTaskCounter] over the task projection. Resolves the session to its project the same way the
- * approval gate does (bound workspace root → [ProjectIdentity.of]) and counts the ready tasks there.
- * Pure read over event-derived state, so it stays replay-safe.
- *
- * Caveat for the execution loop: tasks are keyed by whatever `project` string the decomposer passes
- * to task tools; this counts the project derived from the workspace root. Those must agree for the
- * loop predicate to fire — Slice 2/4 pins that down.
+ * [ReadyTaskCounter] over the task projection. Resolves the session to the project its tasks were
+ * actually filed under ([TaskService.projectForSession] — via the CONTEXT/SESSION link the task
+ * tools record) and counts the ready tasks there, so the loop predicate agrees with the claim stage
+ * regardless of the free-form `project` key the agent chose. Pure read over event-derived state.
 */
 class ProjectReadyTaskCounter(
-    private val eventStore: EventStore,
    private val taskService: TaskService,
 ) : ReadyTaskCounter {
-    override fun count(sessionId: SessionId): Int {
-        val workspaceRoot = eventStore.read(sessionId)
-            .mapNotNull { it.payload as? SessionWorkspaceBoundEvent }
-            .lastOrNull()
-            ?.workspaceRoot
-            ?: return 0
-        return taskService.ready(ProjectIdentity.of(workspaceRoot)).size
-    }
+    override fun count(sessionId: SessionId): Int =
+        taskService.projectForSession(sessionId.value)?.let { taskService.ready(it).size } ?: 0
 }
@@ -62,7 +62,8 @@ class DefaultSessionOrchestrator(
    artifactKindRegistry: ArtifactKindRegistry? = null,
    repoKnowledgeRetriever: RepoKnowledgeRetriever? = null,
    readyTaskCounter: ReadyTaskCounter? = null,
-) : SessionOrchestrator(repositories, engines, artifactStore, decisionJournalRepository, artifactKindRegistry = artifactKindRegistry, repoKnowledgeRetriever = repoKnowledgeRetriever, readyTaskCounter = readyTaskCounter), ApprovalGateway {
+    taskClaimCoordinator: TaskClaimCoordinator? = null,
+) : SessionOrchestrator(repositories, engines, artifactStore, decisionJournalRepository, artifactKindRegistry = artifactKindRegistry, repoKnowledgeRetriever = repoKnowledgeRetriever, readyTaskCounter = readyTaskCounter, taskClaimCoordinator = taskClaimCoordinator), ApprovalGateway {
    override val tokenizer: Tokenizer? = tokenizer
    override val cancellations: ConcurrentHashMap<SessionId, AtomicBoolean> =
        ConcurrentHashMap<SessionId, AtomicBoolean>()
@@ -188,6 +188,7 @@ abstract class SessionOrchestrator(
    private val artifactKindRegistry: ArtifactKindRegistry? = null,
    private val repoKnowledgeRetriever: RepoKnowledgeRetriever? = null,
    private val readyTaskCounter: ReadyTaskCounter? = null,
+    private val taskClaimCoordinator: TaskClaimCoordinator? = null,
 ) {
    private val log = LoggerFactory.getLogger(this::class.java)
    private val eventStore: EventStore = repositories.eventStore
@@ -423,9 +424,29 @@ abstract class SessionOrchestrator(
        val vocabularyEntries = artifactKindRegistry
            ?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" }
            ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList()
+        // Deterministic claim: a stage flagged claimTask gets the kernel to claim the next ready task
+        // (or re-surface the one already claimed) and inject its context bundle as L0 — the loop
+        // advances by recorded fact, not by the agent remembering to claim.
+        val claimedTaskEntries = if (stageConfig.metadata["claimTask"] == "true") {
+            taskClaimCoordinator?.claimNext(sessionId)?.let { bundle ->
+                listOf(
+                    ContextEntry(
+                        id = ContextEntryId(UUID.randomUUID().toString()),
+                        layer = ContextLayer.L0,
+                        content = bundle,
+                        sourceType = "claimedTask",
+                        sourceId = stageId.value,
+                        tokenEstimate = estimateTokens(bundle),
+                        role = EntryRole.SYSTEM,
+                    ),
+                )
+            } ?: emptyList()
+        } else {
+            emptyList()
+        }
        var accumulatedEntries =
            systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries +
-                journalEntries + repoMapEntries +
+                journalEntries + repoMapEntries + claimedTaskEntries +
                needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries +
                clarificationEntries + retryFeedbackEntries
        val contextPack = contextPackBuilder.build(
@@ -708,9 +729,14 @@ abstract class SessionOrchestrator(
                    ),
                )
            }
+            // Per-task write scope: while a task is claimed, narrow the manifest to its affected
+            // paths (recorded on the task) so the implementer can't write outside its unit of work.
+            // Falls back to the stage's static manifest when nothing is claimed.
+            val effectiveManifest = taskClaimCoordinator?.activeScope(sessionId)?.takeIf { it.isNotEmpty() }
+                ?: stageConfig.writeManifest
            val plane2Risk: RiskSummary? = runPlane2Assessment(
                sessionId, stageId, invocationId, toolCall.function.name, request, tool, effectives,
-                stageConfig.writeManifest,
+                effectiveManifest,
            )?.let { assessment ->
                if (assessment.recommendedAction == RiskAction.BLOCK) {
                    emit(
@@ -0,0 +1,22 @@
+package com.correx.core.kernel.orchestration
+
+import com.correx.core.events.types.SessionId
+
+/**
+ * Deterministic, kernel-driven task claiming for the execution loop: the kernel — not the LLM —
+ * picks and claims the next ready task, so the loop terminates by event-recorded fact rather than by
+ * agent compliance (the failure mode the read-before-write logs exposed). Implemented in apps/server
+ * over the task projection; core:kernel stays decoupled from core:tasks (mirrors [ReadyTaskCounter]).
+ * Null injection ⇒ no claiming and unrestricted task scope.
+ */
+interface TaskClaimCoordinator {
+    /**
+     * Ensure the session has a claimed task and return that task's context bundle as L0 text. If the
+     * session already has one claimed (e.g. a review bounced it back), re-injects its context without
+     * re-claiming; otherwise claims the next ready task. Null when nothing is ready to work.
+     */
+    suspend fun claimNext(sessionId: SessionId): String?
+
+    /** Write-scope (affected paths) of the task the session currently has claimed; empty when none. */
+    fun activeScope(sessionId: SessionId): List<String>
+}
@@ -92,6 +92,42 @@ class TaskService(
    fun ready(projectId: ProjectId? = null): List<Task> =
        TaskGraph.ready(projectId?.let { list(it) } ?: listAll())

+    /**
+     * The project a session is working in, inferred from the tasks it created/touched: any task
+     * carrying a CONTEXT/SESSION link back to [sessionId] (recorded by the task tools at create and
+     * claim time). Robust to whatever free-form `project` key the agent passed — it reads back the
+     * project the tasks were actually filed under, so the execution-loop predicate and claim stage
+     * agree without trusting the agent. Picks the earliest such task's project (lowest key sequence)
+     * if a session ever spans projects. Null when the session has created no tasks yet.
+     *
+     * ponytail: scans the whole cross-project board (listAll) per call; fine for local single-project
+     * use. Index by session link if a deployment ever runs many projects in one store.
+     */
+    fun projectForSession(sessionId: String): ProjectId? =
+        listAll()
+            .filter { t -> t.state.links.any(linkToSession(sessionId)) }
+            .minByOrNull(::keySeq)
+            ?.state?.projectId
+
+    /** Next ready task for the session's project, in deterministic claim order (creation sequence, then id). */
+    fun nextReadyForSession(sessionId: String): Task? =
+        projectForSession(sessionId)?.let { ready(it) }.orEmpty()
+            .minWithOrNull(compareBy({ keySeq(it) }, { it.taskId.value }))
+
+    /** The task this session currently has claimed — its in-flight unit of work — or null. */
+    fun activeClaim(sessionId: String): Task? =
+        projectForSession(sessionId)?.let { list(it) }.orEmpty()
+            .firstOrNull { it.state.status == TaskStatus.IN_PROGRESS && it.state.claimant == sessionId }
+
+    private fun linkToSession(sessionId: String): (TaskLink) -> Boolean = { link ->
+        link.type == TaskLinkType.CONTEXT &&
+            link.targetKind == TaskTargetKind.SESSION &&
+            link.targetId == sessionId
+    }
+
+    private fun keySeq(task: Task): Int =
+        task.taskId.value.substringAfterLast('-').toIntOrNull() ?: Int.MAX_VALUE
+
    /** Unfinished tasks that must complete before [taskId] can proceed (resolved within its project). */
    fun blockers(taskId: TaskId): List<Task> {
        val task = getTask(taskId) ?: return emptyList()
@@ -72,6 +72,31 @@ class TaskServiceTest {
        assertNull(service.claim(TaskId("auth-999"), "claude-opus"))
    }

+    @Test
+    fun `session resolution drives the deterministic claim loop`() = runBlocking {
+        // Two tasks filed under this session (via the CONTEXT/SESSION link the tools record), one
+        // unrelated. The loop must resolve the project from those links, regardless of its key.
+        val session = "sess-1"
+        val a = service.createTask(project, "first", "g", affectedPaths = listOf("src/a/**")).taskId
+        val b = service.createTask(project, "second", "g").taskId
+        service.link(a, session, TaskLinkType.CONTEXT, TaskTargetKind.SESSION)
+        service.link(b, session, TaskLinkType.CONTEXT, TaskTargetKind.SESSION)
+        service.createTask(ProjectId("other"), "noise", "g")
+
+        assertEquals(project, service.projectForSession(session))
+        // Deterministic order: lowest key sequence first.
+        assertEquals(a, service.nextReadyForSession(session)?.taskId)
+
+        // Claiming moves it off ready() so the predicate eventually terminates, and becomes the
+        // session's active claim carrying its write scope.
+        service.claim(a, session)
+        assertEquals(b, service.nextReadyForSession(session)?.taskId)
+        assertEquals(a, service.activeClaim(session)?.taskId)
+        assertEquals(listOf("src/a/**"), service.activeClaim(session)?.state?.affectedPaths)
+
+        assertNull(service.projectForSession("unknown-session"))
+    }
+
    @Test
    fun `listAll spans every project's stream`() = runBlocking {
        service.createTask(project, "a", "g")