bambuddy/backend/tests/unit/services/test_notify_live_activities.py

"""Lifecycle protocol tests with real persistence and a fake Notify gateway."""

import json
from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from sqlalchemy import delete, event, select, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

import backend.app.models  # noqa: F401
from backend.app.core.database import Base
from backend.app.models.notification import NotificationProvider
from backend.app.models.notification_live_activity import NotificationLiveActivity
from backend.app.models.printer import Printer
from backend.app.services import notify_live_activities as module
from backend.app.services.notify_client import NotifyError
from backend.app.services.notify_live_activities import NotifyLiveActivityService, PrintSnapshot


@pytest.fixture
async def setup(tmp_path, monkeypatch):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'activities.db'}")

    @event.listens_for(engine.sync_engine, "connect")
    def enable_foreign_keys(connection, _):
        cursor = connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    async with factory() as db:
        db.add(Printer(id=1, name="P1", serial_number="serial1", ip_address="127.0.0.1", access_code="12345678"))
        db.add(
            NotificationProvider(
                id=1,
                name="Phone",
                provider_type="notify",
                enabled=True,
                config=json.dumps({"device_id": "ABC12345", "token": "secret", "live_activities": True}),
                on_print_start=False,
                on_print_progress=False,
                on_print_complete=False,
                daily_digest_enabled=True,
            )
        )
        await db.commit()
    api = SimpleNamespace(
        start_activity=AsyncMock(return_value={"activityId": "LA123456"}),
        get_activity=AsyncMock(return_value={"state": "active"}),
        update_activity=AsyncMock(return_value={"success": True}),
        end_activity=AsyncMock(return_value={"success": True}),
    )
    clock = [datetime(2026, 1, 1, 12)]
    monkeypatch.setattr(module, "_now", lambda: clock[0])
    service = NotifyLiveActivityService(factory, api)
    monkeypatch.setattr(service, "_quiet", lambda provider: False)
    yield service, api, factory, clock
    await engine.dispose()


def state(**values):
    return (
        SimpleNamespace(
            connected=True,
            state="RUNNING",
            subtask_id="job1",
            subtask_name="Test print",
            current_print="test.3mf",
            progress=42,
            remaining_time=15,
            layer_num=42,
            total_layers=100,
            raw_data={},
            hms_errors=[],
            temperatures={"nozzle": 210, "bed": 60},
            stg_cur=-1,
            **values,
        )
        if not values
        else SimpleNamespace(
            **{
                **vars(state()),
                **values,
            }
        )
    )


async def rows(factory):
    async with factory() as db:
        return (await db.scalars(select(NotificationLiveActivity))).all()


async def start(setup):
    service, api, factory, clock = setup
    service.observe(1, state())
    await service.tick()
    return service, api, factory, clock


@pytest.mark.asyncio
async def test_start_independent_of_push_digest_and_group_thread(setup):
    service, api, factory, clock = setup
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config = json.loads(provider.config)
        config["group_type"] = "Workshop"
        provider.config = json.dumps(config)
        await db.commit()
    await start(setup)
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].activity_id == "LA123456"
    await service.tick()
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
async def test_startup_drops_rows_of_provider_deleted_before_cleanup_drained(setup):
    service, api, factory, clock = await start(setup)
    assert len(await rows(factory)) == 1
    # Production SQLite does not enforce the cascade.
    async with factory() as db:
        await db.execute(text("PRAGMA foreign_keys=OFF"))
        await db.execute(delete(NotificationProvider).where(NotificationProvider.id == 1))
        await db.commit()
        await db.execute(text("PRAGMA foreign_keys=ON"))
    restarted = NotifyLiveActivityService(factory, api)
    await restarted._purge_orphans()
    assert await rows(factory) == []


@pytest.mark.asyncio
async def test_restart_reuses_exact_id_without_new_start(setup, monkeypatch):
    service, api, factory, clock = await start(setup)
    clock[0] += timedelta(minutes=2)
    restarted = NotifyLiveActivityService(factory, api)
    monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
    restarted.observe(1, state(progress=46))
    await restarted.tick()
    api.get_activity.assert_awaited_once_with("LA123456", "secret")
    assert api.start_activity.await_count == 1
    assert api.update_activity.call_args.args[0] == "LA123456"


