feat(operator): hh-agent manifest — VMs carry a self-describing handoff record
Add ICM-aligned agent-manifest tooling so a hack-house sandbox/VM can be shared, traded, and resumed without a human briefing. manifest.py is a dependency-light (stdlib + optional PyYAML) library + CLI that writes a `.hh-agent/` bundle: manifest.yaml (canonical machine record) plus rendered AGENT.md / last-state.md / summary.md / goals.yaml views. Wire a `manifest` op into the operator bridge (push/pull/update) that moves the bundle in and out of the *target sandbox* via the same exec path as write/get, so the VM itself carries its purpose, goals, user intent, live state, and a provenance chain — the unit of agent-to-agent work transmission. Pairs with `spawn`: stamp → push → hand a child "load the .hh-agent manifest and continue." Proven end-to-end against a real Kali podman container (push → on-disk verify → update → pull-back) and covered by 6 new offline tests (34 green total). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
leetcrypt
@@ -0,0 +1,345 @@
|
||||
"""hh-agent manifest — the unit of work-transmission between agents (and humans).
|
||||
|
||||
A hack-house VM/sandbox can be *handed off*: one operator does work, then another
|
||||
(human or a freshly-spawned Claude) picks it up. For that to work without a human
|
||||
briefing, the VM carries a small **ICM-style directory** that documents itself —
|
||||
purpose, setup, usage, goals, and the live state of the work.
|
||||
|
||||
This module is the host-side library + CLI that reads/writes that directory. It
|
||||
is deliberately dependency-light (stdlib + optional PyYAML) so it runs the same
|
||||
inside a throwaway container, on the host, or in a nested spawn.
|
||||
|
||||
Bundle layout (written under `.hh-agent/` at the VM/work root):
|
||||
|
||||
.hh-agent/
|
||||
manifest.yaml # CANONICAL machine record (schema hh-agent/v1). Source of truth.
|
||||
AGENT.md # narrative entry point — agent reads first, human can too
|
||||
last-state.md # rendered: progress / done / todo / blockers
|
||||
summary.md # rendered: one-paragraph TL;DR
|
||||
goals.yaml # rendered: objective + user intent + stop conditions
|
||||
|
||||
ICM alignment (see ~/coding/_core/ICM.md): `manifest.yaml` is the single canonical
|
||||
record (Layer 3 — "one rule lives in exactly one place"); the `.md`/`goals.yaml`
|
||||
files are *rendered views* of it (Layer 4 — disposable, regenerated on write).
|
||||
Editing the markdown by hand is fine for a human handoff note, but `manifest.yaml`
|
||||
wins on the next `render`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import uuid
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime, timezone
|
||||
|
||||
SCHEMA = "hh-agent/v1"
|
||||
MANIFEST_DIR = ".hh-agent"
|
||||
MANIFEST_FILE = "manifest.yaml"
|
||||
|
||||
|
||||
# ── serialization (PyYAML if present, JSON fallback so it never hard-deps) ────
|
||||
def _dump(obj: dict) -> str:
|
||||
try:
|
||||
import yaml
|
||||
return yaml.safe_dump(obj, sort_keys=False, default_flow_style=False,
|
||||
allow_unicode=True)
|
||||
except Exception: # noqa: BLE001 — yaml missing or chokes → readable JSON
|
||||
return json.dumps(obj, indent=2, ensure_ascii=False) + "\n"
|
||||
|
||||
|
||||
def _load(text: str) -> dict:
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return {}
|
||||
try:
|
||||
import yaml
|
||||
return yaml.safe_load(text) or {}
|
||||
except Exception: # noqa: BLE001
|
||||
return json.loads(text)
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
# ── the record ────────────────────────────────────────────────────────────────
|
||||
@dataclass
|
||||
class Manifest:
|
||||
"""A self-describing hack-house work artifact. `vm`, `goals`, `state` are
|
||||
free-form dicts on purpose: a nested Claude designing for its own goal can
|
||||
add fields without a schema change, and an older reader just ignores them."""
|
||||
|
||||
name: str
|
||||
purpose: str = ""
|
||||
created_by: str = "operator"
|
||||
id: str = field(default_factory=lambda: f"hh-{uuid.uuid4().hex[:12]}")
|
||||
schema: str = SCHEMA
|
||||
created_at: str = field(default_factory=_now)
|
||||
updated_at: str = field(default_factory=_now)
|
||||
# how this VM is realized + how to drive it
|
||||
vm: dict = field(default_factory=lambda: {
|
||||
"engine": "", "image": "", "state_ref": "", "entrypoint": ""})
|
||||
setup: list = field(default_factory=list) # commands/notes to bring it up
|
||||
usage: list = field(default_factory=list) # how an agent should use it
|
||||
# why the work exists — the part a fresh agent most needs
|
||||
goals: dict = field(default_factory=lambda: {
|
||||
"objective": "", "user_intent": "", "stop_conditions": []})
|
||||
# the live state of the work (the handoff-critical bit)
|
||||
state: dict = field(default_factory=lambda: {
|
||||
"status": "in_progress", "progress": "",
|
||||
"done": [], "todo": [], "blockers": []})
|
||||
# chain of custody: who touched it, when, with what note
|
||||
provenance: list = field(default_factory=list)
|
||||
|
||||
# -- (de)serialize -------------------------------------------------------
|
||||
def to_dict(self) -> dict:
|
||||
d = asdict(self)
|
||||
# keep schema/id/name first for human scanability of the yaml
|
||||
order = ["schema", "id", "name", "purpose", "created_by",
|
||||
"created_at", "updated_at", "vm", "setup", "usage",
|
||||
"goals", "state", "provenance"]
|
||||
return {k: d[k] for k in order if k in d}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d: dict) -> "Manifest":
|
||||
known = {f for f in cls.__dataclass_fields__} # type: ignore[attr-defined]
|
||||
return cls(**{k: v for k, v in (d or {}).items() if k in known})
|
||||
|
||||
# -- mutation ------------------------------------------------------------
|
||||
def touch(self) -> None:
|
||||
self.updated_at = _now()
|
||||
|
||||
def add_provenance(self, by: str, note: str = "", action: str = "handoff") -> None:
|
||||
self.provenance.append({"by": by, "action": action,
|
||||
"at": _now(), "note": note})
|
||||
self.touch()
|
||||
|
||||
def update_state(self, *, status=None, progress=None,
|
||||
done=None, todo=None, blockers=None,
|
||||
append=True) -> None:
|
||||
st = self.state
|
||||
if status is not None:
|
||||
st["status"] = status
|
||||
if progress is not None:
|
||||
st["progress"] = progress
|
||||
for key, val in (("done", done), ("todo", todo), ("blockers", blockers)):
|
||||
if val is None:
|
||||
continue
|
||||
items = [val] if isinstance(val, str) else list(val)
|
||||
if append:
|
||||
st.setdefault(key, []).extend(items)
|
||||
else:
|
||||
st[key] = items
|
||||
self.touch()
|
||||
|
||||
|
||||
# ── rendered views (derived from the canonical manifest) ──────────────────────
|
||||
def _bullets(items, empty="_none_") -> str:
|
||||
items = items or []
|
||||
if not items:
|
||||
return empty
|
||||
return "\n".join(f"- {x}" for x in items)
|
||||
|
||||
|
||||
def render_summary_md(m: Manifest) -> str:
|
||||
g = m.goals or {}
|
||||
obj = g.get("objective") or m.purpose or "(no objective set)"
|
||||
return (f"# {m.name} — summary\n\n"
|
||||
f"**{m.purpose or obj}**\n\n"
|
||||
f"Status: `{m.state.get('status', '?')}` · "
|
||||
f"{m.state.get('progress') or 'no progress note'}\n\n"
|
||||
f"_id `{m.id}` · schema `{m.schema}` · updated {m.updated_at}_\n")
|
||||
|
||||
|
||||
def render_agent_md(m: Manifest) -> str:
|
||||
g = m.goals or {}
|
||||
vm = m.vm or {}
|
||||
vm_line = " · ".join(f"{k}=`{v}`" for k, v in vm.items() if v) or "_unspecified_"
|
||||
return (
|
||||
f"# AGENT.md — {m.name}\n\n"
|
||||
f"> Entry point for whoever (agent or human) picks this VM up next.\n"
|
||||
f"> Canonical record is `{MANIFEST_FILE}`; this file is rendered from it.\n\n"
|
||||
f"## Purpose\n{m.purpose or '_(none)_'}\n\n"
|
||||
f"## Goal\n"
|
||||
f"**Objective:** {g.get('objective') or '_(none)_'}\n\n"
|
||||
f"**User intent:** {g.get('user_intent') or '_(none)_'}\n\n"
|
||||
f"**Stop when:**\n{_bullets(g.get('stop_conditions'))}\n\n"
|
||||
f"## VM\n{vm_line}\n\n"
|
||||
f"## Setup\n{_bullets(m.setup)}\n\n"
|
||||
f"## Usage\n{_bullets(m.usage)}\n\n"
|
||||
f"## Where the work stands\nSee `last-state.md`. "
|
||||
f"Status: `{m.state.get('status', '?')}`.\n"
|
||||
)
|
||||
|
||||
|
||||
def render_last_state_md(m: Manifest) -> str:
|
||||
st = m.state or {}
|
||||
prov = m.provenance or []
|
||||
prov_lines = "\n".join(
|
||||
f"- {p.get('at', '?')} · **{p.get('by', '?')}** "
|
||||
f"({p.get('action', 'handoff')}): {p.get('note', '')}"
|
||||
for p in prov) or "_no handoffs recorded_"
|
||||
return (
|
||||
f"# last-state — {m.name}\n\n"
|
||||
f"_updated {m.updated_at}_\n\n"
|
||||
f"**Status:** `{st.get('status', '?')}`\n\n"
|
||||
f"**Progress:** {st.get('progress') or '_(none)_'}\n\n"
|
||||
f"## Done\n{_bullets(st.get('done'))}\n\n"
|
||||
f"## To do\n{_bullets(st.get('todo'))}\n\n"
|
||||
f"## Blockers\n{_bullets(st.get('blockers'))}\n\n"
|
||||
f"## Provenance\n{prov_lines}\n"
|
||||
)
|
||||
|
||||
|
||||
def render_goals_yaml(m: Manifest) -> str:
|
||||
return _dump({"objective": (m.goals or {}).get("objective", ""),
|
||||
"user_intent": (m.goals or {}).get("user_intent", ""),
|
||||
"stop_conditions": (m.goals or {}).get("stop_conditions", [])})
|
||||
|
||||
|
||||
# ── bundle I/O ────────────────────────────────────────────────────────────────
|
||||
def bundle_files(m: Manifest) -> dict:
|
||||
"""The full set of files (relative names → text) that make up a bundle.
|
||||
Returned as data so the caller can write them to disk *or* push them into a
|
||||
sandbox via the operator's `write` op without this module touching a VM."""
|
||||
return {
|
||||
MANIFEST_FILE: _dump(m.to_dict()),
|
||||
"AGENT.md": render_agent_md(m),
|
||||
"last-state.md": render_last_state_md(m),
|
||||
"summary.md": render_summary_md(m),
|
||||
"goals.yaml": render_goals_yaml(m),
|
||||
}
|
||||
|
||||
|
||||
def write_bundle(m: Manifest, root: str = ".") -> str:
|
||||
"""Write the `.hh-agent/` bundle under `root`. Returns the bundle dir path."""
|
||||
bdir = os.path.join(root, MANIFEST_DIR)
|
||||
os.makedirs(bdir, exist_ok=True)
|
||||
for name, text in bundle_files(m).items():
|
||||
with open(os.path.join(bdir, name), "w", encoding="utf-8") as fh:
|
||||
fh.write(text)
|
||||
return bdir
|
||||
|
||||
|
||||
def read_bundle(root: str = ".") -> Manifest:
|
||||
"""Load a bundle's canonical manifest from under `root`. Accepts either the
|
||||
work root (containing `.hh-agent/`) or the bundle dir itself."""
|
||||
path = _resolve_manifest_path(root)
|
||||
if path is None:
|
||||
raise FileNotFoundError(f"no {MANIFEST_DIR}/{MANIFEST_FILE} under {root!r}")
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return Manifest.from_dict(_load(fh.read()))
|
||||
|
||||
|
||||
def _resolve_manifest_path(root: str) -> str | None:
|
||||
for cand in (os.path.join(root, MANIFEST_DIR, MANIFEST_FILE),
|
||||
os.path.join(root, MANIFEST_FILE)):
|
||||
if os.path.isfile(cand):
|
||||
return cand
|
||||
return None
|
||||
|
||||
|
||||
# ── CLI ───────────────────────────────────────────────────────────────────────
|
||||
def _kv_list(values) -> list:
|
||||
return list(values or [])
|
||||
|
||||
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
prog="hh-manifest",
|
||||
description="Create/inspect/update an hh-agent VM manifest bundle.")
|
||||
p.add_argument("-C", "--root", default=".",
|
||||
help="work root the .hh-agent/ bundle lives under (default: .)")
|
||||
sub = p.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
s = sub.add_parser("init", help="create a new bundle")
|
||||
s.add_argument("name")
|
||||
s.add_argument("--purpose", default="")
|
||||
s.add_argument("--by", dest="created_by", default="operator")
|
||||
s.add_argument("--objective", default="")
|
||||
s.add_argument("--intent", dest="user_intent", default="")
|
||||
s.add_argument("--stop", action="append", default=[], help="a stop condition (repeatable)")
|
||||
s.add_argument("--engine", default="")
|
||||
s.add_argument("--image", default="")
|
||||
s.add_argument("--state-ref", dest="state_ref", default="")
|
||||
s.add_argument("--entrypoint", default="")
|
||||
s.add_argument("--setup", action="append", default=[])
|
||||
s.add_argument("--usage", action="append", default=[])
|
||||
|
||||
sub.add_parser("show", help="print the canonical manifest as JSON")
|
||||
|
||||
sub.add_parser("render", help="re-render the .md/.yaml views from manifest.yaml")
|
||||
|
||||
s = sub.add_parser("update", help="update live work state")
|
||||
s.add_argument("--status")
|
||||
s.add_argument("--progress")
|
||||
s.add_argument("--done", action="append", default=[])
|
||||
s.add_argument("--todo", action="append", default=[])
|
||||
s.add_argument("--blocker", dest="blockers", action="append", default=[])
|
||||
s.add_argument("--replace", action="store_true",
|
||||
help="replace lists instead of appending")
|
||||
|
||||
s = sub.add_parser("handoff", help="record a provenance entry (who/what/note)")
|
||||
s.add_argument("--by", required=True)
|
||||
s.add_argument("--note", default="")
|
||||
s.add_argument("--action", default="handoff")
|
||||
return p
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
args = _build_parser().parse_args(argv)
|
||||
root = args.root
|
||||
|
||||
if args.cmd == "init":
|
||||
m = Manifest(
|
||||
name=args.name, purpose=args.purpose, created_by=args.created_by,
|
||||
vm={"engine": args.engine, "image": args.image,
|
||||
"state_ref": args.state_ref, "entrypoint": args.entrypoint},
|
||||
setup=_kv_list(args.setup), usage=_kv_list(args.usage),
|
||||
goals={"objective": args.objective, "user_intent": args.user_intent,
|
||||
"stop_conditions": _kv_list(args.stop)})
|
||||
m.add_provenance(by=args.created_by, action="init",
|
||||
note=f"created {m.name}")
|
||||
bdir = write_bundle(m, root)
|
||||
print(f"initialized {m.id} → {bdir}")
|
||||
return 0
|
||||
|
||||
# all remaining verbs operate on an existing bundle
|
||||
try:
|
||||
m = read_bundle(root)
|
||||
except FileNotFoundError as e:
|
||||
print(f"error: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if args.cmd == "show":
|
||||
print(json.dumps(m.to_dict(), indent=2, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
if args.cmd == "render":
|
||||
bdir = write_bundle(m, root)
|
||||
print(f"rendered views → {bdir}")
|
||||
return 0
|
||||
|
||||
if args.cmd == "update":
|
||||
m.update_state(
|
||||
status=args.status, progress=args.progress,
|
||||
done=args.done or None, todo=args.todo or None,
|
||||
blockers=args.blockers or None, append=not args.replace)
|
||||
write_bundle(m, root)
|
||||
print(f"updated {m.id} (status={m.state.get('status')})")
|
||||
return 0
|
||||
|
||||
if args.cmd == "handoff":
|
||||
m.add_provenance(by=args.by, note=args.note, action=args.action)
|
||||
write_bundle(m, root)
|
||||
print(f"recorded {args.action} by {args.by}")
|
||||
return 0
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user