The smallest useful system.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install cosmonapse httpx python-dotenv 'mcp<2'
# uv on PATH - the web Effector runs duckduckgo-mcp-server via uvx
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/16-rag-cli
$ python cli.py "what is the raft consensus algorithm"   # one-shot
$ python cli.py                                          # REPL: :memory, :web <q>, :quit
$ python smoke_test.py                                   # offline end-to-end check

? what is the raft consensus algorithm
  ~ recall 'what is the raft consensus algorithm'
    -> 0 hit(s)
  * search {"query": "what is the raft consensus algorithm", "max_results": 5}
  * fetch  {"url": "https://raft.github.io/", "max_length": 6000}
  + imprint 24

answer  [web, 9.4s, 24 chunk(s) indexed from 3 page(s)]
Raft is a consensus algorithm designed to be understandable ... [1][3]

? how does raft elect a leader
  ~ recall 'how does raft elect a leader'
    -> 5 hit(s)

answer  [memory, 1.1s]
A leader election begins when a follower's election timeout ... [2]

16-rag-cli/
  config.py              settings + the llm() factory
  neurons/rag.py         THE Neuron    a stock model + three decorators
  engram/web_memory.py   THE Engram    BM25 over page chunks, via Engram.serve()
  effector/web.py        THE Effector  search + fetch, via Effector.serve()
  brain.py               three Dendrites + ask()
  cli.py                 the terminal
  smoke_test.py          offline end-to-end check

AXON = Axon(
    neuron_id="rag", neuron_fn=llm(), capabilities=["rag"],
    # "web" the memory. directed_id is what lands in the RECALL / IMPRINT
    # envelope, and what the hooks below address.
    engrams=[EngramBinding(name="web", directed_id=ENGRAM_ID)],
    # "web" the tool, with its reachable surface pinned to two names.
    effectors=[EffectorBinding(name="web", directed_id=EFFECTOR_ID,
                               tools=("search", "fetch"))],
    # THE GATE: effectors= is refused without a tool_standard, because the
    # standard is how an Axon would recognise a tool call in a model's raw
    # output. This model answers in prose and the hook issues the calls, so
    # the parser never matches - the standard keeps the binding legal and
    # leaves the door open for a model-driven variant.
    tool_standard="codex",
)


@AXON.host.on_agent_output(neuron="rag")
async def conclude(sig):
    """FINAL resolves the CLI's Pathway. A longer chain would dispatch the
    next TASK here instead - see 14-agent, where this same decorator hands
    work from the planner to the research and coding neurons."""
    await AXON.dendrite.emit_final(
        trace_id=sig.trace_id, parent_id=sig.id, neuron="rag",
        result=sig.payload.get("output", {}),
    )

ENGRAM = Engram.serve(engram_id=ENGRAM_ID, engram_kind=ENGRAM_KIND,
                      capabilities=["bm25", "keyword", "merge_key"])


@ENGRAM.serves
def only_text(query: dict) -> bool:
    """can_serve: decline a vector query if one is ever routed by kind
    rather than by id. The hosting Dendrite skips responding on False."""
    return "text" in query


@ENGRAM.on_recall
async def search(query: dict, *, deadline_ms=None, min_confidence=None):
    """RECALL -> these Hits become the RECALLED payload."""
    STATS["recalls"] += 1
    qterms = query_terms(str(query.get("text", "")))
    n = len(_ENTRIES)
    if not qterms or n == 0:
        return []
    top_k = int(query.get("top_k", 5))
    avgdl = sum(sum(tf.values()) for tf in _TF.values()) / n
    t0, scored = time.monotonic(), []
    for eid, entry in _ENTRIES.items():
        tf = _TF.get(eid, Counter())
        dl = sum(tf.values()) or 1
        score = 0.0
        for term in qterms:
            f = tf.get(term, 0)
            if not f:
                continue
            idf = math.log(1 + (n - _DF[term] + 0.5) / (_DF[term] + 0.5))
            score += idf * (f * (K1 + 1)) / (f + K1 * (1 - B + B * dl / avgdl))
        if score <= 0.0:
            continue
        if min_confidence is not None and score < min_confidence:
            continue
        scored.append(Hit(id=eid, entry=dict(entry), score=score))
        if deadline_ms and (time.monotonic() - t0) * 1000 > deadline_ms:
            break
    scored.sort(key=lambda h: h.score, reverse=True)
    return scored[:top_k]


@ENGRAM.on_imprint
async def write(op: str, entry: dict, *, merge_key=None):
    """IMPRINT -> this id becomes the IMPRINTED receipt.

    The cap runs here because this handler owns the write. Returning None
    falls through, so an op nothing handles reports ``unhandled imprint
    op`` instead of silently succeeding."""
    if op in ("add", "append", "upsert", "merge"):
        eid = _put(entry, merge_key)
        _cap()
        return eid
    if op == "delete":
        eid = _KEYS.get(merge_key) if merge_key else entry.get("id")
        if eid:
            _drop(eid)
        return eid
    return None

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