@pytest.mark.asyncio
@pytest.mark.parametrize("phase,connected,expected", [("PAUSE", True, "Paused"), ("RUNNING", False, "Printer offline")])
async def test_pause_disconnect_clear_countdown_then_resume(setup, phase, connected, expected):
    service, api, factory, clock = await start(setup)
    service.observe(1, state(state=phase, connected=connected))
    await service.tick()
    content = api.update_activity.call_args.args[2]
    assert content["endsIn"] is None
    assert content["status"] == expected
    service.observe(1, state())
    await service.tick()
    assert api.update_activity.call_args.args[2]["endsIn"] == 900
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
async def test_frozen_eta_does_not_drift(setup):
    service, api, factory, clock = await start(setup)
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert api.update_activity.call_args.args[2]["endsIn"] == 780
    service.observe(1, state(remaining_time=10))
    clock[0] += timedelta(minutes=1)
    await service.tick()
    assert api.update_activity.call_args.args[2]["endsIn"] == 600


@pytest.mark.asyncio
async def test_quiet_hours_block_starts_but_not_updates_or_ends(setup, monkeypatch):
    service, api, factory, clock = setup
    monkeypatch.setattr(service, "_quiet", lambda provider: True)
    service.observe(1, state())
    await service.tick()
    api.start_activity.assert_not_awaited()
    monkeypatch.setattr(service, "_quiet", lambda provider: False)
    await service.tick()
    monkeypatch.setattr(service, "_quiet", lambda provider: True)
    clock[0] += timedelta(minutes=2)
    await service.tick()
    api.update_activity.assert_awaited_once()
    service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
    await service.tick()
    assert api.end_activity.call_args.args[2]["progress"] == 100


@pytest.mark.asyncio
@pytest.mark.parametrize("reason", ["dismissed", "dead-token", "never-started", "script"])
async def test_terminal_suppression_survives_restart(setup, monkeypatch, reason):
    service, api, factory, clock = await start(setup)
    api.get_activity.return_value = {"state": "dismissed" if reason == "dismissed" else "ended", "endReason": reason}
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert (await rows(factory))[0].state == "suppressed"
    restarted = NotifyLiveActivityService(factory, api)
    monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
    restarted.observe(1, state())
    await restarted.tick()
    assert api.start_activity.await_count == 1
    restarted.observe(1, state(subtask_id="job2"))
    await restarted.tick()
    assert api.start_activity.await_count == 2


@pytest.mark.asyncio
@pytest.mark.parametrize("reason", ["expired", "overdue", "abandoned"])
async def test_expiry_rolls_over_only_for_live_print(setup, reason):
    service, api, factory, clock = await start(setup)
    api.get_activity.return_value = {"state": "ended", "endReason": reason}
    clock[0] += timedelta(hours=8)
    api.start_activity.return_value = {"activityId": "LA654321"}
    await service.tick()
    assert api.start_activity.await_count == 2
    assert (await rows(factory))[0].activity_id == "LA654321"


@pytest.mark.asyncio
async def test_eight_hour_ceiling_polls_before_rollover(setup):
    service, api, factory, clock = await start(setup)
    clock[0] += timedelta(hours=8)
    await service.tick()
    api.get_activity.assert_awaited_once()
    api.end_activity.assert_awaited_once_with("LA123456", "secret")
    assert api.start_activity.await_count == 2


@pytest.mark.asyncio
async def test_ambiguous_start_with_id_is_reconciled_without_duplication(setup):
    service, api, factory, clock = setup
    api.start_activity.side_effect = NotifyError(
        "unknown", status_code=502, payload={"activityId": "LA123456", "deliveryState": "unknown"}
    )
    await start(setup)
    assert (await rows(factory))[0].activity_id == "LA123456"
    clock[0] += timedelta(minutes=2)
    await service.tick()
    api.get_activity.assert_awaited_once_with("LA123456", "secret")
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
async def test_ambiguous_start_without_id_never_repeats(setup):
    service, api, factory, clock = setup
    api.start_activity.side_effect = NotifyError("timeout", delivery_state="unknown")
    await start(setup)
    clock[0] += timedelta(hours=2)
    await service.tick()
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].state == "suppressed"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "error",
    [
        NotifyError("full", status_code=400, payload={"message": "Maximum 5 activities"}),
        NotifyError("throttled", status_code=429, retry_after_seconds=1800),
        NotifyError("not delivered", status_code=502, retry_after_seconds=1800, delivery_state="not-delivered"),
        NotifyError("unavailable", status_code=503),
    ],
)
async def test_retryable_starts_honor_backoff(setup, error):
    service, api, factory, clock = setup
    api.start_activity.side_effect = error
    await start(setup)
    await service.tick()
    assert api.start_activity.await_count == 1
    clock[0] += timedelta(minutes=31)
    api.start_activity.side_effect = None
    await service.tick()
    assert api.start_activity.await_count == 2


