Bug report (python repo "ledger"):
Catching a projection up more than once crashes.

Reproduction: build an EventStore, receive some stock, call StockProjection.catch_up(store), append another event, then call catch_up(store) again. The second call raises SequenceError complaining that an event has already been applied.

A projection is supposed to be resumable -- repeatedly catching up must be safe, and calling it when nothing new has been appended must be a no-op that applies zero events. Find the underlying cause and fix it. Do not weaken the ledger's ordering guarantees to make the symptom disappear.

Repository files (containing the defects):
README.md: # ledger

A small event-sourced inventory ledger.

Everything that happens to stock is recorded as an immutable event in the
`EventStore`. Current state is never stored directly; it is derived by folding
the event log through a *projection*.

    from ledger import EventStore, StockProjection

    store = EventStore()
    store.receive("widget", qty=100, unit_cost=2.50)
    store.ship("widget", qty=30)

    stock = StockProjection()
    stock.catch_up(store)
    stock.on_hand("widget")   # 70

## Layout

| module | role |
|---|---|
| `events.py` | immutable event types |
| `store.py` | append-only log, owns sequence numbering |
| `projections.py` | `StockProjection` (units) and `Valuation` (FIFO cost) |
| `policies.py` | reorder suggestions and multi-warehouse allocation |
| `errors.py` | exception hierarchy |

## Resuming a projection

Projections are incremental. Each remembers the last sequence number it
applied and asks the store only for events after it, so catching up a long
ledger does not mean replaying it from the beginning.

ledger/__init__.py: """A small event-sourced inventory ledger."""
from .errors import LedgerError, UnknownSku, InsufficientStock
from .events import Event, Received, Shipped, Adjusted
from .store import EventStore
from .projections import StockProjection, Valuation
from .policies import ReorderPolicy, allocate

__all__ = [
    "LedgerError", "UnknownSku", "InsufficientStock",
    "Event", "Received", "Shipped", "Adjusted",
    "EventStore", "StockProjection", "Valuation",
    "ReorderPolicy", "allocate",
]

ledger/errors.py: """Exception types raised across the ledger."""


class LedgerError(Exception):
    """Base class for every error this package raises."""


class UnknownSku(LedgerError):
    """A SKU was referenced that the ledger has never seen."""

    def __init__(self, sku: str):
        super().__init__(f"unknown sku: {sku}")
        self.sku = sku


class InsufficientStock(LedgerError):
    """A shipment was requested for more units than are on hand."""

    def __init__(self, sku: str, wanted: int, available: int):
        super().__init__(
            f"cannot ship {wanted} of {sku}: only {available} on hand")
        self.sku = sku
        self.wanted = wanted
        self.available = available


class SequenceError(LedgerError):
    """Events were presented to a projection out of order."""

ledger/events.py: """Event types.

Every event carries a monotonically increasing sequence number assigned by the
store, plus the SKU it concerns. Events are immutable once created: a ledger
that can be rewritten after the fact is not a ledger.
"""
from dataclasses import dataclass


@dataclass(frozen=True)
class Event:
    """Base event. `seq` is assigned by the store, never by the caller."""
    seq: int
    ts: float
    sku: str

    @property
    def kind(self) -> str:
        return type(self).__name__.lower()

    def describe(self) -> str:
        return f"[{self.seq}] {self.kind} {self.sku}"


@dataclass(frozen=True)
class Received(Event):
    """Stock arrived into the warehouse at a known unit cost."""
    qty: int
    unit_cost: float

    def describe(self) -> str:
        return f"{super().describe()} +{self.qty} @ {self.unit_cost:.2f}"


@dataclass(frozen=True)
class Shipped(Event):
    """Stock left the warehouse."""
    qty: int

    def describe(self) -> str:
        return f"{super().describe()} -{self.qty}"


@dataclass(frozen=True)
class Adjusted(Event):
    """A stocktake correction. `delta` may be negative."""
    delta: int
    reason: str

    def describe(self) -> str:
        return f"{super().describe()} {self.delta:+d} ({self.reason})"

ledger/policies.py: """Decisions taken on top of the read models.

Nothing here touches the event log directly. Policies read a projection and
return a plan; committing that plan is the caller's business.
"""
from dataclasses import dataclass
from typing import Dict, List, Optional, Sequence, Tuple

