Budget Component

budget/
├── __init__.py           # exports BudgetModule, BudgetComponent
├── component.py          # composition root (framework-agnostic)
├── module.py             # catalogue registration metadata
│
├── domain/
│   ├── models/
│   │   ├── token_type.py   # TokenType entity
│   │   ├── budget_plan.py  # BudgetPlan with policies
│   │   ├── period.py       # Period value object (day/week)
│   │   ├── transaction.py  # LedgerTransaction (append-only)
│   │   ├── scheduled_task.py # ScheduledTask with cost
│   │   ├── debt.py         # DebtObligation entity
│   │   └── snapshot.py     # PeriodSnapshot (derived)
│   ├── events.py           # 17 domain events
│   └── errors.py           # domain-specific errors
│
├── application/
│   ├── ports/
│   │   └── repos.py        # 5 repository protocols
│   └── use_cases/
│       ├── token_type_crud.py  # token type CRUD
│       ├── budget_plan_crud.py # budget plan CRUD
│       ├── snapshot.py         # snapshot calculations
│       ├── get_period_snapshot.py # lazy grants
│       ├── check_budget.py     # budget validation
│       ├── schedule_task.py    # task scheduling
│       ├── borrow.py           # debt management
│       └── advanced.py         # split, rollover, auto-balance
│
└── adapters/
    ├── http/
    │   ├── router.py         # FastAPI routes
    │   ├── dependencies.py   # DI wrappers
    │   └── schemas.py        # Pydantic models
    └── persistence/
        └── repo_memory.py    # in-memory implementations

@dataclass
class TokenType:
    id: UUID
    owner_id: UUID
    name: str           # unique per owner, max 32 chars
    color: str | None   # optional hex color
    created_at: datetime
    updated_at: datetime

    def __post_init__(self):
        # Invariant: name is lowercase alphanumeric with underscores
        if not re.match(r'^[a-z0-9_]+$', self.name):
            raise InvalidTokenTypeError("Invalid name format")

    def rename(self, new_name: str) -> "TokenType": ...
    def update_color(self, color: str | None) -> "TokenType": ...

class EnforcementPolicy(Enum):
    HARD_LIMIT = "hard_limit"     # reject if over budget
    SOFT_LIMIT = "soft_limit"     # warn but allow
    AUTO_BORROW = "auto_borrow"   # borrow from future

class RolloverPolicy(Enum):
    NONE = "none"      # unused tokens expire
    FULL = "full"      # all unused carry over
    CAPPED = "capped"  # carry over up to cap

@dataclass
class BudgetPlan:
    id: UUID
    owner_id: UUID
    period_type: PeriodType       # DAY or WEEK
    effective_from: date
    allocations: dict[UUID, int]  # token_type_id -> amount
    enforcement_policy: EnforcementPolicy
    rollover_policy: RolloverPolicy
    rollover_cap: int | None      # for CAPPED policy

    def get_allocation(self, token_type_id: UUID) -> int: ...
    def with_allocation(self, token_type_id: UUID, amount: int) -> "BudgetPlan": ...

class PeriodType(Enum):
    DAY = "day"
    WEEK = "week"

@dataclass(frozen=True)
class Period:
    start_date: date
    period_type: PeriodType

    def contains(self, d: date) -> bool: ...
    def next(self) -> "Period": ...
    def previous(self) -> "Period": ...

    @classmethod
    def for_date(cls, d: date, period_type: PeriodType) -> "Period": ...

    @classmethod
    def today(cls, period_type: PeriodType) -> "Period": ...

class TransactionType(Enum):
    GRANT = "grant"           # budget allocation
    RESERVE = "reserve"       # scheduled task
    RELEASE = "release"       # unscheduled task
    SPEND = "spend"           # completed task
    BORROW = "borrow"         # borrowed from future
    REPAY = "repay"           # debt repayment
    EXPIRE = "expire"         # unused budget expired
    ROLLOVER_IN = "rollover_in"  # carried from previous

@dataclass(frozen=True)
class LedgerTransaction:
    id: UUID
    owner_id: UUID
    period: Period
    token_type_id: UUID
    transaction_type: TransactionType
    amount: int              # positive or negative
    task_id: UUID | None     # reference to Todos component
    source_period: Period | None  # for REPAY transactions
    created_at: datetime

    @property
    def is_credit(self) -> bool: ...
    @property
    def is_debit(self) -> bool: ...