def _shape(tool: str, a: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    """Caller-facing name -> the server's own name and argument shape."""
    if tool == "search":
        return "search", {"query": str(a.get("query", "")),
                          "max_results": int(a.get("max_results", 5))}
    if tool == "fetch":
        return "fetch_content", {"url": str(a.get("url", "")),
                                 "max_length": int(a.get("max_length", 6000))}
    raise ValueError(f"unknown tool {tool!r}; expected search | fetch")


@EFFECTOR.on_tool_call
async def handle(tool: str, args: dict[str, Any], *, trace_id: str | None = None):
    """THE tool. Whatever this returns becomes the TOOL_RESULT; a raise
    becomes ``error`` on it, and a tool error never terminates the parent
    TASK. ``trace_id`` is injected because it is declared - so is
    ``call_id`` and ``deadline_ms`` if a handler asks for them."""
    args = dict(args or {})

    # Within one trace, the same URL is only ever read once.
    if tool == "fetch" and trace_id:
        cached = _SEEN.get(trace_id, {}).get(str(args.get("url", "")))
        if cached is not None:
            return {"response": cached, "cached": True}

    server_tool, server_args = _shape(tool, args)
    out = await _MCP({"tool": server_tool, "arguments": server_args}, [])

    body = str(out.get("response") or "")
    if out.get("is_error"):
        raise RuntimeError(body or f"{tool} failed")

    # The server reports failures as PROSE with a 200 - an empty search, a
    # 403, a timeout - so is_error stays False. Left alone, that error text
    # gets chunked and imprinted as though it were page content, and then
    # ranks against real passages. Promote them to real tool errors so the
    # caller skips the page instead of learning it.
    if body.startswith("Error:") or body.startswith(
            "An error occurred while searching:"):
        raise RuntimeError(body[:200])
    if tool == "search" and body.startswith("No results were found"):
        raise RuntimeError(body[:200])

    if tool == "fetch" and trace_id:
        _SEEN.setdefault(trace_id, {})[str(args.get("url", ""))] = body
    return out


@EFFECTOR.host.on_final
async def forget(sig) -> None:
    """Host-side: the trace is over, drop its fetch memo. Without this the
    dict grows for the life of the process - the reason a per-trace cache
    wants a per-trace cleanup signal, and the reason this decorator
    exists on the action side at all."""
    _SEEN.pop(sig.trace_id, None)

from __future__ import annotations

from cosmonapse import Dendrite, SignalType

from config import HEARTBEAT_S, NAMESPACE, TOP_K
from effector import web
from engram import web_memory
from neurons import rag


def build_rag(synapse):
    """Stand the three Dendrites up on a Synapse. Returns (dendrites, cli).

    ``attach_engram`` / ``attach_effector`` / ``attach_axon`` is the whole
    wiring API - no registry, no handler tables, no manual subscriptions.
    The Axon's chain handler self-registers on announce
    (@AXON.host.on_agent_output, declared in neurons/rag.py), so nothing
    below hand-wires an @node.on_* handler. Effectors and Engrams need no
    registration at all - attaching is enough.

    Liveness is deliberately quiet: a demo does not need 30s beats, so
    `cosmo prism --tail` shows the chain, not chatter.
    """
    liveness = {"heartbeat_s": HEARTBEAT_S, "reregister_on_heartbeat": False}

    host = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="engram-host", role="worker", **liveness)
    host.attach_engram(web_memory.ENGRAM)   # REGISTERs + services it

    tools = Dendrite(synapse=synapse, namespace=NAMESPACE,
                     dendrite_id="web-node", role="worker", **liveness)
    tools.attach_effector(web.EFFECTOR)   # TOOL_CALLs are not role-gated

    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="rag-node", role="orchestrator", **liveness)
    node.attach_axon(rag.AXON)

    return [host, tools, node], node


async def ask(cli, question: str, *, top_k: int = TOP_K,
              force_web: bool = False, timeout_s: float = 180.0) -> dict:
    """One TASK out, one FINAL back. The entire edge of the system."""
    sig = await cli.dispatch_and_wait(
        capabilities=["rag"],
        input={"question": question, "top_k": top_k, "force_web": force_web},
        scope="terminal",       # resolve on FINAL / ERROR only
        finalize=False,         # the Axon's chain handler owns FINAL
        timeout_s=timeout_s,
    )
    if sig.type is SignalType.ERROR:
        raise RuntimeError(sig.payload.get("message") or "ask failed")
    return sig.payload["result"]

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

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python cli.py "what is raft"

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

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python cli.py "what is raft"
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python cli.py "what is raft"