from .errors import InsufficientStock, UnknownSku
from .projections import StockProjection


@dataclass(frozen=True)
class Warehouse:
    """A stocking location, with a priority used to break ties."""
    name: str
    priority: int
    on_hand: int


@dataclass(frozen=True)
class ReorderSuggestion:
    sku: str
    order_qty: int
    reason: str


class ReorderPolicy:
    """Suggests replenishment when a SKU falls to or below its threshold.

    The comparison is deliberately inclusive: a SKU sitting exactly on its
    reorder point has reached it, and waiting for it to drop further is how a
    warehouse runs out during the lead time.
    """

    def __init__(self, thresholds: Dict[str, int], target: Dict[str, int]):
        self.thresholds = dict(thresholds)
        self.target = dict(target)

    def threshold_for(self, sku: str) -> int:
        if sku not in self.thresholds:
            raise UnknownSku(sku)
        return self.thresholds[sku]

    def suggest(self, stock: StockProjection) -> List[ReorderSuggestion]:
        """One suggestion per SKU at or below its reorder point."""
        out: List[ReorderSuggestion] = []
        for sku in sorted(self.thresholds):
            level = stock.levels.get(sku, 0)
            threshold = self.thresholds[sku]
            if level <= threshold:
                target = self.target.get(sku, threshold * 2)
                qty = target - level
                if qty > 0:
                    out.append(ReorderSuggestion(
                        sku=sku, order_qty=qty,
                        reason=f"level {level} at or below threshold {threshold}"))
        return out


def allocate(demand: int, warehouses: Sequence[Warehouse]) -> List[Tuple[str, int]]:
    """Split `demand` across warehouses, highest priority first.

    Ties on priority are broken by name so the result is deterministic. The
    input sequence is never mutated -- callers reuse their warehouse list
    across several allocations and would otherwise see it quietly emptied.
    """
    if demand <= 0:
        raise ValueError("demand must be positive")

    ordered = sorted(warehouses, key=lambda w: (-w.priority, w.name))
    available = sum(w.on_hand for w in ordered)
    if available < demand:
        raise InsufficientStock("<multiple>", demand, available)

    plan: List[Tuple[str, int]] = []
    remaining = demand
    for warehouse in ordered:
        if remaining <= 0:
            break
        take = min(warehouse.on_hand, remaining)
        if take > 0:
            plan.append((warehouse.name, take))
            remaining -= take
    return plan

