feat(tasks): dependency-aware work graph (blockers, ready, blocking)
The DEPENDS_ON/BLOCKS link types were inert metadata — nothing reasoned about
them. TaskGraph turns them into answerable questions: what a task waits on
(its DEPENDS_ON targets plus any task that BLOCKS it), whether it is ready
(TODO with no unmet blocker — a terminal dependency stops blocking), and what
finishing it would unblock. Readiness surfaces workable tasks to claim; it
never assigns (honouring the rejected /tasks/next scheduler).
Surfaced across the stack:
- TaskService.ready / blockers / blocking.
- Context bundle: a `blocked` flag + `blocked_by` list, rendered as a prominent
"BLOCKED — waiting on ..." line, so an agent calling task_context learns it
should wait before charging at the task.
- GET /tasks?ready=true, GET /tasks/{id}/blockers; correx task ready / blockers.
Cross-project dependencies are scoped to a project for now (documented in
TaskGraph). Tests cover both link directions, terminal-dependency clearing,
the ready filter, the reverse blocking direction, the bundle flag, and REST.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskId
|
||||
import com.correx.core.events.types.TaskLinkType
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
|
||||
/**
|
||||
@@ -38,6 +39,8 @@ class TaskContextAssembler(
|
||||
}
|
||||
}
|
||||
|
||||
val blockedBy = computeBlockedBy(taskId, state)
|
||||
|
||||
return TaskContextBundle(
|
||||
id = taskId.value,
|
||||
key = state.key,
|
||||
@@ -46,6 +49,8 @@ class TaskContextAssembler(
|
||||
goal = state.goal,
|
||||
acceptanceCriteria = state.acceptanceCriteria,
|
||||
relevantFiles = state.affectedPaths,
|
||||
blocked = blockedBy.isNotEmpty(),
|
||||
blockedBy = blockedBy,
|
||||
dependencies = resolved.dependencies,
|
||||
documents = resolved.documents,
|
||||
artifacts = resolved.artifacts,
|
||||
@@ -124,6 +129,27 @@ class TaskContextAssembler(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The task's unfinished dependency blockers, resolved over its project board (see [TaskGraph]),
|
||||
* each labelled by the direction of the edge. Empty when the project board can't be read.
|
||||
*/
|
||||
private fun computeBlockedBy(taskId: TaskId, state: TaskState): List<RelatedTask> {
|
||||
val board = state.projectId?.let { runCatching { service.list(it) }.getOrNull() } ?: return emptyList()
|
||||
return TaskGraph.unmetBlockers(Task(taskId, state), board).map { b ->
|
||||
val viaDependsOn = state.links.any {
|
||||
it.targetKind == TaskTargetKind.TASK &&
|
||||
it.type == TaskLinkType.DEPENDS_ON &&
|
||||
it.targetId == b.taskId.value
|
||||
}
|
||||
RelatedTask(
|
||||
id = b.taskId.value,
|
||||
status = b.state.status.name,
|
||||
title = b.state.title,
|
||||
link = if (viaDependsOn) "DEPENDS_ON" else "BLOCKS",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun retrieveKnowledge(state: TaskState): List<KnowledgeHit> {
|
||||
val active = retriever ?: return emptyList()
|
||||
val query = listOfNotNull(state.title, state.goal).joinToString(" ").trim()
|
||||
|
||||
@@ -16,6 +16,10 @@ data class TaskContextBundle(
|
||||
val goal: String?,
|
||||
val acceptanceCriteria: List<String>,
|
||||
val relevantFiles: List<String>,
|
||||
// Dependency-derived readiness (distinct from an explicit BLOCKED status): blockedBy holds the
|
||||
// unfinished tasks this one is waiting on, so an agent calling task_context knows to wait.
|
||||
val blocked: Boolean = false,
|
||||
val blockedBy: List<RelatedTask> = emptyList(),
|
||||
val dependencies: List<RelatedTask>,
|
||||
val documents: List<TaskDocument>,
|
||||
val artifacts: List<TaskArtifact> = emptyList(),
|
||||
@@ -28,6 +32,8 @@ data class TaskContextBundle(
|
||||
fun render(): String = buildString {
|
||||
appendLine("task $id [$status] ${title.orEmpty()}".trimEnd())
|
||||
goal?.let { appendLine("goal: $it") }
|
||||
if (blocked) appendLine("BLOCKED — waiting on ${blockedBy.size} unfinished dependency(ies):")
|
||||
appendList("blocked_by", blockedBy) { " - ${it.id} [${it.status}] ${it.title.orEmpty()} (${it.link})".trimEnd() }
|
||||
appendList("acceptance_criteria", acceptanceCriteria) { " - $it" }
|
||||
if (relevantFiles.isNotEmpty()) appendLine("relevant_files: ${relevantFiles.joinToString(", ")}")
|
||||
appendList("dependencies", dependencies) { " - ${it.id} [${it.status}] ${it.title.orEmpty()} (${it.link})".trimEnd() }
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskLinkType
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
|
||||
/**
|
||||
* Dependency reasoning over the task work graph — turns the `DEPENDS_ON`/`BLOCKS` link types from
|
||||
* inert metadata into answerable questions: what is a task waiting on, what is it ready to be
|
||||
* worked, and what does finishing it unblock.
|
||||
*
|
||||
* Two directions express the same edge, so both count: a `DEPENDS_ON B` link on A means A waits on
|
||||
* B; a `BLOCKS A` link on B means the same. A blocker is "unmet" until it reaches a terminal
|
||||
* status ([FINISHED]); a CANCELLED dependency no longer blocks. All reasoning is over the supplied
|
||||
* board, so callers scope it (one project, or the whole cross-project set); a link to a task absent
|
||||
* from the board is treated as already-resolved rather than an unknown blocker.
|
||||
*/
|
||||
object TaskGraph {
|
||||
|
||||
private val FINISHED = setOf(TaskStatus.DONE, TaskStatus.CANCELLED)
|
||||
|
||||
/** Tasks that must finish before [task] proceeds: its `DEPENDS_ON` targets plus tasks that `BLOCKS` it. */
|
||||
fun blockers(task: Task, board: Collection<Task>): List<Task> {
|
||||
val byId = board.associateBy { it.taskId.value }
|
||||
val dependsOn = task.linkTargets(TaskLinkType.DEPENDS_ON).mapNotNull { byId[it] }
|
||||
val inbound = board.filter { it.hasTaskLink(TaskLinkType.BLOCKS, task.taskId.value) }
|
||||
return (dependsOn + inbound).distinctBy { it.taskId.value }
|
||||
}
|
||||
|
||||
/** The blockers that have not finished — the concrete reason [task] cannot start yet. */
|
||||
fun unmetBlockers(task: Task, board: Collection<Task>): List<Task> =
|
||||
blockers(task, board).filter { it.state.status !in FINISHED }
|
||||
|
||||
/** True when [task] is TODO and has no unmet blockers — workable now (to be claimed, not assigned). */
|
||||
fun isReady(task: Task, board: Collection<Task>): Boolean =
|
||||
task.state.status == TaskStatus.TODO && unmetBlockers(task, board).isEmpty()
|
||||
|
||||
/** Every workable task on the board, in board order. */
|
||||
fun ready(board: Collection<Task>): List<Task> =
|
||||
board.filter { isReady(it, board) }
|
||||
|
||||
/** Tasks [task] is holding up: its own `BLOCKS` targets plus tasks that `DEPENDS_ON` it. */
|
||||
fun blocking(task: Task, board: Collection<Task>): List<Task> {
|
||||
val byId = board.associateBy { it.taskId.value }
|
||||
val downstream = task.linkTargets(TaskLinkType.BLOCKS).mapNotNull { byId[it] }
|
||||
val dependents = board.filter { it.hasTaskLink(TaskLinkType.DEPENDS_ON, task.taskId.value) }
|
||||
return (downstream + dependents).distinctBy { it.taskId.value }
|
||||
}
|
||||
|
||||
private fun Task.linkTargets(type: TaskLinkType): List<String> =
|
||||
state.links.filter { it.targetKind == TaskTargetKind.TASK && it.type == type }.map { it.targetId }
|
||||
|
||||
private fun Task.hasTaskLink(type: TaskLinkType, targetId: String): Boolean =
|
||||
state.links.any { it.targetKind == TaskTargetKind.TASK && it.type == type && it.targetId == targetId }
|
||||
}
|
||||
@@ -84,6 +84,26 @@ class TaskService(
|
||||
fun search(query: String, projectId: ProjectId? = null): List<Task> =
|
||||
TaskSearch.search(projectId?.let { list(it) } ?: listAll(), query)
|
||||
|
||||
/**
|
||||
* Tasks ready to be picked up: TODO with every dependency satisfied (see [TaskGraph]). Surfaces
|
||||
* workable tasks for an agent or human to claim — it never assigns. Scoped to one project when
|
||||
* given, else the whole cross-project board.
|
||||
*/
|
||||
fun ready(projectId: ProjectId? = null): List<Task> =
|
||||
TaskGraph.ready(projectId?.let { list(it) } ?: listAll())
|
||||
|
||||
/** 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()
|
||||
return TaskGraph.unmetBlockers(task, list(projectOf(taskId)))
|
||||
}
|
||||
|
||||
/** Tasks that [taskId] is holding up — what completing it would unblock (within its project). */
|
||||
fun blocking(taskId: TaskId): List<Task> {
|
||||
val task = getTask(taskId) ?: return emptyList()
|
||||
return TaskGraph.blocking(task, list(projectOf(taskId)))
|
||||
}
|
||||
|
||||
suspend fun createTask(
|
||||
projectId: ProjectId,
|
||||
title: String,
|
||||
|
||||
@@ -53,6 +53,27 @@ class TaskContextAssemblerTest {
|
||||
assertEquals("kickoff", bundle.notes.single().body)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bundle flags an unfinished dependency as a blocker and clears it when done`() = runBlocking {
|
||||
val dep = service.createTask(project, "Design auth model", "model") // auth-1
|
||||
service.claim(dep.taskId, "claude-opus") // auth-1 IN_PROGRESS
|
||||
val task = service.createTask(project, "Implement JWT refresh", "auth") // auth-2
|
||||
service.link(task.taskId, dep.taskId.value, TaskLinkType.DEPENDS_ON, TaskTargetKind.TASK)
|
||||
|
||||
val blocked = assembler.assemble(task.taskId)!!
|
||||
assertTrue(blocked.blocked)
|
||||
assertEquals("auth-1", blocked.blockedBy.single().id)
|
||||
assertEquals("IN_PROGRESS", blocked.blockedBy.single().status)
|
||||
assertEquals("DEPENDS_ON", blocked.blockedBy.single().link)
|
||||
assertTrue(blocked.render().contains("BLOCKED"))
|
||||
|
||||
service.submitForReview(dep.taskId)
|
||||
service.complete(dep.taskId)
|
||||
val cleared = assembler.assemble(task.taskId)!!
|
||||
assertTrue(!cleared.blocked)
|
||||
assertTrue(cleared.blockedBy.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `render produces a compact labelled bundle`() = runBlocking {
|
||||
val task = service.createTask(project, "JWT refresh", "stay authed", listOf("rotates"))
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import com.correx.core.events.types.TaskId
|
||||
import com.correx.core.events.types.TaskLinkType
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertFalse
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TaskGraphTest {
|
||||
|
||||
private fun task(id: String, status: TaskStatus = TaskStatus.TODO, links: List<TaskLink> = emptyList()) =
|
||||
Task(TaskId(id), TaskState(status = status, key = id, projectId = ProjectId("p"), title = id, links = links))
|
||||
|
||||
private fun dependsOn(target: String) = TaskLink(target, TaskLinkType.DEPENDS_ON, TaskTargetKind.TASK)
|
||||
private fun blocks(target: String) = TaskLink(target, TaskLinkType.BLOCKS, TaskTargetKind.TASK)
|
||||
|
||||
@Test
|
||||
fun `an unfinished depends_on target blocks the task`() {
|
||||
val a = task("p-1", links = listOf(dependsOn("p-2")))
|
||||
val b = task("p-2", status = TaskStatus.IN_PROGRESS)
|
||||
val board = listOf(a, b)
|
||||
|
||||
assertEquals(listOf("p-2"), TaskGraph.unmetBlockers(a, board).map { it.taskId.value })
|
||||
assertFalse(TaskGraph.isReady(a, board))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a finished dependency no longer blocks`() {
|
||||
val a = task("p-1", links = listOf(dependsOn("p-2")))
|
||||
val done = task("p-2", status = TaskStatus.DONE)
|
||||
val board = listOf(a, done)
|
||||
|
||||
assertTrue(TaskGraph.unmetBlockers(a, board).isEmpty())
|
||||
assertTrue(TaskGraph.isReady(a, board))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an inbound BLOCKS edge blocks the target`() {
|
||||
val a = task("p-1")
|
||||
val blocker = task("p-2", status = TaskStatus.IN_PROGRESS, links = listOf(blocks("p-1")))
|
||||
val board = listOf(a, blocker)
|
||||
|
||||
assertEquals(listOf("p-2"), TaskGraph.unmetBlockers(a, board).map { it.taskId.value })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ready lists only unblocked TODO tasks`() {
|
||||
val a = task("p-1", links = listOf(dependsOn("p-2"))) // blocked
|
||||
val b = task("p-2", status = TaskStatus.IN_PROGRESS) // not TODO
|
||||
val c = task("p-3") // unblocked TODO → ready
|
||||
|
||||
assertEquals(listOf("p-3"), TaskGraph.ready(listOf(a, b, c)).map { it.taskId.value })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `blocking lists what completing a task unblocks`() {
|
||||
val dependent = task("p-1", links = listOf(dependsOn("p-2")))
|
||||
val target = task("p-2")
|
||||
val board = listOf(dependent, target)
|
||||
|
||||
assertEquals(listOf("p-1"), TaskGraph.blocking(target, board).map { it.taskId.value })
|
||||
}
|
||||
}
|
||||
@@ -105,6 +105,22 @@ class TaskServiceTest {
|
||||
assertTrue(lines.last().contains("completed"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ready surfaces unblocked work and blockers explain the wait`() = runBlocking {
|
||||
val a = service.createTask(project, "A", "g").taskId // auth-1
|
||||
val b = service.createTask(project, "B", "g").taskId // auth-2 depends on auth-1
|
||||
service.link(b, a.value, TaskLinkType.DEPENDS_ON, TaskTargetKind.TASK)
|
||||
|
||||
assertEquals(setOf("auth-1"), service.ready(project).map { it.taskId.value }.toSet())
|
||||
assertEquals(listOf("auth-1"), service.blockers(b).map { it.taskId.value })
|
||||
assertEquals(listOf("auth-2"), service.blocking(a).map { it.taskId.value })
|
||||
|
||||
// Finishing auth-1 unblocks auth-2.
|
||||
service.claim(a, "x"); service.submitForReview(a); service.complete(a)
|
||||
assertEquals(setOf("auth-2"), service.ready(project).map { it.taskId.value }.toSet())
|
||||
assertTrue(service.blockers(b).isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `history survives soft-delete but is empty for an unknown task`() = runBlocking {
|
||||
val id = service.createTask(project, "throwaway", "oops").taskId
|
||||
|
||||
Reference in New Issue
Block a user