hooks

• before_create_task
  Before a task is persisted
  outputs: Veto, Patch, SideEffectRequest
• after_create_task
  After a task is persisted
  outputs: SideEffectRequest
• before_update_task
  Before task changes are persisted
  outputs: Veto, Patch, SideEffectRequest
• after_update_task
  After task changes are persisted
  outputs: SideEffectRequest
• before_complete_task
  Before a task is marked complete
  outputs: Veto, Patch, SideEffectRequest
• after_complete_task
  After a task is marked complete
  outputs: SideEffectRequest

• HookRegistered v1
  After a hook handler is registered
  payload: {hook_point, handler_name, priority}
• HookUnregistered v1
  After a hook handler is removed
  payload: {hook_point, handler_name}
• HookDispatched v1
  After hooks are dispatched at a hook point
  payload: {hook_point, entity_type, handlers_run, vetoed}
• HookVetoed v1
  When a hook vetoes an operation
  payload: {hook_point, handler_name, reason, entity_type}

from psp.platform.hooks import HookDispatcher, HookPoint, HookContext, Patch

dispatcher = HookDispatcher()

def auto_tag_emails(ctx: HookContext) -> Patch | None:
    """Add 'email' tag if title contains 'email'."""
    title = ctx.entity_data.get("title", "").lower()
    if "email" in title:
        return Patch(add_items={"tags": ["email", "communication"]})
    return None

dispatcher.register(
    HookPoint.BEFORE_CREATE_TASK,
    auto_tag_emails,
    priority=50,
    name="auto_tag_emails",
)

Add auto-tagging before task creation.

from psp.platform.hooks import HookContext, Veto

def block_empty_titles(ctx: HookContext) -> Veto | None:
    """Reject tasks with empty titles."""
    title = ctx.entity_data.get("title", "").strip()
    if not title:
        return Veto(reason="Task title cannot be empty")
    return None

dispatcher.register(HookPoint.BEFORE_CREATE_TASK, block_empty_titles)

Reject task creation based on business rules.

from psp.platform.hooks import HookContext, SideEffectRequest

def notify_on_complete(ctx: HookContext) -> SideEffectRequest:
    """Send notification when task is completed."""
    return SideEffectRequest(
        kind="send_notification",
        payload={
            "channel": "email",
            "recipient": ctx.entity_data.get("owner_id"),
            "message": f"Task '{ctx.entity_data.get('title')}' completed!",
        },
    )

dispatcher.register(HookPoint.AFTER_COMPLETE_TASK, notify_on_complete)

Schedule notifications after task completion.

async def execute(self, input: CreateTaskInput) -> CreateTaskOutput:
    # Build hook context
    ctx = HookContext(
        hook_point=HookPoint.BEFORE_CREATE_TASK,
        entity_type="Task",
        entity_id=None,
        entity_data={"title": input.title, "tags": input.tags},
    )

    # Dispatch and check for veto
    result = self._dispatcher.dispatch(ctx)
    if result.vetoed:
        raise ValidationError(result.veto_reason)

    # Apply patches from hooks
    tags = input.tags + result.merged_patch.add_items.get("tags", [])

    # Create task with modified data
    task = Task(title=input.title, tags=tags)
    await self._repo.add(task)

    # Execute side effects
    for side_effect in result.side_effects:
        await self._executor.run(side_effect)

Execute hooks and handle results.