ledger/projections.py: """Read models built by folding the event log.

A projection is resumable: it remembers the last sequence number it applied
and asks the store only for what came after. That is what keeps a long ledger
cheap to query, and it is why the store's `since` boundary matters so much.
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

from .errors import SequenceError, UnknownSku
from .events import Adjusted, Event, Received, Shipped
from .store import EventStore


class StockProjection:
    """Units on hand per SKU."""

    def __init__(self) -> None:
        self.levels: Dict[str, int] = {}
        self.last_seq: int = 0

    def apply(self, event: Event) -> None:
        """Fold a single event into the running totals."""
        if event.seq <= self.last_seq:
            raise SequenceError(
                f"event {event.seq} already applied (at {self.last_seq})")
        current = self.levels.get(event.sku, 0)
        if isinstance(event, Received):
            current += event.qty
        elif isinstance(event, Shipped):
            current -= event.qty
        elif isinstance(event, Adjusted):
            current += event.delta
        self.levels[event.sku] = current
        self.last_seq = event.seq

    def catch_up(self, store: EventStore) -> int:
        """Apply everything appended since the last call. Returns the count."""
        pending = store.since(self.last_seq)
        for event in pending:
            self.apply(event)
        return len(pending)

    def on_hand(self, sku: str) -> int:
        if sku not in self.levels:
            raise UnknownSku(sku)
        return self.levels[sku]

    def negative_skus(self) -> List[str]:
        """SKUs the log says we hold a negative quantity of."""
        return sorted(s for s, n in self.levels.items() if n < 0)


@dataclass
class Lot:
    """One receipt still carrying unconsumed units."""
    seq: int
    qty: int
    unit_cost: float


class Valuation:
    """Inventory value under first-in, first-out cost basis."""

    def __init__(self) -> None:
        self.lots: Dict[str, List[Lot]] = {}
        self.last_seq: int = 0
        self.cost_of_goods_sold: float = 0.0

    def apply(self, event: Event) -> None:
        lots = self.lots.setdefault(event.sku, [])
        if isinstance(event, Received):
            lots.append(Lot(seq=event.seq, qty=event.qty,
                            unit_cost=event.unit_cost))
        elif isinstance(event, Shipped):
            self._consume(lots, event.qty)
        elif isinstance(event, Adjusted):
            if event.delta > 0:
                unit_cost = lots[-1].unit_cost if lots else 0.0
                lots.append(Lot(seq=event.seq, qty=event.delta,
                                unit_cost=unit_cost))
            else:
                self._consume(lots, -event.delta)
        self.last_seq = event.seq

    def _consume(self, lots: List[Lot], qty: int) -> None:
        """Draw `qty` units from the oldest lots first."""
        remaining = qty
        while remaining > 0 and lots:
            oldest = lots[0]
            take = min(oldest.qty, remaining)
            self.cost_of_goods_sold += take * oldest.unit_cost
            oldest.qty -= take
            remaining -= take
            if oldest.qty == 0:
                lots.pop(0)

    def catch_up(self, store: EventStore) -> int:
        pending = store.since(self.last_seq)
        for event in pending:
            self.apply(event)
        return len(pending)

    def value_of(self, sku: str) -> float:
        """Total carrying value of the units still on hand for one SKU."""
        if sku not in self.lots:
            raise UnknownSku(sku)
        return sum(lot.qty * lot.unit_cost for lot in self.lots[sku])

    def average_cost(self, sku: str) -> float:
        """Weighted average cost of the units still on hand.

        Zero when nothing is on hand, rather than a division error -- callers
        display this per SKU and a depleted line is normal.
        """
        if sku not in self.lots:
            raise UnknownSku(sku)
        units = sum(lot.qty for lot in self.lots[sku])
        if units == 0:
            return 0.0
        return self.value_of(sku) / units

    def total_value(self) -> float:
        return sum(lot.qty * lot.unit_cost
                   for lots in self.lots.values() for lot in lots)

ledger/store.py: """Append-only event store with snapshot support.

