| from __future__ import annotations |
| import hashlib |
| import io |
| import json |
| import zipfile |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional |
|
|
| from .config import settings |
| from .models import JobResult, A2ACall |
|
|
|
|
| def _canonical_json(obj: Any) -> str: |
| """Deterministic JSON serialization.""" |
| return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(",", ":")) |
|
|
|
|
| def _sha256(text: str) -> str: |
| return hashlib.sha256(text.encode("utf-8")).hexdigest() |
|
|
|
|
| def build_manifest(job: JobResult, a2a_calls: List[A2ACall]) -> Dict[str, Any]: |
| result_dict = job.scorecard.model_dump() if job.scorecard else {} |
| canonical = _canonical_json(result_dict) |
| result_hash = f"sha256:{_sha256(canonical)}" |
|
|
| return { |
| "job_id": job.job_id, |
| "cap_order_id": "", |
| "capability": job.capability, |
| "created_at": job.created_at.isoformat(), |
| "completed_at": job.completed_at.isoformat() if job.completed_at else None, |
| "agent_version": "0.1.0", |
| "input_hash": "", |
| "result_hash": result_hash, |
| "proof_files": [ |
| "result.json", |
| "result.md", |
| "result_hash.sha256", |
| "execution_log.jsonl", |
| "evidence/sources.json", |
| "attestation.json", |
| ], |
| "a2a_calls": [c.model_dump(mode="json") for c in a2a_calls], |
| "limits": { |
| "max_job_seconds": settings.max_job_seconds, |
| "max_repo_mb": settings.max_repo_mb, |
| "network_enabled": settings.allow_network_repro, |
| }, |
| } |
|
|
|
|
| def build_attestation(result_hash: str, log_hash: str) -> Dict[str, Any]: |
| return { |
| "statement": "CAPScore Agent generated this report from the supplied inputs under the recorded runtime constraints.", |
| "result_hash": result_hash, |
| "execution_log_hash": log_hash, |
| "provider_agent": "CAPScore Agent", |
| "version": "0.1.0", |
| "generated_at": datetime.now(timezone.utc).isoformat(), |
| "signature": None, |
| } |
|
|
|
|
| def render_result_md(job: JobResult) -> str: |
| sc = job.scorecard |
| if not sc: |
| return f"# CAPScore Report\n\nJob `{job.job_id}` — No scorecard available.\n" |
|
|
| lines = [ |
| f"# CAPScore Report — Job `{job.job_id}`", |
| f"", |
| f"**Overall Score: {sc.overall_score:.0f}/100**", |
| f"", |
| f"## Judging-Aligned Scorecard", |
| f"", |
| f"| Dimension | Score | Notes |", |
| f"|---|---:|---|", |
| f"| Technical Execution | {sc.technical_execution.score:.0f}/100 | {sc.technical_execution.notes} |", |
| f"| A2A Composability | {sc.a2a_composability.score:.0f}/100 | {sc.a2a_composability.notes} |", |
| f"| Innovation | {sc.innovation.score:.0f}/100 | {sc.innovation.notes} |", |
| f"| Adoption Readiness | {sc.adoption_readiness.score:.0f}/100 | {sc.adoption_readiness.notes} |", |
| f"| Presentation Readiness | {sc.presentation_readiness.score:.0f}/100 | {sc.presentation_readiness.notes} |", |
| f"", |
| ] |
|
|
| if sc.critical_issues: |
| lines += ["## Critical Issues", ""] |
| for i, issue in enumerate(sc.critical_issues, 1): |
| lines.append(f"{i}. {issue}") |
| lines.append("") |
|
|
| if sc.top_fixes: |
| lines += ["## Top Fixes", ""] |
| for i, fix in enumerate(sc.top_fixes, 1): |
| lines.append(f"{i}. {fix}") |
| lines.append("") |
|
|
| if job.worker_results and job.worker_results.claims: |
| lines += ["## Claim Verification", ""] |
| lines += ["| Claim | Status | Confidence | Suggested Rewrite |", "|---|---|---:|---|"] |
| for v in job.worker_results.claims.verifications: |
| status_emoji = {"supported": "✓", "weak": "~", "unsupported": "✗", "misleading": "⚠"}.get(v.status, "?") |
| lines.append(f"| {v.claim[:60]}... | {status_emoji} {v.status} | {v.confidence:.0%} | {v.suggested_rewrite[:60]} |") |
| lines.append("") |
|
|
| lines += [ |
| "## Proof", |
| "", |
| f"- Result hash: `{job.result_hash or 'pending'}`", |
| f"- Proof pack: `proof-pack-{job.job_id}.zip`", |
| f"- A2A calls: {len(job.a2a_calls)}", |
| f"- Generated: {job.completed_at.isoformat() if job.completed_at else 'pending'}", |
| "", |
| "---", |
| "*Generated by CAPScore Agent v0.1.0 — CROO Hackathon 2026*", |
| ] |
|
|
| return "\n".join(lines) |
|
|
|
|
| def build_proof_pack(job: JobResult, execution_log_lines: List[str]) -> bytes: |
| """Build a deterministic ZIP proof pack in memory.""" |
| sc = job.scorecard |
| result_dict = sc.model_dump(mode="json") if sc else {} |
| canonical = _canonical_json(result_dict) |
| result_hash = f"sha256:{_sha256(canonical)}" |
|
|
| log_content = "\n".join(execution_log_lines) |
| log_hash = f"sha256:{_sha256(log_content)}" |
|
|
| manifest = build_manifest(job, job.a2a_calls) |
| manifest["result_hash"] = result_hash |
| attestation = build_attestation(result_hash, log_hash) |
|
|
| result_md = render_result_md(job) |
|
|
| buf = io.BytesIO() |
| with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: |
| zf.writestr("manifest.json", _canonical_json(manifest)) |
| zf.writestr("result.json", _canonical_json(result_dict)) |
| zf.writestr("result.md", result_md) |
| zf.writestr("result_hash.sha256", result_hash) |
| zf.writestr("execution_log.jsonl", log_content) |
| zf.writestr("attestation.json", _canonical_json(attestation)) |
| |
| evidence: Dict[str, Any] = {} |
| if job.worker_results: |
| if job.worker_results.repo: |
| evidence["repo_files"] = job.worker_results.repo.files_present |
| evidence["repo_issues"] = job.worker_results.repo.issues |
| if job.worker_results.security: |
| evidence["security_findings"] = [f.model_dump() for f in job.worker_results.security.findings] |
| zf.writestr("evidence/sources.json", _canonical_json(evidence)) |
| return buf.getvalue() |
|
|
|
|
| def persist_proof_pack(job: JobResult, execution_log_lines: List[str]) -> Path: |
| """Write proof pack to disk and return path.""" |
| runs_dir = settings.runs_dir / job.job_id |
| runs_dir.mkdir(parents=True, exist_ok=True) |
|
|
| zip_bytes = build_proof_pack(job, execution_log_lines) |
| zip_path = runs_dir / f"proof-pack-{job.job_id}.zip" |
| zip_path.write_bytes(zip_bytes) |
|
|
| |
| sc = job.scorecard |
| result_dict = sc.model_dump(mode="json") if sc else {} |
| canonical = json.dumps(result_dict, sort_keys=True, ensure_ascii=False, indent=2) |
| result_hash = f"sha256:{_sha256(json.dumps(result_dict, sort_keys=True, ensure_ascii=False, separators=(',', ':')))}" |
|
|
| (runs_dir / "result.json").write_text(canonical, encoding="utf-8") |
| (runs_dir / "result.md").write_text(render_result_md(job), encoding="utf-8") |
| (runs_dir / "result_hash.sha256").write_text(result_hash, encoding="utf-8") |
| (runs_dir / "execution_log.jsonl").write_text("\n".join(execution_log_lines), encoding="utf-8") |
|
|
| job.result_hash = result_hash |
| job.proof_pack_url = f"{settings.capscore_public_base_url}/jobs/{job.job_id}/proof-pack.zip" |
| return zip_path |
|
|
