def test_projection_catches_up_within_5_seconds():
    """SLA: read model converges within 5s of write."""
    write_model.execute(ChangePriceCommand(sku="sku1", new=150))

    deadline = time.time() + 5.0
    while time.time() < deadline:
        if read_model.get_price("sku1") == 150:
            return  # converged in time
        time.sleep(0.1)

    pytest.fail("Read model did not converge within 5s")

def test_cart_update_converges_within_2s_across_regions():
    cart_service_us.add_item(user_id="u1", sku="sku1")
    deadline = time.time() + 2.0
    while time.time() < deadline:
        if any(i.sku == "sku1" for i in cart_service_eu.get(user_id="u1").items):
            return
        time.sleep(0.05)
    pytest.fail("Cart did not converge across regions within 2s")

def test_monotonic_reads_per_session():
    session = client.connect(read_preference="monotonic")
    initial = session.get("counter")
    for _ in range(100):
        v = session.get("counter")
        assert v >= initial, f"Read regressed: {initial} -> {v}"

def test_bounded_staleness_under_2_seconds():
    leader.write("counter", time.time())
    time.sleep(2.5)  # exceed the bound
    for replica in replicas:
        staleness = time.time() - float(replica.read("counter"))
        assert staleness <= 2.0, f"Replica {replica} stale by {staleness:.2f}s"

def test_zero_downtime_swap():
    # Stand up new projection in parallel
    new_proj = SearchIndexProjectionV2()
    catchup_from_event_log(new_proj, until=current_position)

    # Verify new matches old at the swap point
    assert new_proj.materialize() == old_proj.materialize()

    # Subscribe new to live event stream
    subscribe(new_proj)
    # Switch reads to new - verify no read returns stale state
    swap_query_target(old_proj, new_proj)

def test_post_command_returns_pending_until_projection_catches_up():
    response = api_client.post("/products", {"name": "Phone"})
    assert response.status == 202  # Accepted

    # Get returns "pending" until projection updates
    get1 = api_client.get(f"/products/{response.body['id']}")
    assert get1.body["status"] == "pending"

    wait_for_projection_to_catch_up(timeout=5)

    get2 = api_client.get(f"/products/{response.body['id']}")
    assert get2.body["status"] == "active"

def test_unrelated_aggregates_replay_independently():
    # Two orders; events interleaved in the log
    log = [
        OrderCreated("o1", "c1"),
        OrderCreated("o2", "c2"),
        ItemAdded("o2", "sku2", 1),
        ItemAdded("o1", "sku1", 2),
        OrderConfirmed("o2"),
        OrderConfirmed("o1"),
    ]

    o1 = OrderAggregate.replay(filter(lambda e: e.order_id == "o1", log))
    o2 = OrderAggregate.replay(filter(lambda e: e.order_id == "o2", log))

    assert o1.line_items == [("sku1", 2)]
    assert o2.line_items == [("sku2", 1)]

def test_snapshot_equivalent_to_full_replay():
    events = [...]  # 1000 events

    full_replay = OrderAggregate.replay(events)
    snapshot = OrderAggregate.snapshot_at(events, version=500)
    after_snapshot = OrderAggregate.from_snapshot(snapshot).apply_from(events[500:])

    assert full_replay == after_snapshot

def test_upcasting_v1_to_v2():
    v1_event = {"type": "ItemAdded", "version": 1, "sku": "sku1", "qty": 2}
    v2_event = upcast_v1_v2(v1_event)

    assert v2_event["version"] == 2
    assert v2_event["unit_price"] is None  # or default per business rule
    assert v2_event["sku"] == "sku1"
    assert v2_event["qty"] == 2

def test_replay_handles_mixed_event_versions():
    mixed = [v1_event, v2_event, v1_event]
    upgraded = [upcast(e) for e in mixed]

    state_via_upcast = OrderAggregate.replay(upgraded)
    assert state_via_upcast.line_items_count == 3

def test_projection_rebuild_idempotent():
    events = load_events()
    projection_a = build_projection(events)
    projection_b = build_projection(events)
    assert projection_a == projection_b

    # And rebuild from cleared state matches incremental update
    projection_c = SearchProjection()
    for evt in events:
        projection_c.apply(evt)
    assert projection_c == projection_a

def test_retroactive_correction_recomputes():
    # Original state: order o1 has 2 items
    events_v1 = [OrderCreated("o1"), ItemAdded("o1", "sku1", 2)]
    state_v1 = OrderAggregate.replay(events_v1)
    assert state_v1.total_items == 2

    # Discover ItemAdded was wrong (qty 5, not 2). Two strategies:
    # (a) Append correction event:
    events_v1.append(ItemQtyCorrected("o1", "sku1", new_qty=5))
    state_corrected = OrderAggregate.replay(events_v1)
    assert state_corrected.total_items == 5

    # (b) Or replace the original event in the log + replay:
    events_v2 = [OrderCreated("o1"), ItemAdded("o1", "sku1", 5)]
    state_replayed = OrderAggregate.replay(events_v2)
    assert state_replayed.total_items == 5

def test_replay_mode_suppresses_external_calls():
    email_gateway = MockEmailGateway()

    handler = OrderConfirmedHandler(email_gateway, mode="replay")
    handler.handle(OrderConfirmed("o1"))

    assert email_gateway.sent_count == 0  # replay mode = no real calls

def test_live_mode_invokes_external_calls():
    email_gateway = MockEmailGateway()

    handler = OrderConfirmedHandler(email_gateway, mode="live")
    handler.handle(OrderConfirmed("o1"))

    assert email_gateway.sent_count == 1

def test_concurrent_append_rejected():
    events = [OrderCreated("o1")]
    store.append("o1", events, expected_version=0)  # OK; new version = 1

    # Two concurrent commands both load version=1 + try to append
    new_events_a = [ItemAdded("o1", "sku1", 1)]
    new_events_b = [ItemAdded("o1", "sku2", 1)]

    store.append("o1", new_events_a, expected_version=1)  # OK; new version = 2

    with pytest.raises(ConcurrencyConflict):
        store.append("o1", new_events_b, expected_version=1)  # already at 2