luplo.core.tasks¶

Task domain — a thin wrapper around items where item_type='task'.

Tasks live in items (D3 — no separate table). Each state transition creates a new row that supersedes the previous one (standard supersede pattern); reorder_tasks is the one exception, which updates context.sort_order in place per P10.

User-facing task identity = supersede-chain head. Mutating helpers accept any id in the chain and resolve to the head before acting; the returned Item is the new head.

Concurrency (P7): no DB-level UNIQUE for in-progress tasks. start_task takes a SELECT … FOR UPDATE on the candidate “current in-progress” row(s) per work unit and raises TaskAlreadyInProgressError if any exists. All start paths must go through this function.

Attributes¶

ITEM_TYPE

Functions¶

`create_task`(→ luplo.core.models.Item)	Create a new task in `proposed` status.
`get_task`(→ luplo.core.models.Item \| None)	Fetch the head of the chain containing task_id.
`list_tasks`(→ list[luplo.core.models.Item])	List the heads of every task chain in work_unit_id.
`get_in_progress_task`(→ luplo.core.models.Item \| None)	Return the single in_progress task head for work_unit_id, or `None`.
`start_task`(→ luplo.core.models.Item)	Transition a task to `in_progress`.
`complete_task`(→ luplo.core.models.Item)	Transition a task to `done`.
`block_task`(→ luplo.core.models.Item)	Transition a task to `blocked` with a reason.
`skip_task`(→ luplo.core.models.Item)	Transition a task to `skipped`.
`suggest_decision_from_task`(...)	Build a draft `decision` item from a task without inserting it.
`edit_task`(→ luplo.core.models.Item)	Edit a task's title / body / sort_order by creating a supersede row.
`reorder_tasks`(→ list[luplo.core.models.Item])	Set sort_order for a list of tasks (gap-10 strategy).

Module Contents¶

luplo.core.tasks.ITEM_TYPE = 'task'¶

async luplo.core.tasks.create_task(conn: psycopg.AsyncConnection[Any], *, project_id: str, work_unit_id: str, title: str, actor_id: str, sort_order: int | None = None, systems: list[str] | None = None, body: str | None = None, context_extra: dict[str, Any] | None = None) → luplo.core.models.Item¶

Create a new task in proposed status.

If sort_order is None, it defaults to max(sort_order) + 10 for the work unit (gap-strategy starting at 10).

async luplo.core.tasks.get_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, project_id: str | None = None) → luplo.core.models.Item | None¶

Fetch the head of the chain containing task_id.

Accepts a full UUID or a hex prefix (≥8 chars). Returns None if no row matches; raises AmbiguousIdError if a prefix matches distinct chains.

async luplo.core.tasks.list_tasks(conn: psycopg.AsyncConnection[Any], work_unit_id: str, *, status: str | None = None) → list[luplo.core.models.Item]¶

List the heads of every task chain in work_unit_id.

Ordered by sort_order ASC, then created_at.

async luplo.core.tasks.get_in_progress_task(conn: psycopg.AsyncConnection[Any], work_unit_id: str) → luplo.core.models.Item | None¶: Return the single in_progress task head for work_unit_id, or None.

async luplo.core.tasks.start_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, actor_id: str, project_id: str | None = None) → luplo.core.models.Item¶

Transition a task to in_progress.

P7: enforces the “at most one in_progress per work_unit” invariant in the domain layer (no DB UNIQUE) by taking a row-level lock on any candidate in_progress heads in the same work unit.

Passing project_id scopes prefix resolution to a single project, so an abc12345 prefix that happens to match a task in a different project can never silently mutate that other row.

Raises:: TaskNotFoundError, TaskStateTransitionError, TaskAlreadyInProgressError. –

async luplo.core.tasks.complete_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, actor_id: str, summary: str | None = None, project_id: str | None = None) → luplo.core.models.Item¶

Transition a task to done.

Pass project_id to scope prefix resolution to a single project.

async luplo.core.tasks.block_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, actor_id: str, reason: str, project_id: str | None = None) → luplo.core.models.Item¶

Transition a task to blocked with a reason.

The associated decision item is created by LocalBackend.block_task (cross-cutting), not here. Pass project_id to scope prefix resolution to a single project.

async luplo.core.tasks.skip_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, actor_id: str, reason: str | None = None, project_id: str | None = None) → luplo.core.models.Item¶

Transition a task to skipped.

Pass project_id to scope prefix resolution to a single project.

async luplo.core.tasks.suggest_decision_from_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, project_id: str | None = None) → luplo.core.models.ItemCreate | None¶

Build a draft decision item from a task without inserting it.

Intended for the “on task done propose a decision” flow. The draft is returned to the caller; nothing is written to the database. Inserting the draft remains an explicit user step — the augment-not-replace commitment in the philosophy doc forbids the tool from deciding this moment is the moment to save a decision.

Returns None when the task lacks enough content to support a meaningful draft (no body, no summary, default title). Returning a template in that case would be a confident guess dressed as a suggestion — honesty-over-coverage says no.

Parameters:

conn – Async psycopg connection.
task_id – Any ID in the task’s supersede chain.
project_id – Optional project scope for prefix resolution.

Returns:

An ItemCreate ready to be shown to the user, or None when there is nothing worth suggesting.

Raises:

TaskNotFoundError – If the prefix does not resolve to a task.

async luplo.core.tasks.edit_task(conn: psycopg.AsyncConnection[Any], task_id: str, *, actor_id: str, title: str | None = None, body: str | None = None, sort_order: int | None = None, project_id: str | None = None) → luplo.core.models.Item¶

Edit a task’s title / body / sort_order by creating a supersede row.

The status machine is preserved — an edit never transitions a task between proposed / in_progress / done / blocked / skipped. Changing status is the job of the dedicated start_task(), complete_task(), and similar verbs.

Pass only the fields you want to change; the rest are copied from the current head. Passing no changeable field is a no-op that still returns the current head unchanged (useful when the caller only wants to revalidate existence).

Parameters:

conn – Async psycopg connection.
task_id – Full UUID or hex prefix of any row in the task’s supersede chain.
actor_id – Who is performing the edit.
title – New title, or None to keep the current one.
body – New body, or None to keep the current one.
sort_order – New sort_order, or None to keep the current one.
project_id – Optional project scope for prefix resolution.

Returns:

The new head row (or the unchanged head when no fields changed).

Raises:

TaskNotFoundError – If the prefix does not resolve to a task.
AmbiguousIdError – If the prefix matches multiple distinct heads.

async luplo.core.tasks.reorder_tasks(conn: psycopg.AsyncConnection[Any], work_unit_id: str, task_ids: list[str], *, actor_id: str, project_id: str | None = None) → list[luplo.core.models.Item]¶

Set sort_order for a list of tasks (gap-10 strategy).

Per P10, this is an in-place UPDATE: no new supersede rows, no items_history entries. The audit trail is a single item.update entry recorded by the LocalBackend wrapper.

task_ids may include any id in each chain — they are resolved to heads internally. The returned list mirrors the input order with the refreshed head rows.

Raises TaskNotFoundError if any id cannot be resolved or its head does not belong to work_unit_id.