diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4820443..4806cff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -31,6 +31,35 @@ codes documented in `AGENT.md` are the public API.
 
 ### Fixed
 
+- **Gateway jobs work again, and incoming updates reach the daemon at all.**
+  Four faults stacked up, and each one alone was enough to stop every job:
+
+  - The daemon attached its update handler before the account's client
+    existed, so no incoming update reached the event bus. Jobs, webhooks and
+    `watch` saw the daemon's own sends and nothing else. Every job stopped
+    with `matched=0 skipped=0`.
+  - The bus hands a job the raw TL update, but filters and actions read a
+    Telethon event (`chat_id`, `is_private`, `reply()`). The job now builds
+    that event the way a Telethon handler would, including
+    `NewMessage(incoming=True)`, so an auto-reply never answers the
+    account's own messages.
+  - `filter_chat_id` skipped `@name` refs, expecting the gateway to resolve
+    them, and nothing did, so `chat_id: "@channel"` never matched. Jobs now
+    resolve them at start, and retry once a minute if the account was
+    offline.
+  - A restarted daemon started no jobs until `tlgr job reload`. It now
+    starts them at boot.
+
+- **Accounts no longer go "degraded" on every start** with `AttributeError:
+  'MessageBox' object attribute 'apply_difference' is read-only`. Telethon's
+  `MessageBox` declares `__slots__`, so the too-long hook now installs on a
+  per-client subclass, and the account no longer burns a reconnect or runs
+  without the hook.
+
+- **`job list` reports `enabled` and `running` even when they are at their
+  defaults.** Before, an enabled job that was not running showed `-` in both
+  columns, the same as a disabled one.
+
 - **`chat list` no longer loses a block of dialogs at a page boundary.** The
   dialog walk paired each row with its top message by message id alone, but
   only private chats share one id space: every channel and supergroup numbers
diff --git a/tests/fake_telethon.py b/tests/fake_telethon.py
index c108848..ec06535 100644
--- a/tests/fake_telethon.py
+++ b/tests/fake_telethon.py
@@ -1429,6 +1429,17 @@ def __init__(self, world: World | None = None, session: Any = None, **_: Any) ->
         self._handlers: list[tuple[Any, Any]] = []
         self.disconnected: asyncio.Future[None] = asyncio.get_event_loop().create_future()
         self._requests = 0
+        # What Telethon's events and messages read off a client: `_set_client`
+        # looks entities up here, `build()` takes `_self_id`, and
+        # `Message.text` unparses through `parse_mode` (None means plain text).
+        self.parse_mode = None
+        from telethon._updates import EntityCache
+
+        self._mb_entity_cache = EntityCache(self_id=self.world.me.id, self_bot=False)
+
+    @property
+    def _self_id(self) -> int:
+        return int(self.world.me.id)
 
     # -- connection --------------------------------------------------------
 
