From 19d5b80a2ca84e1a3b7bf413d8a2d0f7491c7e25 Mon Sep 17 00:00:00 2001 From: leetcrypt Date: Fri, 26 Jun 2026 20:28:48 -0700 Subject: [PATCH] =?UTF-8?q?feat(operator):=20hh-agent=20manifest=20?= =?UTF-8?q?=E2=80=94=20VMs=20carry=20a=20self-describing=20handoff=20recor?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cmd_chat/operator/__main__.py | 42 +++++ cmd_chat/operator/bridge.py | 100 ++++++++++ cmd_chat/operator/manifest.py | 345 ++++++++++++++++++++++++++++++++++ skills/hh-operator/SKILL.md | 25 +++ tests/test_operator_bridge.py | 96 ++++++++++ 5 files changed, 608 insertions(+) create mode 100644 cmd_chat/operator/manifest.py diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py index 3d6d6bb..1ebc269 100644 --- a/cmd_chat/operator/__main__.py +++ b/cmd_chat/operator/__main__.py @@ -127,6 +127,30 @@ def _build_parser() -> argparse.ArgumentParser: ky.add_argument("--help-keys", action="store_true", help="print the key vocabulary") ky.add_argument("--session", default=None) + # ── agent-manifest (make a sandbox carry its own handoff record) ───── + mf = sub.add_parser("manifest", help="push/pull/update the sandbox's .hh-agent bundle") + mf.add_argument("action", choices=["push", "pull", "update"]) + mf.add_argument("--root", default="/root", help="dir the .hh-agent/ lives under in the sandbox") + mf.add_argument("--name", default=None) + mf.add_argument("--purpose", default=None) + mf.add_argument("--objective", default=None) + mf.add_argument("--intent", dest="user_intent", default=None) + mf.add_argument("--stop", action="append", default=None, help="stop condition (repeatable)") + mf.add_argument("--engine", default=None) + mf.add_argument("--image", default=None) + mf.add_argument("--state-ref", dest="state_ref", default=None) + mf.add_argument("--entrypoint", default=None) + mf.add_argument("--setup", action="append", default=None) + mf.add_argument("--usage", action="append", default=None) + mf.add_argument("--status", default=None) + mf.add_argument("--progress", default=None) + mf.add_argument("--done", action="append", default=None) + mf.add_argument("--todo", action="append", default=None) + mf.add_argument("--blocker", dest="blockers", action="append", default=None) + mf.add_argument("--note", default=None, help="provenance note for update") + mf.add_argument("--replace", action="store_true", help="replace state lists instead of appending") + mf.add_argument("--session", default=None) + for v in ("roster", "status", "down"): sub.add_parser(v).add_argument("--session", default=None) return ap @@ -356,6 +380,22 @@ def _run_keys(args) -> int: return 0 +def _run_manifest(args) -> int: + req = {"op": "manifest", "action": args.action, "root": args.root} + # forward only the fields the user actually set, so the daemon's defaults hold + for k in ("name", "purpose", "objective", "user_intent", "stop", "engine", + "image", "state_ref", "entrypoint", "setup", "usage", "status", + "progress", "done", "todo", "blockers", "note"): + v = getattr(args, k, None) + if v is not None: + req[k] = v + if args.replace: + req["replace"] = True + resp = _client_request(args, req) + print(json.dumps(resp, indent=2)) + return 0 if resp.get("ok") else 1 + + def _run_simple(args, op: str) -> int: resp = _client_request(args, {"op": op}) print(json.dumps(resp)) @@ -385,6 +425,8 @@ def main(argv: list[str] | None = None) -> int: return _run_keys(args) if verb == "spawn": return _run_spawn(args) + if verb == "manifest": + return _run_manifest(args) if verb == "screen": return _run_screen(args) if verb == "watch": diff --git a/cmd_chat/operator/bridge.py b/cmd_chat/operator/bridge.py index 1ac5435..f6f11b4 100644 --- a/cmd_chat/operator/bridge.py +++ b/cmd_chat/operator/bridge.py @@ -338,6 +338,8 @@ class OperatorBridge(Client): return await self._op_keys(req) if op == "spawn": return await self._op_spawn(req) + if op == "manifest": + return await self._op_manifest(req) if op == "screen": return await self._op_screen(req) if op == "watch": @@ -649,6 +651,104 @@ class OperatorBridge(Client): return {"ok": True, "pid": proc.pid, "plan": plan, "remaining": self.budget.as_dict()} + # ── agent-manifest: make a sandbox carry its own handoff record ────────── + async def _op_manifest(self, req: dict) -> dict: + """Read/write the `.hh-agent/` manifest bundle *inside* the target + sandbox, so a VM travels with a self-describing handoff record. + + Actions: + - ``push`` — build a bundle from the given fields and write all five + files under ``/.hh-agent`` in the sandbox. This is what makes a + VM tradeable: the next agent reads AGENT.md and knows the purpose. + - ``pull`` — cat the canonical ``manifest.yaml`` back out, parsed, so a + handoff/spawn can resume exactly where the last one left off. + - ``update`` — pull, apply a state change (status/progress/done/todo/ + blockers + a provenance entry), and re-push the rendered bundle. + + All manifest logic lives in ``manifest.py``; the bridge only moves bytes + in and out of the sandbox via the same exec path as `write`/`get`.""" + from . import manifest as mf + + action = req.get("action", "push") + root = str(req.get("root", "/root")).rstrip("/") or "/" + bdir = f"{root}/{mf.MANIFEST_DIR}" + engine, name, err = self._target() + if err: + return {"ok": False, "error": err} + prefix = sbx.exec_prefix(engine, name) + if prefix is None: + return {"ok": False, "error": f"can't address {engine}/{name}"} + + async def _put(path: str, data: bytes): + return await sbx.exec_capture( + prefix + ["sh", "-c", 'mkdir -p "$(dirname "$1")" && cat > "$1"', + "hh-manifest", path], stdin=data) + + async def _read_manifest() -> mf.Manifest | None: + raw, rc = await sbx.exec_capture( + prefix + ["cat", "--", f"{bdir}/{mf.MANIFEST_FILE}"], text=False) + if rc != 0: + return None + return mf.Manifest.from_dict(mf._load(raw.decode(errors="replace"))) + + async def _write_bundle(m: mf.Manifest): + for fname, text in mf.bundle_files(m).items(): + _, rc = await _put(f"{bdir}/{fname}", text.encode()) + if rc != 0: + return False, fname + return True, "" + + if action == "pull": + m = await _read_manifest() + if m is None: + return {"ok": False, "error": f"no manifest at {bdir}"} + return {"ok": True, "dir": bdir, "manifest": m.to_dict()} + + if action == "push": + name_ = str(req.get("name") or "").strip() + if not name_: + return {"ok": False, "error": "manifest push needs a name"} + m = mf.Manifest( + name=name_, purpose=str(req.get("purpose", "")), + created_by=self.name, + vm={"engine": req.get("engine") or engine, + "image": req.get("image", ""), + "state_ref": req.get("state_ref", ""), + "entrypoint": req.get("entrypoint", "")}, + setup=list(req.get("setup") or []), + usage=list(req.get("usage") or []), + goals={"objective": req.get("objective", ""), + "user_intent": req.get("user_intent", ""), + "stop_conditions": list(req.get("stop") or [])}) + m.add_provenance(by=self.name, action="init", note=f"created {m.name}") + ok, bad = await _write_bundle(m) + if not ok: + return {"ok": False, "error": f"failed writing {bad} into sandbox"} + await self._emit("manifest", action="push", id=m.id, dir=bdir) + return {"ok": True, "id": m.id, "dir": bdir, + "files": list(mf.bundle_files(m))} + + if action == "update": + m = await _read_manifest() + if m is None: + return {"ok": False, "error": f"no manifest at {bdir} (push first)"} + m.update_state( + status=req.get("status"), progress=req.get("progress"), + done=req.get("done") or None, todo=req.get("todo") or None, + blockers=req.get("blockers") or None, + append=not bool(req.get("replace", False))) + note = req.get("note") + if note: + m.add_provenance(by=self.name, action="update", note=str(note)) + ok, bad = await _write_bundle(m) + if not ok: + return {"ok": False, "error": f"failed writing {bad} into sandbox"} + await self._emit("manifest", action="update", id=m.id, dir=bdir) + return {"ok": True, "id": m.id, "dir": bdir, + "status": m.state.get("status")} + + return {"ok": False, "error": f"unknown manifest action {action!r}"} + def _begin_shutdown(self) -> None: self._stop.set() self.running = False diff --git a/cmd_chat/operator/manifest.py b/cmd_chat/operator/manifest.py new file mode 100644 index 0000000..392c77b --- /dev/null +++ b/cmd_chat/operator/manifest.py @@ -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()) diff --git a/skills/hh-operator/SKILL.md b/skills/hh-operator/SKILL.md index 4f9e5a0..c96a14e 100644 --- a/skills/hh-operator/SKILL.md +++ b/skills/hh-operator/SKILL.md @@ -103,6 +103,31 @@ Tokens combine: literal text types verbatim; named keys send control bytes; **Rule:** end a stuck command with `ctrl-c`; if ignored, `ctrl-\` then `ctrl-c` again. Always know how to exit an interactive program *before* you start it. +## 3b. Agent manifest — make a VM carry its own handoff record + +A sandbox becomes **shareable/tradeable** when it documents itself. `manifest` +writes an ICM-style `.hh-agent/` bundle *inside* the target sandbox so the next +agent (human or a spawned Claude) can resume without a briefing: + +```bash +# stamp a fresh VM with purpose, goal, user intent, stop conditions +$HH manifest push --root /root --name kali-recon-vm \ + --purpose "Kali recon sandbox" \ + --objective "map the target and hand off" \ + --intent "tradeable VM state across handoffs" --stop "scan complete" +# as work progresses, record state + a provenance line +$HH manifest update --root /root --status done --progress "scan done" \ + --done "ran nmap" --todo "exploit phase" --note "oracle → child handoff" +# what a spawn/handoff reads first to resume exactly where you left off +$HH manifest pull --root /root # → parsed manifest.yaml +``` + +Bundle written under `/.hh-agent/`: `manifest.yaml` (canonical machine +record), `AGENT.md` (read this first), `last-state.md`, `summary.md`, +`goals.yaml`. The yaml is the source of truth; the rest are rendered from it. +Pair with `spawn`: stamp → push → spawn a child whose objective is "load the +`.hh-agent` manifest and continue." + ## 4. Delegate to a local model (optional) A room often has an `/ai ` Ollama agent. To offload a task, just `say` diff --git a/tests/test_operator_bridge.py b/tests/test_operator_bridge.py index 0f4f7b3..e6d0686 100644 --- a/tests/test_operator_bridge.py +++ b/tests/test_operator_bridge.py @@ -427,3 +427,99 @@ def test_watch_matches_event_and_idle_and_timeout(): res = await b._op_watch({"for": "never-ever", "timeout": 0.3}) assert res["reason"] == "timeout" and res["matched"] is False asyncio.run(go()) + + +# ── agent-manifest (the unit of work-transmission) ────────────────────────── +from cmd_chat.operator import manifest as mfst # noqa: E402 + + +def test_manifest_roundtrip_dict(): + m = mfst.Manifest(name="box", purpose="p", + goals={"objective": "o", "user_intent": "u", + "stop_conditions": ["done"]}) + d = m.to_dict() + # canonical key order keeps schema/id/name scannable at the top + assert list(d)[:3] == ["schema", "id", "name"] + m2 = mfst.Manifest.from_dict(d) + assert m2.name == "box" and m2.goals["objective"] == "o" + # unknown keys from a newer writer are ignored, not fatal + m3 = mfst.Manifest.from_dict({**d, "future_field": 123}) + assert m3.name == "box" + + +def test_manifest_update_appends_and_replaces(): + m = mfst.Manifest(name="box") + first = m.updated_at + m.update_state(done="wrote tooling", todo=["wire verb", "test"], status="in_progress") + assert m.state["done"] == ["wrote tooling"] + assert m.state["todo"] == ["wire verb", "test"] + m.update_state(done=["more"], append=True) + assert m.state["done"] == ["wrote tooling", "more"] + m.update_state(todo=["only"], append=False) # replace + assert m.state["todo"] == ["only"] + assert m.updated_at >= first + + +def test_manifest_provenance_and_render(): + m = mfst.Manifest(name="recon", purpose="kali box") + m.add_provenance(by="op-a", action="init", note="created") + m.add_provenance(by="op-b", note="handed off") + agent = mfst.render_agent_md(m) + assert "AGENT.md — recon" in agent and "kali box" in agent + last = mfst.render_last_state_md(m) + assert "op-a" in last and "op-b" in last and "handed off" in last + files = mfst.bundle_files(m) + assert set(files) == {"manifest.yaml", "AGENT.md", "last-state.md", + "summary.md", "goals.yaml"} + + +def test_manifest_write_read_bundle(tmp_path): + m = mfst.Manifest(name="box", purpose="p", + goals={"objective": "ship", "user_intent": "i", + "stop_conditions": ["x"]}) + bdir = mfst.write_bundle(m, str(tmp_path)) + assert Path(bdir).name == mfst.MANIFEST_DIR + loaded = mfst.read_bundle(str(tmp_path)) + assert loaded.id == m.id and loaded.goals["objective"] == "ship" + + +def test_op_manifest_push_pull_update_via_local(tmp_path): + async def go(): + b = _bridge("oracle") + b._own_engine, b._own_name = "local", "" # exec on host into tmp_path + root = str(tmp_path) + r = await b._op_manifest({ + "action": "push", "root": root, "name": "demo-box", + "purpose": "build manifest tooling live", "objective": "hand it off", + "stop": ["round-trips"]}) + assert r["ok"] and r["id"].startswith("hh-") + # the VM now carries its own bundle on disk + assert (tmp_path / ".hh-agent" / "manifest.yaml").is_file() + assert (tmp_path / ".hh-agent" / "AGENT.md").is_file() + # pull it back parsed — what a spawn/handoff reads to resume + r = await b._op_manifest({"action": "pull", "root": root}) + assert r["ok"] and r["manifest"]["name"] == "demo-box" + assert r["manifest"]["created_by"] == "oracle" + # update state + provenance, re-rendered into the bundle + r = await b._op_manifest({ + "action": "update", "root": root, "status": "done", + "progress": "tooling shipped", "done": ["wrote it"], + "note": "operator handing to child"}) + assert r["ok"] and r["status"] == "done" + r = await b._op_manifest({"action": "pull", "root": root}) + st = r["manifest"]["state"] + assert st["status"] == "done" and "wrote it" in st["done"] + prov = r["manifest"]["provenance"] + assert any(p["action"] == "update" for p in prov) + asyncio.run(go()) + + +def test_op_manifest_pull_missing_is_clean_error(): + async def go(): + b = _bridge("oracle") + b._own_engine, b._own_name = "local", "" + import tempfile + with tempfile.TemporaryDirectory() as d: + r = await b._op_manifest({"action": "pull", "root": d}) + assert r["ok"] is False and "no manifest" in r["error"] + asyncio.run(go())