Spaces:
Running
Running
| from __future__ import annotations | |
| from datetime import datetime, timezone | |
| from typing import Any | |
| from .schemas import RunExport | |
| def infer_status(state: dict[str, Any]) -> str: | |
| decision = state.get("orchestration_decision", {}).get("decision") | |
| errors = state.get("errors", []) | |
| has_output = bool(state.get("requirements") or state.get("test_designs") or state.get("review")) | |
| if decision == "finish" and not errors: | |
| return "completed" | |
| if has_output: | |
| return "partial" | |
| return "failed" | |
| def build_run_export(state: dict[str, Any]) -> RunExport: | |
| metadata = dict(state.get("run_metadata", {})) | |
| created_at = metadata.get("created_at") or datetime.now(timezone.utc).isoformat() | |
| completed_at = datetime.now(timezone.utc) | |
| start_ms = int(metadata.get("started_monotonic_ms", 0) or 0) | |
| now_ms = int(datetime.now(timezone.utc).timestamp() * 1000) | |
| metadata.update( | |
| { | |
| "run_id": metadata.get("run_id", "run_unknown"), | |
| "created_at": created_at, | |
| "completed_at": completed_at, | |
| "status": infer_status(state), | |
| "total_duration_ms": max(0, now_ms - start_ms) if start_ms else 0, | |
| "app_version": metadata.get("app_version", "langgraph-qa-compare-0.1.0"), | |
| } | |
| ) | |
| payload = { | |
| "run_metadata": metadata, | |
| "title": state.get("title", ""), | |
| "source_requirements": state.get("source_requirements", ""), | |
| "requirements": state.get("requirements", []), | |
| "test_designs": state.get("test_designs", []), | |
| "generated_artifacts": state.get("generated_artifacts", []), | |
| "review": state.get("review") or {}, | |
| "orchestration_decision": state.get("orchestration_decision") or { | |
| "decision": "fail", | |
| "target_agent": None, | |
| "reason": "No orchestration decision was produced.", | |
| "feedback": None, | |
| }, | |
| "iterations": state.get("iterations", 0), | |
| "run_controls": state.get("run_controls", {}), | |
| "agent_configs": state.get("agent_configs", []), | |
| "stage_traces": state.get("stage_traces", []), | |
| "run_session": { | |
| "title": state.get("title", ""), | |
| "source_requirements": state.get("source_requirements", ""), | |
| "working_memory": state.get("working_memory", {"shared": {}, "agent_private": {}, "timeline": []}), | |
| }, | |
| } | |
| return RunExport.model_validate(payload) | |