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 2a04dfaf..1b96d0b8 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 @@ -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), diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/tasks/DefaultTaskClaimCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/DefaultTaskClaimCoordinator.kt new file mode 100644 index 00000000..58e001de --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/DefaultTaskClaimCoordinator.kt @@ -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 = + taskService.activeClaim(sessionId.value)?.state?.affectedPaths ?: emptyList() +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/tasks/ProjectReadyTaskCounter.kt b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/ProjectReadyTaskCounter.kt index e8a3507a..c86a8436 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/tasks/ProjectReadyTaskCounter.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/tasks/ProjectReadyTaskCounter.kt @@ -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 } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index c05cda9d..5fb1659b 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -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 = ConcurrentHashMap() diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 0c490d99..71f8f4ec 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -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( diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/TaskClaimCoordinator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/TaskClaimCoordinator.kt new file mode 100644 index 00000000..2308d565 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/TaskClaimCoordinator.kt @@ -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 +} diff --git a/core/tasks/src/main/kotlin/com/correx/core/tasks/TaskService.kt b/core/tasks/src/main/kotlin/com/correx/core/tasks/TaskService.kt index ad31a48c..886612db 100644 --- a/core/tasks/src/main/kotlin/com/correx/core/tasks/TaskService.kt +++ b/core/tasks/src/main/kotlin/com/correx/core/tasks/TaskService.kt @@ -92,6 +92,42 @@ class TaskService( fun ready(projectId: ProjectId? = null): List = 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 { val task = getTask(taskId) ?: return emptyList() diff --git a/core/tasks/src/test/kotlin/com/correx/core/tasks/TaskServiceTest.kt b/core/tasks/src/test/kotlin/com/correx/core/tasks/TaskServiceTest.kt index fea1f8ec..0c890e47 100644 --- a/core/tasks/src/test/kotlin/com/correx/core/tasks/TaskServiceTest.kt +++ b/core/tasks/src/test/kotlin/com/correx/core/tasks/TaskServiceTest.kt @@ -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")