Integrating Components

The Scenario

Budget wants to:

Veto task completion when the user is over budget (hook)
Debit the budget when a task completes (event)
Track which tasks have costs (store task IDs)

This requires all three interaction patterns: hooks, events, and ID-based references.

Step 1: Store Task IDs

Budget stores the task ID, not the task object:

budget/domain/models.py

@dataclass
class TaskCost:
    """Cost assigned to a task. Budget only stores the ID."""
    id: UUID
    owner_id: UUID
    task_id: UUID        # reference to Todos, not the Task itself
    token_type_id: UUID
    amount: int
                    

Budget doesn't import Task. It doesn't know (or care) what fields Task has.

Step 2: Register a Hook to Veto

Budget registers a hook to block task completion when over budget:

budget/module.py

class BudgetModule(Module):
    def register_hooks(self, dispatcher: HookDispatcher):
        dispatcher.register(
            "todos.before_complete_task",
            self._check_budget
        )

    def _check_budget(self, context: dict) -> HookResponse:
        task_id = context["task_id"]
        owner_id = context["owner_id"]

        # Does this task have a cost?
        cost = self.cost_repo.get_by_task(task_id)
        if not cost:
            return Allow()  # no cost, allow

        # Check budget
        balance = self.ledger.get_balance(owner_id, cost.token_type_id)
        if balance < cost.amount:
            return Veto(
                f"Insufficient budget: need {cost.amount}, have {balance}"
            )

        return Allow()
                    

The hook receives primitive data (task_id, owner_id), not a Task object. This maintains component boundaries.

Step 3: Subscribe to Events

After a task completes, debit the budget:

budget/module.py

class BudgetModule(Module):
    def register_subscribers(self, event_bus: EventBus):
        event_bus.subscribe(
            "todos.TaskCompleted",
            self._on_task_completed
        )

    def _on_task_completed(self, envelope: EventEnvelope):
        event = envelope.payload
        task_id = event["task_id"]
        owner_id = event["owner_id"]

        # Idempotency check
        key = f"budget:task_completed:{envelope.id}"
        if self.idempotency.exists(key):
            return

        # Does this task have a cost?
        cost = self.cost_repo.get_by_task(task_id)
        if not cost:
            return

        # Debit the budget
        self.ledger.debit(
            owner_id=owner_id,
            token_type_id=cost.token_type_id,
            amount=cost.amount,
            reference_id=task_id,
            reason=f"Task completed: {event.get('title', task_id)}"
        )

        self.idempotency.mark(key)
                    

The Complete Flow

Integration Flow

%%{init: {"sequence": {"useMaxWidth": false}}}%% sequenceDiagram participant U as User participant T as Todos participant H as Hook participant B as Budget U->>T: Complete task T->>H: dispatch(before_complete) H->>B: check budget B-->>H: Allow H-->>T: proceed T->>T: task.complete() T->>B: publish(TaskCompleted) B->>B: debit ledger

API for Assigning Costs

Budget provides an endpoint to assign costs to tasks:

budget/adapters/http/router.py

@router.post("/task-costs")
def assign_task_cost(request: AssignTaskCostRequest):
    """Assign a cost to a task (by ID)."""
    cost = TaskCost(
        id=uuid4(),
        owner_id=request.owner_id,
        task_id=request.task_id,  # just the UUID
        token_type_id=request.token_type_id,
        amount=request.amount
    )
    cost_repo.save(cost)
    return {"id": cost.id}
                    

The UI calls this endpoint after creating a task. Budget never validates that the task exists—it just stores the ID.

Testing the Integration

Test each piece independently:

tests/test_budget_integration.py

def test_hook_vetoes_when_over_budget():
    # Setup: user has 5 tokens, task costs 10
    ledger.credit(owner_id, token_type_id, 5)
    cost_repo.save(TaskCost(task_id=task_id, amount=10, ...))

    # Act
    result = budget_module._check_budget({
        "task_id": task_id,
        "owner_id": owner_id
    })

    # Assert
    assert isinstance(result, Veto)
    assert "Insufficient" in result.reason

def test_event_debits_budget():
    # Setup: user has 10 tokens, task costs 3
    ledger.credit(owner_id, token_type_id, 10)
    cost_repo.save(TaskCost(task_id=task_id, amount=3, ...))

    # Act: simulate event
    budget_module._on_task_completed(EventEnvelope(
        id=uuid4(),
        payload={"task_id": task_id, "owner_id": owner_id}
    ))

    # Assert
    assert ledger.get_balance(owner_id, token_type_id) == 7
                    