feat(journal): closeout catch-up on startup when slot already passed
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -111,6 +111,37 @@ def _run_closeout_threadsafe(user_id: int) -> None:
|
|||||||
asyncio.run_coroutine_threadsafe(_do_closeout(user_id), _loop)
|
asyncio.run_coroutine_threadsafe(_do_closeout(user_id), _loop)
|
||||||
|
|
||||||
|
|
||||||
|
async def _closeout_catchup(user_id: int) -> None:
|
||||||
|
"""On startup, run yesterday's closeout once if the slot already passed
|
||||||
|
and no entry for yesterday exists in observations_raw.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
tz_str = await get_user_timezone(user_id)
|
||||||
|
tz = _resolve_tz(tz_str)
|
||||||
|
config = await get_journal_config(user_id)
|
||||||
|
if not config.get("closeout_enabled", True):
|
||||||
|
return
|
||||||
|
rollover_hour = int(config.get("day_rollover_hour", 4))
|
||||||
|
now = datetime.datetime.now(tz)
|
||||||
|
# Slot hasn't passed yet today → wait for the cron.
|
||||||
|
if now.hour < rollover_hour:
|
||||||
|
return
|
||||||
|
yesterday = (now - datetime.timedelta(days=1)).date()
|
||||||
|
|
||||||
|
from fabledassistant.services.user_profile import get_profile
|
||||||
|
profile = await get_profile(user_id)
|
||||||
|
existing_dates = {
|
||||||
|
(e or {}).get("date") for e in (profile.observations_raw or [])
|
||||||
|
}
|
||||||
|
if yesterday.isoformat() in existing_dates:
|
||||||
|
return
|
||||||
|
|
||||||
|
from fabledassistant.services.journal_closeout import run_for_user
|
||||||
|
await run_for_user(user_id=user_id, yesterday=yesterday)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Closeout catch-up failed for user %d", user_id)
|
||||||
|
|
||||||
|
|
||||||
async def update_user_schedule(user_id: int) -> None:
|
async def update_user_schedule(user_id: int) -> None:
|
||||||
"""Add or replace this user's daily-prep + closeout jobs from current config."""
|
"""Add or replace this user's daily-prep + closeout jobs from current config."""
|
||||||
if _scheduler is None:
|
if _scheduler is None:
|
||||||
@@ -155,6 +186,9 @@ async def _register_all_user_jobs() -> None:
|
|||||||
users = (await session.execute(select(User))).scalars().all()
|
users = (await session.execute(select(User))).scalars().all()
|
||||||
for user in users:
|
for user in users:
|
||||||
await update_user_schedule(user.id)
|
await update_user_schedule(user.id)
|
||||||
|
# Fire catch-up asynchronously so a slow LLM call doesn't block startup
|
||||||
|
if _loop is not None:
|
||||||
|
asyncio.run_coroutine_threadsafe(_closeout_catchup(user.id), _loop)
|
||||||
|
|
||||||
|
|
||||||
def start_journal_scheduler(loop: asyncio.AbstractEventLoop) -> None:
|
def start_journal_scheduler(loop: asyncio.AbstractEventLoop) -> None:
|
||||||
|
|||||||
@@ -202,3 +202,48 @@ async def test_update_user_schedule_does_not_register_closeout_when_disabled(mon
|
|||||||
|
|
||||||
added = [c.kwargs.get("id") for c in fake_scheduler.add_job.call_args_list]
|
added = [c.kwargs.get("id") for c in fake_scheduler.add_job.call_args_list]
|
||||||
assert "journal_closeout_7" not in added
|
assert "journal_closeout_7" not in added
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_closeout_catchup_skips_when_already_have_entry(monkeypatch):
|
||||||
|
from fabledassistant.services import journal_scheduler as sched
|
||||||
|
|
||||||
|
monkeypatch.setattr(sched, "get_user_timezone", AsyncMock(return_value="UTC"))
|
||||||
|
monkeypatch.setattr(sched, "get_journal_config", AsyncMock(return_value={
|
||||||
|
"closeout_enabled": True,
|
||||||
|
"day_rollover_hour": 0, # slot has always passed
|
||||||
|
}))
|
||||||
|
yesterday = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1)).date()
|
||||||
|
fake_profile = SimpleNamespace(observations_raw=[{"date": yesterday.isoformat(), "bullets": "..."}])
|
||||||
|
|
||||||
|
from fabledassistant.services import user_profile as up
|
||||||
|
monkeypatch.setattr(up, "get_profile", AsyncMock(return_value=fake_profile))
|
||||||
|
|
||||||
|
run_mock = AsyncMock()
|
||||||
|
from fabledassistant.services import journal_closeout as jc
|
||||||
|
monkeypatch.setattr(jc, "run_for_user", run_mock)
|
||||||
|
|
||||||
|
await sched._closeout_catchup(user_id=7)
|
||||||
|
run_mock.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_closeout_catchup_runs_when_no_entry_yet(monkeypatch):
|
||||||
|
from fabledassistant.services import journal_scheduler as sched
|
||||||
|
|
||||||
|
monkeypatch.setattr(sched, "get_user_timezone", AsyncMock(return_value="UTC"))
|
||||||
|
monkeypatch.setattr(sched, "get_journal_config", AsyncMock(return_value={
|
||||||
|
"closeout_enabled": True,
|
||||||
|
"day_rollover_hour": 0,
|
||||||
|
}))
|
||||||
|
fake_profile = SimpleNamespace(observations_raw=[])
|
||||||
|
|
||||||
|
from fabledassistant.services import user_profile as up
|
||||||
|
monkeypatch.setattr(up, "get_profile", AsyncMock(return_value=fake_profile))
|
||||||
|
|
||||||
|
run_mock = AsyncMock()
|
||||||
|
from fabledassistant.services import journal_closeout as jc
|
||||||
|
monkeypatch.setattr(jc, "run_for_user", run_mock)
|
||||||
|
|
||||||
|
await sched._closeout_catchup(user_id=7)
|
||||||
|
run_mock.assert_awaited_once()
|
||||||
|
|||||||
Reference in New Issue
Block a user