#!/usr/bin/env python3 """Synchronize the explicitly synthetic peripheral work orders into task projections.""" from __future__ import annotations import argparse import asyncio import json import logging import os import sys from pathlib import Path from uuid import NAMESPACE_URL, UUID, uuid5 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 except ModuleNotFoundError: from run_operational_worker import CredentialProvider, WorkerSettings logger = logging.getLogger("peripheral-task-sync") STATUS = {"draft": "scheduled", "open": "assigned", "in_progress": "in_progress", "awaiting_reply": "awaiting_feedback", "completed": "completed", "cancelled": "cancelled"} EVENT_WORK_ORDER_TYPES = frozenset({"water_quality_complaint", "burst_report"}) MAX_TARGETED_ORDERS = 200 def peripheral_order_id(external_id: str | None) -> str | None: """Return only canonical IDs owned by this synchronizer's namespace.""" if not external_id or not external_id.startswith("peripheral_sample:"): return None try: return str(UUID(external_id.removeprefix("peripheral_sample:"))) except ValueError: return None def should_project_work_order(order: dict, linked_order_ids: set[str]) -> bool: """Exclude retired seed checks while preserving user and workflow orders.""" work_order_id = order["work_order_id"] work_order_type = order.get("work_order_type", "general") if work_order_type in EVENT_WORK_ORDER_TYPES or work_order_id in linked_order_ids: return True return not ( work_order_type == "general" and order.get("origin") == "synthetic" and order.get("created_by") == str(SEED_ACTOR_ID) ) def work_order_snapshot(order: dict) -> dict: """Preserve the flat order and its recorded activity without inventing stages.""" required = ( "work_order_id", "revision", "title", "summary", "status", "priority", "origin", "updated_at", ) snapshot = {key: order[key] for key in required} snapshot.update({key: order.get(key) for key in ( "assignee", "scheduled_at", "location", "scenario_code", )}) snapshot["feedback_requirements"] = order.get("feedback_requirements") or [] snapshot["activity"] = order.get("activity") or [] snapshot["work_order_type"] = order.get("work_order_type", "general") return snapshot async def work_order_pages(client, url, headers, work_order_ids, deleted_ids): if work_order_ids is not None: for order_id in sorted(work_order_ids): response = await client.get(f"{url}/{order_id}", headers=headers) if response.status_code == 404: if response.json().get("code") != "WORK_ORDER_NOT_FOUND": raise ValueError("peripheral work order lookup did not confirm deletion") deleted_ids.add(order_id) continue response.raise_for_status() order = response.json() if order["work_order_id"] != order_id: raise ValueError("peripheral work order identity mismatch") yield {"items": [order]} return cursor = None while True: params = {"limit": 100} if cursor: params["cursor"] = cursor response = await client.get(url, headers=headers, params=params) response.raise_for_status() page = response.json() yield page cursor = page.get("next_cursor") if not cursor: return async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, peripheral_url: str, credentials: CredentialProvider, *, work_order_ids: set[str] | None = None) -> int: if work_order_ids is not None: work_order_ids = {str(UUID(order_id)) for order_id in work_order_ids} headers = await credentials.headers(client) tasks_url = f"{settings.server_url.rstrip('/')}/api/v1/operational-tasks" existing: dict[str, dict] = {} event_tasks: dict[str | None, dict] = {} investigations: dict[str, dict] = {} for order_id in sorted(work_order_ids) if work_order_ids is not None else [None]: offset = 0 while True: params = {"limit": 100, "offset": offset} if order_id is not None: params["peripheral_work_order_id"] = order_id response = await client.get(tasks_url, headers=headers, params=params) response.raise_for_status() page = response.json() for task in page["items"]: if task["task_type"] == "external_work_order": existing[task["source"].get("external_id")] = task elif task["task_type"] == "event_investigation": event_tasks[task.get("source", {}).get("external_id")] = task if task["status"] in {"queued", "waiting_human", "running"} or ( work_order_ids is not None and task["status"] == "completed" ): investigations[task["task_id"]] = task if page.get("next_offset") is None: break offset = page["next_offset"] linked: dict[str, list[str]] = {} for task in investigations.values(): response = await client.get(f"{tasks_url}/{task['task_id']}", headers=headers) response.raise_for_status() for action in response.json().get("actions", []): receipt = action.get("receipt") if action["kind"] == "create_work_order" and receipt and receipt["adapter"] == "peripheral_sample": linked.setdefault(receipt["receipt_id"], []).append(task["task_id"]) linked_order_ids = set(linked) synchronized = 0 deleted_ids: set[str] = set() seen_order_ids: set[str] = set() async for page in work_order_pages( client, f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders", headers, work_order_ids, deleted_ids, ): for order in page["items"]: if order["project_id"] != settings.project_id or order["origin"] != "synthetic": raise ValueError("peripheral work order scope or origin mismatch") seen_order_ids.add(order["work_order_id"]) if not should_project_work_order(order, linked_order_ids): continue external_id = f"peripheral_sample:{order['work_order_id']}" # Complaint intake is an event even if no browser was open when the # external system created it. Use the same source key as the UI. event_kind = order.get("work_order_type") if ( event_kind in EVENT_WORK_ORDER_TYPES and order["work_order_id"] not in event_tasks and order["work_order_id"] not in linked_order_ids ): event_request_id = str(uuid5(NAMESPACE_URL, f"{settings.project_id}:{external_id}:investigation")) event = await client.post(tasks_url, headers=headers, json={ "request_id": event_request_id, "task_type": "event_investigation", "workflow": event_kind, "title": "水质投诉调查" if event_kind == "water_quality_complaint" else "爆管事件调查", "summary": f"合成外围事件：{order['summary']}", "priority": "urgent" if order["priority"] in {"high", "urgent"} else "normal", "source": {"kind": "external_system", "external_id": order["work_order_id"], "external_revision": order["revision"], "event_kind": event_kind}, "trigger": {"kind": "event", "occurred_at": order.get("scheduled_at") or order["created_at"], "detail": f"外围样板事件上报 · {order.get('location') or '位置待确认'}"}, }) event.raise_for_status() event_tasks[order["work_order_id"]] = event.json() task = existing.get(external_id) if task is None: request_id = str(uuid5(NAMESPACE_URL, f"{settings.project_id}:{external_id}:create")) result = await client.post(tasks_url, headers=headers, json={ "request_id": request_id, "task_type": "external_work_order", "workflow": "external-work-order", "title": order["title"], "summary": f"合成外围工单：{order['summary']}", "priority": "urgent" if order["priority"] in {"high", "urgent"} else "normal", "source": {"kind": "external_system", "external_id": external_id, "event_kind": order.get("work_order_type", "general")}, "trigger": {"kind": "external_sync", "occurred_at": order["created_at"], "detail": "外围样板工单同步，进度以外部系统反馈为准"}, "scheduled_for": order["scheduled_at"], }) result.raise_for_status() task = result.json() existing[external_id] = task changed = (task["source"].get("external_revision") or 0) < order["revision"] if work_order_ids is None and changed and order["revision"] > 1: # Only changed sources need completed-parent receipt discovery. # Never fetch every historical completed investigation each tick. offset = 0 while True: related = await client.get(tasks_url, headers=headers, params={ "limit": 100, "offset": offset, "peripheral_work_order_id": order["work_order_id"], }) related.raise_for_status() related_page = related.json() for parent in related_page["items"]: if parent["task_type"] != "event_investigation" or parent.get("status") != "completed": continue detail = await client.get(f"{tasks_url}/{parent['task_id']}", headers=headers) detail.raise_for_status() for action in detail.json().get("actions", []): receipt = action.get("receipt") if (action["kind"] == "create_work_order" and receipt and receipt["adapter"] == "peripheral_sample" and receipt["receipt_id"] == order["work_order_id"]): linked.setdefault(order["work_order_id"], []).append(parent["task_id"]) if related_page.get("next_offset") is None: break offset = related_page["next_offset"] event_id = f"{external_id}:{order['revision']}" notes = [activity["detail"] for activity in order.get("activity", []) if activity.get("kind") == "feedback" and activity.get("detail")] evidence_summary = order["summary"] if notes: evidence_summary += "\n现场反馈：\n" + "\n".join(notes[-10:]) feedback = { "request_id": str(uuid5(NAMESPACE_URL, event_id)), "source_event_id": external_id, "source_revision": order["revision"], "status": STATUS[order["status"]], "occurred_at": order["updated_at"], "detail": f"合成外围工单 {order['work_order_id']}：{order['status']}", "evidence": {"evidence_id": event_id, "kind": "synthetic", "summary": evidence_summary, "source_label": "外围样板合成工单", "source_ref": external_id, "observed_at": order["updated_at"]}, } # A child order's creation is already represented by its dispatch # receipt. Only a subsequent source revision is new field input. targets = set(linked.get(order["work_order_id"], [])) if order["revision"] > 1 else set() if work_order_ids is None and not changed: targets.clear() event_task = event_tasks.get(order["work_order_id"]) if (event_task and event_task.get("status") != "cancelled" and order["revision"] > (event_task["source"].get("external_revision") or 0)): targets.add(event_task["task_id"]) # The external projection is the reconciliation cursor. Advance it # only after all investigation feedback was accepted; retry remains # safe if a response is lost or a later target fails. ordered_targets = sorted(targets) if changed: ordered_targets.append(task["task_id"]) for target in ordered_targets: payload = dict(feedback) if target == task["task_id"]: payload["work_order_snapshot"] = work_order_snapshot(order) result = await client.post(f"{tasks_url}/{target}/external-feedback", headers=headers, json=payload) result.raise_for_status() synchronized += 1 for external_id, task in existing.items(): order_id = peripheral_order_id(external_id) if order_id is None: continue if order_id in seen_order_ids: continue if work_order_ids is not None and order_id not in work_order_ids: continue # The list may change during pagination. Only a scoped 404 proves that # the source order is now absent; transient errors must stop this tick. if order_id not in deleted_ids: response = await client.get( f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders/{order_id}", headers=headers, ) if response.status_code == 200: continue if response.status_code != 404: response.raise_for_status() if response.json().get("code") != "WORK_ORDER_NOT_FOUND": raise ValueError("peripheral work order lookup did not confirm deletion") result = await client.post( f"{tasks_url}/{task['task_id']}/withdraw-external-source", headers=headers, json={ "request_id": str(uuid5(NAMESPACE_URL, f"{settings.project_id}:{external_id}:withdraw")), "source_event_id": external_id, }, ) result.raise_for_status() synchronized += 1 return synchronized 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 self.pending_order_ids: set[str] = set() self.full_reconcile = False def _queue_order(self, order_id: str) -> None: if not self.full_reconcile: self.pending_order_ids.add(order_id) if len(self.pending_order_ids) > MAX_TARGETED_ORDERS: self.pending_order_ids.clear() self.full_reconcile = True self.wake.set() 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 self.full_reconcile = True # 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 if (event.topic == "workbench.work_order.changed" and event.aggregate_type == "peripheral_work_order" and event.metadata.get("source") == "peripheral_sample"): self._queue_order(str(UUID(event.aggregate_id))) elif event.topic == "operational.action.changed": # The source create can commit before its action # receipt is saved. Revisit just those receipt IDs. task_id = str(UUID(event.metadata["task_id"])) task = await self.client.get( f"{self.settings.server_url.rstrip('/')}/api/v1/operational-tasks/{task_id}", headers=headers, ) task.raise_for_status() for action in task.json().get("actions", []): receipt = action.get("receipt") if (action["kind"] == "create_work_order" and receipt and receipt["adapter"] == "peripheral_sample"): self._queue_order(str(UUID(receipt["receipt_id"]))) self.cursor = event.event_seq elif not line: event_name = "" await asyncio.sleep(1) except (httpx.HTTPError, ValueError, KeyError): logger.warning("Work-order event stream disconnected: project=%s; reconciliation remains active", self.settings.project_id) 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()) retry_seconds = 5 next_reconcile = asyncio.get_running_loop().time() + reconcile_seconds try: while True: try: remaining = max(0, next_reconcile - asyncio.get_running_loop().time()) await asyncio.wait_for(self.wake.wait(), timeout=remaining) except TimeoutError: self.full_reconcile = True # 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() full_reconcile = self.full_reconcile self.full_reconcile = False order_ids, self.pending_order_ids = self.pending_order_ids, set() try: count = await sync_once( self.client, self.settings, self.peripheral_url, self.credentials, **({"work_order_ids": order_ids} if not full_reconcile else {}), ) logger.info("Peripheral progress synchronization completed: project=%s revisions=%d", self.settings.project_id, count) retry_seconds = 5 if full_reconcile: next_reconcile = asyncio.get_running_loop().time() + reconcile_seconds except (httpx.HTTPError, ValueError, KeyError): logger.error("Peripheral synchronization failed: project=%s; persisted revisions will be retried", self.settings.project_id) self.full_reconcile |= full_reconcile for order_id in order_ids: self._queue_order(order_id) await asyncio.sleep(retry_seconds) self.wake.set() 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"] async with httpx.AsyncClient(timeout=30, follow_redirects=False, trust_env=False) as client: if once: results = await asyncio.gather(*( sync_once(client, scope, peripheral_url, CredentialProvider(scope)) for scope in settings.project_scopes() ), return_exceptions=True) if any(isinstance(result, Exception) for result in results): logger.error("Peripheral synchronization failed; persisted revisions remain available") raise SystemExit(1) from None logger.info("Peripheral progress synchronization completed: %d revisions", sum(results)) return await asyncio.gather(*( PeripheralWorkOrderSynchronizer(client, scope, peripheral_url, CredentialProvider(scope)).serve(reconcile_seconds) for scope in settings.project_scopes() )) 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, args.reconcile_seconds)) except KeyboardInterrupt: pass