diff --git a/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.py b/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.py index bafbf02..4545c57 100755 --- a/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.py +++ b/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.py @@ -4,9 +4,11 @@ Accepts: - Claude ``--output-format stream-json`` (type: assistant / user / result) -- Grok ``--output-format streaming-messages-json`` (same Messages wire) -- Grok native ``--output-format streaming-json`` (ACP: text / thought / - tool_call / tool_call_update / end) +- Grok ``--output-format streaming-messages-json`` (same Messages wire; Grok + tool results arrive as JSON blobs whose ``output`` is a byte array) +- Grok native ``--output-format streaming-json`` (ACP: per-token ``thought`` / + ``text`` deltas, ``tool_call`` / ``tool_call_update``, ``usage``, ``end`` — + the session id only arrives on ``end``) Surfaces a resume id once when first seen and again just above ``—— final ——``: @@ -14,7 +16,8 @@ 📎 session_id= Human-visible progress is teed to AGENT_HUMAN_STREAM_LIVE_LOG (or the -legacy CLAUDE_HUMAN_STREAM_LIVE_LOG). +legacy CLAUDE_HUMAN_STREAM_LIVE_LOG); the final text is also written to +``.final.md`` so a caller never has to scrape the log. """ from __future__ import annotations @@ -24,6 +27,8 @@ import time _live_fp = None +THINK_WIDTH = int(os.environ.get("AGENT_HUMAN_STREAM_THINK_WIDTH") or "160") +TOOL_OUT_WIDTH = int(os.environ.get("AGENT_HUMAN_STREAM_TOOL_WIDTH") or "200") def env(*keys: str, default: str = "") -> str: @@ -55,6 +60,97 @@ def _short(value, n: int = 160) -> str: return str(value or "").replace("\n", " ")[:n] +def _bytes_to_text(value) -> str | None: + """Grok ships command output as a JSON array of byte values.""" + if isinstance(value, list) and value and all(isinstance(b, int) and 0 <= b < 256 for b in value): + try: + return bytes(value).decode("utf-8", errors="replace") + except (ValueError, TypeError): + return None + return None + + +def decode_tool_result(content) -> str: + """Turn a tool_result payload (Claude text, Grok JSON blob) into plain text.""" + if isinstance(content, list): + parts = [] + for c in content: + if isinstance(c, dict) and c.get("type") == "text": + parts.append(str(c.get("text") or "")) + elif isinstance(c, str): + parts.append(c) + content = "\n".join(parts) + if not isinstance(content, str): + content = json.dumps(content, ensure_ascii=False) if content is not None else "" + text = content.strip() + if not text.startswith("{"): + return text + try: + blob = json.loads(text) + except json.JSONDecodeError: + return text + if not isinstance(blob, dict): + return text + kind = blob.get("type") or "" + for key in ("output", "stdout", "stderr"): + decoded = _bytes_to_text(blob.get(key)) + if decoded is not None: + exit_code = blob.get("exit_code") + suffix = f" (exit {exit_code})" if isinstance(exit_code, int) and exit_code != 0 else "" + return f"{decoded.strip()}{suffix}" if decoded.strip() else f"(no {key}){suffix}" + if isinstance(blob.get(key), str) and blob.get(key): + return str(blob[key]) + fc = blob.get("FileContent") + if isinstance(fc, dict) and isinstance(fc.get("content"), str): + return fc["content"] + if isinstance(blob.get("FileNotFound"), str): + return f"not found: {blob['FileNotFound']}" + todo = blob.get("TodosUpdated") + if isinstance(todo, dict) and isinstance(todo.get("summary_for_prompt"), str): + return todo["summary_for_prompt"] + out = blob.get("output") + if isinstance(out, dict): + for key in ("OkayOutput", "ErrorOutput", "content", "text"): + if isinstance(out.get(key), str): + return out[key] + if isinstance(blob.get("content"), str): + return blob["content"] + if kind == "Monitor" and blob.get("taskId"): + return f"monitor task {blob['taskId']} started (timeout {blob.get('timeoutMs')} ms) — the -p turn will end while it waits" + return text + + +_SHELL_TOOLS = {"bash", "run_terminal_cmd", "run_terminal_command", "shell"} +_PATH_KEYS = ("path", "file", "file_path", "target_file", "target_directory", "directory") + + +def _tool_line(name: str, inp) -> str: + if not isinstance(inp, dict): + inp = {} + key = (name or "tool").strip() + low = key.lower() + if low in _SHELL_TOOLS: + cmd = _short(inp.get("command") or inp.get("cmd") or "") + return f"$ {cmd}" if cmd else key + if low == "monitor": + return f"⏳ monitor (background; ends the -p turn) {_short(inp.get('command') or '', 120)}" + if low in {"get_command_or_subagent_output", "get_task_output"}: + return f"⏳ wait {_short(inp.get('task_id') or inp.get('id') or inp, 60)}" + if low in {"spawn_subagent", "task"}: + return f"🧵 subagent {_short(inp.get('description') or inp.get('prompt') or '', 120)}" + for pk in _PATH_KEYS: + if inp.get(pk): + extra = "" + if inp.get("pattern"): + extra = f" /{_short(inp.get('pattern'), 60)}/" + return f"{key} {inp[pk]}{extra}" + if inp.get("pattern"): + return f"{key} /{_short(inp.get('pattern'), 80)}/ {_short(inp.get('glob') or '', 40)}".rstrip() + if inp.get("query"): + return f"{key} {_short(inp.get('query'), 120)}" + return f"{key} {_short(inp, 120)}" + + def text_bits(content) -> str: if isinstance(content, str): return content @@ -68,30 +164,16 @@ def text_bits(content) -> str: if t == "text": parts.append(c.get("text") or "") elif t == "tool_use": - parts.append(_tool_line(c.get("name") or "tool", c.get("input") or {})) + parts.append("🔧 " + _tool_line(c.get("name") or "tool", c.get("input") or {})) elif t == "thinking": - th = _short(c.get("thinking") or "", 80) + th = _short(c.get("thinking") or "", THINK_WIDTH) if th: parts.append(f"(thinking) {th}") elif t == "server_tool_use": - parts.append(_tool_line(c.get("name") or "server_tool", c.get("input") or {})) + parts.append("🔧 " + _tool_line(c.get("name") or "server_tool", c.get("input") or {})) return "\n".join(p for p in parts if p) -def _tool_line(name: str, inp) -> str: - if not isinstance(inp, dict): - inp = {} - key = (name or "tool").strip() - low = key.lower() - if low in {"bash", "run_terminal_cmd", "shell"}: - cmd = _short(inp.get("command") or inp.get("cmd") or "") - return f"$ {cmd}" if cmd else key - path = inp.get("path") or inp.get("file") or inp.get("file_path") - if path: - return f"{key} {path}" - return f"{key} {_short(inp, 120)}" - - def extract_session_id(ev: dict) -> str | None: for key in ("session_id", "sessionId"): sid = ev.get(key) @@ -110,7 +192,10 @@ def backend_name() -> str: return (env("AGENT_HUMAN_STREAM_BACKEND", "CLAUDE_HUMAN_STREAM_BACKEND") or "claude").strip().lower() -def append_registry(event: str, session_id: str | None) -> None: +_MODEL_SEEN: str | None = None + + +def append_registry(event: str, session_id: str | None, extra: dict | None = None) -> None: path = env("AGENT_HUMAN_STREAM_REGISTRY", "CLAUDE_HUMAN_STREAM_REGISTRY") if not path: return @@ -129,9 +214,11 @@ def append_registry(event: str, session_id: str | None) -> None: "resume_from": env("AGENT_HUMAN_STREAM_RESUME_FROM", "CLAUDE_HUMAN_STREAM_RESUME_FROM") or None, "live_log": env("AGENT_HUMAN_STREAM_LIVE_LOG", "CLAUDE_HUMAN_STREAM_LIVE_LOG") or None, - "model": env("AGENT_HUMAN_STREAM_MODEL", "CLAUDE_HUMAN_STREAM_MODEL") or None, + "model": env("AGENT_HUMAN_STREAM_MODEL", "CLAUDE_HUMAN_STREAM_MODEL") or _MODEL_SEEN, "effort": env("AGENT_HUMAN_STREAM_EFFORT", "CLAUDE_HUMAN_STREAM_EFFORT") or None, } + if extra: + rec.update(extra) with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n") except OSError: @@ -157,25 +244,68 @@ def _acp_tool_line(ev: dict) -> str: return f"{line} ({status})" if status and status != "in_progress" else line +def usage_line(ev: dict) -> str | None: + usage = ev.get("usage") if isinstance(ev.get("usage"), dict) else {} + bits = [] + if ev.get("num_turns") is not None: + bits.append(f"turns={ev.get('num_turns')}") + if ev.get("duration_ms") is not None: + bits.append(f"wall={round(ev['duration_ms'] / 1000)}s") + if ev.get("duration_api_ms") is not None: + bits.append(f"api={round(ev['duration_api_ms'] / 1000)}s") + if ev.get("total_cost_usd") is not None: + bits.append(f"cost=${ev['total_cost_usd']:.4f}") + if usage.get("input_tokens") is not None: + bits.append(f"in={usage.get('input_tokens')}") + if usage.get("output_tokens") is not None: + bits.append(f"out={usage.get('output_tokens')}") + if usage.get("cache_read_input_tokens"): + bits.append(f"cached={usage.get('cache_read_input_tokens')}") + if ev.get("stop_reason") or ev.get("stopReason"): + bits.append(f"stop={ev.get('stop_reason') or ev.get('stopReason')}") + return "📊 " + " ".join(bits) if bits else None + + +def usage_fields(ev: dict) -> dict: + usage = ev.get("usage") if isinstance(ev.get("usage"), dict) else {} + return { + "num_turns": ev.get("num_turns"), + "duration_ms": ev.get("duration_ms"), + "total_cost_usd": ev.get("total_cost_usd"), + "input_tokens": usage.get("input_tokens"), + "output_tokens": usage.get("output_tokens"), + "cache_read_input_tokens": usage.get("cache_read_input_tokens"), + "stop_reason": ev.get("stop_reason") or ev.get("stopReason"), + "is_error": bool(ev.get("is_error")), + } + + def format_event(ev: dict) -> tuple[list[str], str | None]: """Return (human lines, final-result-if-terminal).""" + global _MODEL_SEEN typ = ev.get("type") lines: list[str] = [] final: str | None = None if typ == "assistant": msg = ev.get("message") or {} + if isinstance(msg.get("model"), str) and not _MODEL_SEEN: + _MODEL_SEEN = msg["model"] bit = text_bits(msg.get("content")) if bit: - lines.extend(f"🤖 {row}" for row in bit.splitlines() if row) + for row in bit.splitlines(): + if not row: + continue + lines.append(row if row.startswith(("🔧", "(thinking)")) else f"🤖 {row}") elif typ == "user": msg = ev.get("message") or {} content = msg.get("content") if isinstance(content, list): for c in content: if isinstance(c, dict) and c.get("type") == "tool_result": - out = _short(c.get("content") or "", 200) - lines.append(f"📎 tool → {out}") + out = _short(decode_tool_result(c.get("content")), TOOL_OUT_WIDTH) + marker = "❌ tool" if c.get("is_error") else "📎 tool" + lines.append(f"{marker} → {out}") elif typ == "result": final = ev.get("result") if final is None and isinstance(ev.get("content"), str): @@ -187,32 +317,33 @@ def format_event(ev: dict) -> tuple[list[str], str | None]: }: err = ev.get("errors") or ev.get("message") or final lines.append(f"❌ { _short(err, 240)}") + ul = usage_line(ev) + if ul: + lines.append(ul) elif typ == "system": subtype = ev.get("subtype") or "" if subtype == "init": model = ev.get("model") or "" + if isinstance(model, str) and model: + _MODEL_SEEN = model cwd = ev.get("cwd") or "" bits = [b for b in (f"model={model}" if model else "", f"cwd={cwd}" if cwd else "") if b] + tools = ev.get("tools") + if isinstance(tools, list): + bits.append(f"tools={len(tools)}") + mcp = ev.get("mcp_servers") + if isinstance(mcp, list): + ok = sum(1 for m in mcp if isinstance(m, dict) and m.get("status") == "connected") + bits.append(f"mcp={ok}/{len(mcp)}") lines.append("init " + " ".join(bits) if bits else "init") elif subtype == "compact_boundary": lines.append("compact") - elif typ == "text": - data = ev.get("data") - if isinstance(data, str) and data.strip(): - for row in data.splitlines(): - if row: - lines.append(f"🤖 {row}") - final = data - elif typ == "thought": - th = _short(ev.get("data") or "", 80) - if th: - lines.append(f"(thinking) {th}") elif typ == "tool_call": lines.append(f"🔧 {_acp_tool_line(ev)}") elif typ == "tool_call_update": status = ev.get("status") or "update" out = ev.get("rawOutput") - summary = _short(out, 200) if out not in (None, "", [], {}) else "" + summary = _short(decode_tool_result(out) if out not in (None, "", [], {}) else "", TOOL_OUT_WIDTH) if summary: lines.append(f"📎 tool → {summary}") elif status and status != "in_progress": @@ -221,6 +352,9 @@ def format_event(ev: dict) -> tuple[list[str], str | None]: final = ev.get("result") if final is None and isinstance(ev.get("data"), str): final = ev.get("data") + ul = usage_line(ev) + if ul: + lines.append(ul) elif typ == "error": lines.append(f"❌ {_short(ev.get('message') or ev, 240)}") final = ev.get("message") @@ -233,11 +367,45 @@ def format_event(ev: dict) -> tuple[list[str], str | None]: return lines, final +class DeltaBuffer: + """Coalesce ACP per-token ``thought`` / ``text`` deltas into lines.""" + + def __init__(self) -> None: + self.kind: str | None = None + self.buf = "" + + def push(self, kind: str, data: str) -> list[str]: + out: list[str] = [] + if self.kind and self.kind != kind: + out.extend(self.flush()) + self.kind = kind + self.buf += data + while "\n" in self.buf: + row, self.buf = self.buf.split("\n", 1) + if row.strip(): + out.append(self._render(row)) + return out + + def _render(self, row: str) -> str: + if self.kind == "thought": + return f"(thinking) {_short(row, THINK_WIDTH)}" + return f"🤖 {row}" + + def flush(self) -> list[str]: + out = [] + if self.buf.strip(): + out.append(self._render(self.buf)) + self.buf = "" + return out + + def main() -> None: final = None session_id: str | None = None announced = False - last_text = None + last_text_parts: list[str] = [] + usage_extra: dict = {} + deltas = DeltaBuffer() global _live_fp live_path = env("AGENT_HUMAN_STREAM_LIVE_LOG", "CLAUDE_HUMAN_STREAM_LIVE_LOG").strip() @@ -269,12 +437,22 @@ def main() -> None: live_emit(f"… {_short(ev, 200)}") continue + typ = ev.get("type") + if typ in {"thought", "text"} and isinstance(ev.get("data"), str): + for row in deltas.push(typ, ev["data"]): + live_emit(row) + if typ == "text": + last_text_parts.append(ev["data"]) + continue + for row in deltas.flush(): + live_emit(row) + sid = extract_session_id(ev) if sid and not session_id: session_id = sid print_session_id(session_id) announced = True - append_registry("session", session_id) + append_registry("session", session_id, {"model": _MODEL_SEEN or env("AGENT_HUMAN_STREAM_MODEL") or None}) elif sid and session_id and sid != session_id: session_id = sid print_session_id(session_id) @@ -283,13 +461,15 @@ def main() -> None: rows, maybe_final = format_event(ev) for row in rows: live_emit(row) - if ev.get("type") == "text" and isinstance(ev.get("data"), str): - last_text = ev.get("data") - if maybe_final is not None and ev.get("type") in {"result", "end", "error"}: + if maybe_final is not None and typ in {"result", "end", "error"}: final = maybe_final + if typ in {"result", "end"}: + usage_extra = usage_fields(ev) - if final is None: - final = last_text + for row in deltas.flush(): + live_emit(row) + if final is None and last_text_parts: + final = "".join(last_text_parts) live_emit("") if session_id: @@ -297,9 +477,15 @@ def main() -> None: print_session_id(session_id) else: print_session_id(session_id, again=True) - append_registry("end", session_id) + append_registry("end", session_id, usage_extra) live_emit("—— final ——") live_emit(final if final is not None else "(no result field)") + if live_path: + try: + with open(live_path + ".final.md", "w", encoding="utf-8") as f: + f.write((final if final is not None else "") + "\n") + except OSError: + pass finally: if _live_fp is not None: try: diff --git a/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.test.py b/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.test.py index 72d02da..d43501f 100755 --- a/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.test.py +++ b/sume-desk/skills/sume-main-agent-orchestration/bin/agent-human-stream.test.py @@ -1,5 +1,12 @@ #!/usr/bin/env python3 -"""Offline fixtures for agent-human-stream.py (Claude + Grok NDJSON).""" +"""Offline fixtures for agent-human-stream.py (Claude + Grok NDJSON). + +Grok shapes are captured from Grok Build 1.0.11 on 2026-09-02 (sumelabs/sume#5706): +tool results arrive as ``{"type":"Bash","output":[byte,…]}`` blobs, thinking +blocks are present, ``result`` carries usage/cost, and native ACP +``streaming-json`` streams per-token ``thought`` / ``text`` deltas with the +session id only on ``end``. +""" from __future__ import annotations import json @@ -12,111 +19,130 @@ ROOT = Path(__file__).resolve().parent FMT = ROOT / "agent-human-stream.py" +SID1 = "11111111-1111-1111-1111-111111111111" +SID2 = "01a05e1f-90bb-7561-b9a9-8f5e4f9558c6" +SID3 = "01a05f54-7a11-7eb0-8052-70abb0c35704" + CLAUDE_STREAM = [ - { - "type": "system", - "subtype": "init", - "session_id": "11111111-1111-1111-1111-111111111111", - "model": "opus", - "cwd": "/tmp/demo", - }, + {"type": "system", "subtype": "init", "session_id": SID1, "model": "opus", "cwd": "/tmp/demo"}, { "type": "assistant", - "session_id": "11111111-1111-1111-1111-111111111111", + "session_id": SID1, "message": { "role": "assistant", "content": [ {"type": "text", "text": "Checking the repo."}, - { - "type": "tool_use", - "name": "Bash", - "input": {"command": "git status"}, - }, + {"type": "tool_use", "name": "Bash", "input": {"command": "git status"}}, ], }, }, { "type": "user", - "session_id": "11111111-1111-1111-1111-111111111111", - "message": { - "role": "user", - "content": [{"type": "tool_result", "content": "clean"}], - }, - }, - { - "type": "result", - "session_id": "11111111-1111-1111-1111-111111111111", - "result": "Claude done.", + "session_id": SID1, + "message": {"role": "user", "content": [{"type": "tool_result", "content": "clean"}]}, }, + {"type": "result", "session_id": SID1, "result": "Claude done."}, ] +BASH_BYTES = list(b"=== remote ===\norigin\tgit@github.com:sumelabs/sume-com.git (fetch)\n") + GROK_MESSAGES = [ { "type": "system", "subtype": "init", - "session_id": "22222222-2222-2222-2222-222222222222", + "session_id": SID2, + "apiKeySource": "oauth", "model": "grok-4.6", - "cwd": "/tmp/demo", + "cwd": "/Users/x/sume/sume-com", + "permissionMode": "bypassPermissions", + "tools": ["run_terminal_command", "read_file", "grep", "monitor"], + "mcp_servers": [{"name": "railway", "status": "connected"}, {"name": "linear", "status": "failed"}], }, { "type": "assistant", - "session_id": "22222222-2222-2222-2222-222222222222", + "session_id": SID2, "message": { "role": "assistant", + "model": "grok-4.6", "content": [ - {"type": "text", "text": "Reading the file."}, { - "type": "tool_use", - "name": "read_file", - "input": {"path": "src/main.rs"}, + "type": "thinking", + "thinking": "I'm the Grok land babysitter for PR #5697 / issue #5696. I need to check origin/main first, then the MQ label, then draft CI, and only then decide whether to re-label.", + "signature": "x", }, + {"type": "text", "text": "Checking main first."}, + {"type": "tool_use", "name": "run_terminal_command", "input": {"command": "git remote -v | head -4"}}, + {"type": "tool_use", "name": "read_file", "input": {"target_file": "/Users/x/.agents/skills/sume-gt-mq/SKILL.md", "limit": 150}}, + {"type": "tool_use", "name": "grep", "input": {"pattern": "avatar-videos.idempotency", "glob": "**/*.ts"}}, + {"type": "tool_use", "name": "monitor", "input": {"command": "while true; do gh pr view 5697; sleep 60; done"}}, + ], + }, + }, + { + "type": "user", + "session_id": SID2, + "message": { + "role": "user", + "content": [ + {"type": "tool_result", "content": json.dumps({"type": "Bash", "output": BASH_BYTES, "exit_code": 0})}, + {"type": "tool_result", "content": json.dumps({"type": "ReadFile", "FileContent": {"content": "1→---\nname: sume-gt-mq\n"}})}, + {"type": "tool_result", "content": json.dumps({"type": "GrepSearch", "stdout": list(b"apps/api/vitest.config.ts:3\n")})}, + {"type": "tool_result", "content": json.dumps({"type": "Monitor", "taskId": "01a05e22-5af1", "timeoutMs": 3600000, "persistent": False})}, + {"type": "tool_result", "is_error": True, "content": json.dumps({"type": "ReadFile", "FileNotFound": "Error: /nope does not exist."})}, ], }, }, { "type": "result", - "session_id": "22222222-2222-2222-2222-222222222222", - "result": "Grok messages done.", + "subtype": "success", + "is_error": False, + "duration_ms": 51965, + "duration_api_ms": 10224, + "num_turns": 3, + "result": "Watching until origin/main has (#5697).", + "stop_reason": "end_turn", + "total_cost_usd": 0.0186728, + "usage": {"input_tokens": 36334, "output_tokens": 222, "cache_read_input_tokens": 71680}, + "session_id": SID2, }, ] GROK_ACP = [ - {"type": "thought", "data": "Looking around."}, - { - "type": "tool_call", - "toolCallId": "call_1", - "toolName": "run_terminal_cmd", - "status": "in_progress", - "rawInput": {"command": "ls"}, - "sessionId": "33333333-3333-3333-3333-333333333333", - }, - { - "type": "tool_call_update", - "toolCallId": "call_1", - "status": "completed", - "rawOutput": {"lines": 3}, - }, - {"type": "text", "data": "Here is the listing."}, + {"type": "available_commands", "tools": ["run_terminal_command"], "commands": ["compact"]}, + {"type": "thought", "data": "The"}, + {"type": "thought", "data": " user"}, + {"type": "thought", "data": " wants"}, + {"type": "thought", "data": " PONG.\n"}, + {"type": "tool_call", "toolCallId": "call_1", "toolName": "run_terminal_command", "status": "in_progress", "rawInput": {"command": "ls"}}, + {"type": "tool_call_update", "toolCallId": "call_1", "status": "completed", "rawOutput": {"type": "Bash", "output": list(b"a.txt\nb.txt\n")}}, + {"type": "text", "data": "P"}, + {"type": "text", "data": "ONG"}, + {"type": "usage", "usage": {"input_tokens": 35564, "output_tokens": 28, "reasoning_tokens": 22}}, { "type": "end", - "sessionId": "33333333-3333-3333-3333-333333333333", - "result": "ACP done.", "stopReason": "end_turn", + "sessionId": SID3, + "requestId": "r1", + "usage": {"input_tokens": 35564, "output_tokens": 28}, + "num_turns": 1, + "total_cost_usd": 0.0121312, }, ] -def run_stream(events: list[dict], backend: str) -> str: +def run_stream(events: list[dict], backend: str) -> tuple[str, str, str]: payload = "".join(json.dumps(ev) + "\n" for ev in events) env = os.environ.copy() env["AGENT_HUMAN_STREAM_BACKEND"] = backend env["AGENT_HUMAN_STREAM_NAME"] = "self-test" - env.pop("AGENT_HUMAN_STREAM_LIVE_LOG", None) - env.pop("CLAUDE_HUMAN_STREAM_LIVE_LOG", None) - env.pop("AGENT_HUMAN_STREAM_REGISTRY", None) - env.pop("CLAUDE_HUMAN_STREAM_REGISTRY", None) + env["AGENT_HUMAN_STREAM_PID"] = "4242" + for key in ("AGENT_HUMAN_STREAM_MODEL", "CLAUDE_HUMAN_STREAM_MODEL"): + env.pop(key, None) with tempfile.TemporaryDirectory() as tmp: - env["AGENT_HUMAN_STREAM_LIVE_LOG"] = str(Path(tmp) / "live.log") + live = Path(tmp) / "live.log" + reg = Path(tmp) / "reg.jsonl" + env["AGENT_HUMAN_STREAM_LIVE_LOG"] = str(live) + env["AGENT_HUMAN_STREAM_REGISTRY"] = str(reg) proc = subprocess.run( [sys.executable, "-u", str(FMT)], input=payload, @@ -125,9 +151,12 @@ def run_stream(events: list[dict], backend: str) -> str: env=env, check=False, ) - if proc.returncode != 0: - raise SystemExit(f"formatter failed ({backend}): {proc.stderr}") - return proc.stdout + if proc.returncode != 0: + raise SystemExit(f"formatter failed ({backend}): {proc.stderr}") + final_file = live.with_name(live.name + ".final.md") + final = final_file.read_text() if final_file.exists() else "" + registry = reg.read_text() if reg.exists() else "" + return proc.stdout, final, registry def require(haystack: str, needle: str) -> None: @@ -135,30 +164,57 @@ def require(haystack: str, needle: str) -> None: raise SystemExit(f"missing {needle!r} in:\n{haystack}") +def forbid(haystack: str, needle: str) -> None: + if needle in haystack: + raise SystemExit(f"unexpected {needle!r} in:\n{haystack}") + + def main() -> None: - claude = run_stream(CLAUDE_STREAM, "claude") - require(claude, "📎 session_id=11111111-1111-1111-1111-111111111111") + claude, final, reg = run_stream(CLAUDE_STREAM, "claude") + require(claude, f"📎 session_id={SID1}") require(claude, "backend=claude") require(claude, "🤖 Checking the repo.") - require(claude, "$ git status") + require(claude, "🔧 $ git status") require(claude, "📎 tool → clean") require(claude, "—— final ——") require(claude, "Claude done.") require(claude, "claude-human-stream --resume") + require(final, "Claude done.") + require(reg, '"model": "opus"') - grok_msg = run_stream(GROK_MESSAGES, "grok") - require(grok_msg, "📎 session_id=22222222-2222-2222-2222-222222222222") + grok_msg, final, reg = run_stream(GROK_MESSAGES, "grok") + require(grok_msg, f"📎 session_id={SID2}") require(grok_msg, "backend=grok") - require(grok_msg, "read_file src/main.rs") - require(grok_msg, "Grok messages done.") + require(grok_msg, "init model=grok-4.6 cwd=/Users/x/sume/sume-com tools=4 mcp=1/2") + require(grok_msg, "(thinking) I'm the Grok land babysitter for PR #5697 / issue #5696. I need to check origin/main first, then the MQ label") + require(grok_msg, "🔧 $ git remote -v | head -4") + require(grok_msg, "🔧 read_file /Users/x/.agents/skills/sume-gt-mq/SKILL.md") + require(grok_msg, "🔧 grep /avatar-videos.idempotency/ **/*.ts") + require(grok_msg, "⏳ monitor (background; ends the -p turn)") + require(grok_msg, "📎 tool → === remote === origin\tgit@github.com:sumelabs/sume-com.git (fetch)") + forbid(grok_msg, '"output":[61') + require(grok_msg, "📎 tool → 1→--- name: sume-gt-mq") + require(grok_msg, "📎 tool → apps/api/vitest.config.ts:3") + require(grok_msg, "📎 tool → monitor task 01a05e22-5af1 started") + require(grok_msg, "❌ tool → not found: Error: /nope does not exist.") + require(grok_msg, "📊 turns=3 wall=52s api=10s cost=$0.0187 in=36334 out=222 cached=71680 stop=end_turn") + require(grok_msg, "Watching until origin/main has (#5697).") require(grok_msg, "agent-human-stream --backend grok --resume") - - grok_acp = run_stream(GROK_ACP, "grok") - require(grok_acp, "📎 session_id=33333333-3333-3333-3333-333333333333") - require(grok_acp, "(thinking) Looking around.") - require(grok_acp, "$ ls") - require(grok_acp, "📎 tool → {'lines': 3}") - require(grok_acp, "ACP done.") + require(final, "Watching until origin/main has (#5697).") + require(reg, '"model": "grok-4.6"') + require(reg, '"total_cost_usd": 0.0186728') + require(reg, '"event": "end"') + + grok_acp, final, reg = run_stream(GROK_ACP, "grok") + require(grok_acp, f"📎 session_id={SID3}") + require(grok_acp, "(thinking) The user wants PONG.") + forbid(grok_acp, "(thinking) The\n") + require(grok_acp, "🔧 $ ls") + require(grok_acp, "📎 tool → a.txt b.txt") + require(grok_acp, "🤖 PONG") + require(grok_acp, "📊 turns=1 cost=$0.0121 in=35564 out=28 stop=end_turn") + require(grok_acp, "—— final ——\nPONG") + require(final, "PONG") print("agent-human-stream self-test: ok (claude + grok messages + grok acp)")