Add hc_agent MCP run gateway and storage helpers

hc_agent/__init__.py

@@ -0,0 +1 @@
+"""Package for hc agent utilities."""

hc_agent/mcp/__init__.py

@@ -0,0 +1 @@
+"""MCP integration for hc agent."""

hc_agent/mcp/run_gateway.py

@@ -0,0 +1,54 @@
+"""Gateway for hc agent runs."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+from uuid import uuid4
+
+from .storage import write_run_artifacts
+
+
+@dataclass(frozen=True)
+class RunRequest:
+    preview: Any
+    apply: Any
+    evidence: Any
+    run: Any
+    report_md: str
+    run_id: str = field(default_factory=lambda: uuid4().hex)
+
+
+@dataclass(frozen=True)
+class RunResult:
+    run_id: str
+    run_dir: str
+    stored_at: str
+
+
+def hc_run(request: RunRequest, *, base_dir: str = "runs") -> RunResult:
+    """Persist run artifacts and return storage metadata."""
+    run_dir = write_run_artifacts(
+        base_dir=base_dir,
+        run_id=request.run_id,
+        preview=request.preview,
+        apply=request.apply,
+        evidence=request.evidence,
+        run=request.run,
+        report_md=request.report_md,
+    )
+    stored_at = datetime.now(timezone.utc).isoformat()
+    return RunResult(run_id=request.run_id, run_dir=str(run_dir), stored_at=stored_at)
+
+
+def hc_run_demo(*, base_dir: str = "runs") -> RunResult:
+    """Create a demo run with placeholder data."""
+    request = RunRequest(
+        preview={"summary": "Demo preview", "steps": ["analyze", "plan"]},
+        apply={"summary": "Demo apply", "steps": ["execute", "verify"]},
+        evidence={"summary": "Demo evidence", "artifacts": ["log.txt"]},
+        run={"status": "demo", "metadata": {"source": "hc_run_demo"}},
+        report_md="# Demo Report\n\nThis is a demo report.",
+    )
+    return hc_run(request, base_dir=base_dir)

hc_agent/mcp/storage.py

@@ -0,0 +1,86 @@
+"""Run storage helpers for hc agent."""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+
+def sanitize_obj(value: Any) -> Any:
+    """Return a JSON-serializable structure derived from value."""
+    if dataclasses.is_dataclass(value):
+        return sanitize_obj(dataclasses.asdict(value))
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    if isinstance(value, bytes):
+        return value.decode("utf-8", errors="replace")
+    if isinstance(value, Path):
+        return str(value)
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, dict):
+        return {str(key): sanitize_obj(val) for key, val in value.items()}
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [sanitize_obj(item) for item in value]
+    return str(value)
+
+
+def build_min_evidence(evidence: Any) -> Any:
+    """Trim evidence payloads down to a minimal summary when possible."""
+    if evidence is None:
+        return {}
+    if isinstance(evidence, dict):
+        preferred_keys = (
+            "summary",
+            "highlights",
+            "metrics",
+            "files",
+            "artifacts",
+            "errors",
+            "warnings",
+            "notes",
+        )
+        trimmed = {key: evidence[key] for key in preferred_keys if key in evidence}
+        return trimmed if trimmed else evidence
+    if isinstance(evidence, (list, tuple, set, frozenset)):
+        return list(evidence)
+    return evidence
+
+
+def _write_json(path: Path, payload: Any) -> None:
+    path.write_text(
+        json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+    )
+
+
+def init_run_dir(base_dir: Path | str, run_id: str) -> Path:
+    run_dir = Path(base_dir) / run_id
+    run_dir.mkdir(parents=True, exist_ok=True)
+    return run_dir
+
+
+def write_run_artifacts(
+    *,
+    base_dir: Path | str,
+    run_id: str,
+    preview: Any,
+    apply: Any,
+    evidence: Any,
+    run: Any,
+    report_md: str,
+) -> Path:
+    run_dir = init_run_dir(base_dir, run_id)
+
+    _write_json(run_dir / "preview.json", sanitize_obj(preview))
+    _write_json(run_dir / "apply.json", sanitize_obj(apply))
+    _write_json(
+        run_dir / "evidence.json", sanitize_obj(build_min_evidence(evidence))
+    )
+    _write_json(run_dir / "run.json", sanitize_obj(run))
+
+    (run_dir / "report.md").write_text(report_md, encoding="utf-8")
+    return run_dir