From b853b61a0b7a92a635c875e55986871a1857fb03 Mon Sep 17 00:00:00 2001 From: Huarch Date: Wed, 11 Nov 2026 19:10:41 +0800 Subject: [PATCH] feat(investigation): preserve cumulative reference graphs Accumulate investigation paths and preserve failed attempts across rounds. Keep graph annotations separate from execution state, synchronize the API contract, and consolidate documentation. --- README.md | 4 +- app/domain/schemas/operational_task.py | 68 +- app/infra/db/postgresql/operational_tasks.py | 31 + app/infra/db/postgresql/read_queries.py | 34 +- app/mcp/catalog.py | 2 +- app/services/investigation_graph.py | 94 ++ app/services/operational_tasks.py | 66 +- contracts/manifest.json | 2 +- contracts/server-v1.openapi.json | 239 ++++- docs/README.md | 56 + docs/api-style.md | 3 +- docs/dynamic-investigation-plan.md | 6 +- .../history/backend-naming-audit.md | 2 + .../history/retired-api-investigation.md | 4 +- ...investigation-progress-and-presentation.md | 81 +- docs/investigation-report-contract.md | 6 +- docs/mcp-query-design.md | 4 +- docs/mcp-query-implementation.md | 6 +- docs/operational-task-workers.md | 2 +- docs/peripheral-generated-data.md | 5 +- docs/peripheral-sample.md | 53 + docs/peripheral-work-order-events.md | 50 - docs/scada-peripheral-replay-design.md | 2 +- docs/workbench-event-refinement.md | 2 +- docs/workbench-operations-storage.md | 15 - docs/workbench-runtime.md | 18 +- scripts/migrate_investigation_graphs.py | 182 ++++ scripts/missing_api_endpoints.md | 2 + tests/fixtures/chart-publication-cases.json | 992 ++++++++++++++++++ tests/fixtures/unified-presentation.jsonl | 14 + ...test_investigation_reference_projection.py | 84 ++ tests/support/operational_task_repository.py | 13 + tests/unit/test_chart_publication_examples.py | 48 + .../unit/test_investigation_diagram_status.py | 4 +- tests/unit/test_investigation_graph.py | 315 ++++++ .../test_investigation_graph_migration.py | 45 + .../test_investigation_reference_states.py | 27 + tests/unit/test_operational_tasks.py | 11 +- 38 files changed, 2445 insertions(+), 147 deletions(-) create mode 100644 app/services/investigation_graph.py create mode 100644 docs/README.md rename BACKEND_NAMING_AUDIT.md => docs/history/backend-naming-audit.md (94%) rename 失效API排查.md => docs/history/retired-api-investigation.md (89%) delete mode 100644 docs/peripheral-work-order-events.md delete mode 100644 docs/workbench-operations-storage.md create mode 100644 scripts/migrate_investigation_graphs.py create mode 100644 tests/fixtures/chart-publication-cases.json create mode 100644 tests/fixtures/unified-presentation.jsonl create mode 100644 tests/integration/test_investigation_reference_projection.py create mode 100644 tests/unit/test_chart_publication_examples.py create mode 100644 tests/unit/test_investigation_graph.py create mode 100644 tests/unit/test_investigation_graph_migration.py create mode 100644 tests/unit/test_investigation_reference_states.py diff --git a/README.md b/README.md index a4034bc..951b846 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ `next-tjwater-server` 是 NextTJWater 供水系统的 FastAPI 后端，提供认证、项目、管网、模拟、爆管、漏损、SCADA、地图服务、工况任务和受控 MCP 能力。业务数据和时序数据按已认证项目路由，后台工作进程也必须使用明确的项目范围。 +协议、工作进程、数据库、SCADA 与历史资料索引见 [文档目录](docs/README.md)。 + ## 技术栈 - Python 3.12 @@ -213,7 +215,7 @@ Gitea Actions 工作流位于 `.gitea/workflows/package.yml`，可手动触发 ## 产品 MCP 查询 -Server 可选启用 MCP 2026-07-28，默认关闭。当前目录包含 67 项类型化工具，其中 42 项查询、12 项计算、13 项预览式写入。MCP 使用专用 OAuth resource Bearer、宿主绑定上下文和现有项目或任务权限，不提供自由命令执行。 +Server 可选启用 MCP 2026-07-28，默认关闭。目录包含查询、分析计算、预览式写入和计算任务管理工具；可用项以 `app/mcp/catalog.py` 及当前授权目录为准。MCP 使用专用 OAuth resource Bearer、宿主绑定上下文和现有项目或任务权限，不提供自由命令执行。 写入工具默认只执行 `dry_run`。真实写入必须使用同一 `request_id`，带上预览返回的 `expected_digest`，并通过宿主原生许可和项目权限检查。超时或结果不明时先查询操作状态，不盲目重试。实施、启用和授权边界见 [MCP 实施与启用](docs/mcp-query-implementation.md) 和 [协议与旧 CLI 能力对照](docs/mcp-query-design.md)。 diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index c1e41fc..66830be 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -5,7 +5,7 @@ import re from typing import Annotated, Literal from uuid import UUID -from pydantic import Field, JsonValue, model_serializer, model_validator +from pydantic import Field, JsonValue, field_validator, model_serializer, model_validator from app.domain.schemas.answer_ui import ( CHART_SAMPLE_TIME_PATTERN, ReportUiBlock, validate_chart_sample_time, validate_report_ui_blocks, ) @@ -222,25 +222,61 @@ class InvestigationMapTarget(StrictModel): feature_id: TaskFeatureId -class InvestigationDiagramNodeRef(StrictModel): +InvestigationNodeAnnotation = Literal["reference", "recorded", "needs_input", "failed", "not_taken"] +LEGACY_INVESTIGATION_NODE_ANNOTATIONS = { + "pending": "reference", "running": "reference", "completed": "recorded", + "waiting": "needs_input", "waiting_human": "needs_input", + "skipped": "not_taken", "cancelled": "not_taken", +} + + +class _InvestigationDiagramNode(StrictModel): node_id: Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_-]{0,79}$")] label: NonEmptyStr + detail: Annotated[str, Field(min_length=1, max_length=2000)] | None = None evidence_refs: list[NonEmptyStr] = Field(default_factory=list) action_refs: list[UUID] = Field(default_factory=list) - status: Literal[ - "pending", "running", "completed", "waiting", "waiting_human", "failed", "skipped", "cancelled" - ] | None = Field(default=None, description=( - "Evidence-confirmed node state; waiting means prerequisites/external/system result, " - "waiting_human requires manual authorization/approval/feedback. " - "Unknown stays omitted/null; host controls colors." - )) map_targets: list[InvestigationMapTarget] = Field(default_factory=list, max_length=100) +class InvestigationDiagramNodeRef(_InvestigationDiagramNode): + status: InvestigationNodeAnnotation | None = Field(default=None, description=( + "A saved process-reference annotation, not live execution state. " + "recorded means a finding exists, not that a task or field action completed." + )) + + @field_validator("status", mode="before") + @classmethod + def read_legacy_annotation(cls, value: object) -> object: + return LEGACY_INVESTIGATION_NODE_ANNOTATIONS.get(value, value) if isinstance(value, str) else value + + +class InvestigationGraphNodeUpdate(_InvestigationDiagramNode): + status: InvestigationNodeAnnotation | None = Field(default=None, description=( + "Optional reference annotation: reference, recorded finding, needs_input, failed attempt, or not_taken. " + "No live node/task binding or per-step synchronization is required." + )) + + +class InvestigationGraphEdge(StrictModel): + source: Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_-]{0,79}$")] + target: Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_-]{0,79}$")] + label: Annotated[str, Field(min_length=1, max_length=200)] | None = None + + +class InvestigationGraphUpdate(StrictModel): + title: Annotated[str, Field(min_length=1, max_length=200)] | None = None + nodes: list[InvestigationGraphNodeUpdate] = Field(default_factory=list, max_length=100) + edges: list[InvestigationGraphEdge] = Field(default_factory=list, max_length=200) + + class InvestigationDiagram(StrictModel): title: NonEmptyStr mermaid: Annotated[str, Field(min_length=1, max_length=20000)] node_refs: list[InvestigationDiagramNodeRef] = Field(default_factory=list, max_length=100) + # Present only on a graph maintained by the host. Legacy Mermaid stays readable. + edges: list[InvestigationGraphEdge] | None = Field(default=None, max_length=200) + source_mermaid: Annotated[str, Field(min_length=1, max_length=20000)] | None = None @model_validator(mode="after") def validate_description(self): @@ -323,6 +359,16 @@ class InvestigationProgress(StrictModel): diagram: InvestigationDiagram | None = None +class InvestigationProgressInput(InvestigationProgress): + graph_update: InvestigationGraphUpdate | None = None + + @model_validator(mode="after") + def validate_graph_input(self): + if self.graph_update is not None and self.diagram is not None: + raise ValueError("use graph_update without a generated Mermaid diagram") + return self + + class InvestigationProgressSnapshot(StrictModel): revision: PositiveInt run_id: UUID @@ -367,7 +413,7 @@ class AgentArtifactsCommand(StrictModel): evidence: list[EvidenceInput] = Field(default_factory=list, max_length=100) plan: ExecutionPlanInput | None = None report: InvestigationReportInput | None = None - progress: InvestigationProgress | None = None + progress: InvestigationProgressInput | None = None @model_validator(mode="after") def validate_artifacts(self): @@ -377,6 +423,8 @@ class AgentArtifactsCommand(StrictModel): self.report.diagram if self.report is not None else None, self.progress.diagram if self.progress is not None else None, ): + if diagram is not None and (diagram.edges is not None or diagram.source_mermaid is not None): + raise ValueError("structured graph metadata is host-owned; use progress.graph_update") missing = _undeclared_diagram_node_refs(diagram) if missing: raise ValueError( diff --git a/app/infra/db/postgresql/operational_tasks.py b/app/infra/db/postgresql/operational_tasks.py index 407cb57..3a48cc9 100644 --- a/app/infra/db/postgresql/operational_tasks.py +++ b/app/infra/db/postgresql/operational_tasks.py @@ -628,6 +628,37 @@ class OperationalTaskRepository: row = await cursor.fetchone() return row["document"] if row else None + async def get_latest_graph_progress(self, task_id: str) -> dict | None: + availability = await self.connection.execute( + "SELECT to_regclass('workbench.operational_task_progress') AS relation" + ) + if not (await availability.fetchone())["relation"]: + return None + cursor = await self.connection.execute( + """SELECT document FROM workbench.operational_task_progress + WHERE project_id=%s AND task_id=%s + AND jsonb_typeof(document#>'{progress,diagram,edges}')='array' + ORDER BY (document->>'execution_generation')::bigint DESC, revision DESC LIMIT 1""", + (self.project_id, task_id), + ) + row = await cursor.fetchone() + return row["document"] if row else None + + async def get_latest_diagram(self, task_id: str) -> dict | None: + cursor = await self.connection.execute( + """SELECT diagram FROM ( + SELECT document#>'{progress,diagram}' AS diagram, published_at AS occurred_at + FROM workbench.operational_task_progress WHERE project_id=%s AND task_id=%s + UNION ALL + SELECT document#>'{report,diagram}', created_at + FROM workbench.operational_task_reports WHERE project_id=%s AND task_id=%s + ) AS diagrams WHERE jsonb_typeof(diagram)='object' + ORDER BY occurred_at DESC LIMIT 1""", + (self.project_id, task_id, self.project_id, task_id), + ) + row = await cursor.fetchone() + return row["diagram"] if row else None + async def save_progress(self, task_id: str, snapshot: dict) -> None: cursor = await self.connection.execute( """INSERT INTO workbench.operational_task_progress diff --git a/app/infra/db/postgresql/read_queries.py b/app/infra/db/postgresql/read_queries.py index bc47f0b..3e71499 100644 --- a/app/infra/db/postgresql/read_queries.py +++ b/app/infra/db/postgresql/read_queries.py @@ -7,7 +7,9 @@ from uuid import UUID from psycopg import AsyncConnection from psycopg import sql from psycopg.rows import dict_row +from psycopg.types.json import Jsonb from fastapi import HTTPException +from app.domain.schemas.operational_task import LEGACY_INVESTIGATION_NODE_ANNOTATIONS from app.infra.db.postgresql.scada import _SCADA_VIEW_SELECT, _device from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository @@ -329,6 +331,29 @@ async def task_context_summary(conn: AsyncConnection, *, project_id: UUID, task_ return result +def _reference_diagram_projection(stored_sql: str, stored_params: tuple, section: str) -> tuple[str, tuple]: + """Normalize only saved diagram annotations before any data_path or page projection.""" + if section not in {"progress", "report"}: + raise ValueError("reference diagrams only belong to progress or report sections") + path = "{data," + section + ",diagram,node_refs}" + statement = f""" + WITH reference_documents AS ( + {stored_sql} + ), reference_aliases AS ( + SELECT %s::jsonb AS annotations + ) + SELECT CASE WHEN jsonb_typeof(snapshot#>'{path}')='array' THEN + jsonb_set(snapshot,'{path}',COALESCE(( + SELECT jsonb_agg(CASE WHEN annotations ? (node->>'status') + THEN jsonb_set(node,'{{status}}',annotations->(node->>'status')) + ELSE node END ORDER BY position) + FROM jsonb_array_elements(snapshot#>'{path}') WITH ORDINALITY AS nodes(node,position) + ),'[]'::jsonb)) ELSE snapshot END AS snapshot + FROM reference_documents CROSS JOIN reference_aliases + """ + return statement, (*stored_params, Jsonb(LEGACY_INVESTIGATION_NODE_ANNOTATIONS)) + + async def task_context_page( conn: AsyncConnection, *, project_id: UUID, task_id: UUID, section: str, report_revision: int | None, data_path: list[str], limit: int, offset: int, @@ -386,11 +411,16 @@ async def task_context_page( stored = """SELECT jsonb_build_object('section','progress','data',p.document) AS snapshot FROM workbench.operational_task_progress p JOIN workbench.operational_task_runs r ON r.project_id=p.project_id AND r.task_id=p.task_id AND r.run_id=p.run_id - JOIN workbench.operational_tasks t ON t.project_id=p.project_id AND t.task_id=p.task_id AND t.execution_generation=r.generation - WHERE p.project_id=%s AND p.task_id=%s""" + JOIN workbench.operational_tasks t ON t.project_id=p.project_id AND t.task_id=p.task_id + WHERE p.project_id=%s AND p.task_id=%s AND r.generation<=t.execution_generation + AND (t.execution_generation=r.generation + OR jsonb_typeof(p.document#>'{progress,diagram,edges}')='array') + ORDER BY (t.execution_generation=r.generation) DESC,r.generation DESC,p.revision DESC LIMIT 1""" params = (project_id, task_id) else: raise HTTPException(422, "Unknown task context section") + if section in {"progress", "report"}: + stored, params = _reference_diagram_projection(stored, params, section) return await _snapshot_page(conn, stored_sql=stored, stored_params=params, data_path=data_path, limit=limit, offset=projection_offset, mcp_sources=[], max_bytes=8000) diff --git a/app/mcp/catalog.py b/app/mcp/catalog.py index dadd1d8..918bb79 100644 --- a/app/mcp/catalog.py +++ b/app/mcp/catalog.py @@ -217,7 +217,7 @@ _specs = [ ), ( "get_task_context", - "读取绑定任务的有界摘要及handling_skill元数据；section=skill读取Server按任务workflow分发的业务处置Skill正文与版本，属于流程指引而非事件证据；其他section选择task/report/evidence/actions/progress/access_requests/plan/run，data_path按返回路径读取嵌套字段，limit/offset分页；report_revision保留选定报告版本", + "读取绑定任务的有界摘要及handling_skill元数据；section=skill读取Server按任务workflow分发的业务处置Skill正文与版本，属于流程指引而非事件证据；其他section选择task/report/evidence/actions/progress/access_requests/plan/run，data_path按返回路径读取嵌套字段，limit/offset分页；progress/report图节点仅为流程参考标记，旧执行态兼容转换，不代表实时进度；实际执行状态读run、动作状态读actions；report_revision保留选定报告版本", m.TaskContextInput, "operational_task.view", "task", diff --git a/app/services/investigation_graph.py b/app/services/investigation_graph.py new file mode 100644 index 0000000..8f02f3b --- /dev/null +++ b/app/services/investigation_graph.py @@ -0,0 +1,94 @@ +"""Accumulate explanatory investigation paths and generate Mermaid deterministically.""" +import re +from app.domain.domain_errors import require +from app.domain.schemas.operational_task import ( + InvestigationDiagram, InvestigationDiagramNodeRef, InvestigationGraphUpdate, +) + + +def _label(value: str) -> str: + # Encode Mermaid delimiters and HTML before placing untrusted text in quotes. + replacements = {"#": "#35;", "&": "#38;", '"': "#quot;", "<": "#lt;", + ">": "#gt;", "`": "#96;", "%": "#37;", "|": "#124;", "\\": "#92;"} + return "".join(replacements.get(c, " " if c in "\r\n" else c) for c in value) + + +def adopt_legacy_investigation_graph(diagram: InvestigationDiagram) -> InvestigationDiagram: + """Make a saved legacy graph cumulative without parsing or changing its source.""" + if diagram.edges is not None: + return diagram + return diagram.model_copy(deep=True, update={"edges": [], "source_mermaid": diagram.mermaid}) + + +def update_investigation_graph( + previous: InvestigationDiagram | None, update: InvestigationGraphUpdate, +) -> InvestigationDiagram: + """Upsert nodes and append explicit edges; never delete a path or erase a failed attempt. + + A retry uses a new node id. Relationships describe activity for readers and + impose no ordering or completion requirements on the Agent. + """ + if previous is not None: + previous = adopt_legacy_investigation_graph(previous) + source = previous.source_mermaid if previous else None + def in_source(node_id: str) -> bool: + return bool(source and re.search( + rf"(?|"{_label(edge.label)}"| ' if edge.label else " --> " + lines.append(f" {edge.source}{connector}{edge.target}") + mermaid = "\n".join(lines) + require(len(mermaid) <= 20000, "GRAPH_CAPACITY_EXCEEDED", + "Investigation Mermaid exceeds 20000 characters; existing paths were retained", status=422) + return InvestigationDiagram( + title=update.title or (previous.title if previous else "调查脉络"), + mermaid=mermaid, node_refs=list(nodes.values()), edges=edges, source_mermaid=source, + ) diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 48e3949..a8e4e93 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -49,6 +49,8 @@ from app.domain.schemas.operational_task import ( InvestigationReportRevision, InvestigationReportReference, InvestigationProgressSnapshot, + InvestigationProgress, + InvestigationDiagram, IssueCommandPayload, NotificationAcknowledgeCommand, OperationalTaskCreate, @@ -71,6 +73,7 @@ from app.domain.schemas.operational_task import ( ) from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource from app.services.content_digest import digest +from app.services.investigation_graph import adopt_legacy_investigation_graph, update_investigation_graph from app.services.operational_presentation import resolve_overlay from app.services.operational_recovery import accept_completion, track_work_order from app.services.operational_scada_coverage import require_complete_scada_evidence @@ -264,6 +267,8 @@ class OperationalTaskService: active_run = await self._active_run(task) latest_run = await self.repo.get_latest_run(str(task_id)) progress = await self.repo.get_latest_progress(str(task_id), task["execution_generation"]) + if progress is None: + progress = await self.repo.get_latest_graph_progress(str(task_id)) work_order_snapshot = None if task["task_type"] == "external_work_order" and task["source"].get("external_id"): feedback = await self.repo.get_feedback( @@ -850,11 +855,27 @@ class OperationalTaskService: for item in await self.repo.list_actions(str(task_id)) } progress_revision = None + saved_diagram = None + if command.progress is not None or command.report is not None: + stored = await self.repo.get_latest_graph_progress(str(task_id)) + if stored is not None: + saved_diagram = InvestigationProgressSnapshot.model_validate(stored).progress.diagram + elif command.progress is not None: + legacy = await self.repo.get_latest_diagram(str(task_id)) + saved_diagram = InvestigationDiagram.model_validate(legacy) if legacy else None if command.progress is not None: require(command.run_id is not None, "RUN_REQUIRED", "Progress must identify its current execution run", status=422) - if command.progress.diagram: - for node in command.progress.diagram.node_refs: + diagram = command.progress.diagram + if command.progress.graph_update is not None: + diagram = update_investigation_graph(saved_diagram, command.progress.graph_update) + elif saved_diagram is not None: + require(diagram is None or saved_diagram.edges is None, "GRAPH_UPDATE_REQUIRED", + "Use graph_update to extend the saved investigation graph", status=422) + if diagram is None: + diagram = adopt_legacy_investigation_graph(saved_diagram) + if diagram: + for node in diagram.node_refs: self._require_evidence_refs(node.evidence_refs, known_evidence) require(all(str(ref) in actions for ref in node.action_refs), "ACTION_NOT_FOUND", "Progress references an unknown action", status=422) @@ -863,9 +884,11 @@ class OperationalTaskService: snapshot = InvestigationProgressSnapshot( revision=progress_revision, run_id=active_run["run_id"], execution_generation=task["execution_generation"], - source_session_id=command.source_session_id, published_at=now, progress=command.progress, + source_session_id=command.source_session_id, published_at=now, + progress=InvestigationProgress(summary=command.progress.summary, diagram=diagram), ) await self.repo.save_progress(str(task_id), _json(snapshot)) + saved_diagram = diagram await self._append_event( topic="operational.progress.changed", aggregate_type="operational_task_progress", aggregate_id=str(task_id), aggregate_revision=progress_revision, occurred_at=now, @@ -917,21 +940,31 @@ class OperationalTaskService: await self.repo.insert_plan(str(task_id), _json(snapshot)) current_plan = snapshot if command.report: - self._require_evidence_refs(command.report.evidence_refs, known_evidence) - if command.report.stage == "final": + report_input = command.report + if saved_diagram is not None: + diagram = adopt_legacy_investigation_graph(saved_diagram) + report_input = report_input.model_copy(update={ + "diagram": diagram, + "evidence_refs": list(dict.fromkeys([ + *report_input.evidence_refs, + *(ref for node in diagram.node_refs for ref in node.evidence_refs), + ])), + }) + self._require_evidence_refs(report_input.evidence_refs, known_evidence) + if report_input.stage == "final": require_complete_scada_evidence( - task, await self.repo.list_evidence(str(task_id)), command.report.evidence_refs, + task, await self.repo.list_evidence(str(task_id)), report_input.evidence_refs, ) - for gap in command.report.data_gaps: - require(set(gap.evidence_refs) <= set(command.report.evidence_refs), + for gap in report_input.data_gaps: + require(set(gap.evidence_refs) <= set(report_input.evidence_refs), "REPORT_GAP_EVIDENCE_INVALID", "Gap evidence must be included in report evidence", status=422) - if command.report.diagram: - for node in command.report.diagram.node_refs: - self._require_evidence_refs(node.evidence_refs, set(command.report.evidence_refs)) + if report_input.diagram: + for node in report_input.diagram.node_refs: + self._require_evidence_refs(node.evidence_refs, set(report_input.evidence_refs)) require(all(str(ref) in actions for ref in node.action_refs), "ACTION_NOT_FOUND", "Diagram references an unknown action", status=422) - if command.report.outcome == "insufficient_data": + if report_input.outcome == "insufficient_data": require(not any(action.approval_status == "pending" or ( action.approval_status == "approved" and action.execution_status in {"not_submitted", "submitted", "unknown"} ) for action in actions.values()), "REPORT_ACTION_PENDING", @@ -941,7 +974,7 @@ class OperationalTaskService: revision=report_revision, execution_generation=task["execution_generation"], source_session_id=command.source_session_id, - published_at=now, published_by=self._actor(), report=command.report, + published_at=now, published_by=self._actor(), report=report_input, ) report = await self._report_with_progress_diagram(task_id, report) await self.repo.insert_report(str(task_id), _json(report)) @@ -954,11 +987,12 @@ class OperationalTaskService: metadata={"task_id": str(task_id), "run_id": str(active_run["run_id"])}, ) previous_result = previous_report["report"] if previous_report else None - if command.report.stage == "final" and ( + effective_report = _json(report.report) + if report.report.stage == "final" and ( previous_report is None or previous_report.get("execution_generation") != task["execution_generation"] or previous_result.get("stage") != "final" - or any(previous_result.get(key, [] if key == "ui_blocks" else None) != _json(command.report).get(key) + or any(previous_result.get(key, [] if key == "ui_blocks" else None) != effective_report.get(key) for key in ("current_conclusion", "conclusion_boundary", "outcome", "stop_reason", "evidence_refs", "recommendations", "pending_items", "data_gaps", "ui_blocks")) ): @@ -967,7 +1001,7 @@ class OperationalTaskService: now=now, reason="report_superseded", ) await self._notify( - task_id, "report_updated", task["title"], command.report.current_conclusion, now, + task_id, "report_updated", task["title"], report.report.current_conclusion, now, report_reference=InvestigationReportReference( execution_generation=task["execution_generation"], revision=report_revision, ), diff --git a/contracts/manifest.json b/contracts/manifest.json index 50226fc..e930bd3 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "426b8d9170f253aa7132e9376031a7da8c04ac6f6e693948890d6f343420af1b" + "sha256": "7ed3f6485909c3ef7fac8d2083d6931765e4d197e690c3b9914b905a8aa743b2" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 43bbbb4..5d5ea86 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -530,7 +530,7 @@ "progress": { "anyOf": [ { - "$ref": "#/components/schemas/InvestigationProgress" + "$ref": "#/components/schemas/InvestigationProgressInput" }, { "type": "null" @@ -3540,6 +3540,21 @@ "InvestigationDiagram": { "additionalProperties": false, "properties": { + "edges": { + "anyOf": [ + { + "items": { + "$ref": "#/components/schemas/InvestigationGraphEdge" + }, + "maxItems": 200, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Edges" + }, "mermaid": { "maxLength": 20000, "minLength": 1, @@ -3554,6 +3569,19 @@ "title": "Node Refs", "type": "array" }, + "source_mermaid": { + "anyOf": [ + { + "maxLength": 20000, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Mermaid" + }, "title": { "minLength": 1, "title": "Title", @@ -3578,6 +3606,19 @@ "title": "Action Refs", "type": "array" }, + "detail": { + "anyOf": [ + { + "maxLength": 2000, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Detail" + }, "evidence_refs": { "items": { "minLength": 1, @@ -3608,14 +3649,11 @@ "anyOf": [ { "enum": [ - "pending", - "running", - "completed", - "waiting", - "waiting_human", + "reference", + "recorded", + "needs_input", "failed", - "skipped", - "cancelled" + "not_taken" ], "type": "string" }, @@ -3623,7 +3661,7 @@ "type": "null" } ], - "description": "Evidence-confirmed node state; waiting means prerequisites/external/system result, waiting_human requires manual authorization/approval/feedback. Unknown stays omitted/null; host controls colors.", + "description": "A saved process-reference annotation, not live execution state. recorded means a finding exists, not that a task or field action completed.", "title": "Status" } }, @@ -3634,6 +3672,153 @@ "title": "InvestigationDiagramNodeRef", "type": "object" }, + "InvestigationGraphEdge": { + "additionalProperties": false, + "properties": { + "label": { + "anyOf": [ + { + "maxLength": 200, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Label" + }, + "source": { + "pattern": "^[A-Za-z_][A-Za-z0-9_-]{0,79}$", + "title": "Source", + "type": "string" + }, + "target": { + "pattern": "^[A-Za-z_][A-Za-z0-9_-]{0,79}$", + "title": "Target", + "type": "string" + } + }, + "required": [ + "source", + "target" + ], + "title": "InvestigationGraphEdge", + "type": "object" + }, + "InvestigationGraphNodeUpdate": { + "additionalProperties": false, + "properties": { + "action_refs": { + "items": { + "format": "uuid", + "type": "string" + }, + "title": "Action Refs", + "type": "array" + }, + "detail": { + "anyOf": [ + { + "maxLength": 2000, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Detail" + }, + "evidence_refs": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Evidence Refs", + "type": "array" + }, + "label": { + "minLength": 1, + "title": "Label", + "type": "string" + }, + "map_targets": { + "items": { + "$ref": "#/components/schemas/InvestigationMapTarget" + }, + "maxItems": 100, + "title": "Map Targets", + "type": "array" + }, + "node_id": { + "pattern": "^[A-Za-z_][A-Za-z0-9_-]{0,79}$", + "title": "Node Id", + "type": "string" + }, + "status": { + "anyOf": [ + { + "enum": [ + "reference", + "recorded", + "needs_input", + "failed", + "not_taken" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Optional reference annotation: reference, recorded finding, needs_input, failed attempt, or not_taken. No live node/task binding or per-step synchronization is required.", + "title": "Status" + } + }, + "required": [ + "node_id", + "label" + ], + "title": "InvestigationGraphNodeUpdate", + "type": "object" + }, + "InvestigationGraphUpdate": { + "additionalProperties": false, + "properties": { + "edges": { + "items": { + "$ref": "#/components/schemas/InvestigationGraphEdge" + }, + "maxItems": 200, + "title": "Edges", + "type": "array" + }, + "nodes": { + "items": { + "$ref": "#/components/schemas/InvestigationGraphNodeUpdate" + }, + "maxItems": 100, + "title": "Nodes", + "type": "array" + }, + "title": { + "anyOf": [ + { + "maxLength": 200, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Title" + } + }, + "title": "InvestigationGraphUpdate", + "type": "object" + }, "InvestigationMapTarget": { "additionalProperties": false, "properties": { @@ -3691,6 +3876,42 @@ "title": "InvestigationProgress", "type": "object" }, + "InvestigationProgressInput": { + "additionalProperties": false, + "properties": { + "diagram": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationDiagram" + }, + { + "type": "null" + } + ] + }, + "graph_update": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationGraphUpdate" + }, + { + "type": "null" + } + ] + }, + "summary": { + "maxLength": 20000, + "minLength": 1, + "title": "Summary", + "type": "string" + } + }, + "required": [ + "summary" + ], + "title": "InvestigationProgressInput", + "type": "object" + }, "InvestigationProgressSnapshot": { "additionalProperties": false, "properties": { diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 0000000..43633b8 --- /dev/null +++ b/docs/README.md @@ -0,0 +1,56 @@ +# 供水 Server 文档目录 + +环境、启动和发布命令见 [仓库 README](../README.md)，开发规则见 [AGENTS.md](../AGENTS.md)。整理日期：2026-11-10。下列文档按职责组织，不将历史迁移或本地验证当作现有服务状态。 + +## API、认证与数据 + +| 主题 | 维护入口 | +| --- | --- | +| API 命名、字段和生成契约 | [API 风格](api-style.md) · [OpenAPI](../contracts/server-v1.openapi.json) | +| 身份、项目成员、管理接口与元数据 | [认证与用户管理](../AUTHENTICATION_AND_USER_MANAGEMENT.md) | +| MCP 设计依据、启用与认证配置 | [查询设计](mcp-query-design.md) · [实施与启用](mcp-query-implementation.md) · [Keycloak 配置](mcp-keycloak-setup.md) | +| MCP 分析、项目服务和后台计算 | [分析工具](mcp-analysis-tools.md) · [项目服务](mcp-project-services.md) · [计算队列](calculation-queue.md) | +| 项目地名与空间覆盖 | [地名搜索](project-place-search.md) · [节点覆盖](node-coverage-tool.md) | +| 项目 BizDB 与时序模型 | [数据库架构](../resources/db_v2/DATABASE_ARCHITECTURE.md) · [管网结构](../resources/db_v2/WNDB_STRUCTURE.md) | + +工具目录以 `app/mcp/catalog.py` 为准；启用范围、项目权限和会话用途决定实际可用项，不以文档中的历史工具数量推定。 + +## 任务、调查与工作台 + +| 主题 | 维护入口 | +| --- | --- | +| 任务类型、调查轮次与协作 | [工况任务契约](operational-tasks.md) | +| Worker、租约、持久消息、外部动作和部署 | [工况工作进程](operational-task-workers.md) | +| 进展、展示事件与人工反馈续调 | [调查进展与展示](investigation-progress-and-presentation.md) | +| 正式报告与稳定引用 | [调查报告契约](investigation-report-contract.md) | +| 爆管工单到处置闭环 | [爆管处置路线](burst-response-workflow.md) | +| 工作台运行态、外围工单与告警存储 | [运行态与存储](workbench-runtime.md) | +| 事件顺序、SSE 与读取去重 | [工作台事件链路](workbench-event-refinement.md) | +| 通知进入、退出与生命周期 | [通知路由](operational-notification-routing.md) | +| 迁移门禁与逐项目验证 | [数据库变更门禁](workbench-database-rollout.md) | +| 外围工单来源、项目路由和主动同步 | [外围服务](peripheral-sample.md) | + +[旧动态计划协议](dynamic-investigation-plan.md) 用于兼容历史计划，当前任务和报告规则以上述专门契约为准。[15173 UI Keycloak 客户端](integration-ui-keycloak-client.md) 是特定联调环境说明，不能作为全部部署环境的通用配置。 + +## SCADA、回放与数据集 + +| 主题 | 维护入口 | +| --- | --- | +| 回放、时间轴、周期监测与调查边界 | [SCADA 回放设计与实现](scada-peripheral-replay-design.md) | +| 外围测试样本与数据再现 | [合成样本](peripheral-generated-data.md) | +| 凤阳历史爆管模拟 | [凤阳 SCADA 复制记录](fengyang-burst-scada.md) | +| 奉贤独立 Agent benchmark | [tjwater_v2 benchmark](tjwater-v2-burst-benchmark.md) | +| tjwater2 场景选取与观测 | [多距离与弱信号方案](tjwater2-v2-burst-balanced-design.md) | +| tjwater2 水力校准方法 | [爆管模拟校准](tjwater2-v2-burst-simulation-design.md) | +| 监测点算法与资源压力取舍 | [灵敏度优化](sensor-sensitivity-optimization.md) | + +这些数据文档分别描述数据准备、算法设计或历史试算，不能互相替代，也不证明目标数据库已经加载或真实 Agent 已完成验收。 + +## 合并与历史资料 + +- `workbench-operations-storage.md` 已并入运行态文档的“工单与告警存储”。 +- `peripheral-work-order-events.md` 已并入外围服务文档的“工单事件主动同步”，旧固定库限制已按项目路由实现修订。 +- [历史命名审查](history/backend-naming-audit.md) 与 [失效接口排查](history/retired-api-investigation.md) 保留原始审查依据，不能作为当前 API 目录。 +- `scripts/missing_api_endpoints.md` 是历史能力盘点，新增能力前需重新核对当前契约与调用方。 + +设计与部署说明保留分工：设计解释选择与边界，运行文档给出可执行配置；不同数据库、不同数据集和不同生命周期的业务对象不为减少文件数而合并。 diff --git a/docs/api-style.md b/docs/api-style.md index 89351e2..4cda258 100644 --- a/docs/api-style.md +++ b/docs/api-style.md @@ -1,7 +1,8 @@ # TJWater REST API v1 This contract is the public API contract for coordinated TJWater -Server, Agent, Frontend, and CLI releases. +Server, Agent, and Frontend releases. Legacy CLI material is a historical +capability reference, not a current runtime integration. - Paths use lowercase kebab-case, have no trailing slash, and identify resources rather than handler actions. diff --git a/docs/dynamic-investigation-plan.md b/docs/dynamic-investigation-plan.md index 1e5ece5..1e0d0c4 100644 --- a/docs/dynamic-investigation-plan.md +++ b/docs/dynamic-investigation-plan.md @@ -1,6 +1,8 @@ -# 动态调查计划数据契约 +# 旧动态调查计划兼容契约 -调查任务从工单投诉内容或人工请求创建。Agent 在本次执行中生成 `execution-plan@2`，根据新证据追加修订；数据库保存不可变计划快照的 JSONB 文档，并不保存固定流程模板。任务、Run、证据、访问申请、审核动作和报告各有独立生命周期。计划中的 `gate_ref` 只引用已经持久化的访问申请或动作，不能把审批结果写成模型自行判断的文本。 +> 历史兼容文档：字段和示例用于读取已有 `execution-plan@1/@2` 快照，不要求当前调查生成或执行这些计划。当前进展与累计图见 [调查进展](investigation-progress-and-presentation.md)，Agent 生命周期见 [任务执行](../../next-tjwater-agent/docs/execution-plan.md)。 + +早期设计由 Agent 生成 `execution-plan@2` 并按证据追加修订；数据库保存不可变计划快照的 JSONB 文档，不保存固定流程模板。任务、Run、证据、访问申请、审核动作和报告各有独立生命周期。计划中的 `gate_ref` 只引用已经持久化的访问申请或动作，不能把审批结果写成模型自行判断的文本。 ## `execution-plan@2` diff --git a/BACKEND_NAMING_AUDIT.md b/docs/history/backend-naming-audit.md similarity index 94% rename from BACKEND_NAMING_AUDIT.md rename to docs/history/backend-naming-audit.md index 4e76cc6..b51d0c0 100644 --- a/BACKEND_NAMING_AUDIT.md +++ b/docs/history/backend-naming-audit.md @@ -1,5 +1,7 @@ # Backend Naming Audit +> 历史接口盘点，保留旧名称与修正依据。本文中的旧仓库名、URL 和替代建议不作为当前可用接口清单；现行规范见 [API 风格](../api-style.md)，实际契约见 [Server OpenAPI](../../contracts/server-v1.openapi.json)。 + DOC-003 audit for the internal `TJWaterServerBinary` backend. ## Scope diff --git a/失效API排查.md b/docs/history/retired-api-investigation.md similarity index 89% rename from 失效API排查.md rename to docs/history/retired-api-investigation.md index 8233525..17a6394 100644 --- a/失效API排查.md +++ b/docs/history/retired-api-investigation.md @@ -1,5 +1,7 @@ # `app/api/v1/endpoints/` 失效 API 排查与修正 +> 历史接口盘点，保留旧名称与修正依据。本文中的旧仓库名、URL 和替代建议不作为当前可用接口清单；现行规范见 [API 风格](../api-style.md)，实际契约见 [Server OpenAPI](../../contracts/server-v1.openapi.json)。 + 排查范围：`app/api/v1/endpoints/` 结论：本次共确认 5 个问题接口，处理结果如下： @@ -67,7 +69,7 @@ DMA 计算请使用： 服务区计算请使用： -- `/api/v1/calculateservicearea/` +- `/api/v1/calculateservicearea/` 现在返回全部时间步结果，不再接收 `time_index` ## 变更文件 diff --git a/docs/investigation-progress-and-presentation.md b/docs/investigation-progress-and-presentation.md index 19f2e65..6fe300d 100644 --- a/docs/investigation-progress-and-presentation.md +++ b/docs/investigation-progress-and-presentation.md @@ -1,8 +1,9 @@ # 调查进展与实时展示 -调查执行中可持续发布正文和完整 Mermaid 图，尚未形成结论时也能发布。 +调查执行中可按需整理正文和流程参考图，Server 累计保留路径并生成 Mermaid，尚未形成结论时也能发布。 Server 将阅读进展、业务报告和浏览器展示请求分别保存。运行结束仍要求同轮最终报告， -进展中的节点状态不参与调度、审批或结束判断。 +图节点不绑定实际执行任务，标记只解释最近整理的路线、结果和限制；不参与调度、审批、等待或结束判断。 +路线有实质调整、形成重要结果、需要保留失败尝试或整理报告时更新即可，不要求逐步发布、频繁刷新或及时同步。 ## 进展快照 @@ -14,37 +15,73 @@ Server 将阅读进展、业务报告和浏览器展示请求分别保存。运 "run_id": "", "source_session_id": "", "progress": { - "summary": "正在核对投诉时段的监测记录，尚未形成结论。", - "diagram": { - "title": "当前调查进展", - "mermaid": "flowchart LR\nread[核查监测记录] --> judge[判断证据范围]", - "node_refs": [{ + "summary": "整理投诉时段的监测核查路线，尚未形成结论。", + "graph_update": { + "title": "调查流程参考", + "nodes": [{ "node_id": "read", "label": "核查监测记录", - "status": "running", + "status": "reference", "evidence_refs": [], "action_refs": [], "map_targets": [{"source_id": "pipes", "feature_id": ""}] - }] + }], + "edges": [] } } } ``` -示例中的编号由实际任务和资产替换。正文最长 20,000 字符，`diagram` 可空。 +示例中的编号由实际任务和资产替换。正文最长 20,000 字符，`graph_update` 可省略。 图示沿用 [调查报告契约](investigation-report-contract.md) 的 Mermaid 安全限制。 -节点状态可空，或为 `pending/running/completed/waiting_human/failed/skipped/cancelled`。 -`map_targets` 只接受 `junctions/pipes/valves/pumps/tanks/reservoirs/scada` 七类来源和要素编号。 -图中节点引用在本次发布内唯一，证据和动作引用必须属于该任务。跨修订可以修改节点编号、 -分支和文字，Server 不验证依赖执行顺序或节点完工。 +节点标记可省略或为 null，均显示为参考步骤；新增量更新只接受以下五种： -每次传入 `progress` 都完整替换该 run 的快照，传入 `diagram: null` 清除上一版图示。 -省略 `progress` 保留原快照。同一个请求编号重试只发布一次。 +| status | 显示 | 含义 | +| --- | --- | --- | +| reference | 参考步骤 | 路线或分支说明，不断言是否执行 | +| recorded | 已有结果 | 已整理出结果或依据，不表示任务、计算或现场动作完成 | +| needs_input | 待补充条件 | 记录缺少的资料、权限或人工事实，不控制任务等待 | +| failed | 尝试失败 | 保留一次失败尝试及原因，后续路线另建节点 | +| not_taken | 未采用路线 | 未采用、不适用或改走其他分支 | + +旧图读取兼容：pending/running 转为 reference，completed 转为 recorded，waiting/waiting_human 转为 needs_input， +skipped/cancelled 转为 not_taken；failed 保留。仅规范化读取结果，不改写历史文档，不再显示“正在执行”或“记录时执行中”。 +REST 与 MCP 使用相同的旧标记映射。MCP 的 progress/report 图示在分页及 data_path 投影之前完成转换， +读取整份说明、节点列表或单个 status 时含义一致；实际 task/run/action 状态和旧执行计划状态保持各自语义。 +`map_targets` 只接受 `junctions/pipes/valves/pumps/tanks/reservoirs/scada` 七类来源和要素编号。 +图中节点引用唯一，证据和动作引用必须属于该任务。新 Agent 使用下述增量更新，节点编号在任务各轮保持稳定。 +Server 不验证依赖执行顺序或推断节点完工。 + +`progress.graph_update` 包含可选 `title`、`nodes` 和 `edges`。节点沿用 `node_id/label/status/evidence_refs/action_refs/map_targets`， +可用 `detail` 记录实际结果或失败原因；连线为 `{source,target,label?}`。仅提交新增或变化的内容， +`update_investigation_graph` 合并节点、追加明确连线并生成 Mermaid，保留已有路径和引用。 +尝试失败时标记 `failed`，重试或替代路线使用新节点及连线；原失败节点不能改写为成功。 +仅更新正文或传入 `diagram: null` 均保留累计图，原始 Mermaid 不再由新 Agent 生成。 +首次增量更新沿用最近保存的旧图，将原源码保存在 `source_mermaid`，保留原分支和形状后追加内容； +旧图内节点不得改名或重用为其他步骤。已被旧实现覆盖且未存入报告的内容无法还原。 +尚未迁移的旧图仍接受旧客户端显式提交的完整图；仅更新正文或传入 `diagram: null` 时自动接管旧图， +保留原源码并转为累计图。任务启用累计图后拒绝旧客户端重画覆盖，不提供通过空值清图的入口。 +省略 `progress` 保留原快照，同一个请求编号重试只发布一次。 `workbench.operational_task_progress` 每个任务和 run 保存最新一份快照，独立于七天事件清理。 任务详情和 Agent 上下文返回 `latest_progress`，包含 `revision/run_id/execution_generation/` -`published_at/source_session_id/progress`。只读取当前执行代次，新轮次尚未发布时返回空。 +`published_at/source_session_id/progress`。优先返回当前执行代次；新轮次尚未发布时保留最近的累计图及其原始轮次。 +`get_task_context(section="progress")` 同样允许读取上轮累计图以接续，摘要和图示不证明新轮次已执行。 +报告发布时由 Server 存入累计图及其已有依据，历史报告保持独立快照；最多保留 100 个节点、200 条连线，超限显式报错，不静默裁剪。 快照更新发出 `operational.progress.changed`，前端据此读取最新内容。 +### 旧任务图迁移 + +在 Server 仓库运行 `python -m scripts.migrate_investigation_graphs --plan /受保护目录/plan.json --project-id <项目UUID>`； +多个项目重复指定 `--project-id`。预览只读数据库，计划文件以私有权限保存原始进度文档和迁移后的完整文档。 +核对项目、业务库、任务和图数量后，运行 `python -m scripts.migrate_investigation_graphs --apply /受保护目录/plan.json`。 +连接使用 MetaDB 路由，计划中不保存连接凭据。迁移逐项目事务执行，与正常发布使用相同的事件锁和任务锁， +预览后仍为旧图但已发生变化的行会使该项目回滚；重复执行跳过已经接管的图。 + +迁移仅更新已有进度图及其 revision，保留原 summary、发布时间、run、会话、节点说明和引用； +原 Mermaid 字节保存在 `source_mermaid`，旧节点标记转换为参考标记，不解析或猜测原始连线。 +历史报告保持原样，通过读取兼容显示新标记。迁移逐项目校验任务、run 和报告整行摘要未变， +并发出进度变更事件及写入独立执行回执；不产生报告更新通知。备份和回执不得提交到仓库。 + ## 浏览器实时联动 `POST /api/v1/operational-tasks/{task_id}/presentation-events` 接收 @@ -68,7 +105,7 @@ Server 校验当前 run、绑定会话、有效租约和幂等身份，记录 `o 以上十种动作与 Chat 的 `present_view`、Dev JSONL 共用扁平 `command: {action, ...字段}`。Chat 使用 `frontend-action@2` 和执行回执；任务仍只记录展示请求，Dev 只做本地预览。三者使用独立的地图源、SCADA 图层和图表状态。Dev 的动作数据行必须为 `{"record_type":"presentation","command":{...}}`；界面块行使用 `record_type: ui_block`，新示例明确带 `phase: final`。普通调查数据保持来源字段，不执行其中的 action。仓库示例见[前端统一展示夹具](../../next-tjwater-frontend/tests/fixtures/unified-presentation.jsonl)，其中编号来自测试夹具。 -图表限定折线和柱形，包含 `title/chart_type/x_data/series`。最多 100 个横轴值和 8 组有限数值序列， +图表使用普通折线 line、柱状 bar，或当前曲线加历史散点的 time_comparison，包含 `title/chart_type/x_data/series`。最多 100 个横轴值和 8 组有限数值序列， 序列长度与横轴对齐，引用证据必须属于当前任务。地图高亮来源沿用上述七类。 覆盖层的权威数据来自不可变证据。省略 `data` 时，Server 从证据快照提取现有 GeoJSON， 或使用已记录的 `longitude/latitude` 点位，将 `FeatureCollection` 加入事件中的 `command.data`。 @@ -103,6 +140,14 @@ MCP 保存完整查询响应作为不可变证据，展示解析器同时支持 注明来源、范围与实测或模型属性；没有同口径基准时不计算或绘制改善量，缺测不补零。 过程图表通过 `open_chart` 提供 `chart_id/chart/evidence_refs`，每图最多100个横轴值、8组序列， 超限时分图或明确说明抽样、聚合范围；单值或无可比数据用文字或表格。 +同指标、单位和窗口的同期复合图按对象分别发布稳定 chart_id，标题为“对象ID 共同标题”，每条序列名以“对象ID ”开头。共同标题、chart_type、轴名、x_data、当前序列去掉对象前缀后的名称与 sample_times 完全一致时，浏览器将对象数据集聚合为一个子图表：右侧按钮切换对象，下方图例控制序列；窄屏对象列表移到下方。普通趋势总览、柱状图和结构化 Chart 保持独立，不新增 objects 字段。原 chart_id 的 close_chart 只移除对应对象，关闭聚合卡片移除组内对象。 + +浏览器的总图表窗口统一管理过程图组与报告 Chart，子图表仍可手动独立弹窗放大。只有正在跟随的运行中任务新收到 open_chart 时可自动弹出总窗口；后续命令更新内容，不自动打开子窗口，用户关闭后不重弹。任务打开、切换、刷新、已完成任务载入、SSE 重连历史补载及历史导航只恢复数据，不弹窗。窗口意图是浏览器临时状态，不写入事件或历史快照，不新增协议字段；open_chart 的内容加载与窗口是否弹出是两个不同状态，Server 的 recorded 也不表示浏览器已经打开窗口。 + +time_comparison 的 x_data 使用精度一致、按窗口排列的 UTC+8 HH:mm 或 HH:mm:ss 共享槽，允许跨午夜但短于一天，不追加历史日期。series 同时包含当前 line 与历史 scatter，不混入 bar；sample_times 与 data 等长，有效值和真实带时区时间成对保留，缺测值与时间均为 null。时间转换后须与横轴槽一致，当前线使用同一窗口，每条历史序列来自同一历史日的对应窗口，不接受当天或未来日期。新任务展示与报告发布不满足这些规则时返回 TIME_COMPARISON_INVALID，旧保存内容保持结构读取。普通 line/bar 跳过同期语义，但不能将 line/scatter 复合图标为 line。 + +前端维护的脚本 scripts/export-view-command-contract.mjs 同步 Agent/Server 的结构协议及 tests/fixtures 下的统一 JSONL 和 chart-publication-cases.json；JSON Schema 不代替跨字段语义校验。Server 的独立测试用相同正反例校验 Pydantic 边界和 validate_time_comparison_chart；Agent 主机、原生工具及 Dev 的新图表输入也执行对应规则。 + 关键趋势与比较结论保留到报告中，最终报告的图表不替代过程发布。 业务 Skill 说明本步骤应展示什么，调查指令统一说明发布方法、非空几何检查、去重和失败处理。 同一阶段批量展示，数据或方案没有变化时不重复请求；不为地图动作增加无关业务分析。 diff --git a/docs/investigation-report-contract.md b/docs/investigation-report-contract.md index 7344918..4cdc968 100644 --- a/docs/investigation-report-contract.md +++ b/docs/investigation-report-contract.md @@ -47,6 +47,9 @@ node_id 为最多 80 字符的 ASCII 标识符，必须以字母或下划线开 并与 Mermaid 源码中实际节点使用完全相同的 ID；禁止为同一节点另造引用 ID； 最多 100 条引用。证据必须来自报告 evidence_refs，动作必须属于当前任务。 节点可以说明候选方向与条件关系；图中的分支、状态文字不能证明实际并发、动作执行或现场完成。 +可选 status 仅为参考标记：reference（参考步骤）、recorded（已有结果）、needs_input（待补充条件）、 +failed（尝试失败）、not_taken（未采用路线）。recorded 不证明业务完成，标记不与任务或工具运行状态绑定。 +旧运行状态按[兼容映射](investigation-progress-and-presentation.md)读取，未提供标记时保持中性的参考步骤。 图的颜色、字体和交互由前端统一配置。正文不包含 SVG 或 HTML。 ## 业务约束 @@ -150,7 +153,8 @@ failure_code 为 ARTIFACT_INCOMPLETE；调用方须读取返回的业务状态 ## 发布、阅读与协作 -执行中尚未形成结论的正文和 Mermaid 使用独立的 `progress` 快照持续更新， +执行中尚未形成结论的正文和 Mermaid 使用独立的 `progress` 快照按需整理， +在路线调整、重要结果或报告归纳时更新即可，不要求逐步更新或及时同步。 见 [调查进展与实时展示](investigation-progress-and-presentation.md)。 报告的 interim/final 与 concluded/insufficient_data 结果契约保持不变。 diff --git a/docs/mcp-query-design.md b/docs/mcp-query-design.md index 05d00d2..543c514 100644 --- a/docs/mcp-query-design.md +++ b/docs/mcp-query-design.md @@ -1,10 +1,10 @@ # NextTJWater MCP 查询与认证设计 -日期：2026-10-15。状态：查询与认证代码已在本地实现，认证发现、resource 参数和版本头体校验已补齐，默认关闭，未部署；真实 Keycloak 与项目数据库验收待完成。 +初稿日期：2026-10-15。本文保留 MCP 查询与认证的设计依据，后续工具扩展及当前启用步骤见 [实施与启用](mcp-query-implementation.md)。早期本地实现、默认配置和验证快照不代表当前线上启用或部署状态；需核对目标环境。 本设计将现有后端查询能力通过 MCP 开放给普通项目 Agent、调查主 Agent 和调查只读子 Agent。采用 MCP 2026-07-28、Streamable HTTP、现有 Python Server 内挂载服务、可信 Agent 宿主按请求代理认证。旧 CLI 用于查找功能和接口线索，业务实现继续由当前 Server 负责。 -第一版完成查询、权限、取消、分页、调查证据和凭据续期的完整闭环。已有报告发布、工单审批、设备控制、调查运行生命周期继续使用当前受控工具；不通过 MCP 自动开放旧 CLI 中的分析执行命令。 +本文第一版设计聚焦查询、权限、取消、分页、调查证据和凭据续期。后续已扩展受控分析、预览式写入及后台计算队列，详见实施说明；报告发布、审批、设备控制和调查运行仍遵循各自业务边界，不从旧 CLI 清单自动开放能力。 ## 当前系统与功能核对 diff --git a/docs/mcp-query-implementation.md b/docs/mcp-query-implementation.md index df12fef..cbac94f 100644 --- a/docs/mcp-query-implementation.md +++ b/docs/mcp-query-implementation.md @@ -1,6 +1,6 @@ # 产品 MCP 查询：本地实施与启用 -实现使用 MCP 2026-07-28、Python mcp 2.3.0、TypeScript @modelcontextprotocol/client 与 @modelcontextprotocol/server 2.3.1。Server 提供 `/mcp` 和71项工具（42项原查询、12项计算、13项预览式写入、4项计算任务管理）；Agent 将其投影合并到本地 MCP，由 OpenCode 原生发现为 `mcp_tjwater_*`，普通项目会话、调查主 Agent 和已验证的直接只读子 Agent 共享42项查询；12项计算仅向非reader会话开放，13项写入仅向项目成员的普通聊天开放。个人会话不能获取项目凭据。旧 CLI 只作为能力清单，没有运行时 CLI 桥接。 +实现使用 MCP 2026-07-28、Python mcp 2.3.0、TypeScript @modelcontextprotocol/client 与 @modelcontextprotocol/server 2.3.1。Server 通过 `/mcp` 提供查询、分析计算、预览式写入和计算任务管理，具体目录以 `app/mcp/catalog.py` 为准；Agent 投影到本地 MCP，由 OpenCode 原生发现为 `mcp_tjwater_*`。普通项目会话、调查主 Agent 和已验证的直接只读子 Agent 在各自授权范围内查询；计算仅向非 reader 会话开放，写入仅向项目成员的普通聊天开放。个人会话不能获取项目凭据。旧 CLI 只作为能力清单，没有运行时 CLI 桥接。 `run_simulation` 是在线重算，复用 REST 在线计算服务而不复制数据库；按水力步长更新模型条件和实时基线，不创建正式方案。MCP 时长为秒，原 REST 时长仍为分钟。其 annotation 为 `readOnlyHint=false`、`destructiveHint=true`；只读 reader 不能调用，其他方案模拟继续使用隔离临时模型。详细边界见 [MCP 分析工具说明](mcp-analysis-tools.md)。 @@ -107,10 +107,10 @@ Agent： ```bash bun run check -bun test tests/product/mcpAuthDiscovery.test.ts tests/product/mcpQueryClient.test.ts tests/product/credentialRefresh.test.ts tests/runtime/task-agent-policy.test.ts tests/runtime/opencodeConfig.test.ts tests/opencode/mcpPolicy.test.ts tests/opencode/productMcp.test.ts +bun test tests/product/mcpAuthDiscovery.test.ts tests/product/mcpQueryClient.test.ts tests/product/credentialRefresh.test.ts tests/runtime/task-agent-policy.test.ts tests/runtime/opencodeConfig.test.ts tests/opencode/mcpPolicy.test.ts tests/opencode/queryMcp.test.ts tests/mcp/queryServer.test.ts MCP_TEST_PYTHON=<隔离Server环境Python> MCP_TEST_SERVER_ROOT= bun tests/integration/mcp-query-smoke.ts ``` -HTTP smoke 只启动自己的 loopback fixture，使用真实两端 SDK，无生产数据库、Keycloak或模型推理。它验证新版发现、真实资源元数据 HTTP 读取、授权服务器元数据及带 resource 的两种 grant（后两者为桩）、67项工具、主机绑定、查询及撤销；真实环境验收记录见上方实际运行环境一节，不能用 fixture 结果代替。前端严格解析已接受这些 MCP 快照来源，未加入浏览器 MCP 凭据或新 UI。 +HTTP smoke 只启动自己的 loopback fixture，使用真实两端 SDK，无生产数据库、Keycloak或模型推理。它验证新版发现、真实资源元数据 HTTP 读取、授权服务器元数据及带 resource 的两种 grant（后两者为桩）、对应源码版本的工具目录、主机绑定、查询及撤销；真实环境验收记录见上方实际运行环境一节，不能用 fixture 结果代替。前端严格解析已接受这些 MCP 快照来源，未加入浏览器 MCP 凭据或新 UI。 查询限制回归使用 `python -m pytest tests/unit/test_mcp_query_limits.py tests/unit/test_mcp_query.py tests/unit/test_timeseries_history.py tests/unit/test_timeseries_history_schema.py -q`。实际 PostgreSQL SQL 验证可设置 `MCP_QUERY_TEST_DSN` 后运行 `python -m pytest tests/integration/test_mcp_query_limits_postgres.py -q`；该 DSN 必须指向空的、可丢弃的独立数据库，测试创建临时业务同名 schema 并在事务结束时回滚。它验证数据点跨序列翻页、证据路径/Unicode分页、任务摘要与指定报告、章节稳定路径、项目/任务隔离、既有scope边界、字节门槛和合成摘要，不代替真实项目验收。 diff --git a/docs/operational-task-workers.md b/docs/operational-task-workers.md index 272ff3d..4528602 100644 --- a/docs/operational-task-workers.md +++ b/docs/operational-task-workers.md @@ -150,7 +150,7 @@ conda run -n server python scripts/sync_peripheral_work_orders.py --once conda run -n server python scripts/sync_peripheral_work_orders.py ``` -使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器订阅源工单变更事件，收到消息即读取分页工单，将外围版本按系统数据投影到 `external_work_order` 任务；默认每 300 秒兜底对账。工单完成只表示该外部任务完成，不代表其他调查自动完成。事件协议、`tjwater_v2` 范围与启用步骤见 [外围工单主动同步](peripheral-work-order-events.md)。 +使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器订阅源工单变更事件，收到源工单事件后按 ID 定向读取工单，将外围版本按系统数据投影到 `external_work_order` 任务；默认每 300 秒兜底对账。工单完成只表示该外部任务完成，不代表其他调查自动完成。事件协议、项目存储范围与启用步骤见 [外围工单主动同步](peripheral-sample.md#工单事件主动同步)。 工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单由独立同步器使用服务身份响应后端事件，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与同步器固定项目不一致。 diff --git a/docs/peripheral-generated-data.md b/docs/peripheral-generated-data.md index f6ac976..6807ba8 100644 --- a/docs/peripheral-generated-data.md +++ b/docs/peripheral-generated-data.md @@ -52,7 +52,7 @@ conda run -n server python scripts/seed_peripheral_sample.py --project-id ` 只返回同项目的来源任务和动作回执关联任务；外围接口只读取该工单，不再分页遍历全部工单和全部调查详情。重复 ID 合并；目标工单的明确删除只撤回该工单投影，不影响其他任务。 + +内存待处理 ID 最多 200 个，超过上限合并为一次全量对账，不持续扩张队列。这是突发积压保护，不增加公共配置项。 + +默认每 300 秒仍进行一次全量对账，处理漏通知、断线、旧批量脚本直接写库及重启。持续收到事件不会推迟全量对账期限。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前取出待处理 ID，扫描期间新到达的 ID 留给下一轮；失败的 ID 会放回队列，重试不依赖新消息。 + +浏览器收到源事件时不会提前刷新业务快照，而是等待任务投影事件后读取对应任务。工况手动刷新仍只读 Server 权威快照。工单事件不等于 Header 通知；只有需要处理的业务待办或告警才进入通知列表。 + +游标顺序、共享 Hub 查询与兼容边界见 [工作台事件链路](workbench-event-refinement.md)。新增查询参数已从 Server OpenAPI 同步生成 Agent 类型，不另建手写契约。 + +### 验证与启用 + +```bash +conda run -n server python -m pytest tests/unit/test_peripheral_push_sync.py tests/unit/test_peripheral_work_order_events.py tests/unit/test_peripheral_task_sync.py -q +PERIPHERAL_SAMPLE_LIVE_TEST=1 conda run -n server python -m pytest tests/integration/test_peripheral_work_order_events.py -q +``` + +数据库测试强制限定 `tjwater_v2`，使用随机项目键和回滚事务，不保留测试工单或事件。覆盖创建、修改、反馈、删除、幂等重放及回滚无事件。HTTP 隔离测试覆盖源事件即时投影、游标重置、断线续传、失败重试和监听取消。 + +发布时外围服务随主 Server 镜像更新，同步器是独立常驻进程；需按维护者授权的发布流程分别更新 Server、同步器及配套前端。提交代码不证明现有进程已重载。回退应用代码不涉及数据结构回滚，已有事件仍可正常补读；回退到旧同步器后恢复周期对账。 diff --git a/docs/peripheral-work-order-events.md b/docs/peripheral-work-order-events.md deleted file mode 100644 index c2ded1b..0000000 --- a/docs/peripheral-work-order-events.md +++ /dev/null @@ -1,50 +0,0 @@ -# 外围工单主动同步 - -工单的创建、编辑、现场反馈和删除改为由来源事件触发同步。浏览器继续使用已有工作台 SSE，不需要新增连接或定时工单轮询。 - -```text -外围工单写入 - → 同一 PostgreSQL 事务保存源记录、历史版本和变更事件 - → COMMIT 后 NOTIFY - → Server 工作台 SSE 唤醒外围同步器 - → 同步器通过授权 API 更新任务投影 - → task/action/notification 事件推送给浏览器 -``` - -## 数据库范围 - -本次仅启用 `tjwater_v2`。没有新增表、列或迁移；复用 `workbench.domain_event_log` 的 `workbench.work_order.changed` topic。源事件的聚合类型为 `peripheral_work_order`，元数据只包含 `source=peripheral_sample`、工单 ID 和 `created|updated|deleted`，不包含工单正文。事务失败时，源记录、历史、事件及通知一起回滚；同一请求的幂等重放不再产生事件。 - -外围服务只写源记录及源变更事实，不直接改业务任务。任务投影、调查创建和通知仍通过 Server 状态机和现有授权 API 处理。其他数据库不写入该源事件；扩展前遵循 [数据库变更门禁](workbench-database-rollout.md)。事件表缺失或无写权限时，`tjwater_v2` 的工单写入返回存储错误，不会假装成功。 - -## 同步器 - -从 Server 仓库运行已有命令： - -```bash -conda run -n server python scripts/sync_peripheral_work_orders.py -conda run -n server python scripts/sync_peripheral_work_orders.py --once -``` - -继续复用 `OPERATIONAL_WORKER_*` OAuth 工作身份及 `OPERATIONAL_ACTION_PERIPHERAL_URL`，不需要新账号或密钥。启动和游标过期时先读取事件游标，再进行全量对账；扫描期间的事件从该游标继续补读。相同源版本仍由原幂等键去重。源事件按工单 ID 唤醒同步器；动作回执事件只补查该任务关联的工单，避免源创建早于回执保存时漏同步。任务投影事件不再触发全量扫描。 - -正常状态收到源事件即处理对应 ID：`GET /api/v1/operational-tasks?peripheral_work_order_id=` 只返回同项目的来源任务和动作回执关联任务；外围接口只读取该工单，不再分页遍历全部工单和全部调查详情。重复 ID 合并；目标工单的明确删除只撤回该工单投影，不影响其他任务。 - -内存待处理 ID 最多 200 个，超过上限合并为一次全量对账，不持续扩张队列。这是突发积压保护，不增加公共配置项。 - -默认每 300 秒仍进行一次全量对账，处理漏通知、断线、旧批量脚本直接写库及重启。持续收到事件不会推迟全量对账期限。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前取出待处理 ID，扫描期间新到达的 ID 留给下一轮；失败的 ID 会放回队列，重试不依赖新消息。 - -浏览器收到源事件时不会提前刷新业务快照，而是等待任务投影事件后读取对应任务。工况手动刷新仍只读 Server 权威快照。工单事件不等于 Header 通知；只有需要处理的业务待办或告警才进入通知列表。 - -游标顺序、共享 Hub 查询与兼容边界见 [事件链路收敛](workbench-event-refinement.md)。新增查询参数已从 Server OpenAPI 同步生成 Agent 类型，不另建手写契约。 - -## 验证与启用 - -```bash -conda run -n server python -m pytest tests/unit/test_peripheral_push_sync.py tests/unit/test_peripheral_work_order_events.py tests/unit/test_peripheral_task_sync.py -q -PERIPHERAL_SAMPLE_LIVE_TEST=1 conda run -n server python -m pytest tests/integration/test_peripheral_work_order_events.py -q -``` - -数据库测试强制限定 `tjwater_v2`，使用随机项目键和回滚事务，不保留测试工单或事件。覆盖创建、修改、反馈、删除、幂等重放及回滚无事件。HTTP 隔离测试覆盖源事件即时投影、游标重置、断线续传、失败重试和监听取消。 - -上线时更新外围服务、Server 同步器及前端，再按既有发布流程重启对应进程。提交代码不会自动重启现有服务；旧同步器进程仍按旧的 30 秒周期运行。回退应用代码不涉及数据结构回滚，已有事件仍可正常补读；回退到旧同步器后恢复周期对账。 diff --git a/docs/scada-peripheral-replay-design.md b/docs/scada-peripheral-replay-design.md index cfaf51e..3fc621b 100644 --- a/docs/scada-peripheral-replay-design.md +++ b/docs/scada-peripheral-replay-design.md @@ -161,7 +161,7 @@ MetaDB 的 `projects.settings` 为项目级通用 JSONB 对象，默认 `{}`， | [SCADA API](../app/api/v1/endpoints/timeseries/scada.py)、[仓储](../app/infra/db/timescaledb/repositories/scada.py) | 范围读取、最新值读取、批量插入；当前范围读取返回 raw 和 corrected，缺少回放游标及完整流水线 | 增加只读 raw 分页契约；复用存储查询，补齐处理检查点、清洗与检测调度 | | [清洗服务](../app/services/timeseries_analysis.py) | 从原始值计算并更新 `cleaned_value`；目前支持压力和流量 | 在线质量标记、因果修正及选择值依据；水质不能直接套用现有清洗 | | [压力清洗](../app/algorithms/scada_cleaning/pressure_series.py)、[流量清洗](../app/algorithms/scada_cleaning/flow_series.py) | 包含居中窗口、后向补值、Kalman 平滑；流量路径还会补为 1 分钟序列 | 在线路径只使用已到达的数据，保留设备频率，不整月预清洗后再模拟检测 | -| [外围工单事件同步](peripheral-work-order-events.md) | 持久事件、通知唤醒、后台同步和定期对账 | 复用可靠处理方式，SCADA 的批次有自己的持久状态 | +| [外围工单事件同步](peripheral-sample.md#工单事件主动同步) | 持久事件、通知唤醒、后台同步和定期对账 | 复用可靠处理方式，SCADA 的批次有自己的持久状态 | | [调查 worker](../scripts/run_operational_worker.py) | 可靠投递 `event_investigation`，恢复执行租约，管理 Agent 执行代次 | 接收由 SCADA 异常事件建立的调查；不承担每条观测的检测 | | [任务告警](../app/domain/schemas/task_alert.py) | 支持 raised、acknowledged、resolved；当前来源必须携带任务和执行标识 | 增加真实的回放检测来源，不能为调查前告警伪造 `task_id`、`run_id` | | [MCP 上下文](../app/mcp/contexts.py)、[SCADA 查询](../app/mcp/query_service.py) | 校验任务、项目、执行代次和会话，保存查询证据 | 在所有回放调查读取入口执行统一的时间与数据版本约束 | diff --git a/docs/workbench-event-refinement.md b/docs/workbench-event-refinement.md index 7aa180b..8e2b7b7 100644 --- a/docs/workbench-event-refinement.md +++ b/docs/workbench-event-refinement.md @@ -9,7 +9,7 @@ - 这个锁会串行化事件写事务，任务仓储的短命令事务（包括心跳）也使用它。这是当前低频业务规模下的明确取舍，不适用于高频遥测或长事务。扩大业务事件量前应实际测量写入等待时间，不把增加锁误称为写吞吐优化。 - 每个进程/项目仍只有一条专用 LISTEN 连接，不长期占用普通请求池。Hub 按唤醒代次共享同游标读取；最多保留 32 个游标批次，每批最多 200 条，慢客户端不会形成无界缓存。事件通知和每 30 秒一次的共享对账使缓存失效；每个浏览器的 15 秒 keepalive 不再触发独立 SQL 查询。无订阅者 60 秒后仍关闭 Hub。 - `reset_required` 允许权威快照把客户端游标从较大值重置为 0；普通快照仍禁止回退，过时并发响应仍不能覆盖新响应。重置失败会在重连时重试。只有 SSE 请求成功打开后才显示连接正常，HTTP 失败按原退避策略重连；离开或重置时释放流读取器。 -- 工单源事件按 ID 定向同步。任务投影的 `operational.task.changed` 负责更新详情，其伴随的 `workbench.work_order.changed/operational_task` 不再重复拉取整份快照。告警和分析事件仍合并后刷新运行态；SCADA/流量采样通道不变。全量对账见 [外围工单主动同步](peripheral-work-order-events.md)。 +- 工单源事件按 ID 定向同步。任务投影的 `operational.task.changed` 负责更新详情，其伴随的 `workbench.work_order.changed/operational_task` 不再重复拉取整份快照。告警和分析事件仍合并后刷新运行态；SCADA/流量采样通道不变。全量对账见 [外围工单主动同步](peripheral-sample.md#工单事件主动同步)。 - 项目查询按同一身份提供器共享，缓存 5 分钟。每个调用仍重新获取 token，各调用可独立取消；共享查询有 15 秒超时，失败后清除缓存。此缓存不授予权限，Server 每次请求和每次 SSE 重建仍鉴权。 - 事件接收游标与数据读取成功分开处理：增量详情、通知或事件触发的运行态快照失败时，只对失败读取按 1、2、4 秒递增至 30 秒重试，成功后才确认失效事件。被并发游标守卫拒绝的快照不能当作刷新成功；仅在更新的成功快照已覆盖失效游标时停止重试。正常读取成功后不安排定时请求；依赖变化、重置或离开页面会取消旧请求及重试定时器。这不是新增正常轮询通道，也不扩大初始化列表的裁剪/重试语义。 diff --git a/docs/workbench-operations-storage.md b/docs/workbench-operations-storage.md deleted file mode 100644 index 3332d2d..0000000 --- a/docs/workbench-operations-storage.md +++ /dev/null @@ -1,15 +0,0 @@ -# 工作台工单与告警存储契约 - -工作台聚合外围来源工况任务与任务主动推送的告警。浏览器不推导告警，也不直接写工况任务。 - -## 任务告警 - -DDL 位于 `resources/sql/008_workbench_alert_events.sql`。任务通过 `POST /api/v1/workbench/task-alert-events` 推送 `raised`、`acknowledged` 或 `resolved` 事件。来源包含 `task_type`、`task_id` 和 `run_id`，生产者由认证项目身份绑定。`event_id` 在项目和生产者范围内幂等，告警事实在首次 raised 后不可改写。 - -## 预约工单 - -外围样板的 `peripheral_sample.work_orders` 是演示工单的管理来源；`activity` 只记录实际发生的创建、修改和现场反馈，不由总状态补造阶段。同步器将有效工单写入 `workbench.operational_tasks` 的 `external_work_order` 投影。原有三张独立演示工单也已迁入外围样板，并由同一面板管理。 - -`resources/sql/011_workbench_work_orders.sql` 和 `017_workbench_schema_comments.sql` 属于历史基线；`021_retire_workbench_work_orders.sql` 在验证三个迁入目标及其任务投影后，仅在 `tjwater_v2` 删除旧表。不要重放 `011` 重建它。`GET /api/v1/workbench/runtime` 暂保留空的 `work_orders` 字段作为接口兼容。其他项目数据库没有旧表，不需要应用 `021`。 - -告警事件表需要 `SELECT, INSERT`，告警状态表需要 `SELECT, INSERT, UPDATE`。迁移及扩展门禁见 [`workbench-database-rollout.md`](workbench-database-rollout.md)。 diff --git a/docs/workbench-runtime.md b/docs/workbench-runtime.md index 988e4ff..e3a806c 100644 --- a/docs/workbench-runtime.md +++ b/docs/workbench-runtime.md @@ -7,4 +7,20 @@ - `workbench.alert_events` 保存任务显式推送的事件，`workbench.alerts` 保存当前状态。分析结果、测点值和缺测状态不会自动变成告警。 - 每个可选分区独立读取并返回 `ready`、`not_configured` 或 `degraded`，一个分区失败不会污染其他分区。 -原有三张独立演示工单已迁入 `peripheral_sample.work_orders`，仍由外围样板面板管理，并由同步器投影为 `external_work_order` 工况任务。旧表退役前的完整数据库备份由部署记录保存，不能重新运行历史 `011_workbench_work_orders.sql` 或旧种子脚本重建双份数据。 +历史上原有三张独立演示工单已迁入 `peripheral_sample.work_orders`，仍由外围样板面板管理，并由同步器投影为 `external_work_order` 工况任务。旧表退役前的完整数据库备份由部署记录保存，不能重新运行历史 `011_workbench_work_orders.sql` 或旧种子脚本重建双份数据。 + +## 工单与告警存储 + +### 任务告警 + +DDL 位于 `resources/sql/008_workbench_alert_events.sql`。任务通过 `POST /api/v1/workbench/task-alert-events` 推送 `raised`、`acknowledged` 或 `resolved` 事件。来源包含 `task_type`、`task_id` 和 `run_id`，生产者由认证项目身份绑定。`event_id` 在项目和生产者范围内幂等，告警事实在首次 raised 后不可改写。 + +### 外围工单 + +外围服务的 `peripheral_sample.work_orders` 是业务工单的管理来源，当前 `origin` 固定为 `system_data`；`activity` 只记录实际发生的创建、修改和现场反馈，不由总状态补造阶段。同步器将有效工单写入 `workbench.operational_tasks` 的 `external_work_order` 投影。原有三张独立演示工单也已迁入外围样板，并由同一面板管理。 + +`resources/sql/011_workbench_work_orders.sql` 和 `017_workbench_schema_comments.sql` 属于历史基线；`021_retire_workbench_work_orders.sql` 在验证三个迁入目标及其任务投影后，仅在 `tjwater_v2` 删除旧表。不要重放 `011` 重建它。`GET /api/v1/workbench/runtime` 暂保留空的 `work_orders` 字段作为接口兼容。其他项目数据库没有旧表，不需要应用 `021`。 + +告警事件表需要 `SELECT, INSERT`，告警状态表需要 `SELECT, INSERT, UPDATE`。迁移及扩展门禁见 [`workbench-database-rollout.md`](workbench-database-rollout.md)。 + +当前外围工单来源与按项目路由见 [外围服务](peripheral-sample.md)，其中 `sample` 只保留兼容名称。上述旧表迁移的历史数据库范围不表示其他项目已应用迁移，也不限制当前 API 的项目路由。 diff --git a/scripts/migrate_investigation_graphs.py b/scripts/migrate_investigation_graphs.py new file mode 100644 index 0000000..430248d --- /dev/null +++ b/scripts/migrate_investigation_graphs.py @@ -0,0 +1,182 @@ +"""Preview, back up, then adopt saved progress graphs without rewriting reports. + +Run with ``python -m scripts.migrate_investigation_graphs --help`` from the Server. +DSNs are resolved from MetaDB and never written to the plan or console. +""" +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import os +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from uuid import UUID + +from psycopg import AsyncConnection +from psycopg.rows import dict_row +from sqlalchemy import text + +from app.domain.schemas.operational_task import InvestigationDiagram, InvestigationProgressSnapshot +from app.infra.db.metadb.database import SessionLocal, engine +from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository, ProjectDbRouting +from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository +from app.services.investigation_graph import adopt_legacy_investigation_graph + + +MIGRATION = "investigation-graph-reference-v1" + + +def prepare_progress_document(document: dict) -> dict | None: + """Return a new revision, preserving publication provenance and Mermaid bytes.""" + graph = document["progress"].get("diagram") + if graph is None or graph.get("edges") is not None: + return None + diagram = adopt_legacy_investigation_graph(InvestigationDiagram.model_validate(graph)) + result = deepcopy(document) + result["revision"] += 1 + result["progress"]["diagram"] = diagram.model_dump(mode="json") + InvestigationProgressSnapshot.model_validate(result) + return result + + +def _write_private(path: Path, content: str, *, append: bool = False) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | (os.O_APPEND if append else os.O_EXCL) + with os.fdopen(os.open(path, flags, 0o600), "w", encoding="utf-8") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + + +async def _routing(project_id: UUID) -> ProjectDbRouting: + async with SessionLocal() as session: + await session.execute(text("SET TRANSACTION READ ONLY")) + route = await MetadataRepository(session).get_project_db_routing(project_id, "biz_data") + if route is None or route.db_type != "postgresql": + raise ValueError(f"Project {project_id} has no business PostgreSQL routing") + return route + + +async def preview(project_id: UUID) -> dict: + route = await _routing(project_id) + async with await AsyncConnection.connect( + route.dsn, row_factory=dict_row, connect_timeout=8, + options="-c default_transaction_read_only=on -c statement_timeout=30000", + ) as conn: + cursor = await conn.execute("SELECT current_database() AS database") + database = (await cursor.fetchone())["database"] + cursor = await conn.execute( + """SELECT task_id,run_id,document FROM workbench.operational_task_progress + WHERE project_id=%s AND jsonb_typeof(document#>'{progress,diagram}')='object' + AND jsonb_typeof(document#>'{progress,diagram,edges}') IS DISTINCT FROM 'array' + ORDER BY task_id,run_id""", (project_id,), + ) + records = [] + for row in await cursor.fetchall(): + after = prepare_progress_document(row["document"]) + if after is not None: + records.append({"task_id": str(row["task_id"]), "run_id": str(row["run_id"]), + "before": row["document"], "after": after}) + return {"project_id": str(project_id), "database": database, "records": records} + + +async def _protected_digest(conn: AsyncConnection, project_id: UUID) -> dict[str, str]: + """Compare complete protected rows inside the migration transaction.""" + hashes = {} + for table in ("operational_tasks", "operational_task_runs", "operational_task_reports"): + # Table names are fixed here, never supplied by CLI input. + cursor = await conn.execute( + f"SELECT to_jsonb(t) AS row FROM workbench.{table} t WHERE project_id=%s", + (project_id,), + ) + rows = sorted(json.dumps(row["row"], sort_keys=True) for row in await cursor.fetchall()) + hashes[table] = hashlib.sha256(json.dumps(rows).encode()).hexdigest() + return hashes + + +async def apply_project(project: dict) -> dict: + project_id = UUID(project["project_id"]) + route = await _routing(project_id) + async with await AsyncConnection.connect( + route.dsn, row_factory=dict_row, connect_timeout=8, + options="-c lock_timeout=5000 -c statement_timeout=30000", + ) as conn: + repo = OperationalTaskRepository(conn, project_id) + updated = already_managed = 0 + async with repo.transaction(): + cursor = await conn.execute("SELECT current_database() AS database") + if (await cursor.fetchone())["database"] != project["database"]: + raise ValueError("Project database routing changed since preview") + before = await _protected_digest(conn, project_id) + for item in project["records"]: + await repo.lock(f"task:{item['task_id']}") + cursor = await conn.execute( + """SELECT document FROM workbench.operational_task_progress + WHERE project_id=%s AND task_id=%s AND run_id=%s FOR UPDATE""", + (project_id, item["task_id"], item["run_id"]), + ) + row = await cursor.fetchone() + if row and (row["document"]["progress"].get("diagram") or {}).get("edges") is not None: + already_managed += 1 + continue + if row is None or row["document"] != item["before"]: + raise ValueError(f"Progress changed since preview: {item['task_id']}; prepare a new plan") + expected = prepare_progress_document(item["before"]) + if expected is None or expected != item["after"]: + raise ValueError("Migration plan does not match the graph contract") + await repo.save_progress(item["task_id"], expected) + await repo.append_event( + topic="operational.progress.changed", aggregate_type="operational_task_progress", + aggregate_id=item["task_id"], aggregate_revision=expected["revision"], + occurred_at=datetime.now(timezone.utc), metadata={ + "task_id": item["task_id"], "run_id": item["run_id"], + "execution_generation": expected["execution_generation"], "migration": MIGRATION, + }, + ) + updated += 1 + if before != await _protected_digest(conn, project_id): + raise ValueError("Protected task/run/report rows changed; migration rolled back") + return {"project_id": str(project_id), "database": project["database"], "updated": updated, + "already_managed": already_managed, "protected_rows_unchanged": True} + + +async def run(args: argparse.Namespace) -> None: + try: + if args.plan: + if not args.project_id: + raise ValueError("--plan requires at least one --project-id") + projects = [await preview(project_id) for project_id in dict.fromkeys(args.project_id)] + plan = {"migration": MIGRATION, "created_at": datetime.now(timezone.utc).isoformat(), + "projects": projects} + _write_private(args.plan, json.dumps(plan, ensure_ascii=False, indent=2) + "\n") + print(json.dumps({"plan": str(args.plan), "projects": [ + {"project_id": item["project_id"], "database": item["database"], + "progress_graphs": len(item["records"]), + "tasks": len({row["task_id"] for row in item["records"]})} for item in projects + ]})) + else: + plan = json.loads(args.apply.read_text(encoding="utf-8")) + if plan["migration"] != MIGRATION: + raise ValueError("Unsupported migration plan") + for project in plan["projects"]: + result = await apply_project(project) + _write_private(args.apply.with_suffix(".receipts.jsonl"), json.dumps(result) + "\n", append=True) + print(json.dumps(result), flush=True) + finally: + await engine.dispose() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--plan", type=Path, help="Write a preview and full progress backup to a new private JSON file") + mode.add_argument("--apply", type=Path, help="Apply an existing preview; changed rows abort that project's transaction") + parser.add_argument("--project-id", type=UUID, action="append", help="Explicit project scope for --plan") + asyncio.run(run(parser.parse_args())) + + +if __name__ == "__main__": + main() diff --git a/scripts/missing_api_endpoints.md b/scripts/missing_api_endpoints.md index d0410e4..07f4570 100644 --- a/scripts/missing_api_endpoints.md +++ b/scripts/missing_api_endpoints.md @@ -1,5 +1,7 @@ # Missing API Endpoints +> Historical endpoint inventory. Counts and missing-route claims describe the original scan, not the current API. Check [API rules](../docs/api-style.md) and [the generated contract](../contracts/server-v1.openapi.json) before planning changes. + - Legacy endpoints checked: 392 - Current endpoints found: 401 - Missing endpoints: 17 diff --git a/tests/fixtures/chart-publication-cases.json b/tests/fixtures/chart-publication-cases.json new file mode 100644 index 0000000..9561ef4 --- /dev/null +++ b/tests/fixtures/chart-publication-cases.json @@ -0,0 +1,992 @@ +[ + { + "name": "zero and sparse historical data", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "old mixed charts are read only", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "line", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "missing history timestamps", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ] + } + ] + } + }, + { + "name": "missing observed timestamp", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + null, + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "timestamp on a missing value", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + "2026-07-09T08:00:00+08:00", + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "wrong projected slot", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:30:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "mixed historical dates", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-10T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "same day history", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "future history", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-17T08:15:00+08:00", + "2026-07-17T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "duplicate slots", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:00", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "date appended axis", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "07-16 08:00", + "07-16 08:15", + "07-16 08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "inconsistent slot precision", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "bars mixed with comparison", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "bar", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "no historical series", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "no current observations", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + null, + null, + null + ], + "sample_times": [ + null, + null, + null + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "missing samples with missing timestamps", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + null, + 28, + 30 + ], + "sample_times": [ + null, + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "UTC spelling preserves projected instant", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T00:15:00Z", + "2026-07-09T00:30:00Z" + ] + } + ] + } + }, + { + "name": "ordinary line", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "line", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "合成读数", + "data": [ + 0, + null, + 30 + ] + } + ] + } + }, + { + "name": "ordinary bar", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "bar", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "合成读数", + "data": [ + 0, + null, + 30 + ] + } + ] + } + }, + { + "name": "midnight rollover keeps one window", + "accepted": true, + "chart": { + "title": "跨午夜同期对比", + "chart_type": "time_comparison", + "x_data": [ + "23:45", + "00:00" + ], + "series": [ + { + "name": "当前", + "type": "line", + "data": [ + 0, + 25 + ], + "sample_times": [ + "2026-07-16T23:45:00+08:00", + "2026-07-17T00:00:00+08:00" + ] + }, + { + "name": "历史", + "type": "scatter", + "data": [ + 26, + 27 + ], + "sample_times": [ + "2026-07-15T23:45:00+08:00", + "2026-07-16T00:00:00+08:00" + ] + } + ] + } + }, + { + "name": "microseconds are not aligned", + "accepted": false, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00.000001+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "fraction beyond Python microsecond precision", + "accepted": true, + "chart": { + "title": "P1 合成压力七日对照（m）", + "chart_type": "time_comparison", + "x_axis_name": "时间", + "y_axis_name": "压力（m）", + "x_data": [ + "08:00", + "08:15", + "08:30" + ], + "series": [ + { + "name": "P1 当前 2026-07-16", + "type": "line", + "data": [ + 0, + 28, + 30 + ], + "sample_times": [ + "2026-07-16T08:00:00.0000001+08:00", + "2026-07-16T08:15:00+08:00", + "2026-07-16T08:30:00+08:00" + ] + }, + { + "name": "P1 2026-07-09", + "type": "scatter", + "data": [ + null, + 32, + 33 + ], + "sample_times": [ + null, + "2026-07-09T08:15:00+08:00", + "2026-07-09T08:30:00+08:00" + ] + } + ] + } + }, + { + "name": "calendar overflow", + "accepted": false, + "chart": { + "title": "calendar overflow", + "chart_type": "time_comparison", + "x_data": [ + "07:45" + ], + "series": [ + { + "name": "当前", + "type": "line", + "data": [ + 1 + ], + "sample_times": [ + "9999-12-31T23:45:00Z" + ] + }, + { + "name": "历史缺测", + "type": "scatter", + "data": [ + null + ], + "sample_times": [ + null + ] + } + ] + } + }, + { + "name": "calendar underflow", + "accepted": false, + "chart": { + "title": "calendar underflow", + "chart_type": "time_comparison", + "x_data": [ + "08:00" + ], + "series": [ + { + "name": "当前", + "type": "line", + "data": [ + 1 + ], + "sample_times": [ + "0001-01-01T00:00:00+23:59" + ] + }, + { + "name": "历史缺测", + "type": "scatter", + "data": [ + null + ], + "sample_times": [ + null + ] + } + ] + } + } +] diff --git a/tests/fixtures/unified-presentation.jsonl b/tests/fixtures/unified-presentation.jsonl new file mode 100644 index 0000000..7eaacb5 --- /dev/null +++ b/tests/fixtures/unified-presentation.jsonl @@ -0,0 +1,14 @@ +{"record_type":"presentation","command":{"action":"locate_features","layer":"scada","feature_ids":["P1"]}} +{"record_type":"presentation","command":{"action":"zoom_to_map","center":[121.48,31.1],"zoom":14}} +{"record_type":"presentation","command":{"action":"highlight_features","layer":"scada","feature_ids":["P1"],"effect":"pulse"}} +{"record_type":"presentation","command":{"action":"open_chart","chart_id":"fixture-pressure-P1","evidence_refs":["dev:synthetic"],"chart":{"title":"P1 合成压力七日对照（m）","chart_type":"time_comparison","x_axis_name":"时间","y_axis_name":"压力（m）","x_data":["08:00","08:15","08:30"],"series":[{"name":"P1 当前 2026-07-16","type":"line","data":[32,28,30],"sample_times":["2026-07-16T08:00:00+08:00","2026-07-16T08:15:00+08:00","2026-07-16T08:30:00+08:00"]},{"name":"P1 2026-07-09","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-09T08:15:00+08:00","2026-07-09T08:30:00+08:00"]},{"name":"P1 2026-07-10","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-10T08:15:00+08:00","2026-07-10T08:30:00+08:00"]},{"name":"P1 2026-07-11","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-11T08:15:00+08:00","2026-07-11T08:30:00+08:00"]},{"name":"P1 2026-07-12","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-12T08:15:00+08:00","2026-07-12T08:30:00+08:00"]},{"name":"P1 2026-07-13","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-13T08:15:00+08:00","2026-07-13T08:30:00+08:00"]},{"name":"P1 2026-07-14","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-14T08:15:00+08:00","2026-07-14T08:30:00+08:00"]},{"name":"P1 2026-07-15","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-15T08:15:00+08:00","2026-07-15T08:30:00+08:00"]}]}}} +{"record_type":"presentation","command":{"action":"open_chart","chart_id":"fixture-pressure-P2","evidence_refs":["dev:synthetic"],"chart":{"title":"P2 合成压力七日对照（m）","chart_type":"time_comparison","x_axis_name":"时间","y_axis_name":"压力（m）","x_data":["08:00","08:15","08:30"],"series":[{"name":"P2 当前 2026-07-16","type":"line","data":[31,27,29],"sample_times":["2026-07-16T08:00:00+08:00","2026-07-16T08:15:00+08:00","2026-07-16T08:30:00+08:00"]},{"name":"P2 2026-07-09","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-09T08:15:00+08:00","2026-07-09T08:30:00+08:00"]},{"name":"P2 2026-07-10","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-10T08:15:00+08:00","2026-07-10T08:30:00+08:00"]},{"name":"P2 2026-07-11","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-11T08:15:00+08:00","2026-07-11T08:30:00+08:00"]},{"name":"P2 2026-07-12","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-12T08:15:00+08:00","2026-07-12T08:30:00+08:00"]},{"name":"P2 2026-07-13","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-13T08:15:00+08:00","2026-07-13T08:30:00+08:00"]},{"name":"P2 2026-07-14","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-14T08:15:00+08:00","2026-07-14T08:30:00+08:00"]},{"name":"P2 2026-07-15","type":"scatter","data":[null,32,33],"sample_times":[null,"2026-07-15T08:15:00+08:00","2026-07-15T08:30:00+08:00"]}]}}} +{"record_type":"presentation","command":{"action":"set_overlay","overlay_id":"fixture-area","title":"合成空间样例","mode":"geometry","evidence_id":"dev:synthetic","data":{"type":"FeatureCollection","features":[{"type":"Feature","properties":{"origin":"synthetic"},"geometry":{"type":"Point","coordinates":[121.48,31.1]}}]}}} +{"record_type":"presentation","command":{"action":"render_scada_analysis","items":[{"device_id":"P1","level":"high"}]}} +{"record_type":"ui_block","evidence_refs":[],"block":{"version":"answer-ui@1","block_id":"fixture-summary","phase":"final","fallback_text":"合成候选对比，仅供展示测试。","spec":{"root":"table","elements":{"table":{"type":"Table","props":{"title":"合成候选对比","columns":["候选","说明"],"rows":[["DEMO-PIPE-01","仅供展示测试"],["DEMO-PIPE-02","未执行定位分析"]]},"children":[]}}}}} +{"record_type":"candidate","origin":"synthetic","source_evidence_refs":[],"element_id":"DEMO-PIPE-01","rank":1,"status":"example_only"} +{"record_type":"presentation","command":{"action":"close_chart","chart_id":"fixture-pressure-P1"}} +{"record_type":"presentation","command":{"action":"close_chart","chart_id":"fixture-pressure-P2"}} +{"record_type":"presentation","command":{"action":"clear_overlay","overlay_id":"fixture-area"}} +{"record_type":"presentation","command":{"action":"clear_highlights"}} +{"record_type":"presentation","command":{"action":"clear_scada_analysis"}} diff --git a/tests/integration/test_investigation_reference_projection.py b/tests/integration/test_investigation_reference_projection.py new file mode 100644 index 0000000..dbcef90 --- /dev/null +++ b/tests/integration/test_investigation_reference_projection.py @@ -0,0 +1,84 @@ +"""Opt-in read-only JSONB probes; no tables or business records are written.""" +import asyncio +import os +from copy import deepcopy + +import pytest +from psycopg import AsyncConnection +from psycopg.rows import dict_row +from psycopg.types.json import Jsonb + +from app.infra.db.postgresql import read_queries + + +@pytest.mark.parametrize("section", ["progress", "report"]) +def test_reference_annotations_are_normalized_before_nested_paging(section): + dsn = os.environ.get("MCP_QUERY_TEST_DSN") + if not dsn: + pytest.skip("MCP_QUERY_TEST_DSN is required for a read-only PostgreSQL JSONB probe") + + async def check(): + statuses = ["running", "completed", "waiting", "waiting_human", "skipped", "cancelled", "failed", None] + original = {"section": section, "data": { + "revision": 3, "execution_generation": 2, + section: {"diagram": {"mermaid": "flowchart TD\nA[原图]", "node_refs": [ + {"node_id": f"n{index}", "status": status, "detail": "保留原说明", "evidence_refs": ["e1"]} + for index, status in enumerate(statuses) + ] + [{"node_id": "unset"}]}}, + "status": "running", "action": {"status": "completed"}, + }} + untouched = deepcopy(original) + async with await AsyncConnection.connect( + dsn, row_factory=dict_row, options="-c default_transaction_read_only=on", + ) as conn: + async def page(path=(), *, limit=50, offset=0, full=False): + statement, params = read_queries._reference_diagram_projection( + "SELECT %s::jsonb AS snapshot", (Jsonb(original),), section, + ) + return await read_queries._snapshot_page( + conn, stored_sql=statement, stored_params=params, data_path=list(path), + limit=limit, offset=offset, mcp_sources=[], full_snapshot=full, max_bytes=8000, + ) + + scalar = await page([section, "diagram", "node_refs", "0", "status"]) + assert scalar["value"] == "reference" + second = await page([section, "diagram", "node_refs"], limit=1, offset=1) + assert second["value"][0]["status"] == "recorded" + assert second["total"] == 9 and second["nested"] + assert second["value"][0]["evidence_refs"]["data_path"] == [section, "diagram", "node_refs", "1", "evidence_refs"] + snapshot = await page(full=True) + nodes = snapshot["data"][section]["diagram"]["node_refs"] + assert [node.get("status") for node in nodes] == [ + "reference", "recorded", "needs_input", "needs_input", "not_taken", "not_taken", "failed", None, None, + ] + assert "status" not in nodes[-1] + assert nodes[6]["detail"] == "保留原说明" and nodes[6]["evidence_refs"] == ["e1"] + assert snapshot["data"]["status"] == "running" + assert snapshot["data"]["action"]["status"] == "completed" + assert snapshot["data"][section]["diagram"]["mermaid"] == original["data"][section]["diagram"]["mermaid"] + assert original == untouched + + asyncio.run(check()) + + +@pytest.mark.parametrize("diagram", [None, {}, {"node_refs": []}]) +def test_reference_projection_preserves_empty_and_missing_legacy_diagrams(diagram): + dsn = os.environ.get("MCP_QUERY_TEST_DSN") + if not dsn: + pytest.skip("MCP_QUERY_TEST_DSN is required for a read-only PostgreSQL JSONB probe") + + async def check(): + original = {"section": "progress", "data": {"progress": {"diagram": diagram}, "status": "running"}} + async with await AsyncConnection.connect( + dsn, row_factory=dict_row, options="-c default_transaction_read_only=on", + ) as conn: + statement, params = read_queries._reference_diagram_projection( + "SELECT %s::jsonb AS snapshot", (Jsonb(original),), "progress", + ) + result = await read_queries._snapshot_page( + conn, stored_sql=statement, stored_params=params, data_path=[], + limit=20, offset=0, mcp_sources=[], full_snapshot=True, + ) + assert result == original + + asyncio.run(check()) diff --git a/tests/support/operational_task_repository.py b/tests/support/operational_task_repository.py index 741ca2e..a37bb60 100644 --- a/tests/support/operational_task_repository.py +++ b/tests/support/operational_task_repository.py @@ -104,6 +104,19 @@ class TaskRepository: async def get_latest_progress(self, task_id, generation): return self.progress.get((task_id, generation)) + async def get_latest_graph_progress(self, task_id): + snapshots = [value for (stored_id, _), value in self.progress.items() + if stored_id == task_id + and (value["progress"].get("diagram") or {}).get("edges") is not None] + return max(snapshots, key=lambda row: (row["execution_generation"], row["revision"])) if snapshots else None + + async def get_latest_diagram(self, task_id): + diagrams = [(row["published_at"], row["progress"].get("diagram")) + for (stored_id, _), row in self.progress.items() if stored_id == task_id] + diagrams.extend((row["published_at"], row["report"].get("diagram")) for row in self.reports) + return next((diagram for _, diagram in sorted(diagrams, key=lambda item: item[0], reverse=True) + if diagram is not None), None) + async def save_progress(self, task_id, snapshot): self.progress[(task_id, snapshot["execution_generation"])] = snapshot diff --git a/tests/unit/test_chart_publication_examples.py b/tests/unit/test_chart_publication_examples.py new file mode 100644 index 0000000..f02b77c --- /dev/null +++ b/tests/unit/test_chart_publication_examples.py @@ -0,0 +1,48 @@ +"""Generated fixtures keep executable examples and publication semantics aligned across repositories.""" +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from app.domain.domain_errors import DomainError +from app.domain.schemas.operational_task import TaskPresentationEventCommand, TaskSafeChart +from app.services.time_comparison_chart import validate_time_comparison_chart + + +FIXTURES = Path(__file__).parents[1] / "fixtures" +CASES = json.loads((FIXTURES / "chart-publication-cases.json").read_text()) + + +@pytest.mark.parametrize("case", CASES, ids=lambda case: case["name"]) +def test_chart_publication_matches_frontend_and_agent(case): + def validate(): + chart = TaskSafeChart.model_validate(case["chart"]).model_dump(mode="json") + validate_time_comparison_chart(chart) + + if case["accepted"]: + validate() + else: + with pytest.raises((DomainError, ValidationError)): + validate() + + +def test_historical_mixed_charts_remain_structurally_readable(): + chart = next(case["chart"] for case in CASES if case["name"] == "old mixed charts are read only") + TaskSafeChart.model_validate(chart) + + +def test_exported_command_examples_use_the_current_publication_contract(): + for line in (FIXTURES / "unified-presentation.jsonl").read_text().splitlines(): + record = json.loads(line) + if record["record_type"] != "presentation": + continue + event = TaskPresentationEventCommand.model_validate({ + "request_id": "95000000-0000-4000-8000-000000000001", + "run_id": "95000000-0000-4000-8000-000000000002", + "source_session_id": "synthetic-chart-contract", + "command": record["command"], + }) + command = event.command + if command.action == "open_chart": + validate_time_comparison_chart(command.chart.model_dump(mode="json")) diff --git a/tests/unit/test_investigation_diagram_status.py b/tests/unit/test_investigation_diagram_status.py index a970624..84eb0c2 100644 --- a/tests/unit/test_investigation_diagram_status.py +++ b/tests/unit/test_investigation_diagram_status.py @@ -28,7 +28,7 @@ def test_waiting_and_unknown_node_states_survive_progress_and_final_report(): stage="final", outcome="concluded", data_gaps=[], diagram=diagram, ) assert report.diagram == progress.diagram - assert [node.status for node in report.diagram.node_refs] == ["completed", "waiting", None, None] + assert [node.status for node in report.diagram.node_refs] == ["recorded", "needs_input", None, None] assert "status" not in report.diagram.node_refs[-1].model_fields_set @@ -43,7 +43,7 @@ def test_class_attached_nodes_are_valid_declarations_without_accepting_partial_i "node_refs": [{"node_id": "QUERY", "label": "核查", "status": "running"}, {"node_id": "REVIEW", "label": "复核", "status": "pending"}]}, }} - assert AgentArtifactsCommand.model_validate(payload).progress.diagram.node_refs[0].status == "running" + assert AgentArtifactsCommand.model_validate(payload).progress.diagram.node_refs[0].status == "reference" payload["progress"]["diagram"]["node_refs"][0]["node_id"] = "QUER" with pytest.raises(ValidationError, match="QUER"): AgentArtifactsCommand.model_validate(payload) diff --git a/tests/unit/test_investigation_graph.py b/tests/unit/test_investigation_graph.py new file mode 100644 index 0000000..c34c78b --- /dev/null +++ b/tests/unit/test_investigation_graph.py @@ -0,0 +1,315 @@ +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import pytest + +from app.domain.domain_errors import DomainError +from app.domain.schemas.operational_task import AgentArtifactsCommand +from app.domain.schemas.operational_task import InvestigationDiagram, InvestigationGraphUpdate +from app.domain.schemas.operational_data import TaskDataSnapshot +from app.services.investigation_graph import update_investigation_graph +from app.services.operational_tasks import OperationalTaskService +from tests.support.operational_task_repository import ( + TaskRepository, _active_run, _context, _task_document, +) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +def setup_service(): + now = datetime(2026, 11, 10, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repo = TaskRepository(task, _active_run(task, now)) + return task, repo, OperationalTaskService(repo, context, set(), clock=lambda: now) + + +async def publish(service, repo, task_id, *, nodes=(), edges=(), **progress): + command = AgentArtifactsCommand.model_validate({ + "request_id": str(uuid4()), "source_session_id": repo.run["agent_session_id"], + "run_id": repo.run["run_id"], + "progress": {"summary": "更新调查路线", "graph_update": { + "nodes": list(nodes), "edges": list(edges), + }, **progress}, + }) + return await service.publish_artifacts(task_id, command) + + +@pytest.mark.anyio +async def test_failed_path_is_retained_when_a_new_route_is_added(): + task, repo, service = setup_service() + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "read", "label": "读取全量 SCADA", "status": "recorded"}, + {"node_id": "locate_1", "label": "第一次候选定位", "status": "reference"}, + ], edges=[{"source": "read", "target": "locate_1"}]) + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "locate_1", "label": "第一次候选定位", "status": "failed", "detail": "计算超时"}, + {"node_id": "locate_2", "label": "调整窗口后重试", "status": "reference"}, + ], edges=[{"source": "locate_1", "target": "locate_2", "label": "调整后重试"}]) + result = await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "locate_2", "label": "调整窗口后重试", "status": "recorded"}, + ]) + diagram = result.task.latest_progress.progress.diagram + assert [(n.node_id, n.status) for n in diagram.node_refs] == [ + ("read", "recorded"), ("locate_1", "failed"), ("locate_2", "recorded"), + ] + assert diagram.node_refs[1].detail == "计算超时" + assert len(diagram.edges) == 2 + assert "第一次候选定位（失败）" in diagram.mermaid + assert 'locate_1 -->|"调整后重试"| locate_2' in diagram.mermaid + # A text-only update cannot clear the accumulated graph. + summary = AgentArtifactsCommand.model_validate({ + "request_id": str(uuid4()), "source_session_id": repo.run["agent_session_id"], + "run_id": repo.run["run_id"], "progress": {"summary": "等待人工", "diagram": None}, + }) + assert (await service.publish_artifacts(task["task_id"], summary)).task.latest_progress.progress.diagram == diagram + + +@pytest.mark.anyio +async def test_failed_attempt_cannot_be_relabelled_as_a_success(): + task, repo, service = setup_service() + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "attempt_1", "label": "候选定位", "status": "failed", "detail": "超时"}, + ]) + with pytest.raises(DomainError, match="new node"): + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "attempt_1", "label": "候选定位", "status": "recorded"}, + ]) + + +@pytest.mark.anyio +async def test_new_execution_round_continues_the_same_task_graph(): + task, repo, service = setup_service() + first = await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "first", "label": "首次查询", "status": "failed"}, + ]) + archived = first.task.latest_progress.progress.diagram.model_dump(mode="json") + repo.task["execution_generation"] = 2 + repo.task["execution_request_id"] = str(uuid4()) + repo.run["status"] = "succeeded" + repo.run = _active_run(repo.task, datetime(2026, 11, 10, tzinfo=timezone.utc)) + repo.runs[repo.run["run_id"]] = repo.run + assert (await service.detail(task["task_id"])).latest_progress.progress.diagram.node_refs[0].status == "failed" + result = await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "second", "label": "补充信息后查询", "status": "recorded"}, + ], edges=[{"source": "first", "target": "second"}]) + assert result.progress_revision == 1 + assert result.task.latest_progress.execution_generation == 2 + assert len(result.task.latest_progress.progress.diagram.node_refs) == 2 + assert repo.progress[(task["task_id"], 1)]["progress"]["diagram"] == archived + + +@pytest.mark.anyio +async def test_reports_archive_the_host_graph_and_keep_older_versions_unchanged(): + task, repo, service = setup_service() + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "failed", "label": "数据读取", "status": "failed", "detail": "接口超时"}, + ]) + command = AgentArtifactsCommand.model_validate({ + "request_id": str(uuid4()), "source_session_id": repo.run["agent_session_id"], + "run_id": repo.run["run_id"], "report": { + "title": "调查", "current_conclusion": "已有阶段判断", "conclusion_boundary": "限于现有资料", + "stage": "final", "outcome": "concluded", "data_gaps": [], + "diagram": {"title": "重新生成的图", "mermaid": "flowchart TD\nNEW[新图]"}, + }, + }) + first = await service.publish_artifacts(task["task_id"], command) + assert first.task.latest_report.report.diagram.node_refs[0].status == "failed" + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "retry", "label": "重新读取", "status": "recorded"}, + ], edges=[{"source": "failed", "target": "retry"}]) + assert len((await service.report(task["task_id"], 1)).report.diagram.node_refs) == 1 + assert len((await service.detail(task["task_id"])).latest_progress.progress.diagram.node_refs) == 2 + # Idempotent replays do not append nodes or change the archived report. + assert (await service.publish_artifacts(task["task_id"], command)).report_revision == 1 + + +@pytest.mark.anyio +async def test_graph_updates_validate_references_and_cannot_replace_a_managed_graph(): + task, repo, service = setup_service() + with pytest.raises(DomainError, match="unknown evidence"): + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "unknown", "label": "未知依据", "evidence_refs": ["forged"]}, + ]) + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "kept", "label": "实际步骤", "status": "recorded"}, + ]) + with pytest.raises(DomainError, match="graph_update"): + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand.model_validate({ + "request_id": str(uuid4()), "source_session_id": repo.run["agent_session_id"], + "run_id": repo.run["run_id"], "progress": { + "summary": "重画", "diagram": {"title": "替换", "mermaid": "flowchart TD\nNEW[新图]"}, + }, + })) + + +def test_graph_renderer_escapes_labels_and_keeps_explicit_parallel_and_failed_paths(): + graph = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({ + "nodes": [ + {"node_id": "start", "label": '核查 "地点"