todos

• before_task_create
  Before a task is created
  outputs: Patch (modify input), Veto (cancel creation)
• after_task_create
  After a task is created
  outputs: SideEffect (trigger follow-up actions)
• before_task_complete
  Before a task is marked complete
  outputs: Patch (modify completion), Veto (prevent completion)
• after_task_complete
  After a task is completed
  outputs: SideEffect (trigger recurrence, notifications)
• before_task_update
  Before a task is updated
  outputs: Patch (modify changes), Veto (prevent update)
• on_task_overdue
  When a task becomes overdue
  outputs: SideEffect (send notifications, escalate)

• TaskCreated v1
  After a new task is created
  payload: {task_id, owner_id, container_id, title, priority, due_at, tags, created_at}
• TaskUpdated v1
  After a task is modified
  payload: {task_id, changes: {field: (old, new)}, updated_at}
• TaskCompleted v1
  After a task is completed
  payload: {task_id, completed_at, recurrence_triggered}
• TaskReopened v1
  After a completed task is reopened
  payload: {task_id, reopened_at, previous_completed_at}
• TaskDeleted v1
  After a task is soft-deleted
  payload: {task_id, deleted_at}
• TaskMoved v1
  After a task is moved to another container
  payload: {task_id, from_container_id, to_container_id, moved_at}
• TaskOverdue v1
  When a task's due_at passes without completion
  payload: {task_id, owner_id, due_at, overdue_since}
• RecurrenceTriggered v1
  When completing a recurring task creates a new instance
  payload: {source_task_id, new_task_id, triggered_at}
• ReminderRequested v1
  When a reminder is scheduled for a task
  payload: {task_id, reminder_id, remind_at, channel}
• ContainerCreated v1
  After a new container is created
  payload: {container_id, owner_id, name, container_type, parent_id, created_at}
• ContainerArchived v1
  After a container is archived
  payload: {container_id, archived_at, cascade_archived: []}
• TagCreated v1
  After a new tag is created
  payload: {tag_id, owner_id, name, color, created_at}
• TagAddedToTask v1
  After a tag is added to a task
  payload: {task_id, tag_id, added_at}
• TagRemovedFromTask v1
  After a tag is removed from a task
  payload: {task_id, tag_id, removed_at}
• ContainerMoved v1
  After a container is moved to a new parent
  payload: {container_id, old_parent_id, new_parent_id, moved_at}
• NoteAddedToTask v1
  After a note is added to a task
  payload: {task_id, note_id, author_id, created_at}
• TasksLinked v1
  After two tasks are linked
  payload: {link_id, source_task_id, target_task_id, link_type, created_at}
• TasksUnlinked v1
  After a link between tasks is removed
  payload: {link_id, source_task_id, target_task_id, link_type, removed_at}

from uuid import uuid4

from psp.components.todos.application.use_cases import CreateTask
from psp.components.todos.application.dto import CreateTaskInput
from psp.components.todos.domain import Priority
from psp.platform.clock import SystemClock
from psp.platform.eventbus import InMemoryEventBus

# Wire dependencies (in real app, use DI container)
create_task = CreateTask(
    task_repo=task_repo,
    container_repo=container_repo,
    clock=SystemClock(),
    event_bus=InMemoryEventBus(),
)

# Execute use case
result = create_task.execute(
    owner_id=current_user_id,
    input=CreateTaskInput(
        container_id=inbox_id,
        title="Review pull request",
        priority=Priority.HIGH,
        due_at=tomorrow,
    ),
)

print(f"Created task {result.id}: {result.title}")

Inject dependencies and call CreateTask use case.

from psp.platform.query import QuerySpec, FilterOp, SortDir

# Find high-priority incomplete tasks due this week
spec = QuerySpec(
    filters=[
        ("status", FilterOp.NE, "completed"),
        ("priority", FilterOp.IN, ["high", "urgent"]),
        ("due_at", FilterOp.LTE, end_of_week),
    ],
    sort_by="due_at",
    sort_dir=SortDir.ASC,
    limit=20,
)

result = task_repo.query(spec)
for task in result.items:
    print(f"[{task.priority.value}] {task.title} - due {task.due_at}")

Use QuerySpec to filter and paginate tasks.

from psp.components.todos.application.use_cases import CompleteTask

complete_task = CompleteTask(
    task_repo=task_repo,
    clock=clock,
    event_bus=event_bus,
    recurrence_engine=recurrence_engine,  # Optional
)

# Complete the task - if it has recurrence, a new task is created
result = complete_task.execute(task_id=recurring_task_id)

if result.next_task_id:
    print(f"Completed! Next occurrence created: {result.next_task_id}")
else:
    print("Completed (no recurrence)")

Completing a recurring task triggers next instance.

from psp.components.todos.application.use_cases import ExplainTaskPriority

explain = ExplainTaskPriority(
    task_repo=task_repo,
    container_repo=container_repo,
    activity_log=activity_log,  # Optional
)

result = explain.execute(task_id=task.id)

print(f"Priority: {result.priority}")
print(f"Source: {result.source}")  # explicit, inherited, computed
print(f"Reason: {result.reason}")
# e.g., "Inherited from project 'Q4 Goals' which has priority=high"
# e.g., "Computed: due in < 24 hours with no priority set"
for entry in result.history:
    print(f"  {entry.timestamp}: {entry.action}")

Understand why a task has its current priority.