feat(work-orders): synchronize peripheral changes through events

docs/operational-task-workers.md

@@ -79,9 +79,9 @@ conda run -n server python scripts/sync_peripheral_work_orders.py --once
 conda run -n server python scripts/sync_peripheral_work_orders.py
 ```
 
-使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器读取分页工单，将外围版本投影到 `external_work_order` 任务。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。
+使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器订阅源工单变更事件，收到消息即读取分页工单，将外围版本投影到 `external_work_order` 任务；默认每 300 秒兜底对账。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。事件协议、`tjwater_v2` 范围与启用步骤见 [外围工单主动同步](peripheral-work-order-events.md)。
 
-工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单仍由独立 worker 使用服务身份周期同步，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与 worker 固定项目不一致。
+工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单由独立同步器使用服务身份响应后端事件，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与同步器固定项目不一致。
 
 通用合成检查工单不属于预约任务。早期种子脚本创建的通信、压力和零流量检查工单已经退役，种子脚本不再恢复它们；同步器按不可编辑的专用 seed actor 跳过旧种子工单。投诉、爆管事件以及调查动作实际创建的现场工单继续同步。`018_remove_synthetic_check_task_projections.sql` 只清理专用 seed actor 创建的旧检查投影，不按用户可编辑的标题或状态删除任务。三张独立演示工单已迁入外围样板，由同一同步器维护。
 

docs/peripheral-work-order-events.md

@@ -0,0 +1,44 @@
+# 外围工单主动同步
+
+工单的创建、编辑、现场反馈和删除改为由来源事件触发同步。浏览器继续使用已有工作台 SSE，不需要新增连接或定时工单轮询。
+
+```text
+外围工单写入
+  → 同一 PostgreSQL 事务保存源记录、历史版本和变更事件
+  → COMMIT 后 NOTIFY
+  → Server 工作台 SSE 唤醒外围同步器
+  → 同步器通过授权 API 更新任务投影
+  → task/action/notification 事件推送给浏览器
+```
+
+## 数据库范围
+
+本次仅启用 `tjwater_v2`。没有新增表、列或迁移；复用 `workbench.domain_event_log` 的 `workbench.work_order.changed` topic。源事件的聚合类型为 `peripheral_work_order`，元数据只包含 `source=peripheral_sample`、工单 ID 和 `created|updated|deleted`，不包含工单正文。事务失败时，源记录、历史、事件及通知一起回滚；同一请求的幂等重放不再产生事件。
+
+外围服务只写源记录及源变更事实，不直接改业务任务。任务投影、调查创建和通知仍通过 Server 状态机和现有授权 API 处理。其他数据库不写入该源事件；扩展前遵循 [数据库变更门禁](workbench-database-rollout.md)。事件表缺失或无写权限时，`tjwater_v2` 的工单写入返回存储错误，不会假装成功。
+
+## 同步器
+
+从 Server 仓库运行已有命令：
+
+```bash
+conda run -n server python scripts/sync_peripheral_work_orders.py
+conda run -n server python scripts/sync_peripheral_work_orders.py --once
+```
+
+继续复用 `OPERATIONAL_WORKER_*` OAuth 工作身份及 `OPERATIONAL_ACTION_PERIPHERAL_URL`，不需要新账号或密钥。启动和游标过期时先读取事件游标，再进行全量对账；扫描期间的事件从该游标继续补读。相同源版本仍由原幂等键去重。同步器只被 `peripheral_work_order` 源事件唤醒，不被自己写出的任务投影事件再次唤醒。
+
+正常状态收到源事件即扫描；无事件时默认每 300 秒对账一次，处理漏通知、断线、旧批量脚本直接写库及重启。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前清除唤醒标记，扫描期间新到达的事件会触发下一轮，失败后的重试也不依赖新消息。
+
+浏览器收到源事件时不会提前刷新业务快照，而是等待任务投影事件后读取对应任务。工况手动刷新仍只读 Server 权威快照。工单事件不等于 Header 通知；只有需要处理的业务待办或告警才进入通知列表。
+
+## 验证与启用
+
+```bash
+conda run -n server python -m pytest tests/unit/test_peripheral_push_sync.py tests/unit/test_peripheral_work_order_events.py tests/unit/test_peripheral_task_sync.py -q
+PERIPHERAL_SAMPLE_LIVE_TEST=1 conda run -n server python -m pytest tests/integration/test_peripheral_work_order_events.py -q
+```
+
+数据库测试强制限定 `tjwater_v2`，使用随机项目键和回滚事务，不保留测试工单或事件。覆盖创建、修改、反馈、删除、幂等重放及回滚无事件。HTTP 隔离测试覆盖源事件即时投影、游标重置、断线续传、失败重试和监听取消。
+
+上线时更新外围服务、Server 同步器及前端，再按既有发布流程重启对应进程。提交代码不会自动重启现有服务；旧同步器进程仍按旧的 30 秒周期运行。回退应用代码不涉及数据结构回滚，已有事件仍可正常补读；回退到旧同步器后恢复周期对账。

samples/peripheral/store.py

@@ -1,4 +1,4 @@
-"""PostgreSQL persistence isolated to peripheral_sample, never existing business writes."""
+"""Peripheral persistence and transactional work-order change signals."""
 from contextlib import contextmanager
 from copy import deepcopy
 from datetime import datetime, timedelta, timezone
@@ -306,8 +306,31 @@ class PostgresStore:
         conn.execute("""INSERT INTO peripheral_sample.work_order_revisions
             (project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""",
             (who.project_id, document["work_order_id"], document["revision"], Jsonb(document)))
+        self._work_order_event(conn, who, document)
         return document
 
+    def _work_order_event(self, conn, who: Identity, document: dict[str, Any], *, deleted: bool = False) -> None:
+        # This rollout is restricted to the provisioned tjwater_v2 event log.
+        # Task projections remain owned by the Server's authenticated APIs.
+        if self.settings.db_name != "tjwater_v2":
+            return
+        metadata = {
+            "source": "peripheral_sample",
+            "work_order_id": document["work_order_id"],
+            "change": "deleted" if deleted else "created" if document["revision"] == 1 else "updated",
+        }
+        row = conn.execute(
+            """INSERT INTO workbench.domain_event_log
+               (project_id,event_id,topic,aggregate_type,aggregate_id,
+                aggregate_revision,occurred_at,metadata)
+               VALUES (%s,%s,'workbench.work_order.changed','peripheral_work_order',%s,%s,%s,%s)
+               RETURNING event_seq""",
+            (who.project_id, uuid4(), document["work_order_id"], document["revision"],
+             document["updated_at"], Jsonb(metadata)),
+        ).fetchone()
+        conn.execute("SELECT pg_notify('workbench_events', %s)",
+                     (f"{who.project_id}:{row['event_seq']}",))
+
     def list_work_orders(self, who: Identity, limit: int, cursor: UUID | None = None) -> dict[str, Any]:
         return self._list_managed(who, "work_orders", "work_order_id", limit, cursor)
 
@@ -385,6 +408,7 @@ class PostgresStore:
             conn.execute("""INSERT INTO peripheral_sample.work_order_revisions
                 (project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""",
                 (who.project_id, work_order_id, revision, Jsonb(tombstone)))
+            self._work_order_event(conn, who, tombstone, deleted=True)
             return result
         return self._mutate(who, f"delete_work_order:{work_order_id}", command, write)
 

scripts/sync_peripheral_work_orders.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import asyncio
+import json
 import logging
 import os
 import sys
@@ -15,6 +16,7 @@ import httpx
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from samples.peripheral.seed_identity import SEED_ACTOR_ID
+from app.domain.schemas.workbench_events import WorkbenchEvent
 
 try:
     from scripts.run_operational_worker import CredentialProvider, WorkerSettings
@@ -211,30 +213,125 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe
     return synchronized
 
 
-async def run(once: bool):
+class PeripheralWorkOrderSynchronizer:
+    """Wake on committed source events; periodically reconcile missed changes."""
+
+    def __init__(self, client: httpx.AsyncClient, settings: WorkerSettings,
+                 peripheral_url: str, credentials: CredentialProvider) -> None:
+        self.client = client
+        self.settings = settings
+        self.peripheral_url = peripheral_url
+        self.credentials = credentials
+        self.wake = asyncio.Event()
+        self.cursor: int | None = None
+
+    async def _reset_cursor(self, headers: dict[str, str]) -> None:
+        response = await self.client.get(
+            f"{self.settings.server_url.rstrip('/')}/api/v1/workbench/sync-snapshot",
+            headers=headers,
+        )
+        response.raise_for_status()
+        cursor = response.json()["event_cursor"]
+        if not isinstance(cursor, int) or isinstance(cursor, bool) or cursor < 0:
+            raise ValueError("invalid workbench event cursor")
+        self.cursor = cursor
+        # Capture the cursor before scanning sources, so updates during the
+        # initial/reset scan are replayed by the following SSE connection.
+        self.wake.set()
+
+    async def listen(self) -> None:
+        while True:
+            try:
+                headers = await self.credentials.headers(self.client)
+                if self.cursor is None:
+                    await self._reset_cursor(headers)
+                async with self.client.stream(
+                    "GET", f"{self.settings.server_url.rstrip('/')}/api/v1/workbench/events",
+                    headers={**headers, "Accept": "text/event-stream"},
+                    params={"after": self.cursor}, timeout=None,
+                ) as response:
+                    response.raise_for_status()
+                    event_name = ""
+                    async for line in response.aiter_lines():
+                        if line.startswith("event:"):
+                            event_name = line[6:].strip()
+                        elif line.startswith("data:") and event_name == "reset_required":
+                            await self._reset_cursor(headers)
+                            break
+                        elif line.startswith("data:") and event_name == "workbench_event":
+                            event = WorkbenchEvent.model_validate(json.loads(line[5:].strip()))
+                            if str(event.project_id) != self.settings.project_id:
+                                raise ValueError("workbench event project mismatch")
+                            if event.event_seq <= self.cursor:
+                                continue
+                            self.cursor = event.event_seq
+                            if (event.topic == "workbench.work_order.changed"
+                                    and event.aggregate_type == "peripheral_work_order"
+                                    and event.metadata.get("source") == "peripheral_sample"):
+                                self.wake.set()
+                        elif not line:
+                            event_name = ""
+                await asyncio.sleep(1)
+            except (httpx.HTTPError, ValueError, KeyError):
+                logger.warning("Work-order event stream disconnected; reconciliation remains active")
+                await asyncio.sleep(5)
+
+    async def serve(self, reconcile_seconds: int = 300) -> None:
+        if reconcile_seconds < 1:
+            raise ValueError("reconciliation interval must be at least one second")
+        listener = asyncio.create_task(self.listen())
+        wait_seconds = reconcile_seconds
+        retry_seconds = 5
+        try:
+            while True:
+                try:
+                    await asyncio.wait_for(self.wake.wait(), timeout=wait_seconds)
+                except TimeoutError:
+                    pass
+                # Clear before work, so arrivals during a scan request another
+                # pass. A failed scan gets a retry even after its cursor advances.
+                self.wake.clear()
+                try:
+                    count = await sync_once(
+                        self.client, self.settings, self.peripheral_url, self.credentials,
+                    )
+                    logger.info("Peripheral progress synchronization completed: %d revisions", count)
+                    wait_seconds, retry_seconds = reconcile_seconds, 5
+                except (httpx.HTTPError, ValueError, KeyError):
+                    logger.error("Peripheral synchronization failed; persisted revisions will be retried")
+                    wait_seconds = retry_seconds
+                    retry_seconds = min(retry_seconds * 2, 60)
+        finally:
+            listener.cancel()
+            await asyncio.gather(listener, return_exceptions=True)
+
+
+async def run(once: bool, reconcile_seconds: int = 300) -> None:
     settings = WorkerSettings.from_environment()
     peripheral_url = os.environ["OPERATIONAL_ACTION_PERIPHERAL_URL"]
     credentials = CredentialProvider(settings)
     async with httpx.AsyncClient(timeout=30, follow_redirects=False, trust_env=False) as client:
-        while True:
+        if once:
             try:
                 count = await sync_once(client, settings, peripheral_url, credentials)
                 logger.info("Peripheral progress synchronization completed: %d revisions", count)
             except (httpx.HTTPError, ValueError, KeyError):
-                logger.error("Peripheral synchronization failed; next tick will reconcile persisted revisions")
-                if once:
-                    raise SystemExit(1) from None
-            if once:
-                return
-            await asyncio.sleep(settings.interval_seconds)
+                logger.error("Peripheral synchronization failed; persisted revisions remain available")
+                raise SystemExit(1) from None
+            return
+        await PeripheralWorkOrderSynchronizer(client, settings, peripheral_url, credentials).serve(reconcile_seconds)
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--once", action="store_true")
+    parser.add_argument("--reconcile-seconds", type=int, default=300,
+                        help="missed-event reconciliation interval (default: 300 seconds)")
     args = parser.parse_args()
+    if args.reconcile_seconds < 1:
+        parser.error("--reconcile-seconds must be at least 1")
     logging.basicConfig(level=logging.INFO)
     try:
-        asyncio.run(run(args.once))
+        asyncio.run(run(args.once, args.reconcile_seconds))
     except KeyboardInterrupt:
         pass

tests/integration/test_peripheral_sample_live.py

@@ -1,4 +1,4 @@
-"""Opt-in checks against provisioned sample tables, never business-table writes."""
+"""Opt-in checks against provisioned sample tables and transactional source events."""
 from concurrent.futures import ThreadPoolExecutor
 import os
 import subprocess
@@ -39,6 +39,10 @@ def sample():
     yield store, who
     # Only this test's unguessable project key; tables created by the authorized setup.
     with store.connection() as conn:
+        if store.settings.db_name == "tjwater_v2" and conn.execute(
+            "SELECT to_regclass('workbench.domain_event_log') AS name"
+        ).fetchone()["name"] is not None:
+            conn.execute("DELETE FROM workbench.domain_event_log WHERE project_id=%s", (who.project_id,))
         for table in (
             "work_order_revisions", "test_data_record_revisions", "work_orders",
             "test_data_records", "feedback", "requests", "revisions", "tasks", "scenarios",

tests/integration/test_peripheral_work_order_events.py

@@ -0,0 +1,84 @@
+"""tjwater_v2 source/event atomicity checks; every write is rolled back."""
+from contextlib import contextmanager
+import os
+from uuid import uuid4
+
+import pytest
+
+from samples.peripheral.auth import Identity
+from samples.peripheral.config import SampleSettings
+from samples.peripheral.errors import SampleFailure
+from samples.peripheral.schemas import CreateWorkOrder, DeleteResource, UpdateWorkOrder
+from samples.peripheral.store import PostgresStore
+
+
+pytestmark = pytest.mark.skipif(
+    os.getenv("PERIPHERAL_SAMPLE_LIVE_TEST") != "1", reason="explicit live sample opt-in required",
+)
+
+
+def test_work_order_mutations_and_events_are_atomic_idempotent_and_rollback_cleanly(monkeypatch):
+    settings = SampleSettings()
+    assert settings.db_name == "tjwater_v2", "event verification is restricted to tjwater_v2"
+    store = PostgresStore(settings)
+    who = Identity(str(uuid4()), str(uuid4()), True)
+    with store.connection() as conn:
+        @contextmanager
+        def connection(*, readonly=False, timeseries=False):
+            assert timeseries is False
+            with conn.transaction():
+                yield conn
+
+        monkeypatch.setattr(store, "connection", connection)
+        with conn.transaction(force_rollback=True):
+            command = CreateWorkOrder(
+                request_id=uuid4(), title="AUTOTEST push work order", summary="isolated transaction",
+            )
+            created = store.create_work_order(who, command)
+            assert store.create_work_order(who, command) == created
+            order_id = created["work_order_id"]
+            store.update_work_order(who, order_id, UpdateWorkOrder(
+                request_id=uuid4(), expected_revision=1, status="in_progress",
+            ))
+            store.update_work_order(who, order_id, UpdateWorkOrder(
+                request_id=uuid4(), expected_revision=2, progress_note="AUTOTEST field feedback",
+            ))
+            with pytest.raises(SampleFailure) as stale:
+                store.update_work_order(who, order_id, UpdateWorkOrder(
+                    request_id=uuid4(), expected_revision=1, status="completed",
+                ))
+            assert stale.value.code == "REVISION_CONFLICT"
+            store.delete_work_order(who, order_id, DeleteResource(
+                request_id=uuid4(), expected_revision=3,
+            ))
+            rows = conn.execute(
+                """SELECT project_id,topic,aggregate_type,aggregate_id,aggregate_revision,metadata
+                   FROM workbench.domain_event_log WHERE project_id=%s ORDER BY event_seq""",
+                (who.project_id,),
+            ).fetchall()
+            assert len(rows) == 4
+            assert [row["aggregate_revision"] for row in rows] == [1, 2, 3, 4]
+            assert [row["metadata"]["change"] for row in rows] == ["created", "updated", "updated", "deleted"]
+            assert all(row["topic"] == "workbench.work_order.changed"
+                       and row["aggregate_type"] == "peripheral_work_order"
+                       and row["aggregate_id"] == order_id for row in rows)
+            assert store.list_work_orders(who, 20)["items"] == []
+
+            original = store._work_order_event
+
+            def fail_after_event(*args, **kwargs):
+                original(*args, **kwargs)
+                raise RuntimeError("AUTOTEST transaction failure")
+
+            monkeypatch.setattr(store, "_work_order_event", fail_after_event)
+            with pytest.raises(RuntimeError, match="transaction failure"):
+                store.create_work_order(who, command.model_copy(update={"request_id": uuid4()}))
+            assert store.list_work_orders(who, 20)["items"] == []
+            assert conn.execute(
+                "SELECT count(*) AS n FROM workbench.domain_event_log WHERE project_id=%s", (who.project_id,),
+            ).fetchone()["n"] == 4
+
+        for table in ("workbench.domain_event_log", "peripheral_sample.work_orders",
+                      "peripheral_sample.work_order_revisions", "peripheral_sample.requests"):
+            assert conn.execute(f"SELECT count(*) AS n FROM {table} WHERE project_id=%s",
+                                (who.project_id,)).fetchone()["n"] == 0

tests/unit/test_peripheral_push_sync.py

@@ -0,0 +1,284 @@
+import asyncio
+import json
+from uuid import uuid4
+
+import httpx
+import pytest
+
+from scripts.run_operational_worker import CredentialProvider, WorkerSettings
+from scripts import sync_peripheral_work_orders as sync_module
+from scripts.sync_peripheral_work_orders import PeripheralWorkOrderSynchronizer
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+class PushStream(httpx.AsyncByteStream):
+    def __init__(self):
+        self.messages = asyncio.Queue()
+        self.closed = False
+
+    async def __aiter__(self):
+        while True:
+            yield await self.messages.get()
+            self.messages.task_done()
+
+    async def aclose(self):
+        self.closed = True
+
+    def publish(self, project_id, order_id, sequence, *, aggregate_type="peripheral_work_order"):
+        event = {
+            "schema": "workbench-event@1", "event_seq": sequence,
+            "event_id": str(uuid4()), "project_id": project_id,
+            "topic": "workbench.work_order.changed", "aggregate_type": aggregate_type,
+            "aggregate_id": order_id, "aggregate_revision": 1,
+            "occurred_at": "2026-10-08T00:00:00Z",
+            "metadata": {"source": "peripheral_sample", "work_order_id": order_id},
+        }
+        self.messages.put_nowait(
+            f"id: {sequence}\nevent: workbench_event\ndata: {json.dumps(event)}\n\n".encode()
+        )
+
+
+@pytest.mark.anyio
+async def test_source_push_synchronizes_new_and_updated_orders_without_waiting_for_reconciliation():
+    project_id, order_id, task_id = map(str, (uuid4(), uuid4(), uuid4()))
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+    orders, tasks, cursors = [], [], []
+    initial_scan, projected = asyncio.Event(), asyncio.Event()
+    scan_count = 0
+
+    def handler(request):
+        nonlocal scan_count
+        assert request.headers["X-Project-Id"] == project_id
+        assert request.headers["Authorization"] == "Bearer TEST_ONLY"
+        if request.url.path.endswith("sync-snapshot"):
+            return httpx.Response(200, json={"event_cursor": 20})
+        if request.url.path.endswith("/events"):
+            cursors.append(request.url.params["after"])
+            return httpx.Response(200, headers={"Content-Type": "text/event-stream"}, stream=stream)
+        if request.url.host == "sample.test":
+            scan_count += 1
+            initial_scan.set()
+            return httpx.Response(200, json={"items": orders, "next_cursor": None})
+        if request.method == "GET":
+            return httpx.Response(200, json={"items": tasks, "next_offset": None})
+        body = json.loads(request.content)
+        if request.url.path.endswith("external-feedback"):
+            tasks[0]["source"]["external_revision"] = body["source_revision"]
+            tasks[0]["status"] = body["status"]
+            projected.set()
+            return httpx.Response(200, json={"task": tasks[0]})
+        task = {**body, "task_id": task_id}
+        tasks.append(task)
+        return httpx.Response(201, json=task)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        runner = asyncio.create_task(synchronizer.serve())
+        try:
+            await asyncio.wait_for(initial_scan.wait(), timeout=3)
+            orders.append({
+                "project_id": project_id, "work_order_id": order_id, "origin": "synthetic",
+                "revision": 1, "title": "预约维修", "summary": "等待现场处理", "priority": "normal",
+                "status": "open", "created_at": "2026-10-08T00:00:00Z",
+                "updated_at": "2026-10-08T00:00:00Z", "scheduled_at": None,
+                "work_order_type": "general",
+            })
+            stream.publish(project_id, order_id, 21)
+            await asyncio.wait_for(projected.wait(), timeout=3)
+            assert tasks[0]["source"]["external_revision"] == 1
+            projected.clear()
+            orders[0].update(revision=2, status="in_progress")
+            stream.publish(project_id, order_id, 22)
+            await asyncio.wait_for(projected.wait(), timeout=3)
+            assert tasks[0]["source"]["external_revision"] == 2
+            assert tasks[0]["status"] == "in_progress"
+            assert scan_count == 3  # Startup plus two pushed source changes.
+            assert cursors == ["20"]
+        finally:
+            runner.cancel()
+            await asyncio.gather(runner, return_exceptions=True)
+        assert stream.closed
+
+
+@pytest.mark.anyio
+async def test_expired_cursor_rebuilds_baseline_and_wakes_source_reconciliation():
+    project_id = str(uuid4())
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+    refreshed = asyncio.Event()
+    baselines = iter([10, 30])
+    cursors = []
+
+    def handler(request):
+        if request.url.path.endswith("sync-snapshot"):
+            cursor = next(baselines)
+            if cursor == 30:
+                refreshed.set()
+            return httpx.Response(200, json={"event_cursor": cursor})
+        cursors.append(request.url.params["after"])
+        return httpx.Response(200, stream=stream)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        listener = asyncio.create_task(synchronizer.listen())
+        try:
+            await asyncio.wait_for(synchronizer.wake.wait(), timeout=3)
+            synchronizer.wake.clear()
+            stream.messages.put_nowait(b"event: reset_required\ndata: {}\n\n")
+            await asyncio.wait_for(refreshed.wait(), timeout=3)
+            assert synchronizer.wake.is_set()
+            assert synchronizer.cursor == 30
+            assert cursors == ["10"]
+        finally:
+            listener.cancel()
+            await asyncio.gather(listener, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_projection_events_and_duplicate_source_events_do_not_trigger_extra_scans():
+    project_id, order_id = str(uuid4()), str(uuid4())
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+
+    def handler(request):
+        if request.url.path.endswith("sync-snapshot"):
+            return httpx.Response(200, json={"event_cursor": 20})
+        return httpx.Response(200, stream=stream)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        listener = asyncio.create_task(synchronizer.listen())
+        try:
+            await asyncio.wait_for(synchronizer.wake.wait(), timeout=3)
+            synchronizer.wake.clear()
+            stream.publish(project_id, order_id, 21, aggregate_type="operational_action")
+            await asyncio.wait_for(stream.messages.join(), timeout=3)
+            assert not synchronizer.wake.is_set()
+            stream.publish(project_id, order_id, 22)
+            await asyncio.wait_for(stream.messages.join(), timeout=3)
+            assert synchronizer.wake.is_set()
+            synchronizer.wake.clear()
+            stream.publish(project_id, order_id, 22)
+            await asyncio.wait_for(stream.messages.join(), timeout=3)
+            assert not synchronizer.wake.is_set()
+        finally:
+            listener.cancel()
+            await asyncio.gather(listener, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_disconnect_resumes_after_last_consumed_event():
+    project_id, order_id = str(uuid4()), str(uuid4())
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+    stream.publish(project_id, order_id, 21)
+    first_message = stream.messages.get_nowait()
+    stream.messages.task_done()
+    reconnected, cursors = asyncio.Event(), []
+
+    def handler(request):
+        if request.url.path.endswith("sync-snapshot"):
+            return httpx.Response(200, json={"event_cursor": 20})
+        cursors.append(request.url.params["after"])
+        if len(cursors) == 1:
+            return httpx.Response(200, content=first_message)
+        reconnected.set()
+        return httpx.Response(200, stream=stream)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        listener = asyncio.create_task(synchronizer.listen())
+        try:
+            await asyncio.wait_for(reconnected.wait(), timeout=3)
+            assert cursors == ["20", "21"]
+        finally:
+            listener.cancel()
+            await asyncio.gather(listener, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_failed_scan_retries_without_requiring_another_source_message(monkeypatch):
+    project_id = str(uuid4())
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+    recovered, attempts = asyncio.Event(), 0
+
+    def handler(request):
+        if request.url.path.endswith("sync-snapshot"):
+            return httpx.Response(200, json={"event_cursor": 20})
+        return httpx.Response(200, stream=stream)
+
+    async def sync_once(*_args):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise httpx.ConnectError("AUTOTEST temporary source outage")
+        recovered.set()
+        return 0
+
+    monkeypatch.setattr(sync_module, "sync_once", sync_once)
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        runner = asyncio.create_task(synchronizer.serve())
+        try:
+            await asyncio.wait_for(recovered.wait(), timeout=8)
+            assert attempts == 2
+        finally:
+            runner.cancel()
+            await asyncio.gather(runner, return_exceptions=True)
+
+
+@pytest.mark.anyio
+async def test_source_event_during_scan_schedules_another_pass(monkeypatch):
+    project_id, order_id = str(uuid4()), str(uuid4())
+    settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY")
+    stream = PushStream()
+    scanning, release, caught_up = asyncio.Event(), asyncio.Event(), asyncio.Event()
+    attempts = 0
+
+    def handler(request):
+        if request.url.path.endswith("sync-snapshot"):
+            return httpx.Response(200, json={"event_cursor": 20})
+        return httpx.Response(200, stream=stream)
+
+    async def sync_once(*_args):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            scanning.set()
+            await release.wait()
+        else:
+            caught_up.set()
+        return 0
+
+    monkeypatch.setattr(sync_module, "sync_once", sync_once)
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        synchronizer = PeripheralWorkOrderSynchronizer(
+            client, settings, "http://sample.test", CredentialProvider(settings),
+        )
+        runner = asyncio.create_task(synchronizer.serve())
+        try:
+            await asyncio.wait_for(scanning.wait(), timeout=3)
+            stream.publish(project_id, order_id, 21)
+            await asyncio.wait_for(stream.messages.join(), timeout=3)
+            release.set()
+            await asyncio.wait_for(caught_up.wait(), timeout=3)
+            assert attempts == 2
+        finally:
+            runner.cancel()
+            await asyncio.gather(runner, return_exceptions=True)

tests/unit/test_peripheral_work_order_events.py

@@ -0,0 +1,50 @@
+from uuid import uuid4
+
+import pytest
+
+from samples.peripheral.auth import Identity
+from samples.peripheral.config import SampleSettings
+from samples.peripheral.store import PostgresStore
+
+
+class EventConnection:
+    def __init__(self):
+        self.statements = []
+
+    def execute(self, statement, parameters):
+        self.statements.append((statement, parameters))
+        return self
+
+    def fetchone(self):
+        return {"event_seq": 42}
+
+
+@pytest.mark.parametrize(("revision", "deleted", "change"), [
+    (1, False, "created"), (2, False, "updated"), (3, True, "deleted"),
+])
+def test_work_order_signals_use_same_connection_and_contain_only_identity(revision, deleted, change):
+    conn = EventConnection()
+    who = Identity(str(uuid4()), str(uuid4()), True)
+    order = {"work_order_id": str(uuid4()), "revision": revision,
+             "updated_at": "2026-10-08T00:00:00Z", "title": "private title"}
+
+    PostgresStore(SampleSettings())._work_order_event(conn, who, order, deleted=deleted)
+
+    statement, parameters = conn.statements[0]
+    assert "workbench.domain_event_log" in statement
+    assert "'workbench.work_order.changed','peripheral_work_order'" in statement
+    assert parameters[0] == who.project_id
+    assert parameters[2:5] == (order["work_order_id"], revision, order["updated_at"])
+    assert parameters[5].obj == {
+        "source": "peripheral_sample", "work_order_id": order["work_order_id"], "change": change,
+    }
+    assert conn.statements[1] == (
+        "SELECT pg_notify('workbench_events', %s)", (f"{who.project_id}:42",),
+    )
+
+
+def test_event_rollout_does_not_write_to_other_databases():
+    conn = EventConnection()
+    store = PostgresStore(SampleSettings(db_name="another_project"))
+    store._work_order_event(conn, Identity(str(uuid4()), str(uuid4()), True), {})
+    assert conn.statements == []