class TokenTypeRepository(Protocol):
    def get(self, token_type_id: UUID) -> TokenType | None: ...
    def save(self, token_type: TokenType) -> None: ...
    def get_by_name(self, owner_id: UUID, name: str) -> TokenType | None: ...
    def list_by_owner(self, owner_id: UUID) -> list[TokenType]: ...

class BudgetPlanRepository(Protocol):
    def get_effective(
        self, owner_id: UUID, period_type: PeriodType, as_of: date
    ) -> BudgetPlan | None:
        """Get the budget plan effective on a given date."""
        ...

class LedgerRepository(Protocol):
    """Append-only ledger for budget transactions."""
    def append(self, transaction: LedgerTransaction) -> None: ...
    def get_transactions(
        self, owner_id: UUID, period: Period, token_type_id: UUID | None = None
    ) -> list[LedgerTransaction]: ...

class ScheduledTaskRepository(Protocol): ...
class DebtObligationRepository(Protocol): ...

class GetPeriodSnapshot:
    def __init__(
        self,
        ledger_repo: LedgerRepository,
        plan_repo: BudgetPlanRepository,
        clock: Clock,
    ):
        ...

    def execute(self, owner_id: UUID, period: Period) -> PeriodSnapshot:
        # Get all transactions for this period
        transactions = self.ledger_repo.get_transactions(owner_id, period)

        # Check if GRANT exists, if not create it (lazy initialization)
        if not any(t.transaction_type == TransactionType.GRANT for t in transactions):
            plan = self.plan_repo.get_effective(owner_id, period.period_type, period.start_date)
            if plan:
                for token_type_id, amount in plan.allocations.items():
                    self.ledger_repo.append(LedgerTransaction(
                        transaction_type=TransactionType.GRANT,
                        amount=amount,
                        ...
                    ))

        # Aggregate into snapshot
        return self._calculate_snapshot(transactions)

class ScheduleTask:
    def execute(self, input: ScheduleTaskInput) -> ScheduledTaskOutput:
        # 1. Check budget availability
        snapshot = self.get_snapshot.execute(input.owner_id, input.period)
        for token_type_id, amount in input.cost.costs.items():
            if snapshot.remaining(token_type_id) < amount:
                raise InsufficientBudgetError(...)

        # 2. Create scheduled task
        scheduled = ScheduledTask(
            id=uuid4(),
            task_id=input.task_id,
            owner_id=input.owner_id,
            period=input.period,
            cost=input.cost,
            status=ScheduledTaskStatus.RESERVED,
        )
        self.scheduled_repo.save(scheduled)

        # 3. Record RESERVE transactions
        for token_type_id, amount in input.cost.costs.items():
            self.ledger_repo.append(LedgerTransaction(
                transaction_type=TransactionType.RESERVE,
                amount=-amount,  # debit
                task_id=input.task_id,
                ...
            ))

        # 4. Publish event
        self.event_bus.publish(TaskScheduled(...))

class BudgetComponent:
    """Framework-agnostic composition root."""

    def __init__(
        self,
        # Repositories
        token_type_repo: TokenTypeRepository,
        budget_plan_repo: BudgetPlanRepository,
        ledger_repo: LedgerRepository,
        scheduled_task_repo: ScheduledTaskRepository,
        debt_repo: DebtObligationRepository,
        # Platform services
        clock: Clock,
        event_bus: EventBus,
        borrow_policy: BorrowPolicy,
    ) -> None:
        # Token type use cases
        self.create_token_type = CreateTokenType(token_type_repo, clock, event_bus)
        self.list_token_types = ListTokenTypes(token_type_repo)

        # Budget plan use cases
        self.create_budget_plan = CreateBudgetPlan(budget_plan_repo, clock, event_bus)
        self.get_effective_budget_plan = GetEffectiveBudgetPlan(budget_plan_repo)

        # Snapshot use cases
        self.get_period_snapshot = GetPeriodSnapshot(ledger_repo, budget_plan_repo, clock)
        self.check_budget = CheckBudget(self.get_period_snapshot)

        # Scheduling use cases
        self.schedule_task = ScheduleTask(
            scheduled_task_repo, ledger_repo, self.get_period_snapshot, clock, event_bus
        )
        self.spend_task_budget = SpendTaskBudget(
            scheduled_task_repo, ledger_repo, clock, event_bus
        )

        # Debt use cases
        self.borrow_budget = BorrowBudget(debt_repo, ledger_repo, borrow_policy, clock, event_bus)
        self.repay_debt = RepayDebt(debt_repo, ledger_repo, clock, event_bus)
        # ... all 25+ use cases