The store owns sequence numbering. Nothing else in the package is allowed to
assign a sequence number, because the ordering guarantee that projections rely
on is only as good as the single place that hands the numbers out.
"""
from typing import Dict, Iterable, Iterator, List, Optional

from .errors import SequenceError
from .events import Adjusted, Event, Received, Shipped


class EventStore:
    """Holds every event ever appended, in order."""

    def __init__(self) -> None:
        self._events: List[Event] = []
        self._next_seq: int = 1
        self._clock: float = 0.0

    # -- writing ---------------------------------------------------------

    def _tick(self) -> float:
        self._clock += 1.0
        return self._clock

    def _append(self, event: Event) -> Event:
        if event.seq != self._next_seq:
            raise SequenceError(
                f"expected seq {self._next_seq}, got {event.seq}")
        self._events.append(event)
        self._next_seq += 1
        return event

    def receive(self, sku: str, qty: int, unit_cost: float) -> Received:
        """Record stock arriving."""
        if qty <= 0:
            raise ValueError("received quantity must be positive")
        return self._append(
            Received(seq=self._next_seq, ts=self._tick(), sku=sku,
                     qty=qty, unit_cost=unit_cost))

    def ship(self, sku: str, qty: int) -> Shipped:
        """Record stock leaving. Does not itself check availability."""
        if qty <= 0:
            raise ValueError("shipped quantity must be positive")
        return self._append(
            Shipped(seq=self._next_seq, ts=self._tick(), sku=sku, qty=qty))

    def adjust(self, sku: str, delta: int, reason: str) -> Adjusted:
        """Record a stocktake correction."""
        if delta == 0:
            raise ValueError("adjustment delta must not be zero")
        return self._append(
            Adjusted(seq=self._next_seq, ts=self._tick(), sku=sku,
                     delta=delta, reason=reason))

    # -- reading ---------------------------------------------------------

    def __len__(self) -> int:
        return len(self._events)

    def __iter__(self) -> Iterator[Event]:
        return iter(self._events)

    @property
    def head(self) -> int:
        """Sequence number of the most recent event, or 0 when empty."""
        return self._events[-1].seq if self._events else 0

    def since(self, seq: int) -> List[Event]:
        """Every event appended strictly after `seq`.

        A projection resumes by passing the sequence number it last applied, so
        this must be exclusive at the lower bound. Making it inclusive would
        replay one already-applied event on every resume.
        """
        return [e for e in self._events if e.seq >= seq]

    def between(self, low: int, high: int) -> List[Event]:
        """Events with low < seq <= high."""
        return [e for e in self._events if low < e.seq <= high]

    def for_sku(self, sku: str) -> List[Event]:
        return [e for e in self._events if e.sku == sku]

    def skus(self) -> List[str]:
        seen: Dict[str, None] = {}
        for event in self._events:
            seen.setdefault(event.sku, None)
        return list(seen)

    def replay(self, events: Optional[Iterable[Event]] = None) -> List[Event]:
        """Materialise an event iterable, defaulting to the whole log."""
        return list(self._events if events is None else events)


Hidden tests:
tests/test_catch_up.py: import unittest

from ledger import EventStore, StockProjection
from ledger.errors import SequenceError


class TestIncrementalCatchUp(unittest.TestCase):
    def test_catch_up_twice_across_new_events(self):
        store = EventStore()
        store.receive("widget", 100, 2.50)
        stock = StockProjection()
        stock.catch_up(store)
        store.ship("widget", 30)
        stock.catch_up(store)
        self.assertEqual(stock.on_hand("widget"), 70)

    def test_catch_up_with_nothing_new_is_a_noop(self):
        store = EventStore()
        store.receive("widget", 10, 1.0)
        stock = StockProjection()
        self.assertEqual(stock.catch_up(store), 1)
        self.assertEqual(stock.catch_up(store), 0)
        self.assertEqual(stock.on_hand("widget"), 10)

    def test_many_incremental_catch_ups_match_one_big_one(self):
        incremental = StockProjection()
        store = EventStore()
        for i in range(1, 6):
            store.receive("bolt", i * 10, 1.0)
            incremental.catch_up(store)
        oneshot = StockProjection()
        oneshot.catch_up(store)
        self.assertEqual(incremental.levels, oneshot.levels)
        self.assertEqual(incremental.on_hand("bolt"), 150)


class TestOrderingGuaranteesPreserved(unittest.TestCase):
    """Deleting the ordering check would hide the symptom but corrupt totals."""

    def test_replaying_an_applied_event_is_still_rejected(self):
        store = EventStore()
        store.receive("widget", 5, 1.0)
        stock = StockProjection()
        stock.catch_up(store)
        already_applied = list(store)[0]
        with self.assertRaises(SequenceError):
            stock.apply(already_applied)

    def test_store_since_is_exclusive_at_the_lower_bound(self):
        store = EventStore()
        store.receive("a", 1, 1.0)
        store.receive("b", 1, 1.0)
        store.receive("c", 1, 1.0)
        self.assertEqual([e.seq for e in store.since(0)], [1, 2, 3])
        self.assertEqual([e.seq for e in store.since(1)], [2, 3])
        self.assertEqual([e.seq for e in store.since(3)], [])


Reference fix:
README.md: # ledger

A small event-sourced inventory ledger.

Everything that happens to stock is recorded as an immutable event in the
`EventStore`. Current state is never stored directly; it is derived by folding
the event log through a *projection*.

    from ledger import EventStore, StockProjection

    store = EventStore()
    store.receive("widget", qty=100, unit_cost=2.50)
    store.ship("widget", qty=30)

    stock = StockProjection()
    stock.catch_up(store)
    stock.on_hand("widget")   # 70

## Layout

| module | role |
|---|---|
| `events.py` | immutable event types |
| `store.py` | append-only log, owns sequence numbering |
| `projections.py` | `StockProjection` (units) and `Valuation` (FIFO cost) |
| `policies.py` | reorder suggestions and multi-warehouse allocation |
| `errors.py` | exception hierarchy |

## Resuming a projection

Projections are incremental. Each remembers the last sequence number it
applied and asks the store only for events after it, so catching up a long
ledger does not mean replaying it from the beginning.

ledger/__init__.py: """A small event-sourced inventory ledger."""
from .errors import LedgerError, UnknownSku, InsufficientStock
from .events import Event, Received, Shipped, Adjusted
from .store import EventStore
from .projections import StockProjection, Valuation
from .policies import ReorderPolicy, allocate

__all__ = [
    "LedgerError", "UnknownSku", "InsufficientStock",
    "Event", "Received", "Shipped", "Adjusted",
    "EventStore", "StockProjection", "Valuation",
    "ReorderPolicy", "allocate",
]

ledger/errors.py: """Exception types raised across the ledger."""


class LedgerError(Exception):
    """Base class for every error this package raises."""


class UnknownSku(LedgerError):
    """A SKU was referenced that the ledger has never seen."""

    def __init__(self, sku: str):
        super().__init__(f"unknown sku: {sku}")
        self.sku = sku


class InsufficientStock(LedgerError):
    """A shipment was requested for more units than are on hand."""

    def __init__(self, sku: str, wanted: int, available: int):
        super().__init__(
            f"cannot ship {wanted} of {sku}: only {available} on hand")
        self.sku = sku
        self.wanted = wanted
        self.available = available


class SequenceError(LedgerError):
    """Events were presented to a projection out of order."""

ledger/events.py: """Event types.

