A claude-code-style harness.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install cosmonapse httpx python-dotenv 'mcp<2' duckduckgo-mcp-server
# uv and Node 18+ on PATH. LLM_ENDPOINT + LLM_API_KEY in .env: an OpenAI-compatible completions endpoint serving Qwen2.5-Coder

$ cd cosmonapse-examples/15-claude-harness
$ python smoke_test.py      # offline wiring check - no key, no MCP servers
$ python demo.py

❯ read hello.py, then rewrite it with argparse and a --shout flag
❯ what does PEP 723 allow? write a compliant single-file script demoing it
❯ /compact
❯ /memory

  ⏺ read {"path": "hello.py"}
  ⏺ write {"path": "hello.py", "content": "import argparse\n..."}
  ⏺ bash {"command": "python hello.py --shout"}

Rewrote hello.py with argparse; --shout upper-cases the greeting.
[4 step(s), tools: read,write,bash]

15-claude-harness/
  config.py                   settings, llm() - one endpoint, three roles
  neurons/model/assistant.py  the assistant: EffectorBindings + hermes + the chain
  neurons/model/explorer.py   the subagent - research, report back as an observation
  neurons/model/compactor.py  /compact and auto-compaction
  effector/                   files, websearch, fetch (MCP) + shell (hand-written)
  engram/session_memory.py    turns, observations, the summary + a disk mirror
  brain.py                    a node per agent and tool, run_turn(), run_compact()
  demo.py                     the REPL
  smoke_test.py               offline wiring check

AXON = Axon(
    neuron_id="assistant", neuron_fn=llm(MODEL), capabilities=["assistant"],
    tool_standard="hermes",         # Qwen's mother tongue - THE GATE
    effectors=[
        EffectorBinding(name="files", directed_id="files-effector",
                        tools=("read", "write", "edit", "ls", "glob")),
        EffectorBinding(name="shell", directed_id="shell-effector",
                        tools=("bash",),
                        default_deadline_ms=int(SHELL_TIMEOUT * 1000) + 5000),
        EffectorBinding(name="websearch", directed_id="websearch-effector",
                        tools=("websearch",)),
        EffectorBinding(name="fetch", directed_id="fetch-effector",
                        tools=("fetch",)),
        # no binding for "agent" - see the module docstring
    ],
)

async def run_turn(repl, prompt: str, session: str, *,
                   max_steps: int = MAX_STEPS,
                   timeout_s: float = 600.0) -> dict:
    tag = new_trace_id()        # the turn's trace; also tags its obs entries
    await repl.imprint(         # the user's side of the transcript
        engram_id=session_memory.ENGRAM_ID, op="append",
        entry={"content": f"user: {prompt}", "tags": ["turn", session]},
        await_ack=True, deadline_ms=2000)
    sig = await repl.dispatch_and_wait(
        capabilities=["assistant"],
        input={"prompt": prompt, "session": session, "tag": tag,
               "step": 1, "max_steps": max_steps, "tools_used": []},
        trace_id=tag,
        scope="terminal",       # resolve on FINAL / ERROR only
        finalize=False,         # the assistant node owns FINAL
        timeout_s=timeout_s,
    )
    if sig.type is SignalType.ERROR:
        raise RuntimeError(f"turn failed: {sig.payload.get('message')}")
    return sig.payload["result"]

from __future__ import annotations

import asyncio

from cosmonapse import Effector, ToolOutcome

from config import SHELL_TIMEOUT, WORKSPACE

DENY = ("sudo", "shutdown", "reboot", "mkfs", "rm -rf /", ":(){", "dd if=")

EFFECTOR = Effector.serve(effector_id="shell-effector", effector_kind="shell")


@EFFECTOR.on_tool_call
async def run(tool: str, args: dict):
    if tool not in ("bash", "run"):
        return None   # unhandled -> "unhandled tool" error on TOOL_RESULT
    cmd = str(args.get("command", "")).strip()
    if not cmd:
        return ToolOutcome(tool=tool, error="no command given")
    if any(bad in cmd for bad in DENY):
        return ToolOutcome(tool=tool, error=f"command refused (denylist): {cmd}")
    try:
        proc = await asyncio.create_subprocess_shell(
            cmd, cwd=str(WORKSPACE),
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
        out, _ = await asyncio.wait_for(proc.communicate(), SHELL_TIMEOUT)
    except asyncio.TimeoutError:
        proc.kill()
        return ToolOutcome(tool=tool, error=f"timed out after {SHELL_TIMEOUT}s: {cmd}")
    text = (out or b"").decode(errors="replace")
    if proc.returncode != 0:
        return ToolOutcome(tool=tool, error=text or f"exit {proc.returncode}")
    return ToolOutcome(tool=tool, result={"response": text or "(no output)",
                                          "result": {"exit_code": proc.returncode}})

from cosmonapse import InMemoryEngram

from config import WORKSPACE

ENGRAM_ID = "session-memory"

ENGRAM = InMemoryEngram(engram_id=ENGRAM_ID, engram_kind="context")


@ENGRAM.host.on_imprint_signal
async def persist_summaries(sig):
    """Mirror "summary" imprints carrying a path to disk (filesystem)."""
    entry = sig.payload.get("entry") or {}
    path = (sig.meta or {}).get("path")
    if path and "summary" in (entry.get("tags") or []):
        outcome = await ENGRAM.dendrite.call_tool(
            effector_id="files-effector", tool="write_file",
            args={"path": str(WORKSPACE / path),
                  "content": entry.get("content", "")})
        if outcome.error:
            raise RuntimeError(f"SESSION.md mirror failed: {outcome.error}")

# terminal 1 - a dev synapse (TCP, single host)
$ cosmo synapse start memory --namespace=harness

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python demo.py

# terminal 3 - Prism, the live view (http://127.0.0.1:7071)
$ cosmo prism --url=cosmo://127.0.0.1:7070 -n harness

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python demo.py
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python demo.py