diff --git a/app/api/v1/endpoints/operational_tasks.py b/app/api/v1/endpoints/operational_tasks.py index c9d6e5a..ae552ea 100644 --- a/app/api/v1/endpoints/operational_tasks.py +++ b/app/api/v1/endpoints/operational_tasks.py @@ -50,6 +50,8 @@ from app.domain.schemas.operational_task import ( ExternalSourceWithdrawalCommand, HumanApprovalAction, HumanEvidenceCommand, + InvestigationReportPage, + InvestigationReportRevision, NotificationAcknowledgeCommand, OperationalTaskCreate, OperationalTaskDetail, @@ -100,6 +102,34 @@ def _raise_domain(exc: DomainError): raise HTTPException(status_code=exc.status, detail=exc.payload()) from exc +@router.get("/operational-tasks/{task_id}/reports", response_model=InvestigationReportPage) +async def list_investigation_reports( + task_id: UUID, + limit: int = Query(default=20, ge=1, le=100), + offset: int = Query(default=0, ge=0), + ctx: ProjectContext = Depends(require_permission(OPERATIONAL_TASK_VIEW)), + conn: AsyncConnection = Depends(get_project_pg_connection), +): + try: + items, next_offset = await _service(ctx, conn).list_reports(task_id, limit, offset) + return InvestigationReportPage(items=items, next_offset=next_offset) + except DomainError as exc: + _raise_domain(exc) + + +@router.get("/operational-tasks/{task_id}/reports/{revision}", response_model=InvestigationReportRevision) +async def get_investigation_report( + task_id: UUID, + revision: int, + ctx: ProjectContext = Depends(require_permission(OPERATIONAL_TASK_VIEW)), + conn: AsyncConnection = Depends(get_project_pg_connection), +): + try: + return await _service(ctx, conn).report(task_id, revision) + except DomainError as exc: + _raise_domain(exc) + + @router.post( "/operational-tasks", response_model=OperationalTaskDetail, @@ -417,11 +447,12 @@ async def record_operational_task_run_dispatch_failure( ) async def get_operational_task_agent_context( task_id: UUID, + report_revision: int | None = Query(default=None, ge=1), ctx: ProjectContext = Depends(require_permission(OPERATIONAL_TASK_VIEW)), conn: AsyncConnection = Depends(get_project_pg_connection), ): try: - return await _service(ctx, conn).agent_context(task_id) + return await _service(ctx, conn).agent_context(task_id, report_revision) except DomainError as exc: _raise_domain(exc) diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 8499412..9f19349 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -1,12 +1,13 @@ """Contracts for project-scoped operational tasks and human-approved actions.""" from __future__ import annotations +import re from typing import Annotated, Literal from uuid import UUID from pydantic import Field, JsonValue, model_validator -from app.domain.schemas.shared import NonEmptyStr, PositiveInt, StrictModel +from app.domain.schemas.shared import NonEmptyStr, PositiveInt, StrictBool, StrictModel from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource from app.domain.schemas.workbench_runtime import ( WorkbenchDateTime, @@ -190,13 +191,84 @@ class StoredEvidenceRecord(EvidenceRecord): data_snapshot: TaskDataSnapshot | None = None -class InvestigationReportInput(StrictModel): +class InvestigationDataGap(StrictModel): + gap_id: NonEmptyStr + node_id: NonEmptyStr | None = None + missing_data: NonEmptyStr + scope: NonEmptyStr + impact: NonEmptyStr + required_data: NonEmptyStr + evidence_refs: list[NonEmptyStr] = Field(min_length=1) + blocking: StrictBool + + +class InvestigationDiagramNodeRef(StrictModel): + node_id: Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_-]{0,79}$")] + label: NonEmptyStr + evidence_refs: list[NonEmptyStr] = Field(default_factory=list) + action_refs: list[UUID] = Field(default_factory=list) + + +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) + + @model_validator(mode="after") + def validate_description(self): + if not re.match(r"^\s*(?:flowchart|graph)\s+(?:TD|TB|BT|LR|RL)\b", self.mermaid): + raise ValueError("investigation diagrams must be Mermaid flowcharts") + if "```" in self.mermaid or "%%{" in self.mermaid or re.search(r"(?:^|[;\n])\s*click\b", self.mermaid): + raise ValueError("diagram descriptions cannot contain fences, directives or click commands") + ids = [node.node_id for node in self.node_refs] + if len(ids) != len(set(ids)): + raise ValueError("diagram node references must be unique") + return self + + +class InvestigationReport(StrictModel): + """Read model: absent historical facts remain unknown, never inferred.""" + title: NonEmptyStr current_conclusion: NonEmptyStr conclusion_boundary: NonEmptyStr evidence_refs: list[NonEmptyStr] = Field(default_factory=list) recommendations: list[NonEmptyStr] = Field(default_factory=list) pending_items: list[NonEmptyStr] = Field(default_factory=list) + stage: Literal["interim", "final"] | None = None + outcome: Literal["concluded", "insufficient_data"] | None = None + stop_reason: NonEmptyStr | None = None + data_gaps: list[InvestigationDataGap] | None = None + diagram: InvestigationDiagram | None = None + + @model_validator(mode="after") + def validate_result(self): + if self.outcome is None: + if self.stage is not None or self.stop_reason is not None or self.data_gaps is not None: + raise ValueError("structured report fields must be supplied together") + return self + if self.stage is None or self.data_gaps is None: + raise ValueError("structured reports require stage and data_gaps") + ids = [gap.gap_id for gap in self.data_gaps] + if len(ids) != len(set(ids)): + raise ValueError("data gap ids must be unique") + blocking = any(gap.blocking for gap in self.data_gaps) + if self.outcome == "insufficient_data": + if self.stage != "final" or not self.stop_reason or not blocking: + raise ValueError("insufficient data requires a final report, stop reason and blocking gap") + if self.recommendations or self.pending_items: + raise ValueError("insufficient data ends the investigation without downstream recommendations or pending work") + elif self.stop_reason is not None or blocking: + raise ValueError("concluded report cannot contain a stop reason or blocking gap") + return self + + +class InvestigationReportInput(InvestigationReport): + """Write model: every new report explicitly declares its business result.""" + + stage: Literal["interim", "final"] + outcome: Literal["concluded", "insufficient_data"] + data_gaps: list[InvestigationDataGap] class ExecutionPlanSnapshot(StrictModel): @@ -214,7 +286,17 @@ class InvestigationReportRevision(StrictModel): source_session_id: NonEmptyStr published_at: WorkbenchDateTime published_by: NonEmptyStr - report: InvestigationReportInput + report: InvestigationReport + + +class InvestigationReportPage(StrictModel): + items: list[InvestigationReportRevision] + next_offset: int | None = Field(default=None, ge=0) + + +class InvestigationReportReference(StrictModel): + execution_generation: PositiveInt + revision: PositiveInt class AgentArtifactsCommand(StrictModel): @@ -481,6 +563,7 @@ class OperationalTaskRunPage(StrictModel): class AgentTaskContext(StrictModel): task: OperationalTaskDetail approved_data_scopes: list[NonEmptyStr] + selected_report: InvestigationReportRevision | None = None class AgentRunCommand(StrictModel): @@ -610,6 +693,7 @@ class OperationalTaskNotification(StrictModel): lifecycle_status: Literal["active", "resolved"] = "active" subject_type: Literal["task", "run", "access_request", "action"] = "task" subject_id: UUID | None = None + report_reference: InvestigationReportReference | None = None resolved_at: WorkbenchDateTime | None = None resolution_reason: NonEmptyStr | None = None status: Literal["unread", "read"] diff --git a/app/infra/db/postgresql/operational_tasks.py b/app/infra/db/postgresql/operational_tasks.py index d274e3c..a20a189 100644 --- a/app/infra/db/postgresql/operational_tasks.py +++ b/app/infra/db/postgresql/operational_tasks.py @@ -409,6 +409,24 @@ class OperationalTaskRepository: async def get_latest_report(self, task_id: str) -> dict | None: return await self._latest_artifact("operational_task_reports", task_id) + async def get_report(self, task_id: str, revision: int) -> dict | None: + cursor = await self.connection.execute( + """SELECT document FROM workbench.operational_task_reports + WHERE project_id=%s AND task_id=%s AND revision=%s""", + (self.project_id, task_id, revision), + ) + row = await cursor.fetchone() + return row["document"] if row else None + + async def list_reports(self, task_id: str, limit: int, offset: int) -> list[dict]: + cursor = await self.connection.execute( + """SELECT document FROM workbench.operational_task_reports + WHERE project_id=%s AND task_id=%s ORDER BY revision DESC + LIMIT %s OFFSET %s""", + (self.project_id, task_id, limit, offset), + ) + return [row["document"] for row in await cursor.fetchall()] + async def _latest_artifact(self, table: str, task_id: str) -> dict | None: if table not in {"operational_task_plans", "operational_task_reports"}: raise ValueError("unsupported artifact table") @@ -535,7 +553,7 @@ class OperationalTaskRepository: SET document=document || jsonb_build_object( 'lifecycle_status','resolved', 'resolved_at',to_jsonb(%s::timestamptz), - 'resolution_reason',%s + 'resolution_reason',%s::text ) WHERE project_id=%s AND task_id=%s AND COALESCE(document->>'lifecycle_status','active')='active' diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 48344c2..42f9fe5 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -45,6 +45,7 @@ from app.domain.schemas.operational_task import ( HumanApprovalAction, HumanEvidenceCommand, InvestigationReportRevision, + InvestigationReportReference, IssueCommandPayload, NotificationAcknowledgeCommand, OperationalTaskCreate, @@ -248,7 +249,22 @@ class OperationalTaskService: work_order_snapshot=work_order_snapshot, ) - async def agent_context(self, task_id: UUID) -> AgentTaskContext: + async def report(self, task_id: UUID, revision: int) -> InvestigationReportRevision: + await self._task(task_id) + value = await self.repo.get_report(str(task_id), revision) + require(value is not None, "REPORT_NOT_FOUND", "Report revision not found", status=404) + return InvestigationReportRevision.model_validate(value) + + async def list_reports( + self, task_id: UUID, limit: int, offset: int + ) -> tuple[list[InvestigationReportRevision], int | None]: + await self._task(task_id) + rows = await self.repo.list_reports(str(task_id), limit + 1, offset) + return [InvestigationReportRevision.model_validate(row) for row in rows[:limit]], ( + offset + limit if len(rows) > limit else None + ) + + async def agent_context(self, task_id: UUID, report_revision: int | None = None) -> AgentTaskContext: scopes: list[str] = [] if SIMULATION_VIEW in self.permissions: scopes.extend([ @@ -263,6 +279,7 @@ class OperationalTaskService: return AgentTaskContext( task=await self.detail(task_id), approved_data_scopes=list(dict.fromkeys(scopes)), + selected_report=await self.report(task_id, report_revision) if report_revision is not None else None, ) async def can_read_source(self, task_id: UUID, source: TaskDataSource) -> bool: @@ -693,6 +710,10 @@ class OperationalTaskService: "Only a running run accepts Agent artifacts", status=409) now = self._now() self._require_active_run_lease(active_run, now) + previous_report = await self.repo.get_latest_report(str(task_id)) + require(not previous_report or previous_report.get("execution_generation") != task["execution_generation"] + or previous_report["report"].get("outcome") != "insufficient_data", + "INVESTIGATION_ENDED", "Insufficient-data investigation requires a new run", status=409) known_evidence = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))} for item in command.evidence: document = _json(EvidenceRecord( @@ -763,6 +784,20 @@ class OperationalTaskService: current_plan = snapshot if command.report: self._require_evidence_refs(command.report.evidence_refs, known_evidence) + for gap in command.report.data_gaps: + require(set(gap.evidence_refs) <= set(command.report.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)) + 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": + 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", + "Resolve existing actions before ending for insufficient data", status=409) report_revision = (report_revision or 0) + 1 report = InvestigationReportRevision( revision=report_revision, @@ -779,6 +814,25 @@ class OperationalTaskService: occurred_at=now, 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 ( + 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) != _json(command.report).get(key) + for key in ("current_conclusion", "conclusion_boundary", "outcome", "stop_reason", + "evidence_refs", "recommendations", "pending_items", "data_gaps")) + ): + await self._resolve_notifications( + task_id, kinds=["report_updated"], subject_type="task", subject_id=task_id, + now=now, reason="report_superseded", + ) + await self._notify( + task_id, "report_updated", task["title"], command.report.current_conclusion, now, + report_reference=InvestigationReportReference( + execution_generation=task["execution_generation"], revision=report_revision, + ), + ) if current_plan is None: stored_plan = await self.repo.get_latest_plan(str(task_id)) if stored_plan is not None: @@ -804,14 +858,11 @@ class OperationalTaskService: ) else: # Compatibility for pre-015 repository adapters only. - if ( - command.report and current_plan is not None - and current_plan.execution_generation == task["execution_generation"] - and current_plan.source_session_id == command.source_session_id - ): - task["status"] = self._status_from_plan( - current_plan.plan.nodes, actions.values() - ) + if command.report: + if task["task_type"] == "event_investigation": + task["status"] = self._investigation_status(actions.values()) if command.report.stage == "final" else "running" + else: + task["status"] = self._status_from_plan(current_plan.plan.nodes, actions.values()) if current_plan else "running" if task["status"] in {"waiting_human", "completed", "failed", "cancelled"}: task["lifecycle_status"] = task["status"] task["agent_lease_expires_at"] = ( @@ -839,13 +890,21 @@ class OperationalTaskService: task = await self._task(task_id) self._require_actionable_task(task) self._validate_action_payload(command) + if task["task_type"] == "event_investigation": + report = await self.repo.get_latest_report(str(task_id)) + require(report is not None + and report.get("execution_generation") == task["execution_generation"] + and report["report"].get("outcome") == "concluded", + "ACTION_CONCLUSION_REQUIRED", + "Publish a current concluded report before proposing an action", status=409) if command.node_id: - latest_plan = await self.repo.get_latest_plan(str(task_id)) - require(latest_plan is not None, "PLAN_NODE_NOT_FOUND", - "Action cannot be attached before a plan is published", status=409) - node_ids = {node["id"] for node in latest_plan["plan"]["nodes"]} - require(command.node_id in node_ids, "PLAN_NODE_NOT_FOUND", - "Action references an unknown plan node", refs=[command.node_id], status=409) + if task["task_type"] != "event_investigation": + latest_plan = await self.repo.get_latest_plan(str(task_id)) + require(latest_plan is not None, "PLAN_NODE_NOT_FOUND", + "Action cannot be attached before a plan is published", status=409) + node_ids = {node["id"] for node in latest_plan["plan"]["nodes"]} + require(command.node_id in node_ids, "PLAN_NODE_NOT_FOUND", + "Action references an unknown plan node", refs=[command.node_id], status=409) known = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))} self._require_evidence_refs(command.evidence_refs, known) now = self._now() @@ -1658,18 +1717,21 @@ class OperationalTaskService: if report_document is not None else None ) same_generation = ( - plan is not None and report is not None - and plan.execution_generation == run["generation"] + report is not None and report.execution_generation == run["generation"] - and plan.source_session_id == command.source_session_id and report.source_session_id == command.source_session_id + and (task["task_type"] == "event_investigation" or ( + plan is not None + and plan.execution_generation == run["generation"] + and plan.source_session_id == command.source_session_id + )) ) - if not same_generation: + if not same_generation or report.report.stage != "final": final_outcome = "failed" lifecycle = "failed" failure_code = "ARTIFACT_INCOMPLETE" failure_detail = ( - "Run requires a final plan and report from the same generation" + "Run requires a final report from the same generation" ) else: actions = [ @@ -1680,9 +1742,19 @@ class OperationalTaskService: TaskAccessRequest.model_validate(item) for item in await self._list_access_requests(task_id) ] - lifecycle = self._status_from_plan( - plan.plan.nodes, actions, access_requests - ) + if task["task_type"] == "event_investigation": + lifecycle = self._investigation_status(actions) + elif report.report.outcome == "insufficient_data": + terminal = all(node.status in {"completed", "skipped", "cancelled"} + for node in plan.plan.nodes) + pending_action = any(action.approval_status == "pending" or ( + action.approval_status == "approved" and action.execution_status in {"not_submitted", "submitted", "unknown"} + ) for action in actions) + lifecycle = "completed" if terminal and not pending_action else "running" + else: + lifecycle = self._status_from_plan( + plan.plan.nodes, actions, () if task["task_type"] == "event_investigation" else access_requests + ) if lifecycle == "running": final_outcome = "failed" lifecycle = "failed" @@ -1810,6 +1882,16 @@ class OperationalTaskService: return "waiting_human" return "running" + @staticmethod + def _investigation_status(actions: Iterable[HumanApprovalAction]) -> str: + """Only recorded business actions can keep an investigation waiting.""" + return "waiting_human" if any( + action.approval_status == "pending" or ( + action.approval_status == "approved" + and action.execution_status in {"not_submitted", "submitted", "unknown"} + ) for action in actions + ) else "completed" + @staticmethod def _validate_plan_state(plan, known_evidence: set[str], actions: dict[str, HumanApprovalAction]) -> None: nodes = {node.id: node for node in plan.nodes} @@ -1963,6 +2045,7 @@ class OperationalTaskService: self, task_id: UUID, kind: str, title: str, description: str, now: datetime, *, required_permission: str | None = None, subject_type: str = "task", subject_id: UUID | None = None, + report_reference: InvestigationReportReference | None = None, ) -> None: notification = OperationalTaskNotification( notification_id=uuid4(), task_id=task_id, kind=kind, @@ -1970,6 +2053,7 @@ class OperationalTaskService: created_at=now, required_permission=required_permission, lifecycle_status="active", subject_type=subject_type, subject_id=subject_id or task_id, + report_reference=report_reference, ) await self.repo.insert_notification(_json(notification)) await self._append_event( diff --git a/contracts/manifest.json b/contracts/manifest.json index b956256..53f0536 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "0669dd2bec261c6ab9320df5eb506b85945829cb5582504483f89b482d826631" + "sha256": "c85b123b9d736a9558fce129cc8343441d50e9fef99532f2ad68c9241405b034" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 7928ebe..3d7a85a 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -779,6 +779,16 @@ "title": "Approved Data Scopes", "type": "array" }, + "selected_report": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationReportRevision" + }, + { + "type": "null" + } + ] + }, "task": { "$ref": "#/components/schemas/OperationalTaskDetail" } @@ -2852,8 +2862,142 @@ "title": "HumanEvidenceCommand", "type": "object" }, - "InvestigationReportInput": { + "InvestigationDataGap": { "additionalProperties": false, + "properties": { + "blocking": { + "title": "Blocking", + "type": "boolean" + }, + "evidence_refs": { + "items": { + "minLength": 1, + "type": "string" + }, + "minItems": 1, + "title": "Evidence Refs", + "type": "array" + }, + "gap_id": { + "minLength": 1, + "title": "Gap Id", + "type": "string" + }, + "impact": { + "minLength": 1, + "title": "Impact", + "type": "string" + }, + "missing_data": { + "minLength": 1, + "title": "Missing Data", + "type": "string" + }, + "node_id": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Node Id" + }, + "required_data": { + "minLength": 1, + "title": "Required Data", + "type": "string" + }, + "scope": { + "minLength": 1, + "title": "Scope", + "type": "string" + } + }, + "required": [ + "gap_id", + "missing_data", + "scope", + "impact", + "required_data", + "evidence_refs", + "blocking" + ], + "title": "InvestigationDataGap", + "type": "object" + }, + "InvestigationDiagram": { + "additionalProperties": false, + "properties": { + "mermaid": { + "maxLength": 20000, + "minLength": 1, + "title": "Mermaid", + "type": "string" + }, + "node_refs": { + "items": { + "$ref": "#/components/schemas/InvestigationDiagramNodeRef" + }, + "maxItems": 100, + "title": "Node Refs", + "type": "array" + }, + "title": { + "minLength": 1, + "title": "Title", + "type": "string" + } + }, + "required": [ + "title", + "mermaid" + ], + "title": "InvestigationDiagram", + "type": "object" + }, + "InvestigationDiagramNodeRef": { + "additionalProperties": false, + "properties": { + "action_refs": { + "items": { + "format": "uuid", + "type": "string" + }, + "title": "Action Refs", + "type": "array" + }, + "evidence_refs": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Evidence Refs", + "type": "array" + }, + "label": { + "minLength": 1, + "title": "Label", + "type": "string" + }, + "node_id": { + "pattern": "^[A-Za-z_][A-Za-z0-9_-]{0,79}$", + "title": "Node Id", + "type": "string" + } + }, + "required": [ + "node_id", + "label" + ], + "title": "InvestigationDiagramNodeRef", + "type": "object" + }, + "InvestigationReport": { + "additionalProperties": false, + "description": "Read model: absent historical facts remain unknown, never inferred.", "properties": { "conclusion_boundary": { "minLength": 1, @@ -2865,6 +3009,30 @@ "title": "Current Conclusion", "type": "string" }, + "data_gaps": { + "anyOf": [ + { + "items": { + "$ref": "#/components/schemas/InvestigationDataGap" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Data Gaps" + }, + "diagram": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationDiagram" + }, + { + "type": "null" + } + ] + }, "evidence_refs": { "items": { "minLength": 1, @@ -2873,6 +3041,21 @@ "title": "Evidence Refs", "type": "array" }, + "outcome": { + "anyOf": [ + { + "enum": [ + "concluded", + "insufficient_data" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Outcome" + }, "pending_items": { "items": { "minLength": 1, @@ -2889,6 +3072,33 @@ "title": "Recommendations", "type": "array" }, + "stage": { + "anyOf": [ + { + "enum": [ + "interim", + "final" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Stage" + }, + "stop_reason": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Stop Reason" + }, "title": { "minLength": 1, "title": "Title", @@ -2900,9 +3110,159 @@ "current_conclusion", "conclusion_boundary" ], + "title": "InvestigationReport", + "type": "object" + }, + "InvestigationReportInput": { + "additionalProperties": false, + "description": "Write model: every new report explicitly declares its business result.", + "properties": { + "conclusion_boundary": { + "minLength": 1, + "title": "Conclusion Boundary", + "type": "string" + }, + "current_conclusion": { + "minLength": 1, + "title": "Current Conclusion", + "type": "string" + }, + "data_gaps": { + "items": { + "$ref": "#/components/schemas/InvestigationDataGap" + }, + "title": "Data Gaps", + "type": "array" + }, + "diagram": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationDiagram" + }, + { + "type": "null" + } + ] + }, + "evidence_refs": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Evidence Refs", + "type": "array" + }, + "outcome": { + "enum": [ + "concluded", + "insufficient_data" + ], + "title": "Outcome", + "type": "string" + }, + "pending_items": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Pending Items", + "type": "array" + }, + "recommendations": { + "items": { + "minLength": 1, + "type": "string" + }, + "title": "Recommendations", + "type": "array" + }, + "stage": { + "enum": [ + "interim", + "final" + ], + "title": "Stage", + "type": "string" + }, + "stop_reason": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Stop Reason" + }, + "title": { + "minLength": 1, + "title": "Title", + "type": "string" + } + }, + "required": [ + "title", + "current_conclusion", + "conclusion_boundary", + "stage", + "outcome", + "data_gaps" + ], "title": "InvestigationReportInput", "type": "object" }, + "InvestigationReportPage": { + "additionalProperties": false, + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/InvestigationReportRevision" + }, + "title": "Items", + "type": "array" + }, + "next_offset": { + "anyOf": [ + { + "minimum": 0.0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Next Offset" + } + }, + "required": [ + "items" + ], + "title": "InvestigationReportPage", + "type": "object" + }, + "InvestigationReportReference": { + "additionalProperties": false, + "properties": { + "execution_generation": { + "minimum": 1.0, + "title": "Execution Generation", + "type": "integer" + }, + "revision": { + "minimum": 1.0, + "title": "Revision", + "type": "integer" + } + }, + "required": [ + "execution_generation", + "revision" + ], + "title": "InvestigationReportReference", + "type": "object" + }, "InvestigationReportRevision": { "additionalProperties": false, "properties": { @@ -2929,7 +3289,7 @@ "type": "string" }, "report": { - "$ref": "#/components/schemas/InvestigationReportInput" + "$ref": "#/components/schemas/InvestigationReport" }, "revision": { "minimum": 1.0, @@ -3792,6 +4152,16 @@ ], "title": "Read By" }, + "report_reference": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvestigationReportReference" + }, + { + "type": "null" + } + ] + }, "required_permission": { "anyOf": [ { @@ -25748,6 +26118,23 @@ "type": "string" } }, + { + "in": "query", + "name": "report_revision", + "required": false, + "schema": { + "anyOf": [ + { + "minimum": 1, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Report Revision" + } + }, { "in": "header", "name": "X-Project-Id", @@ -26324,6 +26711,252 @@ ] } }, + "/api/v1/operational-tasks/{task_id}/reports": { + "get": { + "operationId": "get_operational_tasks_task_id_reports", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Task Id", + "type": "string" + } + }, + { + "in": "query", + "name": "limit", + "required": false, + "schema": { + "default": 20, + "maximum": 100, + "minimum": 1, + "title": "Limit", + "type": "integer" + } + }, + { + "in": "query", + "name": "offset", + "required": false, + "schema": { + "default": 0, + "minimum": 0, + "title": "Offset", + "type": "integer" + } + }, + { + "in": "header", + "name": "X-Project-Id", + "required": true, + "schema": { + "title": "X-Project-Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvestigationReportPage" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Authentication required" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Insufficient permission" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Resource not found" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Resource conflict" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Validation error" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Dependency unavailable" + } + }, + "security": [ + { + "OAuth2PasswordBearer": [] + } + ], + "summary": "List Investigation Reports", + "tags": [ + "Operational Tasks" + ] + } + }, + "/api/v1/operational-tasks/{task_id}/reports/{revision}": { + "get": { + "operationId": "get_operational_tasks_task_id_reports_revision", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Task Id", + "type": "string" + } + }, + { + "in": "path", + "name": "revision", + "required": true, + "schema": { + "title": "Revision", + "type": "integer" + } + }, + { + "in": "header", + "name": "X-Project-Id", + "required": true, + "schema": { + "title": "X-Project-Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvestigationReportRevision" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Authentication required" + }, + "403": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Insufficient permission" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Resource not found" + }, + "409": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Resource conflict" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Validation error" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetails" + } + } + }, + "description": "Dependency unavailable" + } + }, + "security": [ + { + "OAuth2PasswordBearer": [] + } + ], + "summary": "Get Investigation Report", + "tags": [ + "Operational Tasks" + ] + } + }, "/api/v1/operational-tasks/{task_id}/retry": { "post": { "operationId": "post_operational_tasks_task_id_retry", diff --git a/docs/investigation-report-contract.md b/docs/investigation-report-contract.md new file mode 100644 index 0000000..e6fe151 --- /dev/null +++ b/docs/investigation-report-contract.md @@ -0,0 +1,162 @@ +# 调查报告数据契约 + +本契约适用于供水事件调查，由 Server 的 `InvestigationReportInput`（写入）和 +`InvestigationReport`（读取）定义。Agent 和前端采用同样的字段和约束。 +不增加格式版本标签；原字段的类型保持稳定。`revision` 是报告修订号， +`execution_generation` 是调查轮次，两者均不用于选择格式解析器。 + +## 数据模型 + +| 字段 | 类型 | 新写入规则 | +| --- | --- | --- | +| title | 非空字符串 | 报告标题 | +| current_conclusion | 非空字符串 | 当前判断文字；无结论时明确写无法判断，不编造原因 | +| conclusion_boundary | 非空字符串 | 判断范围及限制 | +| evidence_refs | 字符串数组 | 引用该任务中已持久化、不可变的证据编号 | +| recommendations | 字符串数组 | 保持原类型，处置建议不等于已执行动作 | +| pending_items | 字符串数组 | 保持原类型，不从文字推断状态 | +| stage | interim / final | 必填；阶段性报告不能用于结束运行 | +| outcome | concluded / insufficient_data | 必填；业务调查结果，与运行成功/失败独立 | +| stop_reason | 非空字符串或 null | 证据不足时必填；已形成结论时必须为 null | +| data_gaps | 缺口对象数组 | 必填；空数组表示已记录且没有缺口 | +| diagram | 调查脉络对象或 null | 可选；给人阅读的 Mermaid 说明，不控制执行或任务状态 | + +缺口对象除 node_id 外的字段必填： + +| 字段 | 类型 | 含义 | +| --- | --- | --- | +| gap_id | 非空字符串 | 报告内唯一的缺口编号 | +| node_id | 非空字符串或 null | 可选的展示关联，不要求有执行计划 | +| missing_data | 非空字符串 | 缺失数据内容 | +| scope | 非空字符串 | 查询对象、地点及时间范围 | +| impact | 非空字符串 | 因此无法回答的调查问题 | +| required_data | 非空字符串 | 新一轮调查需要补齐的具体数据 | +| evidence_refs | 至少一个证据编号 | 实际查询或可信输入记录；必须也出现在报告 evidence_refs 中 | +| blocking | boolean | 是否阻止形成结论 | + +禁止未知字段、未知枚举和重复 gap_id。各引用在 Server 中校验。 +查询失败、权限配置异常不能作为“没有数据”的证明。 + +调查脉络对象包含 `title`、`mermaid` 和可选的 `node_refs`（默认空数组）。 +`mermaid` 最长 20,000 字符，必须以 `flowchart` 或 `graph` 及方向开头， +不包含 Markdown 围栏、初始化指令和 click 命令。前端使用 strict 安全级别渲染 SVG。 +完整图语法的解析失败只影响图示，正文、依据和审核仍可阅读。 + +每条 node_ref 包含 `node_id`、`label`、`evidence_refs`、`action_refs`；后两项默认空数组。 +node_id 为最多 80 字符的 ASCII 标识符，必须以字母或下划线开头，在本图引用内唯一； +最多 100 条引用。证据必须来自报告 evidence_refs，动作必须属于当前任务。 +节点可以说明候选方向与条件关系；图中的分支、状态文字不能证明实际并发、动作执行或现场完成。 +图的颜色、字体和交互由前端统一配置。正文不包含 SVG 或 HTML。 + +## 业务约束 + +1. `concluded` 允许 interim/final，stop_reason 为 null，不允许 blocking 缺口。 +2. `insufficient_data` 只能是 final，至少一个 blocking 缺口，必须提供 stop_reason； + recommendations 和 pending_items 必须为空，不发起补数据的工单动作。 +3. 报告必须来自当前执行轮次、会话和有效租约；执行计划与图示均可缺省。 + 历史计划或未结束的解释节点不阻止发布与正常收尾。运行状态来自 Server run， + 审核和业务结果来自真实动作与回执，图示不参与计算。 +4. 尚有待审核、已批准未提交、已提交或结果未知的动作时，不能发布数据不足退出报告。 + 不自动撤销已有动作，不影响独立现场处置。 +5. 数据不足报告保存后，本执行代不再接受新的报告。补齐数据应使用已有 retry 接口 + 创建新 execution_generation，原报告、证据和回执保持可追溯。 +6. 事件调查提议动作前必须已保存本轮 outcome=concluded 的报告。 + 顺序为阶段性结论、动作提议、最终报告、运行收尾。图示可在报告中关联动作编号。 +7. 数据不足时顺序为查询证据、最终报告、运行收尾。 + 运行正常结束为 succeeded，调查生命周期为 completed；页面显示“已结束·证据不足”， + 不显示正常水质、不自动关闭原投诉工单，不等待人工授权或补数据。 +8. 数据读取范围由系统配置；本场景的人工审核只针对具体业务动作。 + 既有数据访问审核接口仅保留兼容其他调用，不作为事件调查的产品入口。 + +## 历史读取与兼容 + +旧报告缺少 stage/outcome/stop_reason/data_gaps 时，Server 统一返回 null。 +null 表示未记录，不能转换成 concluded、空缺口清单或正常水质。 +前端兼容滚动发布期间的缺省/null，未知字段仍严格拒绝；一旦提供 outcome， +其余结构化字段和业务约束必须完整满足。新写入不接受全空的历史格式。 + +历史导入脚本的 dry-run 继续校验原始记录；缺少结构化结果的历史报告不能直接发布为新调查报告，实写在创建任务前拒绝，避免按旧正文编造结果。 + +无需改写历史报告或新增数据库表，新增字段保存在原报告 document 中。 +上线顺序：先部署能读取可选新字段的前端，再协调更新 Server 新写入契约与 Agent。 +在切换写入端前应结束旧 Agent 在途运行，旧写入格式会被明确拒绝，不能静默推断结果。 +回退前须保留支持新字段的读取端；不能直接回退到会拒绝新字段的旧前端。 + +## 接口与错误 + +所有路径均以 `/api/v1` 开头，保留现有 Bearer 身份与项目范围校验。 + +| 方法与路径 | 用途 | +| --- | --- | +| GET /operational-tasks/{task_id} | 读取统一的报告模型，旧结构化字段返回 null | +| GET /operational-tasks/{task_id}/reports?limit=20&offset=0 | 按修订号倒序读取历史，返回 items 与 next_offset | +| GET /operational-tasks/{task_id}/reports/{revision} | 读取指定报告快照，任务及项目范围校验 | +| GET /operational-tasks/{task_id}/agent-context?report_revision=1 | selected_report 是所选报告，task 保留当前真实状态；未传则 selected_report 为 null | +| POST /operational-tasks/{task_id}/agent-artifacts | 写入报告和计划；保留 request_id 幂等、run_id 与会话归属校验 | +| POST /operational-tasks/{task_id}/runs/{run_id}/complete | 事件调查校验本轮最终报告后结束运行；检查 expected_revision | +| POST /operational-tasks/{task_id}/actions | 仅本轮已保存 concluded 报告的事件调查允许提议动作 | +| POST /operational-tasks/{task_id}/retry | 创建新调查轮次，不覆盖上一轮报告 | + +| 错误码 | HTTP | 含义 | +| --- | --- | --- | +| 请求体校验错误 | 422 | 新报告缺字段、未知枚举、无关键缺口或相互矛盾的字段 | +| REPORT_GAP_EVIDENCE_INVALID | 422 | 缺口依据不在报告的证据集合内 | +| REPORT_NOT_FOUND | 404 | 指定任务中的报告修订不存在 | +| REPORT_ACTION_PENDING | 409 | 缺数退出仍有关联的未决动作 | +| INVESTIGATION_ENDED | 409 | 本轮已经发布缺数退出报告，再写成果需要新轮次 | +| ACTION_CONCLUSION_REQUIRED | 409 | 没有本轮已形成结论的报告，不得生成动作 | + +complete 接口保留已有返回方式：不完整成果会把 run 标为 failed， +failure_code 为 ARTIFACT_INCOMPLETE；调用方须读取返回的业务状态，不能只看 HTTP 200。 + +## 缺数报告示例 + +以下为 agent-artifacts 请求中 report 字段的完整示例。 +`sample-query-empty` 必须是同一任务内已持久化的空查询证据；示例本身不构成真实查询事实。 + +```json +{ + "title": "水质投诉调查报告", + "current_conclusion": "缺少投诉时段的末梢采样，无法确定异常来源。", + "conclusion_boundary": "已核对可用记录，本轮未确认水质异常原因。", + "evidence_refs": ["sample-query-empty"], + "recommendations": [], + "pending_items": [], + "stage": "final", + "outcome": "insufficient_data", + "stop_reason": "关键采样数据缺失，本轮调查结束。", + "data_gaps": [{ + "gap_id": "terminal-sample-gap", + "node_id": null, + "missing_data": "投诉时段末梢水样检测结果", + "scope": "投诉小区，2026-09-24 08:00 至 09:00（北京时间）", + "impact": "无法判断异常来自供水管网还是用户端。", + "required_data": "该地点和时段的采样检测记录", + "evidence_refs": ["sample-query-empty"], + "blocking": true + }] +} +``` + +## 验收 + +- 新写入缺字段、不一致结果、无证据的关键缺口、未知业务证据或动作引用均被拒绝。 +- 关键数据不足保存报告并正常结束；无运行失败、无待人工通知、无后续动作。 +- 阶段性报告不能冒充最终报告；旧代结论不能授权新代动作。 +- 已有未决动作阻止矛盾的缺数退出；重试幂等，不重复生成报告。 +- 新轮次保留旧报告历史标识，不能显示成当前轮次已结束。 +- 桌面及 320/375px 展示结果、缺口、证据，审批区不出现数据授权入口。 + +## 发布、阅读与协作 + +每次发布保留 `operational.report.published` 同步事件。首次 final、阶段报告转 final、 +新轮次 final 或判断/边界/依据/建议/缺口实质变化时，生成 `report_updated` 通知， +其中 `report_reference = {execution_generation, revision}`。纯图示修订不生成额外成果提醒。 +新提醒取代同任务之前的成果提醒；读取相应报告后按用户确认已读。 +前端在调查列表显示“新报告”，这类提醒不进入“待办与告警”。 + +正在阅读的报告和图示固定在所选修订，收到更新后显示“查看新报告”，不自动替换正文。 +当前待审核动作单独读取最新真实状态，即使用户正在阅读旧报告也不会审批旧版本。 +协作请求 `task_context` 可携带 report_revision、evidence_id、gap_id；轮次和报告修订号由 +Agent 服务根据 Server 返回的快照生成 `data-task_context`，随回答流和历史消息持久化。 +该引用说明回答依据，不授予访问权限，也不成为执行约束。 diff --git a/scripts/import_agent_investigation.py b/scripts/import_agent_investigation.py index b7d530e..b1f0adc 100644 --- a/scripts/import_agent_investigation.py +++ b/scripts/import_agent_investigation.py @@ -15,7 +15,13 @@ import httpx sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from app.domain.schemas.operational_task import AgentArtifactsCommand +from app.domain.schemas.operational_task import AgentArtifactsCommand, InvestigationReport + + +class HistoricalArtifacts(AgentArtifactsCommand): + """Validate history for inspection without inventing a current outcome.""" + + report: InvestigationReport try: from scripts.run_operational_worker import CredentialProvider, WorkerSettings @@ -65,13 +71,17 @@ async def run(args): plan, evidence, report = convert_plan(snapshot) # Validate the complete payload before reporting readiness or creating a # backend task. A dry run must catch incompatible historical plan states. - artifacts = AgentArtifactsCommand.model_validate({ + artifacts = HistoricalArtifacts.model_validate({ "request_id": str(uuid5(NAMESPACE_URL, f"legacy-artifacts:{metadata.get('projectId')}:{args.session_id}:{snapshot['revision']}")), "source_session_id": args.session_id, "plan": plan, "evidence": evidence, "report": report, }).model_dump(mode="json", by_alias=True) if args.dry_run: - print(f"Ready to import {len(plan['nodes'])} nodes and {len(evidence)} synthetic evidence records; no writes") + print(f"Validated {len(plan['nodes'])} nodes and {len(evidence)} synthetic evidence records; no writes. " + "Historical outcomes are unrecorded; automatic publication as a new report is unavailable.") return + # A historical read is not a new investigation result. Reject before task + # creation, rather than inventing concluded/insufficient_data from old prose. + artifacts = AgentArtifactsCommand.model_validate(artifacts).model_dump(mode="json", by_alias=True) settings = WorkerSettings.from_environment() if metadata.get("projectId") != settings.project_id: raise ValueError("session does not belong to the worker project") diff --git a/tests/api/test_operational_task_dynamic_flow.py b/tests/api/test_operational_task_dynamic_flow.py index 300a269..4757bff 100644 --- a/tests/api/test_operational_task_dynamic_flow.py +++ b/tests/api/test_operational_task_dynamic_flow.py @@ -29,6 +29,10 @@ def test_dynamic_investigation_routes_are_in_the_public_contract(): assert "post" in paths["/api/v1/operational-tasks/{task_id}/access-requests/{access_request_id}/decision"] assert "patch" in paths["/api/v1/operational-tasks/{task_id}/actions/{action_id}/draft"] assert "post" in paths["/api/v1/operational-tasks/{task_id}/actions/{action_id}/confirm-dispatch"] + assert "get" in paths["/api/v1/operational-tasks/{task_id}/reports"] + assert "get" in paths["/api/v1/operational-tasks/{task_id}/reports/{revision}"] + context_params = paths["/api/v1/operational-tasks/{task_id}/agent-context"]["get"]["parameters"] + assert any(item["name"] == "report_revision" and item["in"] == "query" for item in context_params) @pytest.mark.anyio diff --git a/tests/integration/test_operational_tasks_storage.py b/tests/integration/test_operational_tasks_storage.py index 9781c27..25c19b4 100644 --- a/tests/integration/test_operational_tasks_storage.py +++ b/tests/integration/test_operational_tasks_storage.py @@ -104,7 +104,7 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( {"id": "sample", "title": "现场采样", "owner": "human", "status": "waiting", "detail": "等待外部反馈", "evidence_refs": []}], "edges": [{"from": "review", "to": "sample"}]}, - "report": {"title": "调查阶段结论", "current_conclusion": "待现场核验", "conclusion_boundary": "合成输入", + "report": {"stage": "final", "outcome": "concluded", "stop_reason": None, "data_gaps": [], "title": "调查阶段结论", "current_conclusion": "待现场核验", "conclusion_boundary": "合成输入", "evidence_refs": ["synthetic-1"], "pending_items": ["现场采样"]}, }) published = await service.publish_artifacts(task.task_id, artifact) @@ -120,7 +120,7 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( source_session_id="test-investigation-session", ), ) - assert task.status == "waiting_human" + assert task.status == "completed" action = await service.propose_action(task.task_id, ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场采样", evidence_refs=["synthetic-1"], payload={"title": "采样", "description": "合成核验", "work_order_type": "sampling"}, @@ -135,13 +135,18 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( item for item in member_notifications if item.kind == "action_approval_pending" ) - human_input = next( + report_notice = next( item for item in member_notifications - if item.kind == "human_input_required" + if item.kind == "report_updated" ) assert not any(item.kind in { - "task_created", "report_updated", "task_completed" + "task_created", "human_input_required", "task_completed" } for item in member_notifications) + assert report_notice.report_reference.revision == 1 + assert report_notice.report_reference.execution_generation == 1 + history, next_offset = await service.list_reports(task.task_id, 20, 0) + assert next_offset is None and history[0].revision == 1 + assert (await service.report(task.task_id, 1)).report.model_dump() == artifact.report.model_dump() assert action_pending.required_permission == action.required_approval_permission viewer_context = ProjectContext( @@ -158,7 +163,7 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( viewer_notifications, _ = await viewer_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert human_input.notification_id in { + assert report_notice.notification_id in { item.notification_id for item in viewer_notifications } assert action_pending.notification_id not in { @@ -166,9 +171,9 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( } acknowledged = await service.acknowledge_notification( - human_input.notification_id, + report_notice.notification_id, NotificationAcknowledgeCommand( - request_id=uuid4(), expected_version=human_input.version + request_id=uuid4(), expected_version=report_notice.version ), ) assert acknowledged.status == "read" @@ -184,21 +189,21 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( ) persisted_created = next( item for item in persisted_notifications - if item.notification_id == human_input.notification_id + if item.notification_id == report_notice.notification_id ) assert persisted_created.status == "read" - assert persisted_created.version == human_input.version + 1 + assert persisted_created.version == report_notice.version + 1 assert persisted_created.read_by == f"principal:{context.user_id}" unread_notifications, _ = await persisted_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert human_input.notification_id not in { + assert report_notice.notification_id not in { item.notification_id for item in unread_notifications } viewer_after_member_ack, _ = await viewer_service.list_notifications( unread_only=True, limit=100, offset=0 ) - assert human_input.notification_id in { + assert report_notice.notification_id in { item.notification_id for item in viewer_after_member_ack } action = await service.decide_action(task.task_id, action.action_id, ActionDecisionCommand( @@ -259,6 +264,10 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( context, permissions_for_context(context)) with pytest.raises(DomainError): await outsider.detail(task.task_id) + await fresh_service.publish_artifacts(task.task_id, artifact.model_copy(update={ + "request_id": uuid4(), "run_id": resumed.active_run.run_id, + "source_session_id": "test-investigation-followup-session", + })) sampling = await fresh_service.propose_action(task.task_id, ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要补充采样", evidence_refs=["synthetic-1"], @@ -307,5 +316,35 @@ async def test_postgres_task_artifacts_approval_and_project_isolation_roll_back( dispatcher, "Bearer TEST_ONLY") assert calls == ["POST", "GET", "GET"] assert (await fresh_service.detail(task.task_id)).status == "queued" + # A separate investigation ends normally on an evidenced empty query. + missing = await service.create(command.model_copy(update={ + "request_id": uuid4(), "title": "AUTOTEST 缺数退出", + })) + missing_run = await service.claim_run(missing.task_id, missing.active_run.run_id, AgentRunCommand( + request_id=missing.active_run.execution_request_id, + expected_revision=missing.active_run.revision, source_session_id="missing-data-session")) + exit_artifact = AgentArtifactsCommand( + request_id=uuid4(), source_session_id="missing-data-session", run_id=missing_run.run_id, + evidence=[{"evidence_id": "empty-sample", "kind": "synthetic", "summary": "查询成功，无采样记录", + "source_label": "合成空查询", "observed_at": now.isoformat()}], + plan={"schema": "execution-plan@1", "title": "调查", "reason": "关键数据不足", + "nodes": [{"id": "query", "title": "核对采样", "owner": "agent", "status": "completed", + "detail": "查询范围内无记录", "evidence_refs": ["empty-sample"]}]}, + report={"title": "无法形成结论", "current_conclusion": "无法确定投诉原因", + "conclusion_boundary": "缺少末梢采样", "evidence_refs": ["empty-sample"], + "stage": "final", "outcome": "insufficient_data", "stop_reason": "关键采样数据为空", + "data_gaps": [{"gap_id": "sample", "node_id": "query", "missing_data": "末梢采样", + "scope": "投诉时段", "impact": "无法判断异常来源", "required_data": "末梢采样结果", + "evidence_refs": ["empty-sample"], "blocking": True}]}) + exited = await service.publish_artifacts(missing.task_id, exit_artifact) + await service.complete_run(missing.task_id, missing_run.run_id, AgentRunCommand( + request_id=uuid4(), expected_revision=exited.task.active_run.revision, + source_session_id="missing-data-session")) + restored = await fresh_service.detail(missing.task_id) + assert restored.lifecycle_status == "completed" + assert restored.active_run is None + assert restored.latest_report.report.outcome == "insufficient_data" + assert restored.latest_report.report.data_gaps[0].blocking is True + assert not restored.actions finally: await conn.close() diff --git a/tests/unit/test_import_agent_investigation.py b/tests/unit/test_import_agent_investigation.py index d96bcc2..ae125be 100644 --- a/tests/unit/test_import_agent_investigation.py +++ b/tests/unit/test_import_agent_investigation.py @@ -53,8 +53,12 @@ async def test_dry_run_validates_historical_schema_before_any_network_access(tmp args = argparse.Namespace(session_id="ses_test", agent_data=tmp_path, dry_run=True) await run(args) assert "2 nodes and 1 synthetic evidence records; no writes" in capsys.readouterr().out + args.dry_run = False + with pytest.raises(ValidationError): + await run(args) + args.dry_run = True state["nodes"][1]["status"] = "invented_status" state_path.write_text(json.dumps({"executionPlans": [state]})) with pytest.raises(ValidationError): await run(args) - assert "Ready" not in capsys.readouterr().out + assert "Validated" not in capsys.readouterr().out diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index 678de6e..d5bb25b 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -297,6 +297,12 @@ class TaskRepository: async def get_latest_report(self, _task_id): return self.reports[-1] if self.reports else None + async def get_report(self, _task_id, revision): + return next((item for item in self.reports if item["revision"] == revision), None) + + async def list_reports(self, _task_id, limit, offset): + return list(reversed(self.reports))[offset:offset + limit] + async def insert_plan(self, _task_id, snapshot): self.plans.append(snapshot) @@ -504,6 +510,7 @@ def _waiting_plan(action_id=None): def _current_report(): return { + "stage": "final", "outcome": "concluded", "stop_reason": None, "data_gaps": [], "title": "本代调查报告", "current_conclusion": "等待现场处置", "conclusion_boundary": "尚无现场回执", @@ -513,6 +520,125 @@ def _current_report(): } +def _insufficient_report(): + return { + **_current_report(), "outcome": "insufficient_data", + "current_conclusion": "无法确定投诉原因", "stop_reason": "缺少投诉时段的末梢采样", + "recommendations": [], "pending_items": [], "evidence_refs": ["empty-query"], + "data_gaps": [{"gap_id": "sample-gap", "node_id": "read", "missing_data": "末梢采样", + "scope": "投诉小区，2026-09-23 08:00 至 09:00", "impact": "无法判断异常来源", + "required_data": "补齐该时段采样结果", "evidence_refs": ["empty-query"], "blocking": True}], + } + + +def test_report_contract_separates_legacy_reads_from_new_writes(): + from app.domain.schemas.operational_task import InvestigationReport, InvestigationReportInput + legacy = {k: v for k, v in _current_report().items() + if k not in {"stage", "outcome", "stop_reason", "data_gaps"}} + stored = InvestigationReport.model_validate(legacy) + assert stored.outcome is None and stored.data_gaps is None + with pytest.raises(ValueError): + InvestigationReportInput.model_validate(legacy) + assert InvestigationReportInput.model_validate(_insufficient_report()).outcome == "insufficient_data" + for changes in [{"stage": "interim"}, {"stop_reason": None}, {"data_gaps": []}, + {"recommendations": ["自动派采样工单"]}, {"pending_items": ["等待人工"]}, + {"outcome": "concluded"}, {"outcome": "other"}]: + with pytest.raises(ValueError): + InvestigationReportInput.model_validate({**_insufficient_report(), **changes}) + + +@pytest.mark.anyio +async def test_insufficient_data_exits_successfully_and_blocks_actions_until_new_run(): + now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + command = AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", + evidence=[{"evidence_id": "empty-query", "kind": "synthetic", "summary": "查询成功，该范围无采样记录", + "source_label": "合成查询", "observed_at": now.isoformat()}], + plan={"schema": "execution-plan@1", "title": "调查", "reason": "关键数据不足", + "nodes": [{"id": "read", "title": "查询采样", "owner": "agent", "status": "completed", + "detail": "已核对查询范围，无记录", "evidence_refs": ["empty-query"]}, + {"id": "action", "title": "处置建议", "owner": "agent", "status": "skipped", + "detail": "缺少关键数据，未进入"}], + "edges": [{"from": "read", "to": "action"}]}, + report=_insufficient_report(), + ) + published = await service.publish_artifacts(task["task_id"], command) + replay = await service.publish_artifacts(task["task_id"], command) + assert published.report_revision == replay.report_revision == 1 + with pytest.raises(DomainError) as ended: + await service.publish_artifacts(task["task_id"], command.model_copy(update={"request_id": uuid4()})) + assert ended.value.code == "INVESTIGATION_ENDED" + with pytest.raises(DomainError) as exc: + await service.propose_action(task["task_id"], ActionProposalCommand( + request_id=uuid4(), kind="create_work_order", reason="缺数后不应派单", + payload={"title": "采样", "description": "采样", "work_order_type": "sampling"}, + )) + assert exc.value.code == "ACTION_CONCLUSION_REQUIRED" + completed = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1")) + assert completed.lifecycle_status == "completed" + assert repository.run["status"] == "succeeded" + assert completed.latest_report.report.outcome == "insufficient_data" + assert repository.action is None + assert not any(item["kind"] in {"task_failed", "human_input_required", "action_approval_pending"} + for item in repository.notifications) + from app.domain.schemas.operational_task import OperationalTaskRetryCommand + restarted = await service.retry_task(task["task_id"], OperationalTaskRetryCommand( + request_id=uuid4(), reason="已补充新的采样记录")) + assert restarted.execution_generation == 2 + assert restarted.latest_report.execution_generation == 1 + assert restarted.latest_report.report.outcome == "insufficient_data" + with pytest.raises(DomainError) as stale: + await service.propose_action(task["task_id"], ActionProposalCommand( + request_id=uuid4(), kind="create_work_order", reason="新轮次尚未形成结论", + payload={"title": "采样", "description": "采样", "work_order_type": "sampling"})) + assert stale.value.code == "ACTION_CONCLUSION_REQUIRED" + + +@pytest.mark.anyio +@pytest.mark.parametrize("defect,expected", [ + ("unlisted_evidence", "REPORT_GAP_EVIDENCE_INVALID"), ("pending_action", "REPORT_ACTION_PENDING"), +]) +async def test_insufficient_report_rejects_inconsistent_context(defect, expected): + now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + report = _insufficient_report() + node = {"id": "read", "title": "查询", "owner": "agent", "status": "skipped", "detail": "关键数据不足"} + if defect == "unknown_node": report["data_gaps"][0]["node_id"] = "unknown" + if defect == "unlisted_evidence": report["data_gaps"][0]["evidence_refs"] = ["unlisted"] + if defect == "waiting_node": node["status"] = "waiting" + if defect == "pending_action": + _seed_concluded_report(repository, now) + await service.propose_action(task["task_id"], ActionProposalCommand( + request_id=uuid4(), kind="create_work_order", reason="已产生待审动作", + payload={"title": "采样", "description": "采样", "work_order_type": "sampling"})) + with pytest.raises(DomainError) as exc: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", + evidence=[{"evidence_id": "empty-query", "kind": "synthetic", "summary": "查询空结果", + "source_label": "合成查询", "observed_at": now.isoformat()}], + plan={"schema": "execution-plan@1", "title": "调查", "reason": "无数据", "nodes": [node]}, + report=report)) + assert exc.value.code == expected + + +def _seed_concluded_report(repository, now): + if repository.reports and repository.reports[-1].get("execution_generation") == repository.task["execution_generation"]: + return + repository.reports.append({ + "revision": 1, "execution_generation": repository.task["execution_generation"], + "source_session_id": "session-1", "published_at": now.isoformat(), + "published_by": "agent", "report": _current_report(), + }) + + @pytest.mark.anyio async def test_plan_only_keeps_lease_until_same_generation_report_closes_run(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) @@ -541,15 +667,40 @@ async def test_plan_only_keeps_lease_until_same_generation_report_closes_run(): source_session_id="session-1", ), ) - assert completed.status == "waiting_human" + assert completed.status == "completed" assert completed.agent_lease_expires_at is None assert [item["kind"] for item in repository.notifications] == [ - "human_input_required" + "report_updated" ] @pytest.mark.anyio -async def test_report_does_not_close_from_legacy_or_prior_generation_plan(): +async def test_interim_report_cannot_finish_even_with_a_completed_plan(): + now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + plan = _waiting_plan() + plan["nodes"][0]["status"] = "completed" + plan["nodes"][0]["evidence_refs"] = ["field-result"] + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", plan=plan, + evidence=[{"evidence_id": "field-result", "kind": "synthetic", "summary": "模拟现场结果", + "source_label": "测试采样", "observed_at": now.isoformat()}], + report={**_current_report(), "stage": "interim"}, + )) + + completed = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1", + )) + + assert completed.lifecycle_status == "failed" + assert repository.run["failure_code"] == "ARTIFACT_INCOMPLETE" + + +@pytest.mark.anyio +async def test_report_is_not_blocked_by_a_historical_plan(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) @@ -568,9 +719,109 @@ async def test_report_does_not_close_from_legacy_or_prior_generation_plan(): result = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", report=_current_report(), )) - - assert result.task.status == "running" assert result.task.latest_report.execution_generation == 2 + assert result.task.status == "completed" + + +@pytest.mark.anyio +async def test_report_and_action_without_plan_finish_with_real_approval_pending(): + now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=_current_report(), + )) + action = await service.propose_action(task["task_id"], ActionProposalCommand( + request_id=uuid4(), kind="create_work_order", node_id="display-only", + reason="已形成判断，需要现场核查", payload={"title": "现场核查", "description": "核验", "work_order_type": "general"}, + )) + completed = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1", + )) + assert completed.latest_plan is None + assert completed.lifecycle_status == "waiting_human" + assert completed.actions[0].action_id == action.action_id + assert repository.run["status"] == "succeeded" + + +@pytest.mark.anyio +async def test_insufficient_data_without_graph_or_node_ends_the_round(): + now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + report = _insufficient_report() + report["data_gaps"][0].pop("node_id") + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=report, + evidence=[{"evidence_id": "empty-query", "kind": "synthetic", "summary": "成功查询，范围内没有记录", + "source_label": "测试输入", "observed_at": now.isoformat()}], + )) + completed = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1", + )) + assert completed.lifecycle_status == "completed" + assert completed.latest_report.report.data_gaps[0].node_id is None + assert repository.run["status"] == "succeeded" + + +@pytest.mark.anyio +async def test_report_history_and_notifications_keep_selected_and_current_references_separate(): + now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + for conclusion in ["第一次判断", "更新后的判断"]: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", + report={**_current_report(), "current_conclusion": conclusion}, + )) + page, offset = await service.list_reports(task["task_id"], 1, 0) + assert [item.revision for item in page] == [2] and offset == 1 + page, offset = await service.list_reports(task["task_id"], 1, offset) + assert [item.revision for item in page] == [1] and offset is None + selected = await service.agent_context(task["task_id"], report_revision=1) + assert selected.selected_report.revision == 1 + assert selected.task.latest_report.revision == 2 + assert selected.selected_report.report.current_conclusion == "第一次判断" + assert repository.notifications[0]["lifecycle_status"] == "resolved" + assert repository.notifications[-1]["report_reference"] == {"execution_generation": 1, "revision": 2} + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", + report={**_current_report(), "current_conclusion": "更新后的判断", + "diagram": {"title": "说明", "mermaid": 'flowchart TD\nA["核查"] --> B["判断"]'}}, + )) + assert len(repository.notifications) == 2 + with pytest.raises(DomainError) as missing: + await service.report(task["task_id"], 999) + assert missing.value.code == "REPORT_NOT_FOUND" + with pytest.raises(DomainError) as other_task: + await service.list_reports(uuid4(), 20, 0) + assert other_task.value.status == 404 + + +@pytest.mark.anyio +@pytest.mark.parametrize("node_ref,code", [ + ({"node_id": "A", "label": "依据", "evidence_refs": ["forged"]}, "EVIDENCE_NOT_FOUND"), + ({"node_id": "A", "label": "动作", "action_refs": [str(uuid4())]}, "ACTION_NOT_FOUND"), +]) +async def test_optional_diagram_cannot_forge_business_references(node_ref, code): + now = datetime(2026, 10, 9, 8, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + with pytest.raises(DomainError) as invalid: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", + report={**_current_report(), "diagram": {"title": "调查脉络", "mermaid": 'flowchart TD\nA["核查"]', "node_refs": [node_ref]}}, + )) + assert invalid.value.code == code + assert repository.reports == [] @pytest.mark.anyio @@ -586,6 +837,7 @@ async def test_action_proposal_keeps_running_lease_until_plan_and_report_close_r revision_before_proposal = repository.run["revision"] lease_before_proposal = repository.run["lease_expires_at"] + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", node_id="field-work", evidence_refs=[], @@ -633,6 +885,7 @@ async def test_rejected_action_requeues_task_in_a_new_execution_generation(): {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[], @@ -681,6 +934,7 @@ async def test_action_decision_rejects_a_stale_expected_version(): {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[], @@ -847,6 +1101,7 @@ async def test_work_order_draft_edit_creates_new_review_version(): task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now) + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="现场采样", evidence_refs=[], payload={"title": "旧草稿", "description": "采样", "work_order_type": "general"}, @@ -874,6 +1129,7 @@ async def test_one_click_work_order_confirmation_replays_without_duplicate_dispa service = OperationalTaskService(repository, context, { OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, }, clock=lambda: now) + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="现场采样", evidence_refs=[], payload={"title": "采样", "description": "现场核验", "work_order_type": "general"}, @@ -914,6 +1170,7 @@ async def test_new_payload_creates_a_new_pending_action_without_inheriting_appro {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) + _seed_concluded_report(repository, now) original = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要首次核验", evidence_refs=[], @@ -927,6 +1184,7 @@ async def test_new_payload_creates_a_new_pending_action_without_inheriting_appro ), ) + _seed_concluded_report(repository, now) revised = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="内容已修改", evidence_refs=[], @@ -1057,6 +1315,7 @@ async def test_withdrawn_external_task_rejects_action_changes_and_resolves_appro {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) + _seed_concluded_report(repository, now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[],