Every event carries a monotonically increasing sequence number assigned by the
store, plus the SKU it concerns. Events are immutable once created: a ledger
that can be rewritten after the fact is not a ledger.
"""
from dataclasses import dataclass


@dataclass(frozen=True)
class Event:
    """Base event. `seq` is assigned by the store, never by the caller."""
    seq: int
    ts: float
    sku: str

    @property
    def kind(self) -> str:
        return type(self).__name__.lower()

    def describe(self) -> str:
        return f"[{self.seq}] {self.kind} {self.sku}"


@dataclass(frozen=True)
class Received(Event):
    """Stock arrived into the warehouse at a known unit cost."""
    qty: int
    unit_cost: float

    def describe(self) -> str:
        return f"{super().describe()} +{self.qty} @ {self.unit_cost:.2f}"


@dataclass(frozen=True)
class Shipped(Event):
    """Stock left the warehouse."""
    qty: int

    def describe(self) -> str:
        return f"{super().describe()} -{self.qty}"


@dataclass(frozen=True)
class Adjusted(Event):
    """A stocktake correction. `delta` may be negative."""
    delta: int
    reason: str

    def describe(self) -> str:
        return f"{super().describe()} {self.delta:+d} ({self.reason})"

ledger/policies.py: """Decisions taken on top of the read models.