@pytest.mark.asyncio
async def test_end_intent_survives_failure_restart_and_next_job(setup, monkeypatch):
    service, api, factory, clock = await start(setup)
    api.end_activity.side_effect = NotifyError("timeout", delivery_state="unknown")
    service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
    await service.tick()
    assert (await rows(factory))[0].state == "ending"
    await service.tick()
    assert api.end_activity.await_count == 1
    restarted = NotifyLiveActivityService(factory, api)
    monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
    restarted.observe(1, state(subtask_id="job2"))
    clock[0] += timedelta(minutes=2)
    api.end_activity.side_effect = None
    await restarted.tick()
    assert api.end_activity.call_args.args[2]["status"] == "Complete"
    assert api.start_activity.await_count == 2
    assert (await rows(factory))[0].state == "ended"


@pytest.mark.asyncio
async def test_late_complete_does_not_end_new_job(setup):
    service, api, factory, clock = await start(setup)
    service.observe(1, state(subtask_id="job2"))
    await service.tick()
    api.end_activity.reset_mock()
    service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
    await service.tick()
    api.end_activity.assert_not_awaited()


@pytest.mark.asyncio
async def test_fallback_survives_delta_and_late_job_id(setup):
    service, api, factory, clock = setup
    service.print_started(1, state(subtask_id=None, raw_data={"gcode_start_time": 12345}))
    await service.tick()
    clock[0] += timedelta(minutes=2)
    service.observe(1, state(subtask_id=None, raw_data={}))
    await service.tick()
    service.observe(1, state(subtask_id="finally-known"))
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert api.start_activity.await_count == 1
    assert api.end_activity.await_count == 0


