From 6a4478bfe49fecdd4e91ecb6f45f5ba1fb9ceebf Mon Sep 17 00:00:00 2001
From: Huarch <huarch97@outlook.com>
Date: Mon, 9 Nov 2026 17:55:00 +0800
Subject: [PATCH] refactor: simplify task and history services

Remove retired query paths and repository compatibility branches, and share task test fixtures.
Cancel pending questions and resolve their notifications when recovering expired runs.
---
 app/services/operational_tasks.py            | 181 ++++-------
 app/services/task_questions.py               |   4 +-
 app/services/timeseries_analysis.py          | 296 +-----------------
 app/services/timeseries_history.py           |  66 ++--
 tests/support/__init__.py                    |   0
 tests/support/operational_task_repository.py | 252 +++++++++++++++
 tests/unit/test_agent_messages.py            |   6 +
 tests/unit/test_operational_tasks.py         | 304 ++++---------------
 tests/unit/test_project_scada_metadata.py    | 121 ++------
 tests/unit/test_task_questions.py            |  50 +--
 10 files changed, 464 insertions(+), 816 deletions(-)
 create mode 100644 tests/support/__init__.py
 create mode 100644 tests/support/operational_task_repository.py

diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py
index 935a463..48e3949 100644
--- a/app/services/operational_tasks.py
+++ b/app/services/operational_tasks.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import json
 from datetime import datetime, timedelta, timezone
