import sqlite3 from datetime import datetime, timezone import pytest from src import main from src.today_store import TodayPromotionConflict, TodayStore, WeekPlanConflict def sample_days(): return [ { "plan_date": "2026-08-21", "ids": ["issue:secret/repo:3:", "issue:secret/repo:2:"], "capacity_minutes": 150, "estimates": {"issue:secret/repo:3:": 60, "issue:secret/repo:2:": 45}, }, { "plan_date": "2026-08-23", "ids": ["issue:secret/repo:5:"], "capacity_minutes": 90, "estimates": {"issue:secret/repo:5:": 30}, }, ] def test_week_plan_is_encrypted_account_scoped_revisioned_and_preserves_other_dates(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, encryption_key=b"w" * 32) saved = store.replace_week( "Timmy", base_revision=0, days=sample_days(), timezone="America/Los_Angeles" ) assert saved == { "revision": 1, "timezone": "America/Los_Angeles", "days": sample_days(), } assert store.get_week("timmy") == saved assert store.get_week("alexander") == {"revision": 0, "timezone": None, "days": []} retained = path.read_bytes() assert b"issue:secret/repo" not in retained assert b"America/Los_Angeles" not in retained changed = [dict(sample_days()[0], capacity_minutes=180), sample_days()[1]] updated = store.replace_week("timmy", base_revision=1, days=changed, timezone="America/Los_Angeles") assert updated["days"][1] == saved["days"][1] with pytest.raises(WeekPlanConflict) as conflict: store.replace_week("timmy", base_revision=1, days=[], timezone="UTC") assert conflict.value.snapshot == updated def test_week_plan_persists_bounded_weekly_availability_with_the_revisioned_plan(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, encryption_key=b"w" * 32) defaults = [0, 480, 420, 360, 300, 240, 0] saved = store.replace_week( "timmy", base_revision=0, days=sample_days(), timezone="UTC", availability_defaults=defaults, ) assert saved["availability_defaults"] == defaults assert store.get_week("timmy")["availability_defaults"] == defaults retained = path.read_bytes() assert b'"availability_defaults"' not in retained with pytest.raises(ValueError, match="seven weekday capacities"): store.replace_week( "timmy", base_revision=1, days=sample_days(), timezone="UTC", availability_defaults=[60] * 6, ) def test_week_update_model_accepts_zero_capacity_weekday_defaults(): payload = main.WeekPlanUpdate.model_validate({ "base_revision": 2, "timezone": "UTC", "days": [], "availability_defaults": [0, 480, 480, 480, 480, 300, 0], }) assert payload.model_dump()["availability_defaults"] == [0, 480, 480, 480, 480, 300, 0] with pytest.raises(ValueError): main.WeekPlanUpdate.model_validate({ "base_revision": 2, "timezone": "UTC", "days": [], "availability_defaults": [60] * 8, }) def test_start_day_plan_selects_earliest_nonempty_week_date_without_mutation(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"r" * 32, clock=lambda: datetime(2026, 8, 21, 13, tzinfo=timezone.utc).timestamp(), ) days = [ { "plan_date": "2026-08-20", "ids": [], "capacity_minutes": 60, "estimates": {}, }, { "plan_date": "2026-08-21", "ids": ["issue:private/repo:7:"], "capacity_minutes": 90, "estimates": {"issue:private/repo:7:": 45}, }, { "plan_date": "2026-08-22", "ids": ["issue:private/repo:8:"], "capacity_minutes": 120, "estimates": {"issue:private/repo:8:": 30}, }, ] week = store.replace_week( "timmy", base_revision=0, days=days, timezone="America/New_York" ) assert store.get_start_day_plan("Timmy") == { **days[1], "timezone": "America/New_York", } assert store.get_week("timmy") == week def test_start_day_plan_falls_back_to_tomorrow_without_migrating_it(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"r" * 32) tomorrow = store.replace_tomorrow( "timmy", base_revision=0, ids=["issue:private/repo:9:"], capacity_minutes=75, estimates={"issue:private/repo:9:": 45}, plan_date="2026-08-21", timezone="UTC", ) assert store.get_start_day_plan("timmy") == tomorrow assert store.get_tomorrow("timmy") == tomorrow with store._connect() as connection: assert connection.execute( "SELECT COUNT(*) FROM week_plans WHERE login = ?", ("timmy",) ).fetchone()[0] == 0 def test_start_day_plan_uses_due_tomorrow_fallback_before_future_week_day(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"r" * 32, clock=lambda: datetime(2026, 8, 20, 13, tzinfo=timezone.utc).timestamp(), ) future = [{ "plan_date": "2026-08-21", "ids": ["issue:private/repo:10:"], "capacity_minutes": 90, "estimates": {"issue:private/repo:10:": 45}, }] store.replace_week("timmy", base_revision=0, days=future, timezone="UTC") tomorrow = store.replace_tomorrow( "timmy", base_revision=0, ids=["issue:private/repo:9:"], capacity_minutes=75, estimates={"issue:private/repo:9:": 45}, plan_date="2026-08-20", timezone="UTC", ) assert store.get_start_day_plan("timmy") == tomorrow def test_week_plan_rejects_the_same_work_on_multiple_dates_without_advancing_revision(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"w" * 32) original = store.replace_week( "timmy", base_revision=0, days=sample_days(), timezone="UTC" ) duplicated = [ sample_days()[0], { "plan_date": "2026-08-22", "ids": ["issue:secret/repo:3:"], "capacity_minutes": 60, "estimates": {"issue:secret/repo:3:": 30}, }, ] with pytest.raises(ValueError, match="work must be assigned to only one Week Ahead date"): store.replace_week( "timmy", base_revision=original["revision"], days=duplicated, timezone="UTC" ) assert store.get_week("timmy") == original def test_week_promotion_moves_only_due_date_to_today_exactly_once(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"p" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.get("timmy") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") promoted = store.promote_week( "timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ) replay = store.promote_week( "timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ) assert replay == promoted assert promoted["ids"] == sample_days()[0]["ids"] remaining = store.get_week("timmy") assert remaining["revision"] == 2 assert remaining["days"] == [sample_days()[1]] def test_week_early_start_moves_a_future_day_to_empty_today_exactly_once(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"e" * 32, clock=lambda: datetime(2026, 8, 20, 8, tzinfo=timezone.utc).timestamp(), ) today = store.get("timmy") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") promoted = store.promote_week( "timmy", promotion_id="early-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], allow_future=True, ) replay = store.promote_week( "timmy", promotion_id="early-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], allow_future=True, ) assert replay == promoted assert promoted["ids"] == sample_days()[0]["ids"] assert promoted["plan_date"] == "2026-08-21" assert store.get_week("timmy")["days"] == [sample_days()[1]] def test_pull_week_item_atomically_appends_to_active_today_and_replays_exactly_once(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"i" * 32) active_id = "issue:stackchain/dashboard:1:" pulled_id = "issue:secret/repo:3:" today = store.apply("timmy", "seed-active", "add", active_id) week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") result = store.pull_week_item( "timmy", operation_id="pull-secret-3", identity=pulled_id, today_revision=today["revision"], week_revision=week["revision"], ) replay = store.pull_week_item( "timmy", operation_id="pull-secret-3", identity=pulled_id, today_revision=today["revision"], week_revision=week["revision"], ) assert replay == result assert result["today"]["ids"] == [active_id, pulled_id] assert result["today"]["estimates"] == {pulled_id: 60} assert result["week"]["days"][0]["ids"] == ["issue:secret/repo:2:"] assert result["week"]["days"][0]["estimates"] == {"issue:secret/repo:2:": 45} assert result["week"]["days"][1] == sample_days()[1] assert store.get("timmy") == result["today"] assert store.get_week("timmy") == result["week"] def test_pull_week_item_rejects_full_or_changed_today_without_partial_write(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"f" * 32) today = store.get("timmy") for index in range(5): today = store.apply("timmy", f"seed-{index}", "add", f"issue:r:{index + 1}:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") with pytest.raises(main.TodayPlanFull): store.pull_week_item( "timmy", operation_id="full", identity="issue:secret/repo:3:", today_revision=today["revision"], week_revision=week["revision"], ) with pytest.raises(TodayPromotionConflict): store.pull_week_item( "timmy", operation_id="stale", identity="issue:secret/repo:3:", today_revision=today["revision"] - 1, week_revision=week["revision"], ) assert store.get("timmy") == today assert store.get_week("timmy") == week def test_pull_week_item_requires_explicit_today_capacity_overload_confirmation(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"c" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) active_id = "issue:stackchain/dashboard:1:" pulled_id = "issue:stackchain/dashboard:2:" seed = store.replace_week( "timmy", base_revision=0, timezone="UTC", days=[{ "plan_date": "2026-08-22", "ids": [active_id], "capacity_minutes": 60, "estimates": {active_id: 45}, }], ) today = store.promote_week( "timmy", promotion_id="seed-today", week_revision=seed["revision"], plan_date="2026-08-22", today_revision=0, ) week = store.replace_week( "timmy", base_revision=2, timezone="UTC", days=[{ "plan_date": "2026-08-23", "ids": [pulled_id], "capacity_minutes": 60, "estimates": {pulled_id: 30}, }], ) with pytest.raises(ValueError, match="explicit overload confirmation"): store.pull_week_item( "timmy", operation_id="unconfirmed", identity=pulled_id, today_revision=today["revision"], week_revision=week["revision"], ) assert store.get("timmy") == today assert store.get_week("timmy") == week moved = store.pull_week_item( "timmy", operation_id="confirmed", identity=pulled_id, today_revision=today["revision"], week_revision=week["revision"], allow_over_capacity=True, ) assert moved["today"]["capacity_minutes"] == 60 assert moved["today"]["estimates"] == {active_id: 45, pulled_id: 30} def test_normal_week_promotion_still_rejects_a_future_day_without_writes(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"e" * 32, clock=lambda: datetime(2026, 8, 20, 8, tzinfo=timezone.utc).timestamp(), ) today = store.get("timmy") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") with pytest.raises(main.TomorrowPlanNotDue): store.promote_week( "timmy", promotion_id="normal-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ) assert store.get("timmy") == today assert store.get_week("timmy") == week def test_week_promotion_preserves_nonempty_today_and_the_due_week(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"p" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "active", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") with pytest.raises(TodayPromotionConflict) as blocked: store.promote_week( "timmy", promotion_id="week-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ) assert blocked.value.today == today assert blocked.value.week == week assert store.get("timmy") == today assert store.get_week("timmy") == week def test_week_reconciliation_atomically_combines_today_and_consumes_only_due_day(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"c" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "active", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") result = store.reconcile_week( "timmy", promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120, estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60}, ) replay = store.reconcile_week( "timmy", promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120, estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60}, ) assert replay == result assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"] assert result["week"]["days"] == [sample_days()[1]] assert store.get("timmy") == result["today"] assert store.get_week("timmy") == result["week"] def test_week_reconciliation_rejects_work_outside_preserved_plans_without_partial_write(tmp_path): store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"c" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "active", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") with pytest.raises(ValueError, match="selected work must come from Today or the due Week Ahead day"): store.reconcile_week( "timmy", promotion_id="bad", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ids=["issue:other/repo:99:"], capacity_minutes=60, estimates={"issue:other/repo:99:": 30}, ) assert store.get("timmy") == today assert store.get_week("timmy") == week @pytest.mark.anyio async def test_week_promotion_api_returns_both_preserved_plans_for_review(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"p" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "active", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) with pytest.raises(main.HTTPException) as blocked: await main.promote_week_plan(main.WeekPromotion( promotion_id="week-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], )) assert blocked.value.status_code == 409 assert blocked.value.detail == { "code": "week_today_in_progress", "today": today, "week": week, } @pytest.mark.anyio async def test_week_early_start_api_promotes_a_future_day(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"a" * 32, clock=lambda: datetime(2026, 8, 20, 8, tzinfo=timezone.utc).timestamp(), ) today = store.get("timmy") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) result = await main.start_week_day_early(main.WeekPromotion( promotion_id="early-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], )) assert result["ids"] == sample_days()[0]["ids"] assert store.get_week("timmy")["days"] == [sample_days()[1]] @pytest.mark.anyio async def test_pull_week_item_api_returns_both_atomic_plans(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"u" * 32) today = store.apply("timmy", "seed", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) result = await main.pull_week_item_into_today(main.WeekItemPull( operation_id="pull-api", identity="issue:secret/repo:3:", week_revision=week["revision"], today_revision=today["revision"], )) assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"] assert result["week"]["days"][0]["ids"] == ["issue:secret/repo:2:"] @pytest.mark.anyio async def test_week_reconciliation_api_returns_atomic_today_and_remaining_week(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore( tmp_path / "today.sqlite3", encryption_key=b"a" * 32, clock=lambda: datetime(2026, 8, 22, 8, tzinfo=timezone.utc).timestamp(), ) today = store.apply("timmy", "active", "add", "issue:r:active:") week = store.replace_week("timmy", base_revision=0, days=sample_days(), timezone="UTC") monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) result = await main.reconcile_week_plan(main.WeekReconciliation( promotion_id="reconcile-2026-08-21-r1", week_revision=week["revision"], plan_date="2026-08-21", today_revision=today["revision"], ids=["issue:r:active:", "issue:secret/repo:3:"], capacity_minutes=120, estimates={"issue:r:active:": 30, "issue:secret/repo:3:": 60}, )) assert result["today"]["ids"] == ["issue:r:active:", "issue:secret/repo:3:"] assert result["week"]["days"] == [sample_days()[1]] def test_existing_tomorrow_plan_migrates_into_week_without_losing_planning_data(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"m" * 32) tomorrow = store.replace_tomorrow( "timmy", base_revision=0, ids=["issue:r:9:"], capacity_minutes=75, estimates={"issue:r:9:": 45}, plan_date="2026-08-21", timezone="UTC", ) week = store.get_week("timmy") assert week == { "revision": 1, "timezone": "UTC", "days": [{key: tomorrow[key] for key in ("plan_date", "ids", "capacity_minutes", "estimates")}], } assert store.get_tomorrow("timmy") == { "revision": 2, "ids": [], "capacity_minutes": None, "estimates": {} } @pytest.mark.anyio async def test_week_api_round_trip_and_conflict_contract(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"a" * 32) monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) payload = main.WeekPlanUpdate(base_revision=0, days=sample_days(), timezone="UTC") saved = await main.replace_week_plan(payload) assert await main.get_week_plan() == saved with pytest.raises(main.HTTPException) as raised: await main.replace_week_plan(payload) assert raised.value.status_code == 409 assert raised.value.detail == {"code": "week_changed", "snapshot": saved} def test_reschedule_today_item_atomically_moves_it_into_selected_week_day(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"s" * 32) active_id = "issue:stackchain/dashboard:1228:" other_id = "issue:stackchain/dashboard:1229:" today = store.apply("timmy", "add-active", "add", active_id) today = store.apply("timmy", "add-other", "add", other_id) week = store.replace_week( "timmy", base_revision=0, timezone="UTC", availability_defaults=[0, 480, 480, 480, 480, 0, 0], days=[{ "plan_date": "2026-08-24", "ids": [active_id], "capacity_minutes": 120, "estimates": {active_id: 30}, }, { "plan_date": "2026-08-25", "ids": [], "capacity_minutes": 90, "estimates": {}, }], ) result = store.reschedule_today_to_week( "timmy", operation_id="reschedule-active-2026-08-25", identity=active_id, estimate_minutes=45, plan_date="2026-08-25", today_revision=today["revision"], week_revision=week["revision"], ) assert result["today"]["ids"] == [other_id] assert result["week"]["availability_defaults"] == [0, 480, 480, 480, 480, 0, 0] assert result["week"]["days"] == [{ "plan_date": "2026-08-24", "ids": [], "capacity_minutes": 120, "estimates": {}, }, { "plan_date": "2026-08-25", "ids": [active_id], "capacity_minutes": 90, "estimates": {active_id: 45}, }] assert store.get("timmy") == result["today"] assert store.get_week("timmy") == result["week"] def test_reschedule_today_item_is_idempotent_and_rejects_stale_plans_without_partial_write(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"i" * 32) identity = "issue:stackchain/dashboard:1228:" today = store.apply("timmy", "add-active", "add", identity) week = store.replace_week( "timmy", base_revision=0, timezone="UTC", days=[{ "plan_date": "2026-08-25", "ids": [], "capacity_minutes": 60, "estimates": {}, }], ) arguments = { "operation_id": "same-operation", "identity": identity, "estimate_minutes": 30, "plan_date": "2026-08-25", "today_revision": today["revision"], "week_revision": week["revision"], } moved = store.reschedule_today_to_week("timmy", **arguments) assert store.reschedule_today_to_week("timmy", **arguments) == moved another = store.apply("timmy", "add-another", "add", "issue:stackchain/dashboard:1230:") with pytest.raises(TodayPromotionConflict) as conflict: store.reschedule_today_to_week( "timmy", **{**arguments, "operation_id": "stale-operation", "identity": another["ids"][-1]} ) assert conflict.value.today == another assert store.get_week("timmy") == moved["week"] def test_reschedule_today_item_requires_explicit_capacity_overload_confirmation(tmp_path): store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"o" * 32) identity = "issue:stackchain/dashboard:1228:" today = store.apply("timmy", "add-active", "add", identity) week = store.replace_week( "timmy", base_revision=0, timezone="UTC", days=[{ "plan_date": "2026-08-25", "ids": ["issue:stackchain/dashboard:1:"], "capacity_minutes": 60, "estimates": {"issue:stackchain/dashboard:1:": 45}, }], ) arguments = { "identity": identity, "estimate_minutes": 30, "plan_date": "2026-08-25", "today_revision": today["revision"], "week_revision": week["revision"], } with pytest.raises(ValueError, match="explicit overload confirmation"): store.reschedule_today_to_week( "timmy", operation_id="unconfirmed", allow_over_capacity=False, **arguments ) assert store.get("timmy") == today assert store.get_week("timmy") == week moved = store.reschedule_today_to_week( "timmy", operation_id="confirmed", allow_over_capacity=True, **arguments ) assert moved["week"]["days"][0]["ids"][-1] == identity @pytest.mark.anyio async def test_week_reschedule_api_returns_atomic_snapshots_and_conflict_recovery(monkeypatch, tmp_path): async def user(): return {"login": "timmy"} store = TodayStore(tmp_path / "today.sqlite3", encryption_key=b"a" * 32) identity = "issue:stackchain/dashboard:1228:" today = store.apply("timmy", "add-active", "add", identity) week = store.replace_week( "timmy", base_revision=0, timezone="UTC", days=[{ "plan_date": "2026-08-25", "ids": [], "capacity_minutes": 60, "estimates": {}, }], ) monkeypatch.setattr(main, "current_user", user) monkeypatch.setattr(main, "_today_store", lambda: store) payload = main.WeekReschedule( operation_id="move-active", identity=identity, estimate_minutes=30, plan_date="2026-08-25", today_revision=today["revision"], week_revision=week["revision"], ) moved = await main.reschedule_today_to_week(payload) assert moved["today"]["ids"] == [] assert moved["week"]["days"][0]["ids"] == [identity] stale = main.WeekReschedule( operation_id="stale", identity="issue:stackchain/dashboard:1229:", estimate_minutes=30, plan_date="2026-08-25", today_revision=today["revision"], week_revision=week["revision"], ) with pytest.raises(main.HTTPException) as raised: await main.reschedule_today_to_week(stale) assert raised.value.status_code == 409 assert raised.value.detail["code"] == "today_changed" assert raised.value.detail["today"] == moved["today"] assert raised.value.detail["week"] == moved["week"]