@pytest.mark.asyncio
async def test_scope_and_disabled_provider_end_tiles(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = False
        await db.commit()
    await service.tick()
    assert api.end_activity.await_count == 1
    assert (await rows(factory))[0].state == "ended"


@pytest.mark.asyncio
async def test_delete_printer_retains_id_until_remote_end(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        await db.delete(await db.get(Printer, 1))
        await db.commit()
    await service.tick()
    assert api.end_activity.await_count == 1
    assert (await rows(factory))[0].state == "ended"


@pytest.mark.asyncio
async def test_cleanup_uses_old_token_and_removes_ownership(setup):
    service, api, factory, clock = await start(setup)
    await service.cleanup_provider(1, {"device_id": "ABC12345", "token": "secret"})
    api.end_activity.assert_awaited_once_with("LA123456", "secret")
    assert await rows(factory) == []


@pytest.mark.asyncio
async def test_http_never_holds_database_session_and_concurrent_ticks_deduplicate(setup):
    import asyncio

    service, api, factory, clock = setup
    entered = 0

    class TrackedSession:
        async def __aenter__(self):
            nonlocal entered
            entered += 1
            self.db = factory()
            return await self.db.__aenter__()

        async def __aexit__(self, *args):
            nonlocal entered
            await self.db.__aexit__(*args)
            entered -= 1

    service._session = TrackedSession

    async def send(*args):
        assert entered == 0
        await asyncio.sleep(0)
        return {"activityId": "LA123456"}

    api.start_activity.side_effect = send
    service.observe(1, state())
    await asyncio.gather(service.tick(), service.tick(), service.tick())
    assert api.start_activity.await_count == 1


def test_advanced_content_and_utf8_budget():
    snapshot = PrintSnapshot.from_state(state(subtask_name="🔒" * 255, remaining_time=1500))
    config = {
        "live_activity_style": "segments",
        "live_activity_metrics": ["progress", "eta", "layers", "nozzle", "bed", "chamber"],
        "live_activity_button_url": "https://bambuddy.example/" + "a" * 480,
        "live_activity_privacy": True,
    }
    content = snapshot.content("🖨" * 100, config)
    assert "🔒" not in content["body"]
    assert content["steps"] == 10 and content["step"] == 4
    assert content["endsIn"] is None and content["trailing"] == "25h 0m"
    assert len(content["metrics"]) == 5
    assert len(json.dumps(content, ensure_ascii=False).encode()) < 2048
    assert content["button"] is None or content["button"]["open"] is True
    assert snapshot.content("P1", {"live_activity_style": "none"})["progress"] is None


@pytest.mark.asyncio
async def test_fallback_dismissal_follows_late_id_across_restart(setup, monkeypatch):
    service, api, factory, clock = setup
    service.print_started(1, state(subtask_id=None))
    await service.tick()
    clock[0] += timedelta(minutes=2)
    api.get_activity.return_value = {"state": "dismissed", "endReason": "dismissed"}
    await service.tick()
    service.observe(1, state(subtask_id="late-id"))
    await service.tick()
    assert (await rows(factory))[0].job_key == "job:late-id"
    restarted = NotifyLiveActivityService(factory, api)
    monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
    restarted.observe(1, state(subtask_id="late-id"))
    await restarted.tick()
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
async def test_end_retry_without_any_printer_snapshot(setup):
    service, api, factory, clock = await start(setup)
    api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
    service.print_finished(1, {"subtask_id": "job1", "status": "failed"})
    await service.tick()
    restarted = NotifyLiveActivityService(factory, api)
    clock[0] += timedelta(minutes=2)
    api.end_activity.side_effect = None
    await restarted.tick()
    assert api.end_activity.call_args.args[2]["status"] == "Failed"
    assert (await rows(factory))[0].state == "ended"


@pytest.mark.asyncio
async def test_printer_scope_new_provider_separate_ownership(setup):
    service, api, factory, clock = setup
    async with factory() as db:
        db.add(Printer(id=2, name="P2", serial_number="serial2", ip_address="127.0.0.2", access_code="12345678"))
        db.add(
            NotificationProvider(
                id=2,
                name="Other Phone",
                provider_type="notify",
                printer_id=2,
                config=json.dumps({"device_id": "ABC12346", "token": "other", "live_activities": True}),
            )
        )
        await db.commit()
    await start(setup)
    assert api.start_activity.await_count == 1
    async with factory() as db:
        provider = await db.get(NotificationProvider, 2)
        provider.printer_id = 1
        await db.commit()
    await service.tick()
    assert api.start_activity.await_count == 2
    assert {r.provider_id for r in await rows(factory)} == {1, 2}


@pytest.mark.asyncio
async def test_unknown_start_record_after_process_crash_is_never_restarted(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        row = await db.get(NotificationLiveActivity, 1)
        row.activity_id = None
        row.state = "uncertain"
        await db.commit()
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].state == "suppressed"


@pytest.mark.asyncio
async def test_410_is_polled_for_reason_before_any_rollover(setup):
    service, api, factory, clock = await start(setup)
    api.update_activity.side_effect = NotifyError("gone", status_code=410)
    clock[0] += timedelta(minutes=2)
    await service.tick()
    api.get_activity.return_value = {"state": "dismissed", "endReason": "dismissed"}
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].state == "suppressed"


def fault(description="Nozzle temperature malfunction", severity=2, full_code="05004003", actions=None):
    return SimpleNamespace(description=description, severity=severity, full_code=full_code, actions=actions)


@pytest.mark.asyncio
async def test_running_printer_fault_replaces_counter_and_restores_after_clear(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config = json.loads(provider.config)
        config.update(live_activity_metrics=["progress", "eta"], live_activity_stage=True)
        provider.config = json.dumps(config)
        await db.commit()
    service.observe(1, state(hms_errors=[fault()]))
    await service.tick()
    content = api.update_activity.call_args.args[2]
    assert content["status"] == "Printer error"
    assert content["body"] == "Nozzle temperature malfunction"
    assert content["endsIn"] is None
    assert content["metrics"] is None
    service.observe(1, state(hms_errors=[]))
    await service.tick()
    content = api.update_activity.call_args.args[2]
    assert content["status"] == "Printing"
    assert content["endsIn"] == 900
    assert len(content["metrics"]) == 2
    assert api.start_activity.await_count == 1


def test_runout_and_actionable_level_three_prompt():
    snapshot = PrintSnapshot.from_state(state(hms_errors=[fault("Filament has run out", 3)]))
    content = snapshot.content("P1", {"live_activity_privacy": True})
    assert content["status"] == "Filament runout"
    assert content["endsIn"] is None
    assert "Test print" not in content["body"]
    snapshot = PrintSnapshot.from_state(
        state(hms_errors=[fault("Heatbreak fan problem", 3, "0000000012345678", ["retry"])])
    )
    assert snapshot.content("P1")["status"] == "Printer error"


@pytest.mark.parametrize(
    "error",
    [
        fault("The cover is open", 3, "0000000012345678"),
        fault("Invalid level", 0),
        fault("", 2),
    ],
)
def test_advisory_does_not_interrupt_live_countdown(error):
    content = PrintSnapshot.from_state(state(hms_errors=[error])).content("P1")
    assert content["status"] == "Printing"
    assert content["endsIn"] == 900


@pytest.mark.asyncio
@pytest.mark.parametrize("device_id", ["GRP12345", "WB123456", "MC123456"])
async def test_non_ios_devices_never_start_activity(setup, device_id):
    service, api, factory, clock = setup
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config = json.loads(provider.config)
        config["device_id"] = device_id
        provider.config = json.dumps(config)
        await db.commit()
    service.observe(1, state())
    await service.tick()
    api.start_activity.assert_not_awaited()


@pytest.mark.asyncio
async def test_new_start_resets_previous_progress_and_eta_until_fresh_telemetry(setup):
    service, api, factory, clock = setup
    service.print_started(1, state(progress=99, remaining_time=1))
    await service.tick()
    assert api.start_activity.call_args.args[2]["progress"] == 0
    assert api.start_activity.call_args.args[2]["endsIn"] is None
    service.observe(1, state(progress=99, remaining_time=1))
    clock[0] += timedelta(seconds=61)
    await service.tick()
    assert api.update_activity.call_args.args[2]["progress"] == 0
    assert api.update_activity.call_args.args[2]["endsIn"] is None
    service.observe(1, state(progress=2, remaining_time=50))
    clock[0] += timedelta(seconds=61)
    await service.tick()
    assert api.update_activity.call_args.args[2]["progress"] == 2
    assert api.update_activity.call_args.args[2]["endsIn"] == 3000


@pytest.mark.asyncio
async def test_old_tile_must_end_before_starting_next_job(setup):
    service, api, factory, clock = await start(setup)
    api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
    service.observe(1, state(subtask_id="job2"))
    await service.tick()
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].state == "ending"
    api.end_activity.side_effect = None
    clock[0] += timedelta(minutes=2)
    await service.tick()
    assert api.start_activity.await_count == 2


@pytest.mark.asyncio
async def test_complete_matches_learned_job_id_for_fallback_row(setup):
    service, api, factory, clock = setup
    service.print_started(1, state(subtask_id=None))
    await service.tick()
    service.observe(1, state(subtask_id="learned"))
    await service.tick()
    service.print_finished(1, {"raw_data": {"subtask_id": "learned"}, "status": "completed"})
    await service.tick()
    assert api.end_activity.call_args.args[2]["status"] == "Complete"
    assert (await rows(factory))[0].state == "ended"


def test_preparation_does_not_display_previous_print_progress():
    snapshot = PrintSnapshot.from_state(state(layer_num=0, total_layers=100, progress=85))
    content = snapshot.content("P1", {"live_activity_metrics": ["progress"]})
    assert content["status"] == "Preparing"
    assert content["progress"] == 0
    assert content["metrics"][0]["value"] == "0%"


@pytest.mark.asyncio
async def test_reenable_resumes_current_print_after_successful_cleanup(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = False
        await db.commit()
    await service.tick()
    assert (await rows(factory))[0].end_reason == "disabled"
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = True
        await db.commit()
    await service.tick()
    assert api.start_activity.await_count == 1  # Wait for fresh telemetry after dormant time.
    service.observe(1, state())
    await service.tick()
    assert api.start_activity.await_count == 2


@pytest.mark.asyncio
async def test_reenable_waits_for_failed_cleanup_before_restarting(setup):
    service, api, factory, clock = await start(setup)
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = False
        await db.commit()
    api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
    await service.tick()
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = True
        await db.commit()
    await service.tick()
    assert api.start_activity.await_count == 1
    api.end_activity.side_effect = None
    clock[0] += timedelta(minutes=2)
    await service.tick()
    service.observe(1, state())
    await service.tick()
    assert api.start_activity.await_count == 2


@pytest.mark.parametrize("config", [{}, {"live_activity_metrics": []}])
def test_default_tile_uses_native_timer_without_metric_chips(config):
    content = PrintSnapshot.from_state(state()).content("P1", config)
    assert content["endsIn"] == 900
    assert content["metrics"] is None
    assert content["trailing"] is None


@pytest.mark.asyncio
async def test_disabling_optional_metrics_clears_chips_and_preserves_timer(setup):
    service, api, factory, clock = setup
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config = json.loads(provider.config)
        config["live_activity_metrics"] = ["progress", "eta"]
        provider.config = json.dumps(config)
        await db.commit()
    await start(setup)
    content = api.start_activity.call_args.args[2]
    assert content["endsIn"] == 900
    assert content["metrics"][0] == {"label": "Progress", "value": "42%", "unit": ""}
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config["live_activity_metrics"] = []
        provider.config = json.dumps(config)
        await db.commit()
    clock[0] += timedelta(minutes=1)
    await service.tick()
    content = api.update_activity.call_args.args[2]
    assert content["metrics"] is None
    assert content["endsIn"] == 840
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
async def test_dormant_worker_has_no_periodic_sql_or_mqtt_wakeups(setup):
    import asyncio

    from sqlalchemy import event

    service, api, factory, clock = setup
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        provider.enabled = False
        await db.commit()
    queries = []
    engine = factory.kw["bind"]

    def record(connection, cursor, statement, parameters, context, many):
        queries.append(statement)

    event.listen(engine.sync_engine, "before_cursor_execute", record)
    service.worker_interval = 0.01
    service.start()
    try:
        for _ in range(100):
            if not service._initial_discovery:
                break
            await asyncio.sleep(0.001)
        queries.clear()
        service.observe(1, state())
        service.print_started(1, state())
        service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
        await asyncio.sleep(0.05)
        assert queries == []
        assert service._snapshots == {}
        api.start_activity.assert_not_awaited()
    finally:
        await service.close()
        event.remove(engine.sync_engine, "before_cursor_execute", record)


@pytest.mark.asyncio
async def test_startup_waits_for_real_status_before_touching_existing_activity(setup):
    service, api, factory, clock = await start(setup)
    clock[0] += timedelta(minutes=2)
    restarted = NotifyLiveActivityService(factory, api)
    await restarted.tick()
    api.get_activity.assert_not_awaited()
    restarted.observe(1, state(state="unknown", progress=0))
    await restarted.tick()
    api.get_activity.assert_not_awaited()
    restarted.observe(1, state())
    await restarted.tick()
    api.get_activity.assert_awaited_once()


@pytest.mark.asyncio
@pytest.mark.parametrize("connected", [None, False, True])
async def test_startup_grace_expires_without_real_status_and_reconnect_resumes(setup, connected):
    service, api, factory, clock = await start(setup)
    clock[0] += timedelta(minutes=2)
    restarted = NotifyLiveActivityService(factory, api)
    # Importing the service before worker startup must not consume the grace.
    clock[0] += timedelta(minutes=5)

    # Neither complete silence nor broker-only status may extend the grace.
    for seconds in (0, 90, 29):
        clock[0] += timedelta(seconds=seconds)
        if connected is not None:
            restarted.observe(1, state(state="unknown", connected=connected, progress=0))
        await restarted.tick()
        api.get_activity.assert_not_awaited()
        api.update_activity.assert_not_awaited()

    clock[0] += timedelta(seconds=1)
    await restarted.tick()
    api.get_activity.assert_awaited_once_with("LA123456", "secret")
    content = api.update_activity.call_args.args[2]
    assert content["status"] == "Printer offline"
    assert content["endsIn"] is None
    assert content["progress"] == 42
    saved = (await rows(factory))[0]
    assert saved.eta_seconds is None
    assert saved.eta_deadline is None
    assert saved.activity_id == "LA123456"
    api.end_activity.assert_not_awaited()
    assert api.start_activity.await_count == 1

    # A real status resumes the same tile immediately, even during throttling.
    restarted.observe(1, state(progress=46, remaining_time=10))
    await restarted.tick()
    assert api.update_activity.await_count == 2
    assert api.update_activity.call_args.args[0] == "LA123456"
    content = api.update_activity.call_args.args[2]
    assert content["status"] == "Printing"
    assert content["endsIn"] == 600
    assert content["progress"] == 46
    assert api.start_activity.await_count == 1


@pytest.mark.asyncio
@pytest.mark.parametrize("reason", ["expired", "dismissed"])
async def test_startup_grace_never_restarts_an_offline_printers_activity(setup, reason):
    service, api, factory, clock = await start(setup)
    restarted = NotifyLiveActivityService(factory, api)
    await restarted.tick()
    api.get_activity.return_value = {"state": "ended", "endReason": reason}
    clock[0] += timedelta(minutes=2)
    await restarted.tick()
    api.get_activity.assert_awaited_once_with("LA123456", "secret")
    api.update_activity.assert_not_awaited()
    api.end_activity.assert_not_awaited()
    assert api.start_activity.await_count == 1
    assert (await rows(factory))[0].state == ("pending" if reason == "expired" else "suppressed")


@pytest.mark.asyncio
async def test_startup_offline_clears_saved_eta_labels(setup):
    service, api, factory, clock = setup
    async with factory() as db:
        provider = await db.get(NotificationProvider, 1)
        config = json.loads(provider.config)
        config["live_activity_metrics"] = ["progress", "eta", "nozzle"]
        provider.config = json.dumps(config)
        await db.commit()
    service.observe(1, state(remaining_time=25 * 60))
    await service.tick()
    original = api.start_activity.call_args.args[2]
    assert original["trailing"] == "25h 0m"
    assert any(metric["label"] == "Remaining" for metric in original["metrics"])

    restarted = NotifyLiveActivityService(factory, api)
    await restarted.tick()
    clock[0] += timedelta(minutes=2)
    await restarted.tick()
    content = api.update_activity.call_args.args[2]
    assert content["status"] == "Printer offline"
    assert content["endsIn"] is None
    assert content["trailing"] is None
    assert content["metrics"] is None
    assert content["progress"] == 42


@pytest.mark.asyncio
async def test_scheduled_cleanup_does_not_wait_for_create_and_keeps_late_handle(setup):
    import asyncio

    service, api, factory, clock = setup
    entered, release = asyncio.Event(), asyncio.Event()

    async def create(*args):
        entered.set()
        await release.wait()
        return {"activityId": "LA123456"}

    api.start_activity.side_effect = create
    service.observe(1, state())
    tick = asyncio.create_task(service.tick())
    await entered.wait()
    await asyncio.wait_for(service.schedule_cleanup(1, {"device_id": "ABC12345", "token": "secret"}), 0.2)
    async with factory() as db:
        await db.delete(await db.get(NotificationProvider, 1))
        await db.commit()
    release.set()
    await tick
    try:
        for _ in range(100):
            if api.end_activity.await_count:
                break
            await asyncio.sleep(0.005)
        api.end_activity.assert_awaited_once_with("LA123456", "secret")
        assert await rows(factory) == []
    finally:
        await service.close()


@pytest.mark.asyncio
async def test_boot_cleans_disabled_activity_without_printer_queries_or_pruning(setup):
    import asyncio

    from sqlalchemy import event

    service, api, factory, clock = await start(setup)
    async with factory() as db:
        (await db.get(NotificationProvider, 1)).enabled = False
        await db.commit()
    statements = []
    engine = factory.kw["bind"]

    def record(connection, cursor, statement, parameters, context, many):
        statements.append(statement.lower())

    event.listen(engine.sync_engine, "before_cursor_execute", record)
    restarted = NotifyLiveActivityService(factory, api)
    restarted.start()
    try:
        for _ in range(100):
            if (await rows(factory))[0].state == "ended":
                break
            await asyncio.sleep(0.005)
        assert (await rows(factory))[0].state == "ended"
        assert not any("from printers" in statement for statement in statements)
        assert not any(statement.startswith("delete from notification_live_activities") for statement in statements)
    finally:
        await restarted.close()
        event.remove(engine.sync_engine, "before_cursor_execute", record)