build-small-hackathon/hackathon-advisor

1from __future__ import annotations2 3import json4from typing import Any5 6from hackathon_advisor._text import utc_now7 8 9TRACE_SCHEMA_VERSION = 110 11 12def build_trace_jsonl(session: dict[str, Any], metadata: dict[str, Any]) -> str:13    trace = session.get("trace") or []14    ideas = session.get("ideas") or []15    records = [16        {17            "type": "trace_manifest",18            "schema_version": TRACE_SCHEMA_VERSION,19            "generated_at": utc_now(),20            "app": "hackathon-advisor",21            "index": {22                "algorithm": metadata.get("index_algorithm", ""),23                "snapshot_generated_at": metadata.get("snapshot_generated_at", ""),24                "index_generated_at": metadata.get("index_generated_at", ""),25                "snapshot_digest": metadata.get("snapshot_digest", ""),26            },27            "idea_count": len(ideas),28            "turn_count": len(trace),29        }30    ]31    for index, event in enumerate(trace, start=1):32        records.append(33            {34                "type": "agent_turn",35                "schema_version": TRACE_SCHEMA_VERSION,36                "turn_index": index,37                "input": str(event.get("input") or ""),38                "tools": _tools(event),39                "verdict": str(event.get("verdict") or ""),40                "overall": event.get("overall"),41                "plan_steps": int(event.get("plan_steps") or 0),42                "artifact_title": str(event.get("artifact_title") or ""),43                "response": str(event.get("response") or ""),44                "tool_resolution": _tool_resolution(event),45            }46        )47    return "\n".join(json.dumps(record, ensure_ascii=False, sort_keys=True) for record in records) + "\n"48 49 50def trace_metadata(index: Any) -> dict[str, str]:51    metadata = {52        "snapshot_generated_at": index.generated_at,53        "index_generated_at": index.index_generated_at,54        "index_algorithm": index.index_algorithm,55        "snapshot_digest": index.snapshot_digest,56    }57    embedding = getattr(index, "embedding_metadata", None)58    if isinstance(embedding, dict):59        metadata.update(60            {61                "embedding_model_repo": str(embedding.get("model_repo") or ""),62                "embedding_model_file": str(embedding.get("model_file") or ""),63                "embedding_runtime": str(embedding.get("runtime") or ""),64                "embedding_build_source": str(embedding.get("build_source") or ""),65                "embedding_dimensions": str(embedding.get("dimensions") or ""),66                "embedding_builder": str(embedding.get("builder") or ""),67                "embedding_modal_app": str(embedding.get("modal_app") or ""),68            }69        )70    return metadata71 72 73def _tools(event: dict[str, Any]) -> list[dict[str, str]]:74    tools = event.get("tools") or []75    return [76        {77            "name": str(tool.get("name") or ""),78            "summary": str(tool.get("summary") or ""),79        }80        for tool in tools81        if isinstance(tool, dict)82    ]83 84 85def _tool_resolution(event: dict[str, Any]) -> dict[str, Any]:86    resolution = event.get("tool_resolution") or {}87    call = resolution.get("call") if isinstance(resolution, dict) else {}88    return {89        "status": str(resolution.get("status") or "") if isinstance(resolution, dict) else "",90        "call": {91            "name": str(call.get("name") or "") if isinstance(call, dict) else "",92            "arguments": call.get("arguments") if isinstance(call, dict) else {},93        },94        "errors": list(resolution.get("errors") or []) if isinstance(resolution, dict) else [],95    }96