From accad4bdfc13b03a06a7e094d113965ff417e194 Mon Sep 17 00:00:00 2001 From: tianM <46552064+a3384379@users.noreply.github.com> Date: Sat, 3 Oct 2026 09:23:56 +0800 Subject: [PATCH 01/16] fix(reporting): preserve assertions and complex execution evidence --- backend/app/repositories/reporting.py | 22 +- backend/app/schemas/reporting.py | 4 +- backend/app/services/reporting.py | 118 +++++-- backend/tests/test_reporting_api.py | 171 ++++++++++ .../evidence/AssertionEvidence.test.tsx | 16 + .../features/evidence/AssertionEvidence.tsx | 30 +- .../reports/ReportExecutionWorkspace.tsx | 74 ++++- .../src/features/reports/report-service.ts | 5 +- frontend/src/features/reports/use-reports.ts | 15 +- .../workflows/ExecutionCheckpointLog.test.tsx | 2 +- .../workflows/ExecutionCheckpointLog.tsx | 6 +- .../workflows/execution-navigation.test.ts | 35 ++ .../workflows/execution-navigation.ts | 45 +++ frontend/src/flow/WorkflowDesigner.tsx | 13 + .../src/flow/WorkflowRunInspector.test.tsx | 93 +++++- frontend/src/flow/WorkflowRunInspector.tsx | 301 ++++++++++++++++-- frontend/src/lib/api.ts | 2 + frontend/src/pages/ReportsPage.test.tsx | 153 ++++++++- frontend/src/pages/ReportsPage.tsx | 42 ++- frontend/src/pages/WorkflowsPage.tsx | 20 +- 20 files changed, 1073 insertions(+), 94 deletions(-) diff --git a/backend/app/repositories/reporting.py b/backend/app/repositories/reporting.py index 067c7e44..93702f7e 100644 --- a/backend/app/repositories/reporting.py +++ b/backend/app/repositories/reporting.py @@ -5,7 +5,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.models.reporting import NotificationDelivery, NotificationWebhook -from app.models.workflows import Workflow, WorkflowExecution, WorkflowNodeExecution +from app.models.workflows import ( + Workflow, + WorkflowControlRecord, + WorkflowExecution, + WorkflowNodeExecution, +) class ReportingRepository: @@ -88,6 +93,21 @@ async def list_children(self, execution_id: UUID) -> list[WorkflowExecution]: ).all() ) + async def list_control_records(self, execution_id: UUID) -> list[WorkflowControlRecord]: + return list( + ( + await self._session.scalars( + select(WorkflowControlRecord) + .where(WorkflowControlRecord.workflow_execution_id == execution_id) + .order_by( + WorkflowControlRecord.node_id, + WorkflowControlRecord.kind, + WorkflowControlRecord.ordinal, + ) + ) + ).all() + ) + async def get_webhook(self, webhook_id: UUID) -> NotificationWebhook | None: return await self._session.get(NotificationWebhook, webhook_id) diff --git a/backend/app/schemas/reporting.py b/backend/app/schemas/reporting.py index 13af552b..3af60e39 100644 --- a/backend/app/schemas/reporting.py +++ b/backend/app/schemas/reporting.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, ConfigDict, Field, HttpUrl, JsonValue from app.domain.reporting import FailureCategory, NotificationEvent -from app.engine.results import NodeObservation +from app.engine.results import NodeObservation, NodeResult WebhookName = Annotated[str, Field(min_length=1, max_length=160)] @@ -40,6 +40,8 @@ class ReportNodeResponse(BaseModel): extraction: JsonValue = None assertion: JsonValue = None input_mappings: JsonValue = None + output: JsonValue = None + result: NodeResult | None = None error_code: str | None error_message: str | None diff --git a/backend/app/services/reporting.py b/backend/app/services/reporting.py index 57e155cc..0926721f 100644 --- a/backend/app/services/reporting.py +++ b/backend/app/services/reporting.py @@ -1,12 +1,12 @@ import html import json from collections import Counter, defaultdict -from dataclasses import dataclass +from dataclasses import asdict, dataclass from datetime import UTC, date, datetime, timedelta from typing import cast from uuid import UUID -from pydantic import JsonValue +from pydantic import BaseModel, JsonValue from sqlalchemy.ext.asyncio import AsyncSession from app.core.errors import AppError @@ -15,7 +15,9 @@ from app.models.access import User from app.models.artifacts import Artifact from app.models.workflows import WorkflowExecution, WorkflowNodeExecution +from app.repositories.durable_execution import DurableExecutionRepository from app.repositories.reporting import ReportingRepository +from app.schemas.durable_execution import ExecutionCheckpointResponse from app.services.artifacts import ArtifactService from app.services.audit import AuditService from app.services.projects import ProjectService @@ -55,6 +57,8 @@ class NodeReport: input_mappings: JsonValue error_code: str | None error_message: str | None + output: JsonValue = None + result: NodeResult | None = None @dataclass(frozen=True, slots=True) @@ -65,6 +69,23 @@ class ExecutionReportDetail: dataset_children: list[ExecutionReportSummary] +@dataclass(frozen=True, slots=True) +class ExportControlRecord: + node_id: str + kind: str + ordinal: int + status: str + test_verdict: str + payload: JsonValue + + +@dataclass(frozen=True, slots=True) +class ExecutionExportDetail(ExecutionReportDetail): + snapshot: dict[str, JsonValue] + control_records: list[ExportControlRecord] + checkpoints: list[ExecutionCheckpointResponse] + + @dataclass(frozen=True, slots=True) class TrendPoint: date: date @@ -86,6 +107,7 @@ class ReportService: def __init__(self, session: AsyncSession) -> None: self._session = session self._reports = ReportingRepository(session) + self._durable = DurableExecutionRepository(session) self._projects = ProjectService(session) self._artifacts = ArtifactService(session) self._audit = AuditService(session) @@ -143,7 +165,7 @@ async def trend(self, *, actor: User, project_id: UUID, days: int) -> ReportTren ) async def export_html(self, *, actor: User, project_id: UUID, execution_id: UUID) -> Artifact: - detail = await self.get_execution( + detail = await self._export_detail( actor=actor, project_id=project_id, execution_id=execution_id, @@ -167,6 +189,35 @@ async def export_html(self, *, actor: User, project_id: UUID, execution_id: UUID await self._session.refresh(artifact) return artifact + async def _export_detail( + self, *, actor: User, project_id: UUID, execution_id: UUID + ) -> ExecutionExportDetail: + detail = await self.get_execution( + actor=actor, project_id=project_id, execution_id=execution_id + ) + execution = await self._execution(project_id, execution_id) + records = await self._reports.list_control_records(execution_id) + checkpoints = await self._durable.list_checkpoints(execution_id) + return ExecutionExportDetail( + summary=detail.summary, + nodes=detail.nodes, + context=detail.context, + dataset_children=detail.dataset_children, + snapshot=cast(dict[str, JsonValue], execution.snapshot), + control_records=[ + ExportControlRecord( + node_id=item.node_id, + kind=item.kind, + ordinal=item.ordinal, + status=item.status, + test_verdict=item.test_verdict, + payload=cast(JsonValue, item.payload), + ) + for item in records + ], + checkpoints=[ExecutionCheckpointResponse.model_validate(item) for item in checkpoints], + ) + async def _execution(self, project_id: UUID, execution_id: UUID) -> WorkflowExecution: execution = await self._reports.get_execution(execution_id) if execution is None or execution.project_id != project_id: @@ -215,8 +266,11 @@ async def _summary( @staticmethod def _node(execution: WorkflowExecution, node: WorkflowNodeExecution) -> NodeReport: - output = node.output if isinstance(node.output, dict) else {} - observations = _node_observations(node.result) + stored_output = node.output_summary if node.output_summary is not None else node.output + output = stored_output if isinstance(stored_output, dict) else {} + stored_result = node.result_summary if node.result_summary is not None else node.result + result = _node_result(stored_result) + observations = list(result.observations) if result is not None else [] latest = observations[-1] if observations else None request = ( cast(JsonValue, latest.request.model_dump(mode="json")) @@ -240,10 +294,12 @@ def _node(execution: WorkflowExecution, node: WorkflowNodeExecution) -> NodeRepo request=request, response=response, extraction=cast(JsonValue, output) if node.node_type == "extract" else None, - assertion=cast(JsonValue, output) if node.node_type == "assert" else None, + assertion=_node_assertion(node.node_type, output, result), input_mappings=cast(JsonValue, output.get("input_mappings")), error_code=node.error_code, error_message=node.error_message, + output=cast(JsonValue, stored_output), + result=result, ) @staticmethod @@ -276,13 +332,24 @@ def _prepared_request(snapshot: dict[str, object], node_id: str) -> JsonValue: return cast(JsonValue, api.get("prepared_request")) -def _node_observations(result: dict[str, object] | None) -> list[NodeObservation]: +def _node_result(result: dict[str, object] | None) -> NodeResult | None: if result is None: - return [] + return None try: - return list(NodeResult.model_validate(result).observations) + return NodeResult.model_validate(result) except ValueError: - return [] + # Legacy results can be incomplete; retain their separate stored output. + return None + + +def _node_assertion( + node_type: str, output: dict[str, object], result: NodeResult | None +) -> JsonValue: + if node_type == "assert": + return cast(JsonValue, output) + if result is not None and result.assertions: + return [cast(JsonValue, item.model_dump(mode="json")) for item in result.assertions] + return None def _response(node_type: str, output: dict[str, object]) -> JsonValue: @@ -333,24 +400,13 @@ def _render_html(detail: ExecutionReportDetail) -> str: ) payload = html.escape( json.dumps( - { - "context": detail.context, - "nodes": [ - { - "name": node.name, - "request": node.request, - "response": node.response, - "extraction": node.extraction, - "assertion": node.assertion, - } - for node in detail.nodes - ], - }, + asdict(detail), ensure_ascii=False, indent=2, - default=str, + default=_export_value, ) ) + duration = "未提供" if summary.duration_ms is None else f"{summary.duration_ms} ms" return ( '
' "