Nothing here touches the event log directly. Policies read a projection and
return a plan; committing that plan is the caller's business.
"""
from dataclasses import dataclass
from typing import Dict, List, Optional, Sequence, Tuple

from .errors import InsufficientStock, UnknownSku
from .projections import StockProjection


@dataclass(frozen=True)
class Warehouse:
    """A stocking location, with a priority used to break ties."""
    name: str
    priority: int
    on_hand: int


@dataclass(frozen=True)
class ReorderSuggestion:
    sku: str
    order_qty: int
    reason: str


class ReorderPolicy:
    """Suggests replenishment when a SKU falls to or below its threshold.

    The comparison is deliberately inclusive: a SKU sitting exactly on its
    reorder point has reached it, and waiting for it to drop further is how a
    warehouse runs out during the lead time.
    """

    def __init__(self, thresholds: Dict[str, int], target: Dict[str, int]):
        self.thresholds = dict(thresholds)
        self.target = dict(target)

    def threshold_for(self, sku: str) -> int:
        if sku not in self.thresholds:
            raise UnknownSku(sku)
        return self.thresholds[sku]

    def suggest(self, stock: StockProjection) -> List[ReorderSuggestion]:
        """One suggestion per SKU at or below its reorder point."""
        out: List[ReorderSuggestion] = []
        for sku in sorted(self.thresholds):
            level = stock.levels.get(sku, 0)
            threshold = self.thresholds[sku]
            if level <= threshold:
                target = self.target.get(sku, threshold * 2)
                qty = target - level
                if qty > 0:
                    out.append(ReorderSuggestion(
                        sku=sku, order_qty=qty,
                        reason=f"level {level} at or below threshold {threshold}"))
        return out


def allocate(demand: int, warehouses: Sequence[Warehouse]) -> List[Tuple[str, int]]:
    """Split `demand` across warehouses, highest priority first.

    Ties on priority are broken by name so the result is deterministic. The
    input sequence is never mutated -- callers reuse their warehouse list
    across several allocations and would otherwise see it quietly emptied.
    """
    if demand <= 0:
        raise ValueError("demand must be positive")

    ordered = sorted(warehouses, key=lambda w: (-w.priority, w.name))
    available = sum(w.on_hand for w in ordered)
    if available < demand:
        raise InsufficientStock("<multiple>", demand, available)

    plan: List[Tuple[str, int]] = []
    remaining = demand
    for warehouse in ordered:
        if remaining <= 0:
            break
        take = min(warehouse.on_hand, remaining)
        if take > 0:
            plan.append((warehouse.name, take))
            remaining -= take
    return plan

ledger/projections.py: """Read models built by folding the event log.

A projection is resumable: it remembers the last sequence number it applied
and asks the store only for what came after. That is what keeps a long ledger
cheap to query, and it is why the store's `since` boundary matters so much.
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

from .errors import SequenceError, UnknownSku
from .events import Adjusted, Event, Received, Shipped
from .store import EventStore


class StockProjection:
    """Units on hand per SKU."""

    def __init__(self) -> None:
        self.levels: Dict[str, int] = {}
        self.last_seq: int = 0

    def apply(self, event: Event) -> None:
        """Fold a single event into the running totals."""
        if event.seq <= self.last_seq:
            raise SequenceError(
                f"event {event.seq} already applied (at {self.last_seq})")
        current = self.levels.get(event.sku, 0)
        if isinstance(event, Received):
            current += event.qty
        elif isinstance(event, Shipped):
            current -= event.qty
        elif isinstance(event, Adjusted):
            current += event.delta
        self.levels[event.sku] = current
        self.last_seq = event.seq

    def catch_up(self, store: EventStore) -> int:
        """Apply everything appended since the last call. Returns the count."""
        pending = store.since(self.last_seq)
        for event in pending:
            self.apply(event)
        return len(pending)

    def on_hand(self, sku: str) -> int:
        if sku not in self.levels:
            raise UnknownSku(sku)
        return self.levels[sku]

    def negative_skus(self) -> List[str]:
        """SKUs the log says we hold a negative quantity of."""
        return sorted(s for s, n in self.levels.items() if n < 0)


@dataclass
class Lot:
    """One receipt still carrying unconsumed units."""
    seq: int
    qty: int
    unit_cost: float


class Valuation:
    """Inventory value under first-in, first-out cost basis."""

    def __init__(self) -> None:
        self.lots: Dict[str, List[Lot]] = {}
        self.last_seq: int = 0
        self.cost_of_goods_sold: float = 0.0

    def apply(self, event: Event) -> None:
        lots = self.lots.setdefault(event.sku, [])
        if isinstance(event, Received):
            lots.append(Lot(seq=event.seq, qty=event.qty,
                            unit_cost=event.unit_cost))
        elif isinstance(event, Shipped):
            self._consume(lots, event.qty)
        elif isinstance(event, Adjusted):
            if event.delta > 0:
                unit_cost = lots[-1].unit_cost if lots else 0.0
                lots.append(Lot(seq=event.seq, qty=event.delta,
                                unit_cost=unit_cost))
            else:
                self._consume(lots, -event.delta)
        self.last_seq = event.seq

    def _consume(self, lots: List[Lot], qty: int) -> None:
        """Draw `qty` units from the oldest lots first."""
        remaining = qty
        while remaining > 0 and lots:
            oldest = lots[0]
            take = min(oldest.qty, remaining)
            self.cost_of_goods_sold += take * oldest.unit_cost
            oldest.qty -= take
            remaining -= take
            if oldest.qty == 0:
                lots.pop(0)

    def catch_up(self, store: EventStore) -> int:
        pending = store.since(self.last_seq)
        for event in pending:
            self.apply(event)
        return len(pending)

    def value_of(self, sku: str) -> float:
        """Total carrying value of the units still on hand for one SKU."""
        if sku not in self.lots:
            raise UnknownSku(sku)
        return sum(lot.qty * lot.unit_cost for lot in self.lots[sku])

    def average_cost(self, sku: str) -> float:
        """Weighted average cost of the units still on hand.

        Zero when nothing is on hand, rather than a division error -- callers
        display this per SKU and a depleted line is normal.
        """
        if sku not in self.lots:
            raise UnknownSku(sku)
        units = sum(lot.qty for lot in self.lots[sku])
        if units == 0:
            return 0.0
        return self.value_of(sku) / units

    def total_value(self) -> float:
        return sum(lot.qty * lot.unit_cost
                   for lots in self.lots.values() for lot in lots)

