Spaces:
Sleeping
Sleeping
File size: 2,459 Bytes
b85de73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | 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)
|