def create_budget_component(
    *,
    clock: Clock | None = None,
    event_bus: EventBus | None = None,
    borrow_policy: BorrowPolicy | None = None,
) -> BudgetComponent:
    """Factory with default in-memory repositories."""
    return BudgetComponent(
        token_type_repo=InMemoryTokenTypeRepository(),
        budget_plan_repo=InMemoryBudgetPlanRepository(),
        ledger_repo=InMemoryLedgerRepository(),
        scheduled_task_repo=InMemoryScheduledTaskRepository(),
        debt_repo=InMemoryDebtObligationRepository(),
        clock=clock or SystemClock(),
        event_bus=event_bus or InMemoryEventBus(),
        borrow_policy=borrow_policy or DefaultBorrowPolicy(),
    )

# Usage
component = create_budget_component()
snapshot = component.get_period_snapshot.execute(owner_id, period)

@router.get("/snapshot/{period_date}")
def get_snapshot(
    period_date: date,
    period_type: PeriodType = Query(PeriodType.DAY),
    x_owner_id: UUID = Header(...),
    use_case: GetPeriodSnapshot = Depends(get_period_snapshot),
) -> SnapshotResponse:
    period = Period.for_date(period_date, period_type)
    snapshot = use_case.execute(x_owner_id, period)
    return snapshot_to_response(snapshot)

@router.post("/schedule")
def schedule_task(
    request: ScheduleTaskRequest,
    x_owner_id: UUID = Header(...),
    use_case: ScheduleTask = Depends(get_schedule_task),
) -> ScheduledTaskResponse:
    output = use_case.execute(ScheduleTaskInput(
        owner_id=x_owner_id,
        task_id=request.task_id,
        period=Period.for_date(request.period_date, request.period_type),
        cost=TaskCost(request.costs),
    ))
    return scheduled_task_to_response(output)

# 1. Create task in Todos
task = todos.create_task.execute(CreateTaskInput(title="Write report", ...))

# 2. Schedule task in Budget with cost
scheduled = budget.schedule_task.execute(ScheduleTaskInput(
    task_id=task.id,  # Reference to Todos task
    period=Period.today(PeriodType.DAY),
    cost=TaskCost({deep_work_token_id: 4}),  # 4 hours of deep work
))

# 3. When task completes in Todos, spend budget in Budget
budget.spend_task_budget.execute(SpendTaskBudgetInput(
    scheduled_task_id=scheduled.id,
))

# Register hook handler
hooks.register(BEFORE_COMPLETE_TASK, budget_enforcement_handler)

def budget_enforcement_handler(task_id: UUID) -> HookResult:
    scheduled = budget.scheduled_task_repo.get_by_task_id(task_id)
    if not scheduled:
        return HookResult.allow()  # Unscheduled task, allow

    # Check if spending would exceed budget
    result = budget.check_budget.execute(
        scheduled.owner_id,
        scheduled.period,
        scheduled.cost,
    )
    if result.exceeds_hard_limit:
        return HookResult.veto("Insufficient budget for this task")

    return HookResult.allow()

# Token type events
TokenTypeCreated, TokenTypeUpdated, TokenTypeDeleted

# Budget plan events
BudgetPlanCreated, BudgetPlanUpdated, BudgetGranted

# Transaction events
@dataclass(frozen=True)
class TransactionRecorded:
    transaction_id: UUID
    owner_id: UUID
    period: Period
    transaction_type: TransactionType
    amount: int

# Scheduling events
TaskScheduled, TaskUnscheduled, TaskRescheduled, BudgetSpent, TaskSplit

# Debt events
@dataclass(frozen=True)
class BudgetBorrowed:
    debt_id: UUID
    owner_id: UUID
    token_type_id: UUID
    amount: int
    source_period: Period

DebtRepaid, DebtForgiven

# Period events
BudgetRolledOver, BudgetAutoBalanced

from psp.runtime.module import Module

class BudgetModule(Module):
    name = "budget"
    version = "0.1.0"
    status = ModuleStatus.ALPHA

    @property
    def capabilities(self) -> list[str]:
        return ["commands", "queries", "events"]

    def router(self, ctx: RuntimeContext) -> APIRouter:
        return create_router()

    def describe(self) -> ComponentSpec:
        """Return catalogue metadata."""
        return ComponentSpec(
            name=self.name,
            ports=self._get_ports(),
            schemas=self._get_schemas(),
            events=self._get_events(),
            stories=self._get_stories(),
            examples=self._get_examples(),
        )