ledger/store.py: """Append-only event store with snapshot support.

The store owns sequence numbering. Nothing else in the package is allowed to
assign a sequence number, because the ordering guarantee that projections rely
on is only as good as the single place that hands the numbers out.
"""
from typing import Dict, Iterable, Iterator, List, Optional

from .errors import SequenceError
from .events import Adjusted, Event, Received, Shipped


class EventStore:
    """Holds every event ever appended, in order."""

    def __init__(self) -> None:
        self._events: List[Event] = []
        self._next_seq: int = 1
        self._clock: float = 0.0

    # -- writing ---------------------------------------------------------

    def _tick(self) -> float:
        self._clock += 1.0
        return self._clock

    def _append(self, event: Event) -> Event:
        if event.seq != self._next_seq:
            raise SequenceError(
                f"expected seq {self._next_seq}, got {event.seq}")
        self._events.append(event)
        self._next_seq += 1
        return event

    def receive(self, sku: str, qty: int, unit_cost: float) -> Received:
        """Record stock arriving."""
        if qty <= 0:
            raise ValueError("received quantity must be positive")
        return self._append(
            Received(seq=self._next_seq, ts=self._tick(), sku=sku,
                     qty=qty, unit_cost=unit_cost))

    def ship(self, sku: str, qty: int) -> Shipped:
        """Record stock leaving. Does not itself check availability."""
        if qty <= 0:
            raise ValueError("shipped quantity must be positive")
        return self._append(
            Shipped(seq=self._next_seq, ts=self._tick(), sku=sku, qty=qty))

    def adjust(self, sku: str, delta: int, reason: str) -> Adjusted:
        """Record a stocktake correction."""
        if delta == 0:
            raise ValueError("adjustment delta must not be zero")
        return self._append(
            Adjusted(seq=self._next_seq, ts=self._tick(), sku=sku,
                     delta=delta, reason=reason))

    # -- reading ---------------------------------------------------------

    def __len__(self) -> int:
        return len(self._events)

    def __iter__(self) -> Iterator[Event]:
        return iter(self._events)

    @property
    def head(self) -> int:
        """Sequence number of the most recent event, or 0 when empty."""
        return self._events[-1].seq if self._events else 0

    def since(self, seq: int) -> List[Event]:
        """Every event appended strictly after `seq`.

        A projection resumes by passing the sequence number it last applied, so
        this must be exclusive at the lower bound. Making it inclusive would
        replay one already-applied event on every resume.
        """
        return [e for e in self._events if e.seq > seq]

    def between(self, low: int, high: int) -> List[Event]:
        """Events with low < seq <= high."""
        return [e for e in self._events if low < e.seq <= high]

    def for_sku(self, sku: str) -> List[Event]:
        return [e for e in self._events if e.sku == sku]

    def skus(self) -> List[str]:
        seen: Dict[str, None] = {}
        for event in self._events:
            seen.setdefault(event.sku, None)
        return list(seen)

    def replay(self, events: Optional[Iterable[Event]] = None) -> List[Event]:
        """Materialise an event iterable, defaulting to the whole log."""
        return list(self._events if events is None else events)