diff --git a/tests/test_daemon_jobs.py b/tests/test_daemon_jobs.py
new file mode 100644
index 0000000..073d4a4
--- /dev/null
+++ b/tests/test_daemon_jobs.py
@@ -0,0 +1,70 @@
+"""The daemon starts the jobs in `jobs.yaml` when it boots.
+
+Until 2.0.1 it did not: `reload_jobs()` ran only on `tlgr job reload`, so a
+restarted daemon (launchd, an upgrade, a crash) ran no jobs at all and said
+nothing about it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from fake_telethon import fake_client_factory
+
+JOBS = """\
+jobs:
+  - name: archive
+    account: {account}
+    actions:
+      - forward:
+          to: ["@archive"]
+  - name: parked
+    account: {account}
+    enabled: false
+    actions:
+      - reply: "parked"
+"""
+
+
+async def _until(predicate, timeout: float = 5.0) -> None:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() > deadline:
+            raise AssertionError("condition never became true")
+        await asyncio.sleep(0.02)
+
+
+class TestJobsAtBoot:
+    async def test_enabled_jobs_are_running_once_the_daemon_is_up(
+        self, tlgr_home: Path, stub_account: str, world
+    ):
+        from tlgr.daemon.app import Daemon
+
+        (tlgr_home / "jobs.yaml").write_text(JOBS.format(account=stub_account))
+        daemon = Daemon(tlgr_home, client_factory=fake_client_factory(world))
+        runner = asyncio.create_task(daemon.run())
+        try:
+            await _until(lambda: any(job.get("running") for job in daemon.list_jobs()))
+            jobs = {job["name"]: job for job in daemon.list_jobs()}
+            assert jobs["archive"]["running"] is True
+            assert "parked" not in jobs
+        finally:
+            daemon.request_shutdown()
+            await asyncio.wait_for(runner, timeout=10)
+
+    async def test_a_broken_jobs_file_does_not_stop_the_daemon(
+        self, tlgr_home: Path, stub_account: str, world
+    ):
+        from tlgr.daemon.app import Daemon
+
+        (tlgr_home / "jobs.yaml").write_text("jobs: [unclosed")
+        daemon = Daemon(tlgr_home, client_factory=fake_client_factory(world))
+        runner = asyncio.create_task(daemon.run())
+        try:
+            await _until(lambda: daemon.ready)
+            await _until(lambda: daemon._jobs_task is not None and daemon._jobs_task.done())
+            assert daemon.list_jobs() == []
+        finally:
+            daemon.request_shutdown()
+            await asyncio.wait_for(runner, timeout=10)
diff --git a/tests/test_daemon_updates.py b/tests/test_daemon_updates.py
new file mode 100644
index 0000000..49c5431
--- /dev/null
+++ b/tests/test_daemon_updates.py
@@ -0,0 +1,63 @@
+"""Incoming updates reach the daemon's event bus.
+
+Until 2.0.1 none did: the daemon attached its update handler straight after
+`AccountSession.start()`, which only schedules the supervisor, so it always
+found `session.client is None` and returned. Jobs, webhooks and `watch` saw
+the daemon's own sends and nothing else, and every job stopped with
+`matched=0 skipped=0`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from telethon.tl import types
+
+from tlgr.daemon.session import SessionState
+
+
+async def _online(daemon, alias: str):
+    session = await daemon.sessions.ensure(alias)
+    deadline = asyncio.get_running_loop().time() + 2.0
+    while session.state != SessionState.ONLINE:
+        if asyncio.get_running_loop().time() > deadline:
+            raise AssertionError(f"{alias} never came online: {session.state}")
+        await asyncio.sleep(0.01)
+    return session
+
+
+def _incoming(message_id: int = 7) -> types.UpdateNewChannelMessage:
+    message = types.Message(
+        id=message_id,
+        peer_id=types.PeerChannel(channel_id=1234),
+        date=None,
+        message="hello",
+    )
+    return types.UpdateNewChannelMessage(message=message, pts=1, pts_count=1)
+
+
+class TestIncomingUpdates:
+    async def test_an_incoming_message_is_numbered_on_the_bus(self, daemon, stub_account):
+        session = await _online(daemon, stub_account)
+        before = daemon.bus.latest_seq(stub_account)
+
+        await session.client.feed(_incoming())
+
+        assert daemon.bus.latest_seq(stub_account) == before + 1
+        [envelope], _gap = daemon.bus.replay(stub_account, before)
+        assert envelope.type == "message_new"
+        assert envelope.chat_id == -1000000001234
+
+    async def test_the_handler_is_attached_once_across_reconnects(self, daemon, stub_account):
+        """The client outlives a reconnect, so a second handler would double every event."""
+        session = await _online(daemon, stub_account)
+        handlers = len(session.client._handlers)
+
+        await session.client.disconnect()
+        deadline = asyncio.get_running_loop().time() + 5.0
+        while session.reconnects == 0 or session.state != SessionState.ONLINE:
+            if asyncio.get_running_loop().time() > deadline:
+                raise AssertionError("the session never reconnected")
+            await asyncio.sleep(0.01)
+
+        assert len(session.client._handlers) == handlers
diff --git a/tests/test_gateway_bus.py b/tests/test_gateway_bus.py
new file mode 100644
index 0000000..b3735f2
--- /dev/null
+++ b/tests/test_gateway_bus.py
@@ -0,0 +1,274 @@
+"""A gateway job fed from the daemon's event bus, with real TL updates.
+
+The bus hands a job the raw TL `Update*` beside the envelope, while every
+filter and action reads a high-level Telethon event (`event.chat_id`,
+`event.is_private`, `event.reply()`). These tests drive a job the way the
+daemon does, so a shape mismatch fails here instead of in production, where
+it showed up as `matched=0` for days.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from fake_telethon import FakeTelegramClient, World
+from telethon.tl import types
+
+from tlgr.gateway.config import _parse_job
+from tlgr.gateway.engine import Gateway
+
+SOURCE = 1234
+SOURCE_MARKED = -1000000001234
+ALICE = 4242
+
+
+class _Bus:
+    def __init__(self) -> None:
+        self.handlers: list = []
+
+    def add_handler(self, handler) -> None:
+        self.handlers.append(handler)
+
+    def remove_handler(self, handler) -> None:
+        self.handlers.remove(handler)
+
+    async def deliver(self, update, *, event_type: str = "message_new", account: str = "Neo"):
+        envelope = SimpleNamespace(type=event_type, account=account)
+        for handler in list(self.handlers):
+            await handler(envelope, update)
+
+
+class _JobClient:
+    def __init__(self, world: World, chats: dict[str, int]) -> None:
+        self.client = FakeTelegramClient(world)
+        self.client.forward_messages = AsyncMock()
+        self.client.send_message = AsyncMock()
+        self.resolve_chat = AsyncMock(side_effect=lambda ref: chats[ref])
+
+
+def _channel_post(text: str = "breaking", *, channel: int = SOURCE, out: bool = False):
+    message = types.Message(
+        id=5, peer_id=types.PeerChannel(channel_id=channel), date=None, message=text, out=out
+    )
+    return types.UpdateNewChannelMessage(message=message, pts=1, pts_count=1)
+
+
+def _private(text: str = "hi", *, out: bool = False):
+    message = types.Message(
+        id=6,
+        peer_id=types.PeerUser(user_id=ALICE),
+        from_id=None if out else types.PeerUser(user_id=ALICE),
+        date=None,
+        message=text,
+        out=out,
+    )
+    return types.UpdateNewMessage(message=message, pts=1, pts_count=1)
+
+
+async def _running(job: Gateway, bus: _Bus) -> None:
+    job.start()
+    for _ in range(100):
+        if bus.handlers:
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError("the job never subscribed to the bus")
+
+
+@pytest.fixture
+def bus() -> _Bus:
+    return _Bus()
+
+
+@pytest.fixture
+async def client() -> _JobClient:
+    return _JobClient(World(), {"@source": SOURCE_MARKED, "@archive": -1000000009999})
+
+
+class TestForwardFromTheBus:
+    async def test_a_post_in_the_watched_channel_is_forwarded(self, bus, client):
+        config = _parse_job(
+            {
+                "name": "archive",
+                "account": "Neo",
+                "filters": {"chat_id": "@source"},
+                "actions": [{"forward": {"to": ["@archive"]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_channel_post())
+        finally:
+            await job.stop()
+
+        assert job._stats == {"matched": 1, "skipped": 0, "errors": 0}
+        client.client.forward_messages.assert_awaited_once()
+        assert client.client.forward_messages.await_args.args[0] == -1000000009999
+
+    async def test_a_post_elsewhere_is_skipped(self, bus, client):
+        config = _parse_job(
+            {
+                "name": "archive",
+                "account": "Neo",
+                "filters": {"chat_id": "@source"},
+                "actions": [{"forward": {"to": ["@archive"]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_channel_post(channel=777))
+        finally:
+            await job.stop()
+
+        assert job._stats["matched"] == 0
+        assert job._stats["skipped"] == 1
+        client.client.forward_messages.assert_not_awaited()
+
+    async def test_the_processed_copy_is_sent_with_the_rewritten_text(self, bus, client):
+        config = _parse_job(
+            {
+                "name": "rename",
+                "account": "Neo",
+                "filters": {"chat_id": "@source"},
+                "actions": [
+                    {
+                        "forward": {
+                            "to": ["@archive"],
+                            "processors": [
+                                {"type": "regex", "pattern": "wbnet", "replacement": "VPEON"}
+                            ],
+                        }
+                    }
+                ],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_channel_post("join wbnet today"))
+        finally:
+            await job.stop()
+
+        client.client.send_message.assert_awaited_once_with(-1000000009999, "join VPEON today")
+
+    async def test_another_accounts_post_is_ignored(self, bus, client):
+        config = _parse_job(
+            {
+                "name": "archive",
+                "account": "Neo",
+                "filters": {"chat_id": "@source"},
+                "actions": [{"forward": {"to": ["@archive"]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_channel_post(), account="other")
+        finally:
+            await job.stop()
+
+        assert job._stats == {"matched": 0, "skipped": 0, "errors": 0}
+
+
+class TestReplyFromTheBus:
+    @pytest.fixture
+    def config(self):
+        return _parse_job(
+            {
+                "name": "away",
+                "account": "Neo",
+                "filters": {"chat_type": "private"},
+                "actions": [{"reply": "away"}],
+            }
+        )
+
+    async def test_an_incoming_private_message_is_answered(self, bus, client, config):
+        client.client.send_message = AsyncMock()
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_private())
+        finally:
+            await job.stop()
+
+        assert job._stats["matched"] == 1
+        client.client.send_message.assert_awaited_once()
+
+    async def test_the_accounts_own_message_is_not_answered(self, bus, client, config):
+        """`new_message` means incoming, as it did when jobs used `NewMessage(incoming=True)`."""
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            await bus.deliver(_private(out=True))
+        finally:
+            await job.stop()
+
+        assert job._stats == {"matched": 0, "skipped": 0, "errors": 0}
+        client.client.send_message.assert_not_awaited()
+
+
+class TestChatRefResolution:
+    async def test_numeric_refs_are_left_alone(self, bus, client):
+        config = _parse_job(
+            {
+                "name": "ids",
+                "account": "Neo",
+                "filters": {"chat_id": [SOURCE_MARKED, "@source"]},
+                "actions": [{"forward": {"to": ["@archive"]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await job.setup()
+
+        assert config.filters.filter_value == [SOURCE_MARKED, SOURCE_MARKED]
+        client.resolve_chat.assert_awaited_once_with("@source")
+
+    async def test_an_unresolvable_ref_is_logged_and_kept(self, bus, client, caplog):
+        client.resolve_chat = AsyncMock(side_effect=ValueError("no such chat"))
+        config = _parse_job(
+            {
+                "name": "typo",
+                "account": "Neo",
+                "filters": {"chat_id": "@nope"},
+                "actions": [{"forward": {"to": ["@archive"]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await job.setup()
+
+        assert config.filters.filter_value == "@nope"
+        assert "cannot resolve chat @nope" in caplog.text
+
+    async def test_a_ref_that_failed_at_boot_is_retried_when_events_arrive(self, bus, client):
+        """An account still offline when the job starts must not leave it deaf for good."""
+        answers = [ValueError("offline"), SOURCE_MARKED]
+
+        async def resolve(ref):
+            answer = answers.pop(0)
+            if isinstance(answer, Exception):
+                raise answer
+            return answer
+
+        client.resolve_chat = AsyncMock(side_effect=resolve)
+        config = _parse_job(
+            {
+                "name": "archive",
+                "account": "Neo",
+                "filters": {"chat_id": "@source"},
+                "actions": [{"forward": {"to": [-1000000009999]}}],
+            }
+        )
+        job = Gateway(config, client, bus=bus)
+        await _running(job, bus)
+        try:
+            job._next_resolve = 0.0  # the retry interval has passed
+            await bus.deliver(_channel_post())
+        finally:
+            await job.stop()
+
+        assert job._stats["matched"] == 1
+        assert config.filters.filter_value == SOURCE_MARKED
diff --git a/tests/test_ops_daemon.py b/tests/test_ops_daemon.py
index 7f34452..f801351 100644
--- a/tests/test_ops_daemon.py
+++ b/tests/test_ops_daemon.py
@@ -520,6 +520,17 @@ async def test_an_unclaimed_prefix_is_not_found(self, live_daemon, client, in_th
 
 
 class TestJobs:
+    def test_a_job_row_always_says_whether_it_is_enabled_and_running(self):
+        """With `omit_defaults`, an enabled, stopped job printed `-` in both columns."""
+        import msgspec
+
+        from tlgr.models.daemon import JobState
+
+        row = msgspec.to_builtins(JobState(name="archive", enabled=True, running=False))
+        assert row["enabled"] is True
+        assert row["running"] is False
+        assert row["matched"] == 0
+
     def test_a_job_is_added_from_flags(self, tlgr_home):
         added = local(
             "job.add",
diff --git a/tests/test_telethon_compat.py b/tests/test_telethon_compat.py
new file mode 100644
index 0000000..617cf28
--- /dev/null
+++ b/tests/test_telethon_compat.py
@@ -0,0 +1,56 @@
+"""`install_too_long_hook` against Telethon's real `MessageBox`.
+
+The fake client has no message box, so nothing caught that `MessageBox`
+declares `__slots__` and refuses instance assignment: every account logged
+`AttributeError: 'MessageBox' object attribute 'apply_difference' is
+read-only` on its first connect, burned one reconnect, and ran without the
+hook.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+
+from telethon._updates.messagebox import MessageBox
+from telethon.tl import types
+
+from tlgr.core import telethon_compat as compat
+
+
+class _Client:
+    def __init__(self) -> None:
+        self._message_box = MessageBox(logging.getLogger("test.messagebox"))
+
+
+def _too_long() -> types.updates.DifferenceTooLong:
+    return types.updates.DifferenceTooLong(pts=10)
+
+
+class TestTooLongHook:
+    def test_it_installs_on_a_real_message_box(self):
+        client = _Client()
+        assert compat.install_too_long_hook(client, lambda scope, channel: None)
+        assert isinstance(client._message_box, MessageBox)
+
+    def test_it_reports_a_global_gap(self):
+        client = _Client()
+        seen: list[tuple[str, int | None]] = []
+        compat.install_too_long_hook(client, lambda scope, channel: seen.append((scope, channel)))
+
+        # Telethon's own handling of the diff is not under test.
+        with contextlib.suppress(Exception):
+            client._message_box.apply_difference(_too_long(), None)
+
+        assert seen == [(compat.TOO_LONG_GLOBAL, None)]
+
+    def test_one_clients_hook_does_not_fire_for_another(self):
+        hooked, plain = _Client(), _Client()
+        seen: list[str] = []
+        compat.install_too_long_hook(hooked, lambda scope, channel: seen.append(scope))
+
+        with contextlib.suppress(Exception):
+            plain._message_box.apply_difference(_too_long(), None)
+
+        assert seen == []
+        assert type(plain._message_box) is MessageBox
diff --git a/tlgr/core/telethon_compat.py b/tlgr/core/telethon_compat.py
index 6d221cd..a4ed4a6 100644
--- a/tlgr/core/telethon_compat.py
+++ b/tlgr/core/telethon_compat.py
@@ -141,9 +141,14 @@ def install_too_long_hook(client: Any, callback: Callable[[str, int | None], Any
     Telethon consumes both inside `MessageBox` and delivers nothing to
     handlers, so a client that was offline long enough to blow the server's
     difference window silently skips history (checklist 9). The hook wraps
-    `MessageBox.apply_difference` / `apply_channel_difference` **on this
-    client's own box instance**, never on the class, so one account's hook
-    cannot fire for another's.
+    `MessageBox.apply_difference` / `apply_channel_difference` **for this
+    client's own box only**, so one account's hook cannot fire for another's.
+
+    `MessageBox` declares `__slots__`, so the methods cannot be replaced on
+    the instance: that raised `AttributeError: ... is read-only` on every
+    account's first connect. Instead the box is moved onto a subclass built
+    for this one client. The subclass adds no slots, so the layout matches and
+    `__class__` assignment is allowed.
 
     `callback(scope, channel_id)` is called with `TOO_LONG_GLOBAL`/`None` or
     `TOO_LONG_CHANNEL`/`<id>`.
@@ -153,38 +158,42 @@ def install_too_long_hook(client: Any, callback: Callable[[str, int | None], Any
         _warn_once("_message_box")
         return False
 
-    installed = False
+    base = type(box)
+    overrides: dict[str, Any] = {"__slots__": ()}
 
-    original_global = getattr(box, "apply_difference", None)
+    original_global = getattr(base, "apply_difference", None)
     if callable(original_global):
 
-        def apply_difference(diff: Any, chat_hashes: Any, _orig: Any = original_global) -> Any:
+        def apply_difference(self: Any, diff: Any, chat_hashes: Any) -> Any:
             if type(diff).__name__ == "DifferenceTooLong":
                 _safe(callback, TOO_LONG_GLOBAL, None)
-            return _orig(diff, chat_hashes)
+            return original_global(self, diff, chat_hashes)
 
-        box.apply_difference = apply_difference
-        installed = True
+        overrides["apply_difference"] = apply_difference
     else:  # pragma: no cover - present in 1.44
         _warn_once("MessageBox.apply_difference")
 
-    original_channel = getattr(box, "apply_channel_difference", None)
+    original_channel = getattr(base, "apply_channel_difference", None)
     if callable(original_channel):
 
-        def apply_channel_difference(
-            request: Any, diff: Any, chat_hashes: Any, _orig: Any = original_channel
-        ) -> Any:
+        def apply_channel_difference(self: Any, request: Any, diff: Any, chat_hashes: Any) -> Any:
             if type(diff).__name__ == "ChannelDifferenceTooLong":
                 channel = getattr(request, "channel", None)
                 _safe(callback, TOO_LONG_CHANNEL, getattr(channel, "channel_id", None))
-            return _orig(request, diff, chat_hashes)
+            return original_channel(self, request, diff, chat_hashes)
 
-        box.apply_channel_difference = apply_channel_difference
-        installed = True
+        overrides["apply_channel_difference"] = apply_channel_difference
     else:  # pragma: no cover - present in 1.44
         _warn_once("MessageBox.apply_channel_difference")
 
-    return installed
+    if len(overrides) == 1:
+        return False
+    try:
+        box.__class__ = type(base.__name__, (base,), overrides)
+    except TypeError as exc:
+        _warn_once("MessageBox subclassing", str(exc))
+        return False
+    return True
 
 
 def _safe(callback: Callable[..., Any], *args: Any) -> None:
diff --git a/tlgr/daemon/app.py b/tlgr/daemon/app.py
index 49eb58f..bb6e68c 100644
--- a/tlgr/daemon/app.py
+++ b/tlgr/daemon/app.py
@@ -96,7 +96,7 @@ def __init__(
             self.paths,
             self.config,
             accounts=self.accounts,
-            on_session_ready=self._attach_handlers,
+            on_client=self._attach_handlers,
             **factory_kwargs,
         )
         self.webhook_config = load_webhook_config(self.paths.base)
@@ -119,6 +119,7 @@ def __init__(
         self._start_time = time.time()
         self._runner: web.AppRunner | None = None
         self._idle_task: asyncio.Task[None] | None = None
+        self._jobs_task: asyncio.Task[None] | None = None
         self._token: str | None = None
         self.idle_timeout = effective_idle_timeout(
             self.config.daemon.idle_timeout,
@@ -141,7 +142,7 @@ def token(self) -> str | None:
     # -- session plumbing --------------------------------------------------
 
     async def _attach_handlers(self, session: Any) -> None:
-        """Feed a newly connected account's updates into the bus.
+        """Feed an account's updates into the bus, once its client is built.
 
         The handler body is deliberately tiny: normalise, number, fan out. Any
         real work happens on a bus worker lane, because with
@@ -358,9 +359,29 @@ async def run(self) -> None:
             extra={"count": len(results)},
         )
         self._idle_task = asyncio.create_task(self._idle_monitor(), name="tlgr-idle")
+        self._jobs_task = asyncio.create_task(self._start_jobs(), name="tlgr-jobs")
         await self._shutdown_event.wait()
         await self.shutdown()
 
+    async def _start_jobs(self) -> None:
+        """Load `jobs.yaml` and start every enabled job, once accounts are up.
+
+        Without this a restarted daemon ran no jobs until someone ran
+        `tlgr job reload`, which is how the gateway sat idle from a restart
+        on 2026-09-30 until it was noticed. It runs in the background because
+        each job may wait for its account to come online, and readiness must
+        not wait for that.
+        """
+        try:
+            result = await self.reload_jobs()
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            log.exception("could not start jobs from jobs.yaml")
+            return
+        if result.get("added"):
+            log.info("started %d job(s) from jobs.yaml", len(result["added"]))
+
     async def _idle_monitor(self) -> None:
         while not self._shutdown_event.is_set():
             await asyncio.sleep(5)
@@ -385,10 +406,11 @@ async def shutdown(self, *, drain: float | None = None) -> None:
             drain if drain is not None else self.config.daemon.drain_seconds
         )
 
-        if self._idle_task is not None:
-            self._idle_task.cancel()
-            with contextlib.suppress(asyncio.CancelledError, Exception):
-                await self._idle_task
+        for task in (self._idle_task, self._jobs_task):
+            if task is not None:
+                task.cancel()
+                with contextlib.suppress(asyncio.CancelledError, Exception):
+                    await task
 
         # Wait for in-flight requests rather than cancelling them: a ten
         # minute scan that is killed at second 599 has cost the account the
diff --git a/tlgr/daemon/session.py b/tlgr/daemon/session.py
index a7d6bd8..4718149 100644
--- a/tlgr/daemon/session.py
+++ b/tlgr/daemon/session.py
@@ -153,6 +153,7 @@ def __init__(
         client_factory: Callable[[Path, ClientOptions], Any] = build_client,
         on_state: Callable[[str, str, int | None], None] | None = None,
         on_event: Callable[[str, Any], None] | None = None,
+        on_client: Callable[[AccountSession], Awaitable[None]] | None = None,
         state_save_interval: int = 60,
         presence: str = "off",
         resync_depth: int = 50,
@@ -166,6 +167,7 @@ def __init__(
         self._factory = client_factory
         self._on_state = on_state
         self._on_event = on_event
+        self._on_client = on_client
         self.state_save_interval = state_save_interval
         self.presence = presence
         self.resync_depth = resync_depth
@@ -307,6 +309,7 @@ async def _connect_once(self) -> None:
         """Connect, authorise, catch up, then wait for the connection to end."""
         if self.client is None:
             self.client = self._factory(self.session_path, self.options)
+            await self._client_built()
             self._install_hooks()
 
         await self.client.connect()
@@ -371,6 +374,30 @@ async def catch_up(self) -> None:
 
     # -- hooks -------------------------------------------------------------
 
+    async def _client_built(self) -> None:
+        """Hand the new client to whoever feeds its updates into the bus.
+
+        This runs here, inside the supervisor, because the client does not
+        exist until the first connect attempt builds it. The daemon used to
+        attach its handler straight after `start()`, which only schedules the
+        supervisor, so it always found `client is None` and no incoming update
+        ever reached a job, a webhook or `watch`. The client is built once and
+        kept across reconnects, so this runs once per session.
+        """
+        if self._on_client is None:
+            return
+        try:
+            await self._on_client(self)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            log.warning(
+                "could not attach update handlers for %s: %s",
+                self.alias,
+                exc,
+                extra={"account": self.alias},
+            )
+
     def _install_hooks(self) -> None:
         compat.install_reconnect_hook(self.client, self._on_reconnect)
         compat.install_too_long_hook(self.client, self._on_too_long)
diff --git a/tlgr/daemon/sessions.py b/tlgr/daemon/sessions.py
index 4bd3e53..e51c1be 100644
--- a/tlgr/daemon/sessions.py
+++ b/tlgr/daemon/sessions.py
@@ -40,13 +40,13 @@ def __init__(
         *,
         accounts: AccountManager | None = None,
         client_factory: Any = build_client,
-        on_session_ready: Any = None,
+        on_client: Any = None,
     ) -> None:
         self.paths = paths
         self.config = config
         self.accounts = accounts or AccountManager(paths.base)
         self._factory = client_factory
-        self._on_ready = on_session_ready
+        self._on_client = on_client
         self._sessions: dict[str, AccountSession] = {}
         self._limiters: dict[str, RateLimiter] = {}
         self._locks: dict[str, asyncio.Lock] = {}
@@ -133,6 +133,7 @@ def on_state(state: str, reason: str, user_id: int | None) -> None:
             options=self._options(alias),
             client_factory=self._factory,
             on_state=on_state,
+            on_client=self._on_client,
             state_save_interval=self.config.daemon.state_save_interval,
             presence=self.config.presence.mode,
             resync_depth=self.config.daemon.resync_depth,
@@ -160,9 +161,6 @@ async def ensure(self, alias: str) -> AccountSession:
             session = self._make(alias)
             await session.start()
             self._sessions[alias] = session
-            if self._on_ready is not None:
-                with contextlib.suppress(Exception):
-                    await self._on_ready(session)
             return session
 
     async def connect_all(self, aliases: list[str]) -> dict[str, str]:
diff --git a/tlgr/gateway/engine.py b/tlgr/gateway/engine.py
index 3abb83e..b6447f2 100644
--- a/tlgr/gateway/engine.py
+++ b/tlgr/gateway/engine.py
@@ -10,9 +10,13 @@
 order is preserved, the update loop is never blocked, and a job that falls
 behind is bounded rather than unbounded.
 
-The pipeline itself is unchanged, and filters still read the raw Telethon
-event, which is why the bus carries it beside the normalised envelope. The
-full move to model-based filters belongs to the updates group (PR-4).
+The pipeline itself is unchanged, and filters still read a high-level
+Telethon event (`NewMessage.Event` and friends). The bus carries the raw TL
+update beside the normalised envelope, so the job builds that event itself
+with the same builder a Telethon handler would have used, including
+`NewMessage(incoming=True)`, so a job never answers the account's own
+messages. The full move to model-based filters belongs to the updates group
+(PR-4).
 
 Without a bus — a unit test, or a daemon that has not started one — the job
 falls back to registering Telethon handlers exactly as v1 did.
@@ -21,12 +25,15 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import logging
+import time
+from typing import Any
 
 from telethon import events
 
 from tlgr.actions import get_action
-from tlgr.filters.compose import evaluate
+from tlgr.filters.compose import FilterNode, Op, evaluate
 from tlgr.gateway.config import ActionConfig, GatewayConfig
 from tlgr.gateway.event import Event
 from tlgr.jobs.base import BaseJob
@@ -34,6 +41,10 @@
 
 log = logging.getLogger("tlgr.gateway")
 
+#: How long a job waits before retrying a `@chat` ref it could not resolve,
+#: typically because its account was still offline when the job started.
+_RESOLVE_RETRY_SECONDS = 60.0
+
 
 class _GatewayJobConfig:
     """Minimal shim so Gateway can sit on top of BaseJob.
@@ -49,8 +60,8 @@ def __init__(self, gw: GatewayConfig) -> None:
         self.account = gw.account
 
 
-#: v1's job event names → the bus taxonomy, and back for the pipeline, which
-#: still labels envelopes with v1's names. The expansion is the taxonomy's own
+#: v1's job event names → the bus taxonomy. The pipeline still labels
+#: envelopes with v1's names, taken from the builder that matched. The expansion is the taxonomy's own
 #: alias table (`core.eventtypes.ALIASES`), so a job and a `watch` accept the
 #: same words; a job may also name any v2 type directly.
 def _bus_types(names: list[str]) -> set[str]:
@@ -62,16 +73,6 @@ def _bus_types(names: list[str]) -> set[str]:
     return wanted
 
 
-_V1_TYPE_MAP = {
-    "message_new": "new_message",
-    "message_edited": "message_edited",
-    "message_deleted": "message_deleted",
-    "message_service": "chat_action",
-    "user_status": "user_joined",
-    "read_inbox": "message_read",
-    "read_outbox": "message_read",
-}
-
 _EVENT_TYPE_MAP = {
     "new_message": (events.NewMessage, {}),
     "message_edited": (events.MessageEdited, {}),
@@ -82,6 +83,49 @@ def _bus_types(names: list[str]) -> set[str]:
 }
 
 
+def _builder_for(event_type_name: str) -> Any | None:
+    mapping = _EVENT_TYPE_MAP.get(event_type_name)
+    if not mapping:
+        return None
+    event_cls, kwargs = mapping
+    if event_type_name == "new_message":
+        kwargs = {"incoming": True}
+    return event_cls(**kwargs)
+
+
+def _builders(names: list[str]) -> list[tuple[str, Any]]:
+    """One Telethon event builder per v1 event name, in the job's order."""
+    out: list[tuple[str, Any]] = []
+    for name in names:
+        builder = _builder_for(name)
+        if builder is not None:
+            out.append((name, builder))
+    return out
+
+
+def _needs_resolving(ref: Any) -> bool:
+    if not isinstance(ref, str):
+        return False
+    try:
+        int(ref)
+    except ValueError:
+        return True
+    return False
+
+
+def _chat_id_leaves(nodes: list[FilterNode | None]) -> list[FilterNode]:
+    leaves: list[FilterNode] = []
+    stack = [node for node in nodes if node is not None]
+    while stack:
+        node = stack.pop()
+        if node.op is Op.LEAF:
+            if node.filter_name == "chat_id":
+                leaves.append(node)
+        else:
+            stack.extend(node.children)
+    return leaves
+
+
 class Gateway(BaseJob):
     """Generic pipeline job: filters -> processors -> actions."""
 
@@ -99,8 +143,12 @@ def __init__(
         self._bus = bus
         self._bus_handler = None
         self._stats: dict[str, int] = {"matched": 0, "skipped": 0, "errors": 0}
+        self._refs_pending = False
+        self._next_resolve = 0.0
+        self._unresolved_logged: set[str] = set()
 
     async def setup(self) -> None:
+        await self._resolve_chat_refs()
         log.info(
             "[%s] events=%s filters=%s actions=%s",
             self.name,
@@ -109,6 +157,35 @@ async def setup(self) -> None:
             [a.name for a in self._gw.actions],
         )
 
+    async def _resolve_chat_refs(self) -> None:
+        """Turn every `chat_id: "@name"` into the marked id the filter compares.
+
+        `filter_chat_id` compares `event.chat_id`, an int, and skips string
+        refs on the promise that the gateway resolves them first. Nothing
+        did, so a job filtered on `@channel` could never match. A ref that
+        fails (the account is offline at boot, say) stays a string and is
+        retried from `_handle`, at most once a minute.
+        """
+        nodes = [self._gw.filters] + [action.filters for action in self._gw.actions]
+        self._refs_pending = False
+        self._next_resolve = time.monotonic() + _RESOLVE_RETRY_SECONDS
+        for leaf in _chat_id_leaves(nodes):
+            refs = leaf.filter_value if isinstance(leaf.filter_value, list) else [leaf.filter_value]
+            resolved: list = []
+            for ref in refs:
+                if not _needs_resolving(ref):
+                    resolved.append(ref)
+                    continue
+                try:
+                    resolved.append(await self.client.resolve_chat(str(ref)))
+                except Exception as exc:
+                    level = logging.DEBUG if ref in self._unresolved_logged else logging.WARNING
+                    self._unresolved_logged.add(ref)
+                    log.log(level, "[%s] cannot resolve chat %s: %s", self.name, ref, exc)
+                    self._refs_pending = True
+                    resolved.append(ref)
+            leaf.filter_value = resolved if isinstance(leaf.filter_value, list) else resolved[0]
+
     async def run(self) -> None:
         if self._bus is not None:
             await self._run_on_bus()
@@ -119,6 +196,7 @@ async def _run_on_bus(self) -> None:
         """Subscribe to the daemon's bus instead of the update loop (ROB-02)."""
         wanted = _bus_types(list(self._gw.events))
         account = self._gw.account
+        builders = _builders(list(self._gw.events))
 
         async def on_event(envelope, raw) -> None:
             if envelope.type not in wanted:
@@ -129,7 +207,11 @@ async def on_event(envelope, raw) -> None:
                 # A self-origin echo or a synthesised event: the filters read
                 # the raw Telethon object, so there is nothing to evaluate.
                 return
-            await self._handle(raw, _V1_TYPE_MAP.get(envelope.type, envelope.type))
+            built = await self._build(builders, raw)
+            if built is None:
+                return
+            tg_event, event_type = built
+            await self._handle(tg_event, event_type)
 
         self._bus_handler = on_event
         self._bus.add_handler(on_event)
@@ -138,19 +220,42 @@ async def on_event(envelope, raw) -> None:
         except asyncio.CancelledError:
             raise
 
+    async def _build(self, builders: list[tuple[str, Any]], update: Any) -> tuple[Any, str] | None:
+        """The high-level event Telethon would have handed a handler, or None.
+
+        This mirrors `EventBuilderDict` in Telethon's `_dispatch_update`: build,
+        attach the update's entities and the client, then apply the builder's
+        own filter, which is where `incoming=True` drops outgoing messages.
+        """
+        client = self.client.client
+        self_id = getattr(client, "_self_id", None)
+        for event_type, builder in builders:
+            event = builder.build(update, None, self_id)
+            if not event:
+                continue
+            if isinstance(event, events.common.EventCommon):
+                event.original_update = update
+                event._entities = getattr(update, "_entities", None) or {}
+                event._set_client(client)
+            if not builder.resolved:
+                await builder.resolve(client)
+            passed = builder.filter(event)
+            if inspect.isawaitable(passed):
+                passed = await passed
+            if passed:
+                return event, event_type
+        return None
+
     async def _run_on_client(self) -> None:
         for event_type_name in self._gw.events:
-            mapping = _EVENT_TYPE_MAP.get(event_type_name)
-            if not mapping:
+            builder = _builder_for(event_type_name)
+            if builder is None:
                 log.warning("[%s] unknown event type: %s", self.name, event_type_name)
                 continue
-            event_cls, kwargs = mapping
-            if event_type_name == "new_message":
-                kwargs = {"incoming": True}
 
             et = event_type_name
 
-            @self.client.client.on(event_cls(**kwargs))
+            @self.client.client.on(builder)
             async def handler(tg_event, _et=et):
                 await self._handle(tg_event, _et)
 
@@ -177,6 +282,9 @@ async def teardown(self) -> None:
         )
 
     async def _handle(self, tg_event, event_type: str = "new_message") -> None:
+        if self._refs_pending and time.monotonic() >= self._next_resolve:
+            await self._resolve_chat_refs()
+
         envelope = Event(
             source="telegram",
             raw=tg_event,
diff --git a/tlgr/models/daemon.py b/tlgr/models/daemon.py
index 8eeba03..20564d5 100644
--- a/tlgr/models/daemon.py
+++ b/tlgr/models/daemon.py
@@ -250,8 +250,13 @@ class DeadLetterResult(Model):
     dry_run: bool = False
 
 
-class JobState(Model):
-    """One gateway job."""
+class JobState(Model, omit_defaults=False):
+    """One gateway job.
+
+    `omit_defaults=False` because `enabled` and `running` are the answer:
+    with the default, an enabled job dropped `enabled` and a stopped one
+    dropped `running`, and `job list` printed `-` for both.
+    """
 
     name: str
     account: str = ""