diff --git a/hc_agent/__init__.py b/hc_agent/__init__.py new file mode 100644 index 0000000000..60d16aa775 --- /dev/null +++ b/hc_agent/__init__.py @@ -0,0 +1 @@ +"""Package for hc agent utilities.""" diff --git a/hc_agent/mcp/__init__.py b/hc_agent/mcp/__init__.py new file mode 100644 index 0000000000..0d59ed0b5d --- /dev/null +++ b/hc_agent/mcp/__init__.py @@ -0,0 +1 @@ +"""MCP integration for hc agent.""" diff --git a/hc_agent/mcp/run_gateway.py b/hc_agent/mcp/run_gateway.py new file mode 100644 index 0000000000..8e2f9ad1bc --- /dev/null +++ b/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) diff --git a/hc_agent/mcp/storage.py b/hc_agent/mcp/storage.py new file mode 100644 index 0000000000..70c1efc086 --- /dev/null +++ b/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