feat(tasks): native task tracking — aggregate, agent tools, context bundle
Adds task tracking as a first-class event-sourced aggregate (ADR-0012). The
event log stays the only source of truth; the board is a projection, and all of
a project's task events live in one stream (tasks:<projectId>).
core:
- new core:tasks module mirroring core:sessions: TaskStatus/State/Reducer/
BoardProjector/CounterProjection/Repository + TaskService write path that
appends events via the existing EventStore (no new storage)
- task events in core:events (sealed marker TaskEvent), registered in
Serialization.kt; status is derived by the reducer from lifecycle facts
- work-graph edges: TaskLinkType + typed TaskTargetKind on the link, so bundle
resolution dispatches on the recorded kind instead of guessing from the id
- delete is a soft tombstone (TaskDeletedEvent); cancel stays a visible outcome
surfaces:
- agent tools (Tier T2 mutators, T1 read): task_create / task_update /
task_delete / task_context, registered in both the main and per-workspace
tool registries
- REST GET /tasks/{id}/context for external agents
context bundle (TaskContextAssembler):
- task fields + acceptance criteria + relevant files + resolved dependency
tasks (live status) + raw related links + notes, with a compact render()
- semantic enrichment via a TaskKnowledgeRetriever port bridged to the existing
L3 retriever (late-bound holder for composition-root ordering)
- ADR/doc resolution via a TaskDocumentResolver port + filesystem adapter
(docs/decisions/adr-*.md, padding-agnostic; *.md paths)
Unit-verified across core:tasks / infrastructure:tools / apps:server; full
gradlew check green. Not yet live-QA'd end-to-end against a running server.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,12 @@
|
||||
plugins {
|
||||
id 'java-library'
|
||||
id 'org.jetbrains.kotlin.jvm'
|
||||
id 'org.jetbrains.kotlin.plugin.serialization'
|
||||
}
|
||||
|
||||
dependencies {
|
||||
implementation(project(":core:events"))
|
||||
testImplementation(testFixtures(project(":testing:contracts")))
|
||||
}
|
||||
|
||||
tasks.named("koverVerify").configure { enabled = false }
|
||||
@@ -0,0 +1,147 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskAcceptanceCriteriaSetEvent
|
||||
import com.correx.core.events.events.TaskAffectedPathsSetEvent
|
||||
import com.correx.core.events.events.TaskBlockedEvent
|
||||
import com.correx.core.events.events.TaskCancelledEvent
|
||||
import com.correx.core.events.events.TaskClaimedEvent
|
||||
import com.correx.core.events.events.TaskCompletedEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.events.events.TaskDeletedEvent
|
||||
import com.correx.core.events.events.TaskEditedEvent
|
||||
import com.correx.core.events.events.TaskEvent
|
||||
import com.correx.core.events.events.TaskLinkedEvent
|
||||
import com.correx.core.events.events.TaskNoteAddedEvent
|
||||
import com.correx.core.events.events.TaskReleasedEvent
|
||||
import com.correx.core.events.events.TaskReopenedEvent
|
||||
import com.correx.core.events.events.TaskSubmittedForReviewEvent
|
||||
import com.correx.core.events.events.TaskUnblockedEvent
|
||||
import com.correx.core.events.events.TaskUnlinkedEvent
|
||||
import kotlinx.datetime.Instant
|
||||
|
||||
/**
|
||||
* Folds task events into [TaskState]. Field-update events always apply; lifecycle events
|
||||
* apply only on a legal transition — an illegal transition is ignored and counted in
|
||||
* [TaskState.invalidTransitions] (mirrors how `SessionState` tracks invalid transitions).
|
||||
*/
|
||||
class DefaultTaskReducer : TaskReducer {
|
||||
|
||||
override fun reduce(state: TaskState, event: StoredEvent): TaskState {
|
||||
val payload = event.payload as? TaskEvent ?: return state
|
||||
val ts = event.metadata.timestamp
|
||||
val base = state.copy(
|
||||
createdAt = state.createdAt ?: ts,
|
||||
updatedAt = ts,
|
||||
)
|
||||
return reduceContent(state, base, payload, ts)
|
||||
?: reduceTransition(state, base, payload)
|
||||
}
|
||||
|
||||
/** Field updates that do not touch status. Returns null for lifecycle events. */
|
||||
private fun reduceContent(
|
||||
state: TaskState,
|
||||
base: TaskState,
|
||||
payload: TaskEvent,
|
||||
ts: Instant,
|
||||
): TaskState? = when (payload) {
|
||||
is TaskCreatedEvent -> base.copy(
|
||||
status = TaskStatus.TODO,
|
||||
key = payload.key,
|
||||
projectId = payload.projectId,
|
||||
title = payload.title,
|
||||
goal = payload.goal,
|
||||
acceptanceCriteria = payload.acceptanceCriteria,
|
||||
affectedPaths = payload.affectedPaths,
|
||||
)
|
||||
|
||||
is TaskEditedEvent -> base.copy(
|
||||
title = payload.title ?: state.title,
|
||||
goal = payload.goal ?: state.goal,
|
||||
)
|
||||
|
||||
is TaskAcceptanceCriteriaSetEvent -> base.copy(acceptanceCriteria = payload.criteria)
|
||||
|
||||
is TaskAffectedPathsSetEvent -> base.copy(affectedPaths = payload.paths)
|
||||
|
||||
is TaskLinkedEvent -> base.copy(
|
||||
links = (state.links + TaskLink(payload.targetId, payload.type, payload.targetKind)).distinct(),
|
||||
)
|
||||
|
||||
is TaskUnlinkedEvent -> base.copy(
|
||||
links = state.links.filterNot {
|
||||
it.targetId == payload.targetId && it.type == payload.type
|
||||
},
|
||||
)
|
||||
|
||||
is TaskNoteAddedEvent -> base.copy(
|
||||
notes = state.notes + TaskNote(payload.author, payload.body, ts),
|
||||
)
|
||||
|
||||
is TaskDeletedEvent -> base.copy(deleted = true)
|
||||
|
||||
else -> null
|
||||
}
|
||||
|
||||
/** Lifecycle events: apply on a legal transition, else count it invalid. */
|
||||
private fun reduceTransition(
|
||||
state: TaskState,
|
||||
base: TaskState,
|
||||
payload: TaskEvent,
|
||||
): TaskState {
|
||||
val from = state.status
|
||||
return when (payload) {
|
||||
is TaskClaimedEvent ->
|
||||
base.transitionTo(TaskStatus.IN_PROGRESS, from == TaskStatus.TODO) {
|
||||
it.copy(claimant = payload.claimant)
|
||||
}
|
||||
|
||||
is TaskReleasedEvent ->
|
||||
base.transitionTo(TaskStatus.TODO, from == TaskStatus.IN_PROGRESS) {
|
||||
it.copy(claimant = null)
|
||||
}
|
||||
|
||||
is TaskBlockedEvent ->
|
||||
base.transitionTo(TaskStatus.BLOCKED, from in WORKING)
|
||||
|
||||
is TaskUnblockedEvent ->
|
||||
base.transitionTo(
|
||||
if (state.claimant != null) TaskStatus.IN_PROGRESS else TaskStatus.TODO,
|
||||
from == TaskStatus.BLOCKED,
|
||||
)
|
||||
|
||||
is TaskSubmittedForReviewEvent ->
|
||||
base.transitionTo(TaskStatus.IN_REVIEW, from == TaskStatus.IN_PROGRESS)
|
||||
|
||||
is TaskCompletedEvent ->
|
||||
base.transitionTo(TaskStatus.DONE, from == TaskStatus.IN_REVIEW)
|
||||
|
||||
is TaskReopenedEvent ->
|
||||
base.transitionTo(TaskStatus.TODO, from in TERMINAL) {
|
||||
it.copy(claimant = null)
|
||||
}
|
||||
|
||||
is TaskCancelledEvent ->
|
||||
base.transitionTo(TaskStatus.CANCELLED, from in CANCELLABLE)
|
||||
|
||||
else -> state
|
||||
} ?: state.copy(invalidTransitions = state.invalidTransitions + 1)
|
||||
}
|
||||
|
||||
private inline fun TaskState.transitionTo(
|
||||
to: TaskStatus,
|
||||
legal: Boolean,
|
||||
mutate: (TaskState) -> TaskState = { it },
|
||||
): TaskState? = if (legal) mutate(copy(status = to)) else null
|
||||
|
||||
private companion object {
|
||||
val WORKING = setOf(TaskStatus.TODO, TaskStatus.IN_PROGRESS, TaskStatus.IN_REVIEW)
|
||||
val TERMINAL = setOf(TaskStatus.DONE, TaskStatus.CANCELLED)
|
||||
val CANCELLABLE = setOf(
|
||||
TaskStatus.TODO,
|
||||
TaskStatus.IN_PROGRESS,
|
||||
TaskStatus.IN_REVIEW,
|
||||
TaskStatus.BLOCKED,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.TaskId
|
||||
import com.correx.core.sessions.projections.replay.EventReplayer
|
||||
|
||||
/**
|
||||
* Reads a project's task board by replaying its stream. Mirrors `DefaultSessionRepository`:
|
||||
* the repository is a thin read model over an [EventReplayer]; the event log is the source
|
||||
* of truth and the board is rebuilt from it.
|
||||
*/
|
||||
class DefaultTaskRepository(
|
||||
private val replayer: EventReplayer<TaskBoard>,
|
||||
private val streamId: SessionId,
|
||||
) {
|
||||
|
||||
fun board(): TaskBoard = replayer.rebuild(streamId)
|
||||
|
||||
fun getTask(taskId: TaskId): Task? =
|
||||
board()[taskId]?.takeUnless { it.deleted }?.let { Task(taskId, it) }
|
||||
|
||||
fun list(): List<Task> =
|
||||
board().filterValues { !it.deleted }.map { (id, state) -> Task(id, state) }
|
||||
|
||||
fun count(): Int = list().size
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskId
|
||||
|
||||
data class Task(
|
||||
val taskId: TaskId,
|
||||
val state: TaskState,
|
||||
)
|
||||
@@ -0,0 +1,6 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskId
|
||||
|
||||
/** All tasks in a project stream, keyed by id — the unit a single replay produces. */
|
||||
typealias TaskBoard = Map<TaskId, TaskState>
|
||||
@@ -0,0 +1,22 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskEvent
|
||||
import com.correx.core.sessions.projections.Projection
|
||||
|
||||
/**
|
||||
* Folds a per-project task stream into a [TaskBoard]. Each event is routed to its owning
|
||||
* task by [TaskEvent.taskId]; non-task payloads leave the board untouched.
|
||||
*/
|
||||
class TaskBoardProjector(
|
||||
private val reducer: TaskReducer,
|
||||
) : Projection<TaskBoard> {
|
||||
|
||||
override fun initial(): TaskBoard = emptyMap()
|
||||
|
||||
override fun apply(state: TaskBoard, event: StoredEvent): TaskBoard {
|
||||
val taskId = (event.payload as? TaskEvent)?.taskId ?: return state
|
||||
val current = state[taskId] ?: TaskState()
|
||||
return state + (taskId to reducer.reduce(current, event))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskId
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
|
||||
/**
|
||||
* Assembles a [TaskContextBundle] for a task. Links are partitioned by their recorded
|
||||
* [com.correx.core.events.types.TaskTargetKind]: TASK targets are resolved to [RelatedTask]
|
||||
* dependencies (with live status), DOC targets are resolved inline via [documentResolver], and
|
||||
* anything unresolved (or ARTIFACT/SESSION, not inlined yet) becomes a raw [RelatedLink].
|
||||
*
|
||||
* When a [TaskKnowledgeRetriever] is supplied, the bundle is additionally enriched with
|
||||
* semantically-relevant snippets over the task's title+goal. Retrieval and doc-resolution failures
|
||||
* are swallowed (the bundle still returns) so context assembly never fails on a flaky dependency.
|
||||
*/
|
||||
class TaskContextAssembler(
|
||||
private val service: TaskService,
|
||||
private val retriever: TaskKnowledgeRetriever? = null,
|
||||
private val documentResolver: TaskDocumentResolver? = null,
|
||||
private val knowledgeLimit: Int = DEFAULT_KNOWLEDGE_LIMIT,
|
||||
) {
|
||||
|
||||
suspend fun assemble(taskId: TaskId): TaskContextBundle? {
|
||||
val state = service.getTask(taskId)?.state ?: return null
|
||||
|
||||
val dependencies = mutableListOf<RelatedTask>()
|
||||
val documents = mutableListOf<TaskDocument>()
|
||||
val related = mutableListOf<RelatedLink>()
|
||||
for (link in state.links) {
|
||||
// Resolution dispatches on the link's recorded targetKind — no guessing from the id.
|
||||
// A kind whose resolution comes up empty (missing task, unresolvable doc, or an
|
||||
// ARTIFACT/SESSION we don't inline yet) falls back to a raw related link.
|
||||
when (link.targetKind) {
|
||||
TaskTargetKind.TASK -> addTask(link, dependencies, related)
|
||||
TaskTargetKind.DOC -> addDocument(link, documents, related)
|
||||
TaskTargetKind.ARTIFACT, TaskTargetKind.SESSION ->
|
||||
related += RelatedLink(link.targetId, link.type.name)
|
||||
}
|
||||
}
|
||||
|
||||
return TaskContextBundle(
|
||||
id = taskId.value,
|
||||
key = state.key,
|
||||
status = state.status.name,
|
||||
title = state.title,
|
||||
goal = state.goal,
|
||||
acceptanceCriteria = state.acceptanceCriteria,
|
||||
relevantFiles = state.affectedPaths,
|
||||
dependencies = dependencies,
|
||||
documents = documents,
|
||||
relatedLinks = related,
|
||||
notes = state.notes.map { NoteEntry(it.author.name, it.body) },
|
||||
relevantKnowledge = retrieveKnowledge(state),
|
||||
)
|
||||
}
|
||||
|
||||
private fun addTask(
|
||||
link: TaskLink,
|
||||
dependencies: MutableList<RelatedTask>,
|
||||
related: MutableList<RelatedLink>,
|
||||
) {
|
||||
val linked = runCatching { service.getTask(TaskId(link.targetId)) }.getOrNull()
|
||||
if (linked != null) {
|
||||
dependencies += RelatedTask(
|
||||
id = linked.taskId.value,
|
||||
status = linked.state.status.name,
|
||||
title = linked.state.title,
|
||||
link = link.type.name,
|
||||
)
|
||||
} else {
|
||||
related += RelatedLink(link.targetId, link.type.name)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun addDocument(
|
||||
link: TaskLink,
|
||||
documents: MutableList<TaskDocument>,
|
||||
related: MutableList<RelatedLink>,
|
||||
) {
|
||||
val doc = resolveDocument(link.targetId)
|
||||
if (doc != null) {
|
||||
documents += TaskDocument(
|
||||
targetId = link.targetId,
|
||||
type = link.type.name,
|
||||
title = doc.title,
|
||||
path = doc.path,
|
||||
excerpt = doc.excerpt,
|
||||
)
|
||||
} else {
|
||||
related += RelatedLink(link.targetId, link.type.name)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun resolveDocument(targetId: String): ResolvedDocument? {
|
||||
val resolver = documentResolver ?: return null
|
||||
return runCatching { resolver.resolve(targetId) }.getOrNull()
|
||||
}
|
||||
|
||||
private suspend fun retrieveKnowledge(state: TaskState): List<KnowledgeHit> {
|
||||
val active = retriever ?: return emptyList()
|
||||
val query = listOfNotNull(state.title, state.goal).joinToString(" ").trim()
|
||||
if (query.isEmpty()) return emptyList()
|
||||
return runCatching { active.retrieve(query, knowledgeLimit) }.getOrDefault(emptyList())
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val DEFAULT_KNOWLEDGE_LIMIT = 5
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/**
|
||||
* Token-optimized context bundle for a task — the single payload an agent needs to start work:
|
||||
* the task itself, its acceptance criteria, resolved dependency tasks, the related (non-task)
|
||||
* links it should fetch, the relevant files, and notes. Assembled by [TaskContextAssembler].
|
||||
*/
|
||||
@Serializable
|
||||
data class TaskContextBundle(
|
||||
val id: String,
|
||||
val key: String?,
|
||||
val status: String,
|
||||
val title: String?,
|
||||
val goal: String?,
|
||||
val acceptanceCriteria: List<String>,
|
||||
val relevantFiles: List<String>,
|
||||
val dependencies: List<RelatedTask>,
|
||||
val documents: List<TaskDocument>,
|
||||
val relatedLinks: List<RelatedLink>,
|
||||
val notes: List<NoteEntry>,
|
||||
val relevantKnowledge: List<KnowledgeHit> = emptyList(),
|
||||
) {
|
||||
/** Compact, label-per-line rendering for tool output (cheaper than JSON for the model). */
|
||||
fun render(): String = buildString {
|
||||
appendLine("task $id [$status] ${title.orEmpty()}".trimEnd())
|
||||
goal?.let { appendLine("goal: $it") }
|
||||
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() }
|
||||
appendList("documents", documents) { " - ${it.targetId} (${it.type}) ${it.title} [${it.path}]: ${it.excerpt}" }
|
||||
appendList("related", relatedLinks) { " - ${it.targetId} (${it.type})" }
|
||||
appendList("relevant_knowledge", relevantKnowledge) { " - ${it.source}: ${it.text}" }
|
||||
appendList("notes", notes) { " - [${it.author}] ${it.body}" }
|
||||
}.trimEnd()
|
||||
|
||||
private fun <T> StringBuilder.appendList(label: String, items: List<T>, line: (T) -> String) {
|
||||
if (items.isEmpty()) return
|
||||
appendLine("$label:")
|
||||
items.forEach { appendLine(line(it)) }
|
||||
}
|
||||
}
|
||||
|
||||
/** A linked task resolved against the board: enough for the agent to judge readiness. */
|
||||
@Serializable
|
||||
data class RelatedTask(
|
||||
val id: String,
|
||||
val status: String,
|
||||
val title: String?,
|
||||
val link: String,
|
||||
)
|
||||
|
||||
/** A link whose target resolved to a document (ADR/doc): title + excerpt inlined for the agent. */
|
||||
@Serializable
|
||||
data class TaskDocument(
|
||||
val targetId: String,
|
||||
val type: String,
|
||||
val title: String,
|
||||
val path: String,
|
||||
val excerpt: String,
|
||||
)
|
||||
|
||||
/** A link whose target is neither a task nor a resolvable doc — left for the agent to fetch. */
|
||||
@Serializable
|
||||
data class RelatedLink(
|
||||
val targetId: String,
|
||||
val type: String,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class NoteEntry(
|
||||
val author: String,
|
||||
val body: String,
|
||||
)
|
||||
|
||||
/** A semantically-retrieved snippet relevant to the task's goal (e.g. a repo file or doc). */
|
||||
@Serializable
|
||||
data class KnowledgeHit(
|
||||
val source: String,
|
||||
val text: String,
|
||||
val score: Double,
|
||||
)
|
||||
@@ -0,0 +1,24 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.sessions.projections.Projection
|
||||
|
||||
/**
|
||||
* Counts created tasks in a project stream. The numeric suffix of a human key
|
||||
* (`<prefix>-<n>`) is taken from this count, so ids never collide even after a task is
|
||||
* cancelled — it counts creations, not currently-live tasks.
|
||||
*/
|
||||
class TaskCounterProjection(
|
||||
private val projectId: String,
|
||||
) : Projection<TaskCounterState> {
|
||||
|
||||
override fun initial(): TaskCounterState =
|
||||
TaskCounterState(projectId, 0)
|
||||
|
||||
override fun apply(
|
||||
state: TaskCounterState,
|
||||
event: StoredEvent,
|
||||
): TaskCounterState =
|
||||
if (event.payload is TaskCreatedEvent) state.copy(count = state.count + 1) else state
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
data class TaskCounterState(
|
||||
val projectId: String,
|
||||
val count: Int,
|
||||
)
|
||||
@@ -0,0 +1,21 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/**
|
||||
* Port for resolving a link target (e.g. "adr-7" or a "docs/…/x.md" path) to an inline document
|
||||
* summary, so the context bundle can carry the doc's title + excerpt instead of a bare id the
|
||||
* agent has to fetch separately. Implemented in a higher layer (apps/server reads the repo's
|
||||
* docs); `core:tasks` stays decoupled from the filesystem. Returns null when the target is not a
|
||||
* resolvable document — the link then stays a raw related link.
|
||||
*/
|
||||
interface TaskDocumentResolver {
|
||||
suspend fun resolve(targetId: String): ResolvedDocument?
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class ResolvedDocument(
|
||||
val title: String,
|
||||
val path: String,
|
||||
val excerpt: String,
|
||||
)
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
/**
|
||||
* Port for semantic enrichment of a task context bundle. Implemented in a higher layer
|
||||
* (apps/server bridges it to the L3 vector retriever) and injected into [TaskContextAssembler];
|
||||
* `core:tasks` stays decoupled from the retrieval stack. Implementations must be side-effect-free
|
||||
* and tolerate being called with an arbitrary natural-language query.
|
||||
*/
|
||||
interface TaskKnowledgeRetriever {
|
||||
suspend fun retrieve(query: String, limit: Int): List<KnowledgeHit>
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskLinkType
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
|
||||
/** One edge in the work graph: a typed pointer from a task to another entity, tagged by kind. */
|
||||
data class TaskLink(
|
||||
val targetId: String,
|
||||
val type: TaskLinkType,
|
||||
val targetKind: TaskTargetKind,
|
||||
)
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.TaskNoteAuthor
|
||||
import kotlinx.datetime.Instant
|
||||
|
||||
/** A note on a task, kept in agent/human lanes so consumers can filter by author. */
|
||||
data class TaskNote(
|
||||
val author: TaskNoteAuthor,
|
||||
val body: String,
|
||||
val at: Instant,
|
||||
)
|
||||
@@ -0,0 +1,10 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
|
||||
interface TaskReducer {
|
||||
fun reduce(
|
||||
state: TaskState,
|
||||
event: StoredEvent,
|
||||
): TaskState
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.EventMetadata
|
||||
import com.correx.core.events.events.EventPayload
|
||||
import com.correx.core.events.events.NewEvent
|
||||
import com.correx.core.events.events.TaskAcceptanceCriteriaSetEvent
|
||||
import com.correx.core.events.events.TaskAffectedPathsSetEvent
|
||||
import com.correx.core.events.events.TaskBlockedEvent
|
||||
import com.correx.core.events.events.TaskCancelledEvent
|
||||
import com.correx.core.events.events.TaskClaimedEvent
|
||||
import com.correx.core.events.events.TaskCompletedEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.events.events.TaskDeletedEvent
|
||||
import com.correx.core.events.events.TaskEditedEvent
|
||||
import com.correx.core.events.events.TaskLinkedEvent
|
||||
import com.correx.core.events.events.TaskNoteAddedEvent
|
||||
import com.correx.core.events.events.TaskReleasedEvent
|
||||
import com.correx.core.events.events.TaskReopenedEvent
|
||||
import com.correx.core.events.events.TaskSubmittedForReviewEvent
|
||||
import com.correx.core.events.events.TaskUnblockedEvent
|
||||
import com.correx.core.events.events.TaskUnlinkedEvent
|
||||
import com.correx.core.events.stores.EventStore
|
||||
import com.correx.core.events.types.EventId
|
||||
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.TaskNoteAuthor
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
|
||||
import kotlinx.datetime.Clock
|
||||
import java.util.UUID
|
||||
|
||||
/**
|
||||
* Write path for tasks: turns intents into events appended to the shared [EventStore].
|
||||
* The event log stays the single source of truth — the board is rebuilt by replay, never
|
||||
* stored. All task events for a project live in one stream ([TaskStreams.forProject]).
|
||||
*
|
||||
* Keys are `<project>-<n>`, so a task's project is recoverable from its id; mutating
|
||||
* methods therefore need only the [TaskId]. They return the rebuilt [Task], or null when the
|
||||
* task does not exist (so callers — e.g. tools — can report a clean failure). Mutations that
|
||||
* are illegal for the current status are still appended but are no-ops on replay (the reducer
|
||||
* counts them in `invalidTransitions`).
|
||||
*/
|
||||
@Suppress("TooManyFunctions")
|
||||
class TaskService(
|
||||
private val eventStore: EventStore,
|
||||
private val clock: Clock = Clock.System,
|
||||
) {
|
||||
private val boardReplayer =
|
||||
DefaultEventReplayer(eventStore, TaskBoardProjector(DefaultTaskReducer()))
|
||||
|
||||
fun board(projectId: ProjectId): TaskBoard =
|
||||
boardReplayer.rebuild(TaskStreams.forProject(projectId))
|
||||
|
||||
fun list(projectId: ProjectId): List<Task> =
|
||||
board(projectId).filterValues { !it.deleted }.map { (id, state) -> Task(id, state) }
|
||||
|
||||
fun getTask(taskId: TaskId): Task? {
|
||||
val state = board(projectOf(taskId))[taskId] ?: return null
|
||||
return if (state.deleted) null else Task(taskId, state)
|
||||
}
|
||||
|
||||
suspend fun createTask(
|
||||
projectId: ProjectId,
|
||||
title: String,
|
||||
goal: String,
|
||||
acceptanceCriteria: List<String> = emptyList(),
|
||||
affectedPaths: List<String> = emptyList(),
|
||||
): Task {
|
||||
val key = nextKey(projectId)
|
||||
val taskId = TaskId(key)
|
||||
append(
|
||||
projectId,
|
||||
TaskCreatedEvent(
|
||||
taskId = taskId,
|
||||
projectId = projectId,
|
||||
key = key,
|
||||
title = title,
|
||||
goal = goal,
|
||||
acceptanceCriteria = acceptanceCriteria,
|
||||
affectedPaths = affectedPaths,
|
||||
),
|
||||
)
|
||||
return requireNotNull(getTask(taskId)) { "task $key missing after creation" }
|
||||
}
|
||||
|
||||
suspend fun edit(taskId: TaskId, title: String? = null, goal: String? = null): Task? =
|
||||
mutate(taskId) { TaskEditedEvent(it, title = title, goal = goal) }
|
||||
|
||||
suspend fun setAcceptanceCriteria(taskId: TaskId, criteria: List<String>): Task? =
|
||||
mutate(taskId) { TaskAcceptanceCriteriaSetEvent(it, criteria) }
|
||||
|
||||
suspend fun setAffectedPaths(taskId: TaskId, paths: List<String>): Task? =
|
||||
mutate(taskId) { TaskAffectedPathsSetEvent(it, paths) }
|
||||
|
||||
suspend fun claim(taskId: TaskId, claimant: String): Task? =
|
||||
mutate(taskId) { TaskClaimedEvent(it, claimant) }
|
||||
|
||||
suspend fun release(taskId: TaskId): Task? =
|
||||
mutate(taskId) { TaskReleasedEvent(it) }
|
||||
|
||||
suspend fun block(taskId: TaskId, reason: String): Task? =
|
||||
mutate(taskId) { TaskBlockedEvent(it, reason) }
|
||||
|
||||
suspend fun unblock(taskId: TaskId): Task? =
|
||||
mutate(taskId) { TaskUnblockedEvent(it) }
|
||||
|
||||
suspend fun submitForReview(taskId: TaskId): Task? =
|
||||
mutate(taskId) { TaskSubmittedForReviewEvent(it) }
|
||||
|
||||
suspend fun complete(taskId: TaskId): Task? =
|
||||
mutate(taskId) { TaskCompletedEvent(it) }
|
||||
|
||||
suspend fun reopen(taskId: TaskId): Task? =
|
||||
mutate(taskId) { TaskReopenedEvent(it) }
|
||||
|
||||
suspend fun cancel(taskId: TaskId, reason: String? = null): Task? =
|
||||
mutate(taskId) { TaskCancelledEvent(it, reason) }
|
||||
|
||||
suspend fun link(taskId: TaskId, targetId: String, type: TaskLinkType, targetKind: TaskTargetKind): Task? =
|
||||
mutate(taskId) { TaskLinkedEvent(it, targetId, type, targetKind) }
|
||||
|
||||
suspend fun unlink(taskId: TaskId, targetId: String, type: TaskLinkType): Task? =
|
||||
mutate(taskId) { TaskUnlinkedEvent(it, targetId, type) }
|
||||
|
||||
suspend fun addNote(taskId: TaskId, author: TaskNoteAuthor, body: String): Task? =
|
||||
mutate(taskId) { TaskNoteAddedEvent(it, author, body) }
|
||||
|
||||
/** Soft delete: appends a tombstone. After this, [getTask] returns null and [list] omits it. */
|
||||
suspend fun delete(taskId: TaskId): Boolean {
|
||||
if (getTask(taskId) == null) return false
|
||||
append(projectOf(taskId), TaskDeletedEvent(taskId))
|
||||
return true
|
||||
}
|
||||
|
||||
private suspend fun mutate(taskId: TaskId, event: (TaskId) -> EventPayload): Task? {
|
||||
if (getTask(taskId) == null) return null
|
||||
append(projectOf(taskId), event(taskId))
|
||||
return getTask(taskId)
|
||||
}
|
||||
|
||||
private fun nextKey(projectId: ProjectId): String {
|
||||
val count = DefaultEventReplayer(eventStore, TaskCounterProjection(projectId.value))
|
||||
.rebuild(TaskStreams.forProject(projectId)).count
|
||||
return "${projectId.value}-${count + 1}"
|
||||
}
|
||||
|
||||
private fun projectOf(taskId: TaskId): ProjectId =
|
||||
ProjectId(taskId.value.substringBeforeLast('-'))
|
||||
|
||||
private suspend fun append(projectId: ProjectId, payload: EventPayload) {
|
||||
eventStore.append(
|
||||
NewEvent(
|
||||
metadata = EventMetadata(
|
||||
eventId = EventId(UUID.randomUUID().toString()),
|
||||
sessionId = TaskStreams.forProject(projectId),
|
||||
timestamp = clock.now(),
|
||||
schemaVersion = 1,
|
||||
causationId = null,
|
||||
correlationId = null,
|
||||
),
|
||||
payload = payload,
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import kotlinx.datetime.Instant
|
||||
|
||||
/**
|
||||
* Projected state of a single task, rebuilt by folding its events through [TaskReducer].
|
||||
* Mirrors the shape of `SessionState`: defaults make the empty/pre-creation state valid.
|
||||
*/
|
||||
data class TaskState(
|
||||
val status: TaskStatus = TaskStatus.TODO,
|
||||
val key: String? = null,
|
||||
val projectId: ProjectId? = null,
|
||||
val title: String? = null,
|
||||
val goal: String? = null,
|
||||
val acceptanceCriteria: List<String> = emptyList(),
|
||||
val affectedPaths: List<String> = emptyList(),
|
||||
val links: List<TaskLink> = emptyList(),
|
||||
val claimant: String? = null,
|
||||
val notes: List<TaskNote> = emptyList(),
|
||||
val createdAt: Instant? = null,
|
||||
val updatedAt: Instant? = null,
|
||||
val invalidTransitions: Int = 0,
|
||||
/** Soft-deleted tombstone — repository reads hide these; the events remain in the log. */
|
||||
val deleted: Boolean = false,
|
||||
)
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
/**
|
||||
* Derived task lifecycle status. Never serialized on events — the reducer computes it from
|
||||
* the lifecycle facts. Workflow: TODO → IN_PROGRESS → IN_REVIEW → DONE, with BLOCKED
|
||||
* (reversible) and CANCELLED (terminal).
|
||||
*/
|
||||
enum class TaskStatus {
|
||||
TODO,
|
||||
IN_PROGRESS,
|
||||
IN_REVIEW,
|
||||
BLOCKED,
|
||||
DONE,
|
||||
CANCELLED,
|
||||
;
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import com.correx.core.events.types.SessionId
|
||||
|
||||
/**
|
||||
* Task events live in one event stream per project, keyed `tasks:<projectId>`. The
|
||||
* `tasks:` prefix keeps these streams distinguishable from real agent sessions (the
|
||||
* EventStore is partitioned by [SessionId], and [SessionId]/`TaskId` are both `TypeId`).
|
||||
*/
|
||||
object TaskStreams {
|
||||
const val PREFIX: String = "tasks:"
|
||||
|
||||
fun forProject(projectId: ProjectId): SessionId =
|
||||
SessionId("$PREFIX${projectId.value}")
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.EventMetadata
|
||||
import com.correx.core.events.events.EventPayload
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskAcceptanceCriteriaSetEvent
|
||||
import com.correx.core.events.events.TaskBlockedEvent
|
||||
import com.correx.core.events.events.TaskClaimedEvent
|
||||
import com.correx.core.events.events.TaskCompletedEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.events.events.TaskLinkedEvent
|
||||
import com.correx.core.events.events.TaskNoteAddedEvent
|
||||
import com.correx.core.events.events.RefinementIterationEvent
|
||||
import com.correx.core.events.events.TaskSubmittedForReviewEvent
|
||||
import com.correx.core.events.events.TaskUnblockedEvent
|
||||
import com.correx.core.events.types.EventId
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.TaskId
|
||||
import com.correx.core.events.types.TaskLinkType
|
||||
import com.correx.core.events.types.TaskNoteAuthor
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class DefaultTaskReducerTest {
|
||||
|
||||
private val reducer = DefaultTaskReducer()
|
||||
private val taskId = TaskId("auth-1")
|
||||
private val projectId = ProjectId("auth")
|
||||
|
||||
private fun stored(payload: EventPayload, seq: Long = 1L) = StoredEvent(
|
||||
metadata = EventMetadata(
|
||||
eventId = EventId("e-$seq"),
|
||||
sessionId = SessionId("tasks:auth"),
|
||||
timestamp = Instant.parse("2026-01-0${seq}T00:00:00Z"),
|
||||
schemaVersion = 1,
|
||||
causationId = null,
|
||||
correlationId = null,
|
||||
),
|
||||
sequence = seq,
|
||||
sessionSequence = seq,
|
||||
payload = payload,
|
||||
)
|
||||
|
||||
private fun reduceAll(vararg payloads: EventPayload): TaskState =
|
||||
payloads.foldIndexed(TaskState()) { i, state, payload ->
|
||||
reducer.reduce(state, stored(payload, (i + 1).toLong()))
|
||||
}
|
||||
|
||||
private fun created() = TaskCreatedEvent(
|
||||
taskId = taskId,
|
||||
projectId = projectId,
|
||||
key = "auth-1",
|
||||
title = "Implement JWT refresh flow",
|
||||
goal = "users stay authenticated",
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `TaskCreatedEvent sets fields and TODO status`() {
|
||||
val state = reduceAll(created())
|
||||
|
||||
assertEquals(TaskStatus.TODO, state.status)
|
||||
assertEquals("auth-1", state.key)
|
||||
assertEquals(projectId, state.projectId)
|
||||
assertEquals("Implement JWT refresh flow", state.title)
|
||||
assertEquals("users stay authenticated", state.goal)
|
||||
assertEquals(Instant.parse("2026-01-01T00:00:00Z"), state.createdAt)
|
||||
assertEquals(0, state.invalidTransitions)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `claiming moves TODO to IN_PROGRESS and records claimant`() {
|
||||
val state = reduceAll(created(), TaskClaimedEvent(taskId, claimant = "claude-opus"))
|
||||
|
||||
assertEquals(TaskStatus.IN_PROGRESS, state.status)
|
||||
assertEquals("claude-opus", state.claimant)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `happy path created to claimed to review to done`() {
|
||||
val state = reduceAll(
|
||||
created(),
|
||||
TaskClaimedEvent(taskId, claimant = "claude-opus"),
|
||||
TaskSubmittedForReviewEvent(taskId),
|
||||
TaskCompletedEvent(taskId),
|
||||
)
|
||||
|
||||
assertEquals(TaskStatus.DONE, state.status)
|
||||
assertEquals(0, state.invalidTransitions)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `unblock restores the working status of a claimed task`() {
|
||||
val state = reduceAll(
|
||||
created(),
|
||||
TaskClaimedEvent(taskId, claimant = "claude-opus"),
|
||||
TaskBlockedEvent(taskId, reason = "waiting on auth-138"),
|
||||
TaskUnblockedEvent(taskId),
|
||||
)
|
||||
|
||||
assertEquals(TaskStatus.IN_PROGRESS, state.status)
|
||||
assertEquals(0, state.invalidTransitions)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `illegal transition is ignored and counted`() {
|
||||
val state = reduceAll(created(), TaskCompletedEvent(taskId))
|
||||
|
||||
assertEquals(TaskStatus.TODO, state.status)
|
||||
assertEquals(1, state.invalidTransitions)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `links and notes accumulate without affecting status`() {
|
||||
val state = reduceAll(
|
||||
created(),
|
||||
TaskAcceptanceCriteriaSetEvent(taskId, listOf("refresh endpoint exists")),
|
||||
TaskLinkedEvent(taskId, targetId = "auth-138", type = TaskLinkType.DEPENDS_ON, targetKind = TaskTargetKind.TASK),
|
||||
TaskNoteAddedEvent(taskId, author = TaskNoteAuthor.AGENT, body = "migration failed"),
|
||||
)
|
||||
|
||||
assertEquals(TaskStatus.TODO, state.status)
|
||||
assertEquals(listOf("refresh endpoint exists"), state.acceptanceCriteria)
|
||||
assertEquals(1, state.links.size)
|
||||
assertEquals(TaskLink("auth-138", TaskLinkType.DEPENDS_ON, TaskTargetKind.TASK), state.links.single())
|
||||
assertEquals(1, state.notes.size)
|
||||
assertEquals(TaskNoteAuthor.AGENT, state.notes.single().author)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `non-task payload leaves state untouched`() {
|
||||
val before = reduceAll(created())
|
||||
val nonTask = RefinementIterationEvent(
|
||||
sessionId = SessionId("s-1"),
|
||||
cycleKey = "implement->review",
|
||||
iteration = 1,
|
||||
maxIterations = 3,
|
||||
)
|
||||
val after = reducer.reduce(before, stored(nonTask, seq = 9L))
|
||||
|
||||
assertEquals(before, after)
|
||||
assertNull(after.claimant)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.NewEvent
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.stores.EventStore
|
||||
import com.correx.core.events.types.SessionId
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.emptyFlow
|
||||
|
||||
/** Minimal append-capable in-memory event store for core:tasks tests, keyed per session. */
|
||||
internal class InMemoryEventStore : EventStore {
|
||||
private val events = mutableListOf<StoredEvent>()
|
||||
private var global = 0L
|
||||
private val perSession = mutableMapOf<SessionId, Long>()
|
||||
|
||||
override suspend fun append(event: NewEvent): StoredEvent {
|
||||
global += 1
|
||||
val seq = (perSession[event.metadata.sessionId] ?: 0L) + 1
|
||||
perSession[event.metadata.sessionId] = seq
|
||||
val stored = StoredEvent(
|
||||
metadata = event.metadata,
|
||||
sequence = global,
|
||||
sessionSequence = seq,
|
||||
payload = event.payload,
|
||||
)
|
||||
events += stored
|
||||
return stored
|
||||
}
|
||||
|
||||
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = events.map { append(it) }
|
||||
|
||||
override fun read(sessionId: SessionId): List<StoredEvent> =
|
||||
events.filter { it.metadata.sessionId == sessionId }.sortedBy { it.sessionSequence }
|
||||
|
||||
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
|
||||
read(sessionId).filter { it.sessionSequence >= fromSequence }
|
||||
|
||||
override fun lastSequence(sessionId: SessionId): Long? =
|
||||
read(sessionId).maxOfOrNull { it.sessionSequence }
|
||||
|
||||
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
|
||||
|
||||
override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
|
||||
|
||||
override suspend fun lastGlobalSequence(): Long = global
|
||||
|
||||
override fun allEvents(): Sequence<StoredEvent> = events.asSequence()
|
||||
|
||||
override fun allSessionIds(): Set<SessionId> = events.map { it.metadata.sessionId }.toSet()
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.EventMetadata
|
||||
import com.correx.core.events.events.EventPayload
|
||||
import com.correx.core.events.events.RefinementIterationEvent
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskClaimedEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.events.types.EventId
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.TaskId
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TaskBoardProjectorTest {
|
||||
|
||||
private val projector = TaskBoardProjector(DefaultTaskReducer())
|
||||
private val projectId = ProjectId("auth")
|
||||
|
||||
private fun stored(payload: EventPayload, seq: Long) = StoredEvent(
|
||||
metadata = EventMetadata(
|
||||
eventId = EventId("e-$seq"),
|
||||
sessionId = SessionId("tasks:auth"),
|
||||
timestamp = Instant.parse("2026-01-01T00:00:0${seq}Z"),
|
||||
schemaVersion = 1,
|
||||
causationId = null,
|
||||
correlationId = null,
|
||||
),
|
||||
sequence = seq,
|
||||
sessionSequence = seq,
|
||||
payload = payload,
|
||||
)
|
||||
|
||||
private fun created(id: TaskId) =
|
||||
TaskCreatedEvent(id, projectId, id.value, "title ${id.value}", "goal")
|
||||
|
||||
private fun fold(vararg events: StoredEvent): TaskBoard =
|
||||
events.fold(projector.initial()) { board, event -> projector.apply(board, event) }
|
||||
|
||||
@Test
|
||||
fun `interleaved events update each task independently`() {
|
||||
val one = TaskId("auth-1")
|
||||
val two = TaskId("auth-2")
|
||||
|
||||
val board = fold(
|
||||
stored(created(one), 1),
|
||||
stored(created(two), 2),
|
||||
stored(TaskClaimedEvent(one, claimant = "claude-opus"), 3),
|
||||
)
|
||||
|
||||
assertEquals(2, board.size)
|
||||
assertEquals(TaskStatus.IN_PROGRESS, board[one]?.status)
|
||||
assertEquals("claude-opus", board[one]?.claimant)
|
||||
assertEquals(TaskStatus.TODO, board[two]?.status)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `non-task payload leaves the board unchanged`() {
|
||||
val one = TaskId("auth-1")
|
||||
val withTask = fold(stored(created(one), 1))
|
||||
|
||||
val nonTask = RefinementIterationEvent(
|
||||
sessionId = SessionId("s-1"),
|
||||
cycleKey = "implement->review",
|
||||
iteration = 1,
|
||||
maxIterations = 3,
|
||||
)
|
||||
val after = projector.apply(withTask, stored(nonTask, 2))
|
||||
|
||||
assertEquals(withTask, after)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
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.TaskNoteAuthor
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TaskContextAssemblerTest {
|
||||
|
||||
private val store = InMemoryEventStore()
|
||||
private val service = TaskService(store)
|
||||
private val assembler = TaskContextAssembler(service)
|
||||
private val project = ProjectId("auth")
|
||||
|
||||
@Test
|
||||
fun `bundle resolves linked tasks as dependencies and keeps non-task links raw`() = runBlocking {
|
||||
val dep = service.createTask(project, "Design auth model", "model")
|
||||
service.claim(dep.taskId, "claude-opus")
|
||||
service.submitForReview(dep.taskId)
|
||||
service.complete(dep.taskId) // dep is DONE
|
||||
|
||||
val task = service.createTask(
|
||||
project,
|
||||
title = "Implement JWT refresh",
|
||||
goal = "users stay authenticated",
|
||||
acceptanceCriteria = listOf("refresh endpoint exists"),
|
||||
affectedPaths = listOf("backend/auth/**"),
|
||||
)
|
||||
service.link(task.taskId, dep.taskId.value, TaskLinkType.DEPENDS_ON, TaskTargetKind.TASK)
|
||||
// DOC-tagged, but this assembler has no document resolver → stays a raw related link
|
||||
service.link(task.taskId, "adr-7", TaskLinkType.IMPLEMENTS, TaskTargetKind.DOC)
|
||||
service.addNote(task.taskId, TaskNoteAuthor.AGENT, "kickoff")
|
||||
|
||||
val bundle = assembler.assemble(task.taskId)!!
|
||||
|
||||
assertEquals("auth-2", bundle.id)
|
||||
assertEquals("users stay authenticated", bundle.goal)
|
||||
assertEquals(listOf("refresh endpoint exists"), bundle.acceptanceCriteria)
|
||||
assertEquals(listOf("backend/auth/**"), bundle.relevantFiles)
|
||||
|
||||
val dependency = bundle.dependencies.single()
|
||||
assertEquals("auth-1", dependency.id)
|
||||
assertEquals("DONE", dependency.status)
|
||||
assertEquals("DEPENDS_ON", dependency.link)
|
||||
|
||||
assertEquals(RelatedLink("adr-7", "IMPLEMENTS"), bundle.relatedLinks.single())
|
||||
assertEquals("kickoff", bundle.notes.single().body)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `render produces a compact labelled bundle`() = runBlocking {
|
||||
val task = service.createTask(project, "JWT refresh", "stay authed", listOf("rotates"))
|
||||
val text = assembler.assemble(task.taskId)!!.render()
|
||||
|
||||
assertTrue(text.startsWith("task auth-1 [TODO] JWT refresh"))
|
||||
assertTrue(text.contains("goal: stay authed"))
|
||||
assertTrue(text.contains("- rotates"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `assemble returns null for an unknown task`() = runBlocking {
|
||||
assertNull(assembler.assemble(TaskId("auth-999")))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bundle is enriched with knowledge retrieved over the goal`() = runBlocking {
|
||||
val captured = mutableListOf<String>()
|
||||
val fakeRetriever = object : TaskKnowledgeRetriever {
|
||||
override suspend fun retrieve(query: String, limit: Int): List<KnowledgeHit> {
|
||||
captured += query
|
||||
return listOf(KnowledgeHit("backend/auth/Jwt.kt", "fun refresh(token)", 0.91))
|
||||
}
|
||||
}
|
||||
val enriched = TaskContextAssembler(service, fakeRetriever)
|
||||
val task = service.createTask(project, "JWT refresh", "users stay authenticated")
|
||||
|
||||
val bundle = enriched.assemble(task.taskId)!!
|
||||
|
||||
assertEquals(listOf("JWT refresh users stay authenticated"), captured)
|
||||
assertEquals("backend/auth/Jwt.kt", bundle.relevantKnowledge.single().source)
|
||||
assertTrue(bundle.render().contains("relevant_knowledge"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `linked docs are resolved inline and dropped from raw related links`() = runBlocking {
|
||||
val docs = object : TaskDocumentResolver {
|
||||
override suspend fun resolve(targetId: String): ResolvedDocument? =
|
||||
if (targetId == "adr-7") {
|
||||
ResolvedDocument("ADR 7: use redis", "docs/decisions/adr-0007-redis.md", "faster invalidation")
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}
|
||||
val enriched = TaskContextAssembler(service, documentResolver = docs)
|
||||
val task = service.createTask(project, "JWT refresh", "auth")
|
||||
service.link(task.taskId, "adr-7", TaskLinkType.IMPLEMENTS, TaskTargetKind.DOC) // resolves to a doc
|
||||
service.link(task.taskId, "ext-123", TaskLinkType.RELATES_TO, TaskTargetKind.ARTIFACT) // stays raw
|
||||
|
||||
val bundle = enriched.assemble(task.taskId)!!
|
||||
|
||||
val doc = bundle.documents.single()
|
||||
assertEquals("adr-7", doc.targetId)
|
||||
assertEquals("ADR 7: use redis", doc.title)
|
||||
assertEquals("IMPLEMENTS", doc.type)
|
||||
assertEquals(RelatedLink("ext-123", "RELATES_TO"), bundle.relatedLinks.single())
|
||||
assertTrue(bundle.render().contains("documents:"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `retrieval failure is swallowed and the bundle still returns`() = runBlocking {
|
||||
val flaky = object : TaskKnowledgeRetriever {
|
||||
override suspend fun retrieve(query: String, limit: Int): List<KnowledgeHit> =
|
||||
error("embedder offline")
|
||||
}
|
||||
val task = service.createTask(project, "JWT refresh", "stay authed")
|
||||
|
||||
val bundle = TaskContextAssembler(service, flaky).assemble(task.taskId)!!
|
||||
|
||||
assertTrue(bundle.relevantKnowledge.isEmpty())
|
||||
assertEquals("JWT refresh", bundle.title)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.correx.core.tasks
|
||||
|
||||
import com.correx.core.events.events.EventMetadata
|
||||
import com.correx.core.events.events.EventPayload
|
||||
import com.correx.core.events.events.StoredEvent
|
||||
import com.correx.core.events.events.TaskCancelledEvent
|
||||
import com.correx.core.events.events.TaskCreatedEvent
|
||||
import com.correx.core.events.types.EventId
|
||||
import com.correx.core.events.types.ProjectId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.TaskId
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TaskCounterProjectionTest {
|
||||
|
||||
private val projection = TaskCounterProjection("auth")
|
||||
private val projectId = ProjectId("auth")
|
||||
|
||||
private fun stored(payload: EventPayload, seq: Long) = StoredEvent(
|
||||
metadata = EventMetadata(
|
||||
eventId = EventId("e-$seq"),
|
||||
sessionId = SessionId("tasks:auth"),
|
||||
timestamp = Instant.parse("2026-01-01T00:00:0${seq}Z"),
|
||||
schemaVersion = 1,
|
||||
causationId = null,
|
||||
correlationId = null,
|
||||
),
|
||||
sequence = seq,
|
||||
sessionSequence = seq,
|
||||
payload = payload,
|
||||
)
|
||||
|
||||
private fun created(id: TaskId) =
|
||||
TaskCreatedEvent(id, projectId, id.value, "title", "goal")
|
||||
|
||||
@Test
|
||||
fun `counts only created events and survives cancellation`() {
|
||||
val events = listOf(
|
||||
stored(created(TaskId("auth-1")), 1),
|
||||
stored(created(TaskId("auth-2")), 2),
|
||||
stored(TaskCancelledEvent(TaskId("auth-1")), 3),
|
||||
)
|
||||
|
||||
val state = events.fold(projection.initial()) { s, e -> projection.apply(s, e) }
|
||||
|
||||
assertEquals("auth", state.projectId)
|
||||
assertEquals(2, state.count)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
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.TaskNoteAuthor
|
||||
import com.correx.core.events.types.TaskTargetKind
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertFalse
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TaskServiceTest {
|
||||
|
||||
private val store = InMemoryEventStore()
|
||||
private val service = TaskService(store)
|
||||
private val project = ProjectId("auth")
|
||||
|
||||
@Test
|
||||
fun `createTask allocates sequential human keys per project`() = runBlocking {
|
||||
val first = service.createTask(project, "Add JWT refresh", "users stay authenticated")
|
||||
val second = service.createTask(project, "Rotate keys", "tokens rotate")
|
||||
|
||||
assertEquals("auth-1", first.taskId.value)
|
||||
assertEquals("auth-2", second.taskId.value)
|
||||
assertEquals(TaskStatus.TODO, first.state.status)
|
||||
assertEquals("Add JWT refresh", first.state.title)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `lifecycle round-trips through append and replay`() = runBlocking {
|
||||
val task = service.createTask(project, "Add JWT refresh", "users stay authenticated")
|
||||
val id = task.taskId
|
||||
|
||||
assertEquals(TaskStatus.IN_PROGRESS, service.claim(id, "claude-opus")?.state?.status)
|
||||
assertEquals(TaskStatus.IN_REVIEW, service.submitForReview(id)?.state?.status)
|
||||
val done = service.complete(id)
|
||||
|
||||
assertEquals(TaskStatus.DONE, done?.state?.status)
|
||||
assertEquals("claude-opus", done?.state?.claimant)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `links and notes are persisted on the task`() = runBlocking {
|
||||
val id = service.createTask(project, "Add JWT refresh", "auth").taskId
|
||||
|
||||
service.link(id, targetId = "adr-7", type = TaskLinkType.IMPLEMENTS, targetKind = TaskTargetKind.DOC)
|
||||
val withNote = service.addNote(id, TaskNoteAuthor.AGENT, "migration failed, retrying")
|
||||
|
||||
assertEquals(
|
||||
TaskLink("adr-7", TaskLinkType.IMPLEMENTS, TaskTargetKind.DOC),
|
||||
withNote?.state?.links?.single(),
|
||||
)
|
||||
assertEquals("migration failed, retrying", withNote?.state?.notes?.single()?.body)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `delete is a soft tombstone hidden from reads`() = runBlocking {
|
||||
val id = service.createTask(project, "throwaway", "oops").taskId
|
||||
|
||||
assertTrue(service.delete(id))
|
||||
assertNull(service.getTask(id))
|
||||
assertTrue(service.list(project).none { it.taskId == id })
|
||||
// A second delete is a no-op: the task is already gone from reads.
|
||||
assertFalse(service.delete(id))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `mutating an unknown task returns null`() = runBlocking {
|
||||
assertNull(service.claim(TaskId("auth-999"), "claude-opus"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `keys keep incrementing even after deletion`() = runBlocking {
|
||||
val first = service.createTask(project, "one", "g").taskId
|
||||
service.delete(first)
|
||||
val third = service.createTask(project, "two", "g")
|
||||
|
||||
// count() is creation-based, so the deleted id is never reused.
|
||||
assertEquals("auth-2", third.taskId.value)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user