-from typing import Iterable
+from typing import TYPE_CHECKING, Iterable
 from uuid import NAMESPACE_URL, UUID, uuid4, uuid5
 
 from app.auth.permissions import (
@@ -77,6 +77,10 @@ from app.services.operational_scada_coverage import require_complete_scada_evide
 from app.services.time_comparison_chart import validate_time_comparison_chart
 
 
+if TYPE_CHECKING:
+    from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository
+
+
 LEASE_DURATION = timedelta(seconds=60)
 
 
@@ -89,7 +93,7 @@ def _task_json(model: OperationalTaskSummary) -> dict:
 
 
 class OperationalTaskService:
-    def __init__(self, repository, context, permissions: Iterable[str], *, clock=None,
+    def __init__(self, repository: OperationalTaskRepository, context, permissions: Iterable[str], *, clock=None,
                  execution_scope: str = "production"):
         self.repo = repository
         self.context = context
@@ -114,42 +118,14 @@ class OperationalTaskService:
             require(owner == command.agent_instance_id,
                     "AGENT_INSTANCE_MISMATCH", "Agent instance does not own this run", status=409)
 
-    @property
-    def _supports_runs(self) -> bool:
-        return hasattr(self.repo, "get_active_run") and hasattr(self.repo, "update_run")
-
     async def _active_run(self, task: dict) -> dict | None:
-        if self._supports_runs:
-            return await self.repo.get_active_run(str(task["task_id"]))
-        if task.get("status") not in {"queued", "running"}:
-            return None
-        now = self._now()
-        return _json(OperationalTaskRun(
-            run_id=uuid5(NAMESPACE_URL, f"legacy-run:{task['task_id']}:{task['execution_generation']}"),
-            task_id=task["task_id"],
-            generation=task["execution_generation"],
-            execution_request_id=task["execution_request_id"],
-            status="running" if task["status"] == "running" else "queued",
-            revision=task["revision"],
-            agent_session_id=task.get("agent_session_id"),
-            lease_expires_at=task.get("agent_lease_expires_at"),
-            heartbeat_at=task.get("updated_at") if task["status"] == "running" else None,
-            started_at=task.get("created_at") if task["status"] == "running" else None,
-            dispatch_attempts=0,
-            next_dispatch_at=task.get("created_at") or now,
-            created_at=task.get("created_at") or now,
-            updated_at=task.get("updated_at") or now,
-        ))
+        return await self.repo.get_active_run(str(task["task_id"]))
 
-    async def _append_event(self, **kwargs) -> int | None:
-        append_event = getattr(self.repo, "append_event", None)
-        return await append_event(**kwargs) if append_event is not None else None
+    async def _append_event(self, **kwargs) -> int:
+        return await self.repo.append_event(**kwargs)
 
     async def _list_access_requests(self, task_id: UUID | str) -> list[dict]:
-        list_access_requests = getattr(self.repo, "list_access_requests", None)
-        if list_access_requests is None:
-            return []
-        return await list_access_requests(str(task_id))
+        return await self.repo.list_access_requests(str(task_id))
 
     def _permission_view(self) -> OperationalTaskPermissions:
         permissions = self.permissions
@@ -286,8 +262,7 @@ class OperationalTaskService:
         actions = await self.repo.list_actions(str(task_id))
         access_requests = await self._list_access_requests(task_id)
         active_run = await self._active_run(task)
-        latest_run = (await self.repo.get_latest_run(str(task_id))
-                      if hasattr(self.repo, "get_latest_run") else active_run)
+        latest_run = await self.repo.get_latest_run(str(task_id))
         progress = await self.repo.get_latest_progress(str(task_id), task["execution_generation"])
         work_order_snapshot = None
         if task["task_type"] == "external_work_order" and task["source"].get("external_id"):
@@ -397,9 +372,8 @@ class OperationalTaskService:
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return TaskAccessRequest.model_validate(replay["access_request"])
-            availability = getattr(self.repo, "access_requests_available", None)
             require(
-                availability is None or await availability(),
+                await self.repo.access_requests_available(),
                 "DEPENDENCY_UNAVAILABLE",
                 "Task data access requests are not enabled for this project database",
                 status=503,
@@ -616,8 +590,7 @@ class OperationalTaskService:
         return OperationalTaskRun.model_validate(await self._run(task_id, run_id))
 
     async def _fail_run_messages(self, run_id, reason: str, now: datetime) -> None:
-        if hasattr(self.repo, "fail_run_messages"):
-            await self.repo.fail_run_messages(str(run_id), reason, now)
+        await self.repo.fail_run_messages(str(run_id), reason, now)
 
     async def fail_run(
         self, task_id: UUID, run_id: UUID, command: AgentRunFailureCommand
@@ -772,7 +745,7 @@ class OperationalTaskService:
                 # Lease recovery already holds a row lock. Try rather than wait
                 # for the shared run lock to avoid reversing the lock order of
                 # message/finish transactions; a busy run is checked next tick.
-                if hasattr(self.repo, "try_lock") and not await self.repo.try_lock(f"run:{run['run_id']}"):
+                if not await self.repo.try_lock(f"run:{run['run_id']}"):
                     continue
                 task_id = UUID(str(run["task_id"]))
                 expected = run["revision"]
@@ -784,6 +757,7 @@ class OperationalTaskService:
                 timed_out = OperationalTaskRun.model_validate(run)
                 await self.repo.update_run(_json(timed_out), expected)
                 await self._fail_run_messages(run["run_id"], "Run lease expired", now)
+                await task_questions.cancel_pending(self, task_id, now)
                 await self._run_event(timed_out, now)
                 task = await self._task(task_id)
                 timeout_count = await self.repo.consecutive_timed_out_runs(str(task_id))
@@ -1017,25 +991,9 @@ class OperationalTaskService:
                 lease_expires_at=now + LEASE_DURATION,
                 updated_at=now,
             )
-            if self._supports_runs:
-                await self.repo.update_run(
-                    _json(OperationalTaskRun.model_validate(active_run)), run_expected
-                )
-            else:
-                # Compatibility for pre-015 repository adapters only.
-                if command.report:
-                    if task["task_type"] == "event_investigation":
-                        task["status"] = self._investigation_status(actions.values(), task) if command.report.stage == "final" else "running"
-                    else:
-                        task["status"] = self._status_from_plan(current_plan.plan.nodes, actions.values()) if current_plan else "running"
-                    if task["status"] in {"waiting_human", "completed", "failed", "cancelled"}:
-                        task["lifecycle_status"] = task["status"]
-                    task["agent_lease_expires_at"] = (
-                        now + LEASE_DURATION if task["status"] == "running" else None
-                    )
-                    await self.repo.update_task(
-                        _task_json(OperationalTaskSummary.model_validate(task))
-                    )
+            await self.repo.update_run(
+                _json(OperationalTaskRun.model_validate(active_run)), run_expected
+            )
             await self._task_event(task, now)
             response = {"plan_revision": plan_revision, "report_revision": report_revision}
             if progress_revision is not None:
@@ -1988,13 +1946,12 @@ class OperationalTaskService:
         return DispatchDueSchedulesResult(task_ids=task_ids, dispatched_at=now)
 
     async def _require_no_pending_calculations(self, task_id: UUID, run: dict) -> None:
-        if hasattr(self.repo, "pending_calculation_jobs"):
-            pending = await self.repo.pending_calculation_jobs(
-                str(task_id), str(run["run_id"]), str(run["execution_request_id"]),
-            )
-            require(not pending, "PENDING_CALCULATIONS",
-                    "Wait for submitted calculations before publishing a final report or completing the run",
-                    refs=pending, status=409, recovery="retry_same_request")
+        pending = await self.repo.pending_calculation_jobs(
+            str(task_id), str(run["run_id"]), str(run["execution_request_id"]),
+        )
+        require(not pending, "PENDING_CALCULATIONS",
+                "Wait for submitted calculations before publishing a final report or completing the run",
+                refs=pending, status=409, recovery="retry_same_request")
 
     async def _finish_run(
         self, task_id: UUID, run_id: UUID, command: AgentRunCommand,
@@ -2024,11 +1981,10 @@ class OperationalTaskService:
             actions = []
             if outcome == "succeeded":
                 await self._require_no_pending_calculations(task_id, run)
-                if hasattr(self.repo, "pending_run_messages"):
-                    pending_messages = await self.repo.pending_run_messages(str(run_id))
-                    require(not pending_messages, "PENDING_AGENT_MESSAGES",
-                            "Run has messages not yet entered into Agent context",
-                            status=409, recovery="retry_same_request", refs=pending_messages)
+                pending_messages = await self.repo.pending_run_messages(str(run_id))
+                require(not pending_messages, "PENDING_AGENT_MESSAGES",
+                        "Run has messages not yet entered into Agent context",
+                        status=409, recovery="retry_same_request", refs=pending_messages)
                 plan_document = await self.repo.get_latest_plan(str(task_id))
                 report_document = await self.repo.get_latest_report(str(task_id))
                 plan = (
@@ -2158,8 +2114,6 @@ class OperationalTaskService:
         from app.infra.db.postgresql.task_alerts import TaskAlertRepository
         from app.services.task_alerts import TaskAlertService
 
-        if not hasattr(self.repo, "list_investigation_alerts"):
-            return
         for alert in await self.repo.list_investigation_alerts(task_id):
             command = TaskAlertEvent.model_validate({
                 **{key: value for key, value in alert.items() if key not in {"status", "updated_at", "revision"}},
@@ -2364,42 +2318,41 @@ class OperationalTaskService:
 
     async def _requeue_task(self, task: dict, now: datetime) -> None:
         await task_questions.cancel_pending(self, UUID(str(task["task_id"])), now)
-        if self._supports_runs:
-            active_run = await self.repo.get_active_run(str(task["task_id"]))
-            while active_run is not None:
-                await self.repo.lock(f"run:{active_run['run_id']}")
-                # Completion may have won the run lock after our first read.
-                refreshed_run = await self.repo.get_active_run(str(task["task_id"]))
-                if refreshed_run is None or str(refreshed_run["run_id"]) == str(active_run["run_id"]):
-                    active_run = refreshed_run
-                    break
+        active_run = await self.repo.get_active_run(str(task["task_id"]))
+        while active_run is not None:
+            await self.repo.lock(f"run:{active_run['run_id']}")
+            # Completion may have won the run lock after our first read.
+            refreshed_run = await self.repo.get_active_run(str(task["task_id"]))
+            if refreshed_run is None or str(refreshed_run["run_id"]) == str(active_run["run_id"]):
                 active_run = refreshed_run
-            if active_run is not None and active_run["status"] == "queued":
-                # Multiple inputs can arrive before the worker claims the next
-                # generation. They are all visible when that queued run starts,
-                # so coalesce them instead of violating the one-active-run rule.
-                task.update(
-                    lifecycle_status="open", status="queued",
-                    revision=task["revision"] + 1,
-                    execution_generation=active_run["generation"],
-                    execution_request_id=str(active_run["execution_request_id"]),
-                    agent_session_id=None, agent_lease_expires_at=None,
-                    updated_at=now,
-                )
-                return
-            if active_run is not None:
-                expected = active_run["revision"]
-                active_run.update(
-                    status="cancelled", revision=expected + 1,
-                    lease_expires_at=None, finished_at=now,
-                    failure_code="SUPERSEDED_BY_REQUEUE",
-                    failure_detail="New task input superseded the active run",
-                    updated_at=now,
-                )
-                cancelled = OperationalTaskRun.model_validate(active_run)
-                await self.repo.update_run(_json(cancelled), expected)
-                await self._fail_run_messages(active_run["run_id"], "Run superseded by requeue", now)
-                await self._run_event(cancelled, now)
+                break
+            active_run = refreshed_run
+        if active_run is not None and active_run["status"] == "queued":
+            # Multiple inputs can arrive before the worker claims the next
+            # generation. They are all visible when that queued run starts,
+            # so coalesce them instead of violating the one-active-run rule.
+            task.update(
+                lifecycle_status="open", status="queued",
+                revision=task["revision"] + 1,
+                execution_generation=active_run["generation"],
+                execution_request_id=str(active_run["execution_request_id"]),
+                agent_session_id=None, agent_lease_expires_at=None,
+                updated_at=now,
+            )
+            return
+        if active_run is not None:
+            expected = active_run["revision"]
+            active_run.update(
+                status="cancelled", revision=expected + 1,
+                lease_expires_at=None, finished_at=now,
+                failure_code="SUPERSEDED_BY_REQUEUE",
+                failure_detail="New task input superseded the active run",
+                updated_at=now,
+            )
+            cancelled = OperationalTaskRun.model_validate(active_run)
+            await self.repo.update_run(_json(cancelled), expected)
+            await self._fail_run_messages(active_run["run_id"], "Run superseded by requeue", now)
+            await self._run_event(cancelled, now)
         generation = task["execution_generation"] + 1
         task_id = UUID(task["task_id"])
         task.update(
@@ -2409,9 +2362,8 @@ class OperationalTaskService:
             agent_session_id=None, agent_lease_expires_at=None, updated_at=now,
         )
         run = self._new_run(task_id, generation, now)
-        if self._supports_runs:
-            await self.repo.insert_run(_json(run))
-            await self._run_event(run, now, queued=True)
+        await self.repo.insert_run(_json(run))
+        await self._run_event(run, now, queued=True)
 
     async def _notify(
         self, task_id: UUID, kind: str, title: str, description: str, now: datetime,
@@ -2447,10 +2399,7 @@ class OperationalTaskService:
         now: datetime,
         reason: str,
     ) -> None:
-        resolve = getattr(self.repo, "resolve_notifications", None)
-        if resolve is None:
-            return
-        notifications = await resolve(
+        notifications = await self.repo.resolve_notifications(
             str(task_id),
             kinds=kinds,
             subject_type=subject_type,
diff --git a/app/services/task_questions.py b/app/services/task_questions.py
index a66280c..b10d64f 100644
--- a/app/services/task_questions.py
+++ b/app/services/task_questions.py
@@ -13,9 +13,7 @@ if TYPE_CHECKING:
 
 
 async def list_questions(service: "OperationalTaskService", task_id: UUID) -> list[dict]:
-    # Legacy test repositories do not implement question persistence.
-    reader = getattr(service.repo, "list_questions", None)
-    return await reader(str(task_id)) if reader else []
+    return await service.repo.list_questions(str(task_id))
 
 
 async def cancel_pending(service: "OperationalTaskService", task_id: UUID, now: datetime) -> None:
diff --git a/app/services/timeseries_analysis.py b/app/services/timeseries_analysis.py
index 5070938..286d7cf 100644
--- a/app/services/timeseries_analysis.py
+++ b/app/services/timeseries_analysis.py
@@ -1,11 +1,10 @@
 import time
 from datetime import datetime, timedelta
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List
 
 import numpy as np
 import pandas as pd
 from psycopg import AsyncConnection
-from uuid import UUID
 
 from app.algorithms.pipe_health_prediction.survival_predictor import (
     PipeHealthSurvivalPredictor,
@@ -15,14 +14,11 @@ from app.algorithms.scada_cleaning.pressure_series import clean_pressure_data_df
 from app.infra.db.postgresql.network_assets import NetworkAssetRepository
 from app.infra.db.postgresql.scada import ScadaInfoRepository
 from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
-from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
 from app.infra.db.timescaledb.repositories.scada import ScadaRepository
 
 
 class TimeseriesAnalysisService:
-    """
-    复合查询类，提供跨表查询功能
-    """
+    """SCADA 清洗与管道健康预测；历史查询由 TimeseriesHistoryService 负责。"""
 
     @staticmethod
     async def _get_project_scada_index(
@@ -31,294 +27,6 @@ class TimeseriesAnalysisService:
         scadas = await ScadaInfoRepository.get_scadas(postgres_conn)
         return {scada["device_id"]: scada for scada in scadas}
 
-    @staticmethod
-    async def get_scada_associated_realtime_simulation_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        device_ids: List[str],
-        start_time: datetime,
-        end_time: datetime,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 SCADA 关联的 link/node 模拟值
-
-        根据传入的 SCADA device_ids，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            device_ids: SCADA 设备ID列表
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 device_id 为键，值为数据列表，每个数据包含 time, value 和 scada_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        link_devices: dict[str, str] = {}
-        node_devices: dict[str, str] = {}
-        for device_id in device_ids:
-            target_scada = scada_by_id.get(device_id)
-            if not target_scada:
-                raise ValueError(f"SCADA device {device_id} not found")
-            scada_type = target_scada["device_type"]
-            if scada_type in {"pipe_flow", "flow"}:
-                link_devices[device_id] = target_scada["link_id"]
-            elif scada_type == "pressure":
-                node_devices[device_id] = target_scada["node_id"]
-            else:
-                raise ValueError(f"Unknown SCADA type: {scada_type}")
-
-        link_series = await RealtimeRepository.get_link_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(link_devices.values())), "flow",
-        )
-        node_series = await RealtimeRepository.get_node_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(node_devices.values())), "pressure",
-        )
-        return {
-            device_id: [
-                {**item, "scada_id": device_id}
-                for item in (
-                    link_series.get(element_id, [])
-                    if device_id in link_devices
-                    else node_series.get(element_id, [])
-                )
-            ]
-            for device_id, element_id in (link_devices | node_devices).items()
-        }
-
-    @staticmethod
-    async def get_scada_associated_analysis_simulation_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        device_ids: List[str],
-        start_time: datetime,
-        end_time: datetime,
-        run_id: UUID,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 SCADA 关联的 link/node 分析模拟值
-
-        根据传入的 SCADA device_ids，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            device_ids: SCADA 设备ID列表
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 device_id 为键，值为数据列表，每个数据包含 time, value 和 scada_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        link_devices: dict[str, str] = {}
-        node_devices: dict[str, str] = {}
-        for device_id in device_ids:
-            target_scada = scada_by_id.get(device_id)
-            if not target_scada:
-                raise ValueError(f"SCADA device {device_id} not found")
-
-            scada_type = target_scada["device_type"]
-            if scada_type in {"pipe_flow", "flow"}:
-                link_devices[device_id] = target_scada["link_id"]
-            elif scada_type == "pressure":
-                node_devices[device_id] = target_scada["node_id"]
-            else:
-                raise ValueError(f"Unknown SCADA type: {scada_type}")
-
-        link_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "link",
-            list(dict.fromkeys(link_devices.values())), start_time, end_time, "flow",
-        )
-        node_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "node",
-            list(dict.fromkeys(node_devices.values())), start_time, end_time, "pressure",
-        )
-        return {
-            device_id: [
-                {**item, "scada_id": device_id}
-                for item in (
-                    link_series.get(element_id, [])
-                    if device_id in link_devices
-                    else node_series.get(element_id, [])
-                )
-            ]
-            for device_id, element_id in (link_devices | node_devices).items()
-        }
-
-    @staticmethod
-    async def get_realtime_simulation_data(
-        timescale_conn: AsyncConnection,
-        feature_infos: List[Tuple[str, str]],
-        start_time: datetime,
-        end_time: datetime,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 link/node 模拟值
-
-        根据传入的 feature_infos，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 feature_id 为键，值为数据列表，每个数据包含 time, value 和 feature_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        pipe_ids: list[str] = []
-        junction_ids: list[str] = []
-        for feature_id, feature_type in feature_infos:
-            if feature_type.lower() == "pipe":
-                pipe_ids.append(feature_id)
-            elif feature_type.lower() == "junction":
-                junction_ids.append(feature_id)
-            else:
-                raise ValueError(f"Unknown type: {feature_type}")
-        link_series = await RealtimeRepository.get_link_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time, list(dict.fromkeys(pipe_ids)), "flow"
-        )
-        node_series = await RealtimeRepository.get_node_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(junction_ids)), "pressure",
-        )
-        return {
-            feature_id: [
-                {**item, "feature_id": feature_id}
-                for item in (
-                    link_series.get(feature_id, [])
-                    if feature_type.lower() == "pipe"
-                    else node_series.get(feature_id, [])
-                )
-            ]
-            for feature_id, feature_type in feature_infos
-        }
-
-    @staticmethod
-    async def get_analysis_simulation_data(
-        timescale_conn: AsyncConnection,
-        feature_infos: List[Tuple[str, str]],
-        start_time: datetime,
-        end_time: datetime,
-        run_id: UUID,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 link/node 分析模拟值
-
-        根据传入的 feature_infos，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
-            start_time: 开始时间
-            end_time: 结束时间
-            run_id: 分析运行 ID
-
-        Returns:
-            模拟数据字典，以 feature_id 为键，值为数据列表，每个数据包含 time, value 和 feature_id
-
-        Raises:
-            ValueError: 当类型无效时
-        """
-        pipe_ids: list[str] = []
-        junction_ids: list[str] = []
-        for feature_id, feature_type in feature_infos:
-            if feature_type.lower() == "pipe":
-                pipe_ids.append(feature_id)
-            elif feature_type.lower() == "junction":
-                junction_ids.append(feature_id)
-            else:
-                raise ValueError(f"Unknown type: {feature_type}")
-        link_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "link", list(dict.fromkeys(pipe_ids)),
-            start_time, end_time, "flow",
-        )
-        node_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "node", list(dict.fromkeys(junction_ids)),
-            start_time, end_time, "pressure",
-        )
-        return {
-            feature_id: [
-                {**item, "feature_id": feature_id}
-                for item in (
-                    link_series.get(feature_id, [])
-                    if feature_type.lower() == "pipe"
-                    else node_series.get(feature_id, [])
-                )
-            ]
-            for feature_id, feature_type in feature_infos
-        }
-
-    @staticmethod
-    async def get_element_associated_scada_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        element_id: str,
-        start_time: datetime,
-        end_time: datetime,
-        use_cleaned: bool = False,
-    ) -> Optional[Any]:
-        """
-        获取 link/node 关联的 SCADA 监测值
-
-        根据传入的 link/node id，匹配 SCADA 信息，
-        如果存在关联的 SCADA device_id，获取实际的监测数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            element_id: link 或 node 的 ID
-            start_time: 开始时间
-            end_time: 结束时间
-            use_cleaned: 是否使用清洗后的数据 (True: "cleaned_value", False: "monitored_value")
-
-        Returns:
-            SCADA 监测数据值，如果没有找到则返回 None
-
-        Raises:
-            ValueError: 当元素类型无效时
-        """
-
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        associated_scada = next(
-            (
-                scada
-                for scada in scada_by_id.values()
-                if (scada.get("node_id") or scada.get("link_id")) == element_id
-            ),
-            None,
-        )
-
-        if not associated_scada:
-            return None
-
-        device_id = associated_scada["device_id"]
-
-        data_field = "cleaned_value" if use_cleaned else "monitored_value"
-
-        res = await ScadaRepository.get_scada_field_by_id_time_range(
-            timescale_conn, [device_id], start_time, end_time, data_field
-        )
-
-        return {element_id: res.get(device_id, [])}
-
     @staticmethod
     async def clean_scada_data(
         timescale_conn: AsyncConnection,
diff --git a/app/services/timeseries_history.py b/app/services/timeseries_history.py
index da58440..213ccf1 100644
--- a/app/services/timeseries_history.py
+++ b/app/services/timeseries_history.py
@@ -57,6 +57,33 @@ def _points(
     ]
 
 
+def _history_series(
+    *, element_type: HistoryElementType, element_id: str,
+    source: HistorySource, source_unit: str, device_id: str | None = None,
+    unit_inferred: bool = False, points: list[HistoryPoint] | None = None,
+) -> ElementHistorySeries:
+    metric = _target_metric(element_type)
+    return ElementHistorySeries(
+        element_id=element_id, element_type=element_type, device_id=device_id,
+        metric=metric, source=source, source_unit=source_unit,
+        display_unit=display_unit_for_metric(metric.value),
+        unit_inferred=unit_inferred, points=points if points is not None else [],
+    )
+
+
+async def _resolve_result_units(
+    postgres_conn: AsyncConnection, query: ElementHistoryQuery,
+) -> tuple[dict[str, Any], bool]:
+    units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+    if query.mode.value != "analysis_comparison":
+        return units, False
+    run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
+    if run is None:
+        raise ValueError(f"analysis run does not exist: {query.run_id}")
+    recorded = (run.get("parameters") or {}).get("result_units")
+    return (recorded, False) if isinstance(recorded, dict) else (units, True)
+
+
 async def _query_page(timescale_conn, postgres_conn, query, selected_devices, limit, offset):
     descriptors = []
     series = []
@@ -66,16 +93,14 @@ async def _query_page(timescale_conn, postgres_conn, query, selected_devices, li
             "source": source, "element_type": element_type,
             "id": device_id if source == "scada" else element_id, "field": field,
         })
-        metric = _target_metric(element_type)
-        series.append(ElementHistorySeries(
+        series.append(_history_series(
             element_id=element_id, element_type=element_type, device_id=device_id,
-            metric=metric, source=(
+            source=(
                 HistorySource.SCADA_RAW if field == "monitored_value" else HistorySource.SCADA_CLEANED
             ) if source == "scada" else (
                 HistorySource.ANALYSIS_SIMULATION if source == "analysis" else HistorySource.REALTIME_SIMULATION
             ),
-            source_unit=unit, display_unit=display_unit_for_metric(metric.value),
-            unit_inferred=unit_inferred, points=[],
+            source_unit=unit, unit_inferred=unit_inferred,
         ))
 
     # Stable series ordering makes a page crossing raw/cleaned/model boundaries repeatable.
@@ -84,17 +109,8 @@ async def _query_page(timescale_conn, postgres_conn, query, selected_devices, li
             add(source="scada", element_type=element_type, element_id=element_id,
                 device_id=device_id, field=field, unit=device["measurement_unit"])
     if query.mode.value != "observed":
-        units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+        units, inferred = await _resolve_result_units(postgres_conn, query)
         source = "analysis" if query.mode.value == "analysis_comparison" else "realtime"
-        inferred = False
-        if source == "analysis":
-            run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
-            if run is None:
-                raise ValueError(f"analysis run does not exist: {query.run_id}")
-            recorded = (run.get("parameters") or {}).get("result_units")
-            inferred = not isinstance(recorded, dict)
-            if not inferred:
-                units = recorded
         targets = {(t.element_type, t.element_id) for t in query.elements}
         for element_type, element_id in sorted(targets):
             metric = _target_metric(element_type)
@@ -202,21 +218,18 @@ class TimeseriesHistoryService:
             rows = scada_by_device.get(device_id, [])
             if not rows:
                 continue
-            metric = _target_metric(element_type)
             unit = device["measurement_unit"]
             for source, field in (
                 (HistorySource.SCADA_RAW, "monitored_value"),
                 (HistorySource.SCADA_CLEANED, "cleaned_value"),
             ):
                 series.append(
-                    ElementHistorySeries(
+                    _history_series(
                         element_id=element_id,
                         element_type=element_type,
                         device_id=device_id,
-                        metric=metric,
                         source=source,
                         source_unit=unit,
-                        display_unit=display_unit_for_metric(metric.value),
                         points=_points(
                             rows,
                             value_field=field,
@@ -227,15 +240,8 @@ class TimeseriesHistoryService:
         if query.mode.value == "observed":
             return ElementHistoryResponse(series=series)
 
-        current_units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+        result_units, unit_inferred = await _resolve_result_units(postgres_conn, query)
         if query.mode.value == "analysis_comparison":
-            run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
-            if run is None:
-                raise ValueError(f"analysis run does not exist: {query.run_id}")
-            parameters = run.get("parameters") or {}
-            recorded_units = parameters.get("result_units")
-            unit_inferred = not isinstance(recorded_units, dict)
-            result_units = recorded_units if not unit_inferred else current_units
             link_values = await AnalysisResultsRepository.get_series_by_ids(
                 timescale_conn,
                 query.run_id,
@@ -256,8 +262,6 @@ class TimeseriesHistoryService:
             )
             source = HistorySource.ANALYSIS_SIMULATION
         else:
-            unit_inferred = False
-            result_units = current_units
             link_values = await RealtimeRepository.get_link_fields_by_ids_time_range(
                 timescale_conn,
                 query.start_time,
@@ -285,13 +289,11 @@ class TimeseriesHistoryService:
                 continue
             source_unit = str(result_units.get(metric.value, "")).strip()
             series.append(
-                ElementHistorySeries(
+                _history_series(
                     element_id=target.element_id,
                     element_type=target.element_type,
-                    metric=metric,
                     source=source,
                     source_unit=source_unit,
-                    display_unit=display_unit_for_metric(metric.value),
                     unit_inferred=unit_inferred,
                     points=_points(
                         rows,
diff --git a/tests/support/__init__.py b/tests/support/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/support/operational_task_repository.py b/tests/support/operational_task_repository.py
new file mode 100644
index 0000000..741ca2e
--- /dev/null
+++ b/tests/support/operational_task_repository.py
@@ -0,0 +1,252 @@
+"""In-memory task/run fixtures exercising the production repository contract."""
+from contextlib import asynccontextmanager
+from copy import deepcopy
+from datetime import datetime, timedelta
+from uuid import uuid4
+from app.auth.project_dependencies import ProjectContext
+from app.domain.schemas.operational_task import OperationalTaskSummary, OperationalTaskRun
+
+def _context(user_id=None, *, system_role="user"):
+    return ProjectContext(
+        project_id=uuid4(), project_code="test", user_id=user_id or uuid4(),
+        project_role="member", system_role=system_role,
+    )
+
+def _task_document(context, now, *, lease=None):
+    task_id = uuid4()
+    return OperationalTaskSummary(
+        execution_scope="production",
+        project_id=context.project_id,
+        task_id=task_id,
+        task_type="event_investigation",
+        workflow="water_quality_complaint",
+        status="running",
+        priority="normal",
+        revision=2,
+        execution_generation=1,
+        execution_request_id=uuid4(),
+        title="调查",
+        summary="验证状态机",
+        source={"kind": "operator", "event_kind": "water_quality_complaint"},
+        trigger={"kind": "manual", "occurred_at": now, "detail": "test"},
+        scheduled_for=None,
+        agent_session_id="session-1",
+        agent_lease_expires_at=lease,
+        current_plan_revision=None,
+        current_report_revision=None,
+        created_at=now,
+        updated_at=now,
+    ).model_dump(mode="json")
+
+def _active_run(task, now):
+    return OperationalTaskRun(
+        run_id=uuid4(), task_id=task["task_id"], generation=task["execution_generation"],
+        execution_request_id=task["execution_request_id"], status="running", revision=2,
+        agent_session_id="session-1", lease_expires_at=now + timedelta(minutes=1),
+        heartbeat_at=now, started_at=now, dispatch_attempts=0,
+        next_dispatch_at=now, created_at=now, updated_at=now,
+    ).model_dump(mode="json")
+
+class TaskRepository:
+    def __init__(self, task, run=None, *, action=None):
+        self.task = task
+        self.action = action
+        self.requests = {}
+        self.evidence = {}
+        self.feedback = {}
+        self.plans = []
+        self.reports = []
+        self.progress = {}
+        self.notifications = []
+        self.locks = []
+        self.access_requests = {}
+        self.events = []
+        self.questions = {}
+        now = datetime.fromisoformat(task["created_at"])
+        seeded = _active_run(task, now) if run is None else run
+        if run is None:
+            seeded.update(
+                status=task["status"] if task["status"] in {"queued", "running"} else "succeeded",
+                agent_session_id=task.get("agent_session_id"),
+                lease_expires_at=task.get("agent_lease_expires_at"),
+            )
+        self.run = seeded
+        self.runs = {seeded["run_id"]: seeded}
+
+    @asynccontextmanager
+    async def transaction(self):
+        yield self
+
+    async def lock(self, key):
+        self.locks.append(key)
+
+    async def get_request(self, scope, request_id):
+        return self.requests.get((scope, request_id))
+
+    async def save_request(self, scope, request_id, request_digest, response):
+        self.requests[(scope, request_id)] = {
+            "request_digest": request_digest,
+            "response": response,
+        }
+
+    async def get_task(self, task_id):
+        return self.task if task_id == self.task["task_id"] else None
+
+    async def update_task(self, task):
+        self.task = task
+
+    async def get_latest_plan(self, _task_id):
+        return self.plans[-1] if self.plans else None
+
+    async def get_latest_report(self, _task_id):
+        return self.reports[-1] if self.reports else None
+
+    async def get_latest_progress(self, task_id, generation):
+        return self.progress.get((task_id, generation))
+
+    async def save_progress(self, task_id, snapshot):
+        self.progress[(task_id, snapshot["execution_generation"])] = snapshot
+
+    async def get_report(self, _task_id, revision):
+        return next((item for item in self.reports if item["revision"] == revision), None)
+
+    async def list_reports(self, _task_id, limit, offset):
+        return list(reversed(self.reports))[offset:offset + limit]
+
+    async def insert_plan(self, _task_id, snapshot):
+        self.plans.append(snapshot)
+
+    async def insert_report(self, _task_id, snapshot):
+        self.reports.append(snapshot)
+
+    async def list_evidence(self, _task_id):
+        return list(self.evidence.values())
+
+    async def get_evidence(self, _task_id, evidence_id):
+        return self.evidence.get(evidence_id)
+
+    async def insert_evidence(self, _task_id, evidence):
+        self.evidence[evidence["evidence_id"]] = evidence
+
+    async def list_access_requests(self, _task_id):
+        return list(self.access_requests.values())
+
+    async def get_access_request(self, _task_id, access_request_id):
+        return self.access_requests.get(access_request_id)
+
+    async def insert_access_request(self, _task_id, document):
+        self.access_requests[document["access_request_id"]] = document
+
+    async def update_access_request(self, _task_id, document, expected_version):
+        assert document["version"] == expected_version + 1
+        self.access_requests[document["access_request_id"]] = document
+
+    async def list_actions(self, _task_id):
+        return [self.action] if self.action else []
+
+    async def get_action(self, _task_id, action_id):
+        if self.action and self.action["action_id"] == action_id:
+            return self.action
+        return None
+
+    async def update_action(self, _task_id, action):
+        self.action = action
+
+    async def insert_action(self, _task_id, action):
+        self.action = action
+
+    async def insert_notification(self, notification):
+        self.notifications.append(notification)
+
+    async def resolve_notifications(
+        self, task_id, *, kinds, subject_type, subject_id, resolved_at, resolution_reason
+    ):
+        resolved = []
+        for notification in self.notifications:
+            if (
+                notification["task_id"] == task_id
+                and notification["kind"] in kinds
+                and notification.get("lifecycle_status", "active") == "active"
+                and notification.get("subject_type") == subject_type
+                and notification.get("subject_id") == subject_id
+            ):
+                notification.update(
+                    lifecycle_status="resolved",
+                    resolved_at=resolved_at.isoformat(),
+                    resolution_reason=resolution_reason,
+                )
+                resolved.append(notification.copy())
+        return resolved
+
+    async def get_feedback(self, task_id, source_event_id):
+        return self.feedback.get((task_id, source_event_id))
+
+    async def upsert_feedback(self, task_id, feedback, accepted_at):
+        self.feedback[(task_id, feedback["source_event_id"])] = {
+            "source_revision": feedback["source_revision"],
+            "document": feedback,
+            "accepted_at": accepted_at,
+        }
+
+    async def get_active_run(self, task_id):
+        return next((run for run in sorted(self.runs.values(), key=lambda item: item["generation"], reverse=True)
+                     if run["task_id"] == task_id and run["status"] in {"queued", "claimed", "running"}), None)
+
+    async def get_latest_run(self, task_id):
+        return next((run for run in sorted(self.runs.values(), key=lambda item: item["generation"], reverse=True) if run["task_id"] == task_id), None)
+
+    async def get_run(self, run_id):
+        return self.runs.get(run_id)
+
+    async def require_execution_storage(self):
+        return None
+
+    async def update_run(self, run, expected_revision):
+        assert run["revision"] == expected_revision + 1
+        self.runs[run["run_id"]] = run
+        self.run = max(self.runs.values(), key=lambda item: item["generation"])
+
+    async def insert_run(self, run):
+        self.runs[run["run_id"]] = run
+        self.run = max(self.runs.values(), key=lambda item: item["generation"])
+
+    async def append_event(self, **event):
+        self.events.append(event)
+        return len(self.events)
+
+    async def current_event_cursor(self):
+        return len(self.events)
+
+    async def list_presentation_events(self, task_id, after, through, limit):
+        return [{"event_seq": index, **event} for index, event in enumerate(self.events, 1)
+                if after < index <= through and event["topic"] == "operational.presentation.requested"
+                and event["aggregate_id"] == task_id][:limit]
+    async def access_requests_available(self):
+        return True
+
+    async def fail_run_messages(self, run_id, reason, now):
+        return None
+
+    async def try_lock(self, key):
+        await self.lock(key)
+        return True
+
+    async def pending_calculation_jobs(self, task_id, run_id, execution_request_id):
+        return []
+
+    async def pending_run_messages(self, run_id):
+        return []
+
+    async def list_investigation_alerts(self, task_id):
+        return []
+
+    async def list_questions(self, task_id):
+        return [deepcopy(q) for q in self.questions.values() if q["task_id"] == task_id]
+
+    async def get_question(self, task_id, question_id):
+        q = self.questions.get(question_id)
+        return deepcopy(q) if q and q["task_id"] == task_id else None
+
+    async def save_question(self, task_id, document):
+        assert document["task_id"] == task_id
+        self.questions[document["question_request_id"]] = deepcopy(document)
diff --git a/tests/unit/test_agent_messages.py b/tests/unit/test_agent_messages.py
index c4a50f1..5dc703f 100644
--- a/tests/unit/test_agent_messages.py
+++ b/tests/unit/test_agent_messages.py
@@ -44,6 +44,12 @@ class Repository:
     async def get_task(self, task_id):
         return {"task_type": "event_investigation"} if task_id == str(self.task_id) else None
 
+    async def list_questions(self, task_id):
+        return []
+
+    async def pending_calculation_jobs(self, task_id, run_id, execution_request_id):
+        return []
+
     async def get_message(self, message_id):
         return self.messages.get(message_id)
 
diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py
index 4746170..19100bc 100644
--- a/tests/unit/test_operational_tasks.py
+++ b/tests/unit/test_operational_tasks.py
@@ -1,3 +1,4 @@
+from tests.support.operational_task_repository import TaskRepository, _context, _task_document, _active_run
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
@@ -114,11 +115,6 @@ async def test_access_request_read_uses_table_after_migration():
     assert connection.execute.await_count == 2
 
 
-def _context(user_id=None, *, system_role="user"):
-    return ProjectContext(
-        project_id=uuid4(), project_code="test", user_id=user_id or uuid4(),
-        project_role="member", system_role=system_role,
-    )
 
 
 @pytest.mark.anyio
@@ -232,208 +228,12 @@ async def test_targeted_notification_cannot_be_acknowledged_without_permission()
     assert exc.value.code == "PERMISSION_DENIED"
 
 
-def _task_document(context, now, *, lease=None):
-    task_id = uuid4()
-    return OperationalTaskSummary(
-        execution_scope="production",
-        project_id=context.project_id,
-        task_id=task_id,
-        task_type="event_investigation",
-        workflow="water_quality_complaint",
-        status="running",
-        priority="normal",
-        revision=2,
-        execution_generation=1,
-        execution_request_id=uuid4(),
-        title="调查",
-        summary="验证状态机",
-        source={"kind": "operator", "event_kind": "water_quality_complaint"},
-        trigger={"kind": "manual", "occurred_at": now, "detail": "test"},
-        scheduled_for=None,
-        agent_session_id="session-1",
-        agent_lease_expires_at=lease,
-        current_plan_revision=None,
-        current_report_revision=None,
-        created_at=now,
-        updated_at=now,
-    ).model_dump(mode="json")
 
 
-class TaskRepository:
-    def __init__(self, task, *, action=None):
-        self.task = task
-        self.action = action
-        self.requests = {}
-        self.evidence = {}
-        self.feedback = {}
-        self.plans = []
-        self.reports = []
-        self.progress = {}
-        self.notifications = []
-        self.locks = []
-        self.access_requests = {}
-
-    @asynccontextmanager
-    async def transaction(self):
-        yield self
-
-    async def lock(self, key):
-        self.locks.append(key)
-
-    async def get_request(self, scope, request_id):
-        return self.requests.get((scope, request_id))
-
-    async def save_request(self, scope, request_id, request_digest, response):
-        self.requests[(scope, request_id)] = {
-            "request_digest": request_digest,
-            "response": response,
-        }
-
-    async def get_task(self, task_id):
-        return self.task if task_id == self.task["task_id"] else None
-
-    async def update_task(self, task):
-        self.task = task
-
-    async def get_latest_plan(self, _task_id):
-        return self.plans[-1] if self.plans else None
-
-    async def get_latest_report(self, _task_id):
-        return self.reports[-1] if self.reports else None
-
-    async def get_latest_progress(self, task_id, generation):
-        return self.progress.get((task_id, generation))
-
-    async def save_progress(self, task_id, snapshot):
-        self.progress[(task_id, snapshot["execution_generation"])] = snapshot
-
-    async def get_report(self, _task_id, revision):
-        return next((item for item in self.reports if item["revision"] == revision), None)
-
-    async def list_reports(self, _task_id, limit, offset):
-        return list(reversed(self.reports))[offset:offset + limit]
-
-    async def insert_plan(self, _task_id, snapshot):
-        self.plans.append(snapshot)
-
-    async def insert_report(self, _task_id, snapshot):
-        self.reports.append(snapshot)
-
-    async def list_evidence(self, _task_id):
-        return list(self.evidence.values())
-
-    async def get_evidence(self, _task_id, evidence_id):
-        return self.evidence.get(evidence_id)
-
-    async def insert_evidence(self, _task_id, evidence):
-        self.evidence[evidence["evidence_id"]] = evidence
-
-    async def list_access_requests(self, _task_id):
-        return list(self.access_requests.values())
-
-    async def get_access_request(self, _task_id, access_request_id):
-        return self.access_requests.get(access_request_id)
-
-    async def insert_access_request(self, _task_id, document):
-        self.access_requests[document["access_request_id"]] = document
-
-    async def update_access_request(self, _task_id, document, expected_version):
-        assert document["version"] == expected_version + 1
-        self.access_requests[document["access_request_id"]] = document
-
-    async def list_actions(self, _task_id):
-        return [self.action] if self.action else []
-
-    async def get_action(self, _task_id, action_id):
-        if self.action and self.action["action_id"] == action_id:
-            return self.action
-        return None
-
-    async def update_action(self, _task_id, action):
-        self.action = action
-
-    async def insert_action(self, _task_id, action):
-        self.action = action
-
-    async def insert_notification(self, notification):
-        self.notifications.append(notification)
-
-    async def resolve_notifications(
-        self, task_id, *, kinds, subject_type, subject_id, resolved_at, resolution_reason
-    ):
-        resolved = []
-        for notification in self.notifications:
-            if (
-                notification["task_id"] == task_id
-                and notification["kind"] in kinds
-                and notification.get("lifecycle_status", "active") == "active"
-                and notification.get("subject_type") == subject_type
-                and notification.get("subject_id") == subject_id
-            ):
-                notification.update(
-                    lifecycle_status="resolved",
-                    resolved_at=resolved_at.isoformat(),
-                    resolution_reason=resolution_reason,
-                )
-                resolved.append(notification.copy())
-        return resolved
-
-    async def get_feedback(self, task_id, source_event_id):
-        return self.feedback.get((task_id, source_event_id))
-
-    async def upsert_feedback(self, task_id, feedback, accepted_at):
-        self.feedback[(task_id, feedback["source_event_id"])] = {
-            "source_revision": feedback["source_revision"],
-            "document": feedback,
-            "accepted_at": accepted_at,
-        }
 
 
-class RunTaskRepository(TaskRepository):
-    def __init__(self, task, run):
-        super().__init__(task)
-        self.run = run
-        self.events = []
-
-    async def get_active_run(self, task_id):
-        if task_id == self.task["task_id"] and self.run["status"] in {"queued", "claimed", "running"}:
-            return self.run
-        return None
-
-    async def get_run(self, run_id):
-        return self.run if run_id == self.run["run_id"] else None
-
-    async def require_execution_storage(self):
-        return None
-
-    async def update_run(self, run, expected_revision):
-        assert run["revision"] == expected_revision + 1
-        self.run = run
-
-    async def insert_run(self, run):
-        self.run = run
-
-    async def append_event(self, **event):
-        self.events.append(event)
-        return len(self.events)
-
-    async def current_event_cursor(self):
-        return len(self.events)
-
-    async def list_presentation_events(self, task_id, after, through, limit):
-        return [{"event_seq": index, **event} for index, event in enumerate(self.events, 1)
-                if after < index <= through and event["topic"] == "operational.presentation.requested"
-                and event["aggregate_id"] == task_id][:limit]
 
 
-def _active_run(task, now):
-    return OperationalTaskRun(
-        run_id=uuid4(), task_id=task["task_id"], generation=task["execution_generation"],
-        execution_request_id=task["execution_request_id"], status="running", revision=2,
-        agent_session_id="session-1", lease_expires_at=now + timedelta(minutes=1),
-        heartbeat_at=now, started_at=now, dispatch_attempts=0,
-        next_dispatch_at=now, created_at=now, updated_at=now,
-    ).model_dump(mode="json")
 
 
 @pytest.mark.anyio
@@ -457,7 +257,7 @@ async def test_run_claim_persists_agent_instance_ownership(task_scope, worker_sc
     run = _active_run(task, now)
     run.update(status="queued", revision=1, agent_session_id=None,
                lease_expires_at=None, heartbeat_at=None, started_at=None)
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now,
                                      execution_scope=worker_scope)
 
@@ -488,7 +288,7 @@ async def test_progress_can_publish_without_conclusion_and_replaces_the_complete
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     first = AgentArtifactsCommand.model_validate({
         "request_id": str(uuid4()), "source_session_id": "session-1", "run_id": repository.run["run_id"],
@@ -535,7 +335,7 @@ async def test_reading_publications_reject_stale_execution_ownership(kind, inval
         run["lease_expires_at"] = now.isoformat()
     elif invalid == "finished":
         run["status"] = "succeeded"
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     identity = {"request_id": uuid4(), "run_id": uuid4() if invalid == "run" else run["run_id"],
                 "source_session_id": "other-session" if invalid == "session" else "session-1"}
@@ -556,7 +356,7 @@ async def test_progress_unknown_evidence_does_not_publish_or_authorize_actions()
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     with pytest.raises(DomainError):
         await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
@@ -582,7 +382,7 @@ async def test_presentation_cursor_and_event_are_read_only_and_idempotent():
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     baseline = await service.presentation_cursor(task["task_id"])
     assert baseline.event_cursor == 0 and str(baseline.active_run.run_id) == repository.run["run_id"]
@@ -608,7 +408,7 @@ async def test_replay_investigation_can_publish_chart_and_propose_work_order():
     task["source"]["replay_context"] = {
         "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
     }
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
         source="query_scada_readings", source_ref="mcp:frozen-scada", queried_at=now,
@@ -645,7 +445,7 @@ async def test_comparison_chart_publication_is_validated_at_both_write_boundarie
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
         source="query_scada_readings", source_ref="mcp:comparison", queried_at=now,
@@ -715,7 +515,7 @@ async def test_replay_cannot_propose_approve_or_submit_device_commands(operation
         required_execution_permission="command.execute", approval_status="pending",
         execution_status="not_submitted", proposed_by="principal:test", proposed_at=now,
     )
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.action = action.model_dump(mode="json")
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     dispatcher = SimpleNamespace(dispatch=AsyncMock())
@@ -743,7 +543,7 @@ async def test_unified_commands_record_canonical_events_without_mutating_task(pa
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     command = TaskPresentationEventCommand(
         request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1", command=payload,
@@ -762,7 +562,7 @@ async def test_saved_mcp_coverage_envelope_publishes_groups_without_agent_geomet
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     coverage = build_node_coverage(
         [{"group_id": "one", "title": "第一组", "node_ids": ["A"]},
@@ -795,7 +595,7 @@ async def test_overlay_event_uses_stored_coordinates_and_rejects_untrusted_or_ov
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     snapshot = TaskDataSnapshot(source="scada_devices", source_ref="verified-points", queried_at=now,
                                 origin="system_data", truncated=False,
@@ -844,7 +644,7 @@ async def test_completed_investigation_resumes_once_for_linked_new_field_feedbac
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     _seed_concluded_report(repository, now)
     action = await service.propose_action(task["task_id"], ActionProposalCommand(
@@ -891,7 +691,7 @@ async def test_completed_investigation_rejects_unlinked_or_initial_child_feedbac
     task.update(status="completed", lifecycle_status="completed")
     run = _active_run(task, now)
     run["status"] = "succeeded"
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     with pytest.raises(DomainError) as unlinked:
         await service.external_feedback(task["task_id"], ExternalFeedbackCommand(
@@ -923,7 +723,7 @@ async def test_original_complaint_feedback_advances_its_source_revision():
                 source={"kind": "external_system", "external_id": source_id, "external_revision": 1})
     run = _active_run(task, now)
     run["status"] = "succeeded"
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     command = ExternalFeedbackCommand(request_id=uuid4(), source_event_id=f"peripheral_sample:{source_id}",
                                       source_revision=2, status="in_progress", occurred_at=now, detail="新投诉信息")
@@ -940,7 +740,7 @@ async def test_run_heartbeat_does_not_change_task_business_revision():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
     updated = await service.heartbeat_run(
@@ -962,7 +762,7 @@ async def test_run_heartbeat_rejects_a_different_agent_instance_with_same_sessio
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     run = _active_run(task, now)
     run["agent_instance_id"] = "agent-a"
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
     with pytest.raises(DomainError) as mismatch:
@@ -982,7 +782,7 @@ async def test_complete_without_same_generation_artifacts_fails_run_explicitly()
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
     completed = await service.complete_run(
@@ -1014,7 +814,7 @@ async def test_external_feedback_coalesces_into_an_already_queued_run():
         status="queued", revision=1, agent_session_id=None,
         lease_expires_at=None, heartbeat_at=None, started_at=None,
     )
-    repository = RunTaskRepository(task, run)
+    repository = TaskRepository(task, run)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
     result = await service.external_feedback(
@@ -1069,7 +869,7 @@ async def test_scada_final_report_requires_complete_current_input_not_selected_d
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     task["workflow"] = "scada_anomaly"
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     data = {"devices": [{"device_id": "P1"}], "readings": [], "coverage": {
         "mode": "all_scada", "query_complete": True,
@@ -1124,7 +924,7 @@ async def test_insufficient_data_exits_successfully_and_blocks_actions_until_new
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     command = AgentArtifactsCommand(
         request_id=uuid4(), source_session_id="session-1",
@@ -1179,7 +979,7 @@ async def test_insufficient_report_rejects_inconsistent_context(defect, expected
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     report = _insufficient_report()
     node = {"id": "read", "title": "查询", "owner": "agent", "status": "skipped", "detail": "关键数据不足"}
@@ -1229,7 +1029,7 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
         task["source"]["replay_context"] = {
             "replay_id": str(uuid4()), "knowledge_cutoff": incident.isoformat(), "batch_sequence": 3,
         }
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.connection = object()
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     monkeypatch.setattr(service, "_resolve_investigation_alerts", AsyncMock())
@@ -1308,8 +1108,10 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
     assert repository.evidence[revised_id]["data_snapshot"]["data"]["verified_through"] == (returned + timedelta(minutes=30)).isoformat()
     evidence_id = revised_id
 
-    repository.task.update(status="running", agent_session_id="session-1")
-    repository.run = _active_run(repository.task, now)
+    await service.claim_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
+        request_id=repository.run["execution_request_id"], expected_revision=repository.run["revision"],
+        source_session_id="session-1",
+    ))
     _seed_concluded_report(repository, now)
     with pytest.raises(DomainError) as rejected:
         await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
@@ -1334,7 +1136,7 @@ async def test_plan_only_keeps_lease_until_same_generation_report_closes_run():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
     planned = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
@@ -1369,7 +1171,7 @@ async def test_interim_report_cannot_finish_even_with_a_completed_plan():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     plan = _waiting_plan()
     plan["nodes"][0]["status"] = "completed"
@@ -1410,7 +1212,11 @@ async def test_report_is_not_blocked_by_a_historical_plan():
         request_id=uuid4(), source_session_id="session-1", report=_current_report(),
     ))
     assert result.task.latest_report.execution_generation == 2
-    assert result.task.status == "completed"
+    assert result.task.status == "running"
+    completed = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
+        request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1",
+    ))
+    assert completed.status == "completed"
 
 
 @pytest.mark.anyio
@@ -1418,7 +1224,7 @@ async def test_report_and_action_without_plan_finish_with_real_approval_pending(
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
         request_id=uuid4(), source_session_id="session-1", report=_current_report(),
@@ -1441,7 +1247,7 @@ async def test_insufficient_data_without_graph_or_node_ends_the_round():
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     report = _insufficient_report()
     report["data_gaps"][0].pop("node_id")
@@ -1463,7 +1269,7 @@ async def test_report_history_and_notifications_keep_selected_and_current_refere
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     for conclusion in ["第一次判断", "更新后的判断"]:
         await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
@@ -1499,7 +1305,7 @@ async def test_final_report_archives_its_progress_diagram_across_new_rounds():
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     diagram = {"title": "调查过程", "mermaid": "flowchart LR\nA[核查] --> B[形成判断]",
                "node_refs": [{"node_id": "B", "label": "形成判断", "status": "running"}]}
@@ -1526,7 +1332,7 @@ async def test_legacy_report_diagram_reads_only_matching_prepublication_progress
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     repository.reports.append({
         "revision": 1, "execution_generation": 1, "source_session_id": "session-1",
@@ -1559,7 +1365,7 @@ async def test_optional_diagram_cannot_forge_business_references(node_ref, code)
     now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     with pytest.raises(DomainError) as invalid:
         await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
@@ -1575,7 +1381,7 @@ async def test_action_proposal_keeps_running_lease_until_plan_and_report_close_r
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
         request_id=uuid4(), source_session_id="session-1", plan=_waiting_plan(),
@@ -1624,7 +1430,7 @@ async def test_rejected_action_requeues_task_in_a_new_execution_generation():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now).copy())
+    repository = TaskRepository(task, _active_run(task, now).copy())
     service = OperationalTaskService(
         repository,
         context,
@@ -1673,7 +1479,7 @@ async def test_action_decision_rejects_a_stale_expected_version():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(
         repository,
         context,
@@ -1737,7 +1543,7 @@ async def test_task_access_grant_is_project_scoped_reviewable_and_requeues():
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     initial_revision = task["revision"]
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     agent = OperationalTaskService(repository, context, set(), clock=lambda: now)
     request = await agent.request_access(task["task_id"], TaskAccessRequestCreate(
         request_id=uuid4(), source_session_id="session-1", node_id="read-scada",
@@ -1771,7 +1577,7 @@ async def test_task_access_request_rejects_a_second_pending_request():
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     await service.request_access(task["task_id"], TaskAccessRequestCreate(
         request_id=uuid4(), source_session_id="session-1", node_id="read-scada",
@@ -1794,7 +1600,7 @@ async def test_task_access_request_fails_closed_before_project_migration():
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.access_requests_available = AsyncMock(return_value=False)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
@@ -1813,7 +1619,7 @@ async def test_completed_gate_requires_access_request_decision():
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     request = await service.request_access(task["task_id"], TaskAccessRequestCreate(
         request_id=uuid4(), source_session_id="session-1", node_id="read-scada",
@@ -1850,7 +1656,7 @@ async def test_work_order_draft_edit_creates_new_review_version(replay):
         task["source"]["replay_context"] = {
             "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
         }
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now)
     _seed_concluded_report(repository, now)
     action = await service.propose_action(task["task_id"], ActionProposalCommand(
@@ -1881,7 +1687,7 @@ async def test_one_click_work_order_confirmation_replays_without_duplicate_dispa
         task["source"]["replay_context"] = {
             "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
         }
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {
         OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
     }, clock=lambda: now)
@@ -2125,7 +1931,7 @@ async def test_expired_owned_lease_is_rejected_with_stable_code():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now - timedelta(seconds=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.run["lease_expires_at"] = now - timedelta(seconds=1)
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
 
@@ -2253,7 +2059,7 @@ async def test_interactive_report_publication_is_idempotent_and_versioned_with_e
     now = datetime(2026, 10, 20, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     report = {**rich_report(), "stage": "final"}
     command = AgentArtifactsCommand(
@@ -2290,7 +2096,7 @@ async def test_presentation_history_is_task_scoped_paginated_and_readable_after_
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now)
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.run["status"] = "succeeded"
     for index in range(4):
         await repository.append_event(
@@ -2322,7 +2128,7 @@ async def test_final_report_rejected_while_original_run_has_pending_calculations
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     pending_job = str(uuid4())
     repository.pending_calculation_jobs = AsyncMock(return_value=[pending_job])
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
@@ -2346,7 +2152,7 @@ async def test_completion_rejected_until_calculations_reach_terminal_state():
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.pending_calculation_jobs = AsyncMock(return_value=[str(uuid4())])
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     command = AgentRunCommand(request_id=uuid4(), expected_revision=repository.run['revision'], source_session_id='session-1')
@@ -2380,7 +2186,7 @@ async def test_interim_report_can_explain_progress_while_calculation_is_pending(
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repository = RunTaskRepository(task, _active_run(task, now))
+    repository = TaskRepository(task, _active_run(task, now))
     repository.pending_calculation_jobs = AsyncMock(return_value=[str(uuid4())])
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     result = await service.publish_artifacts(task['task_id'], AgentArtifactsCommand(
diff --git a/tests/unit/test_project_scada_metadata.py b/tests/unit/test_project_scada_metadata.py
index ffd847a..67cedaa 100644
--- a/tests/unit/test_project_scada_metadata.py
+++ b/tests/unit/test_project_scada_metadata.py
@@ -1,10 +1,10 @@
 import asyncio
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock
-from uuid import uuid4
 
 from app.api.v1.endpoints import scada as scada_endpoint
-from app.services import timeseries_analysis as composite_queries
+from app.services import timeseries_history as history
+from app.domain.schemas.timeseries_history import ElementHistoryQuery
 
 
 PROJECT_SCADA = {
@@ -26,107 +26,22 @@ START_TIME = datetime(2026, 6, 1, tzinfo=timezone.utc)
 END_TIME = datetime(2026, 6, 2, tzinfo=timezone.utc)
 
 
-def _patch_project_scadas(monkeypatch):
-    monkeypatch.setattr(
-        composite_queries.ScadaInfoRepository,
-        "get_scadas",
-        AsyncMock(return_value=[PROJECT_SCADA.copy()]),
-    )
-
-
-def test_realtime_scada_simulation_uses_current_project_metadata(monkeypatch):
-    _patch_project_scadas(monkeypatch)
-    query_mock = AsyncMock(
-        return_value={"J1": [{"time": START_TIME, "value": 26.5}]}
-    )
-    monkeypatch.setattr(
-        composite_queries.RealtimeRepository,
-        "get_node_fields_by_ids_time_range",
-        query_mock,
-    )
-    monkeypatch.setattr(
-        composite_queries.RealtimeRepository,
-        "get_link_fields_by_ids_time_range",
-        AsyncMock(return_value={}),
-    )
-
-    result = asyncio.run(
-        composite_queries.TimeseriesAnalysisService.get_scada_associated_realtime_simulation_data(
-            object(),
-            object(),
-            [PROJECT_SCADA["device_id"]],
-            START_TIME,
-            END_TIME,
-        )
-    )
-
-    assert result[PROJECT_SCADA["device_id"]][0]["scada_id"] == PROJECT_SCADA["device_id"]
-    assert query_mock.await_count == 1
-    assert query_mock.await_args.args[1:] == (
-        START_TIME,
-        END_TIME,
-        ["J1"],
-        "pressure",
-    )
-
-
-def test_analysis_scada_simulation_uses_current_project_metadata(monkeypatch):
-    _patch_project_scadas(monkeypatch)
-    async def query_series(_conn, _run_id, element_type, element_ids, *_args):
-        if element_type == "node":
-            return {"J1": [{"time": START_TIME, "value": 26.5}]}
-        return {}
-
-    query_mock = AsyncMock(side_effect=query_series)
-    monkeypatch.setattr(
-        composite_queries.AnalysisResultsRepository,
-        "get_series_by_ids",
-        query_mock,
-    )
-
-    result = asyncio.run(
-        composite_queries.TimeseriesAnalysisService.get_scada_associated_analysis_simulation_data(
-            object(),
-            object(),
-            [PROJECT_SCADA["device_id"]],
-            START_TIME,
-            END_TIME,
-            uuid4(),
-        )
-    )
-
-    assert result[PROJECT_SCADA["device_id"]][0]["scada_id"] == PROJECT_SCADA["device_id"]
-    assert query_mock.await_count == 2
-    node_call = next(
-        call for call in query_mock.await_args_list if call.args[2] == "node"
-    )
-    assert node_call.args[3:] == (["J1"], START_TIME, END_TIME, "pressure")
-
-
-def test_element_scada_query_uses_current_project_metadata(monkeypatch):
-    _patch_project_scadas(monkeypatch)
-    query_mock = AsyncMock(
-        return_value={
-            PROJECT_SCADA["device_id"]: [{"time": START_TIME, "value": 26.5}]
-        }
-    )
-    monkeypatch.setattr(
-        composite_queries.ScadaRepository,
-        "get_scada_field_by_id_time_range",
-        query_mock,
-    )
-
-    result = asyncio.run(
-        composite_queries.TimeseriesAnalysisService.get_element_associated_scada_data(
-            object(),
-            object(),
-            "J1",
-            START_TIME,
-            END_TIME,
-        )
-    )
-
-    assert result == {"J1": [{"time": START_TIME, "value": 26.5}]}
+def test_history_uses_current_project_metadata(monkeypatch):
+    postgres = object()
+    mappings = AsyncMock(return_value=[PROJECT_SCADA.copy()])
+    monkeypatch.setattr(history.ScadaInfoRepository, "get_scadas_for_elements", mappings)
+    monkeypatch.setattr(history.ScadaRepository, "get_scada_by_ids_time_range", AsyncMock(return_value=[{
+        "time": START_TIME, "device_id": PROJECT_SCADA["device_id"],
+        "monitored_value": 26.5, "cleaned_value": 26.0,
+    }]))
+    result = asyncio.run(history.TimeseriesHistoryService.query(object(), postgres, ElementHistoryQuery(
+        start_time=START_TIME, end_time=END_TIME,
+        elements=[{"element_id": "J1", "element_type": "junction", "device_ids": [PROJECT_SCADA["device_id"]]}],
+    )))
+    assert mappings.await_args.args == (postgres, ["J1"], [])
+    assert len(result.series) == 2
+    assert all(item.device_id == PROJECT_SCADA["device_id"] and item.element_id == "J1" for item in result.series)
+    assert [item.points[0].value for item in result.series] == [26.5, 26.0]
 
 
 def test_scada_info_endpoint_uses_current_project_connection(monkeypatch):
diff --git a/tests/unit/test_task_questions.py b/tests/unit/test_task_questions.py
index 3f52e1a..86c8520 100644
--- a/tests/unit/test_task_questions.py
+++ b/tests/unit/test_task_questions.py
@@ -1,6 +1,7 @@
 from copy import deepcopy
 from datetime import datetime, timedelta, timezone
 from uuid import UUID, uuid4
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -10,7 +11,7 @@ from app.domain.schemas.operational_task import AgentRunCommand, AgentRunFailure
 from app.domain.schemas.task_question import TaskQuestionCreate, TaskQuestionReply
 from app.services.operational_tasks import OperationalTaskService
 from app.services.task_questions import create_question, answer_question
-from tests.unit.test_operational_tasks import RunTaskRepository, _context, _task_document, _active_run
+from tests.support.operational_task_repository import TaskRepository, _context, _task_document, _active_run
 
 
 @pytest.fixture
@@ -18,28 +19,11 @@ def anyio_backend():
     return "asyncio"
 
 
-class QuestionRepository(RunTaskRepository):
-    def __init__(self, task, run):
-        super().__init__(task, run)
-        self.questions = {}
-
-    async def list_questions(self, task_id):
-        return [deepcopy(q) for q in self.questions.values() if q["task_id"] == task_id]
-
-    async def get_question(self, task_id, question_id):
-        q = self.questions.get(question_id)
-        return deepcopy(q) if q and q["task_id"] == task_id else None
-
-    async def save_question(self, task_id, document):
-        assert document["task_id"] == task_id
-        self.questions[document["question_request_id"]] = deepcopy(document)
-
-
 def setup():
     now = datetime(2026, 10, 30, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
-    repo = QuestionRepository(task, _active_run(task, now))
+    repo = TaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repo, context, {OPERATIONAL_TASK_EXECUTE, OPERATIONAL_TASK_WRITE}, clock=lambda: now)
     command = TaskQuestionCreate(request_id=uuid4(), source_session_id="session-1", run_id=repo.run["run_id"],
         questions=[dict(question_id="flow", title="漏水量依据", body="是否有计量记录？", options=[{"id": "unknown", "label": "目前未知"}],
@@ -162,6 +146,34 @@ async def test_real_execution_failure_cancels_question_instead_of_disguising_fai
     assert repo.notifications[0]["lifecycle_status"] == "resolved"
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("timeout_count", [1, 3])
+async def test_lease_timeout_cancels_old_question_and_unblocks_the_next_generation(timeout_count):
+    service, repo, task_id, command = setup()
+    question = await create_question(service, task_id, command)
+    service.clock = lambda: question.created_at + timedelta(minutes=2)
+    repo.claim_expired_runs = AsyncMock(return_value=[deepcopy(repo.run)])
+    repo.consecutive_timed_out_runs = AsyncMock(return_value=timeout_count)
+
+    await service.recover_expired_leases(1)
+
+    assert repo.questions[str(question.question_request_id)]["status"] == "cancelled"
+    assert repo.notifications[0]["lifecycle_status"] == "resolved"
+    with pytest.raises(DomainError):
+        await answer_question(service, task_id, question.question_request_id, reply())
+    if timeout_count == 3:
+        assert repo.task["lifecycle_status"] == "failed"
+        return
+
+    await service.claim_run(task_id, UUID(repo.run["run_id"]), AgentRunCommand(
+        request_id=UUID(repo.run["execution_request_id"]), expected_revision=1,
+        source_session_id="session-2"))
+    next_question = await create_question(service, task_id, command.model_copy(update={
+        "request_id": uuid4(), "run_id": UUID(repo.run["run_id"]), "source_session_id": "session-2"}))
+    assert next_question.status == "pending"
+    assert next_question.execution_generation == 2
+
+
 @pytest.mark.anyio
 async def test_multiple_selection_and_optional_question_are_preserved():
     service, repo, task_id, command = setup()