An agent with no loop.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install 'cosmonapse[receptor]' httpx python-dotenv 'mcp<2'
# uv (for uvx) and Node 18+ on PATH - the tools are MCP servers
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/14-agent
$ python brain.py "Research the Collatz conjecture, then write a Python CLI that prints the sequence for N"
$ python brain.py --stream "..."     # every Signal of the chain as it happens
$ python brain.py memory             # what it remembers this session
$ python brain.py                    # REPL, plus the chat page at http://127.0.0.1:8000

  . get_current_time
  . search
  . fetch
  . get_current_time
  . write_file
  . get_current_time
# Research the Collatz conjecture, then write a Python CLI ...

## Research note 1
The Collatz conjecture asks whether repeatedly applying n/2 or 3n+1 ...

## Generated solution
Saved to `.../14-agent/report/solution.py`

## Sources
- https://en.wikipedia.org/wiki/Collatz_conjecture

--- 2 step(s), source: web
report : report/answer.md
code   : report/solution.py

14-agent/
  config.py                 settings, hf_token()
  neurons/model/planner.py  stock LLM Neuron + hooks + its chain handler
  neurons/model/research.py search, fetch, note, hand back
  neurons/model/coding.py   write the script, save it, hand back
  effector/                 websearch, fetch, clock, files - MCP servers as Effectors
  engram/agent_memory.py    the memory + its on_imprint_signal mirror to disk
  receptors/terminal.py     CliReceptor - a goal becomes one TASK
  receptors/chat.py         ChatReceptor - the same TASK from a browser
  helpers.py                MCPEffector, task_input(), result_of()
  brain.py                  a node per agent, per tool group, per interface

from cosmonapse import CliReceptor, SignalType

from config import MAX_STEPS
from engram import agent_memory
from helpers import memory_summary, result_of, task_input

RECEPTOR = CliReceptor(
    capabilities=["planner"],
    prog="agent",
    description="Drive the choreographed Cosmonapse agent from a terminal.",
    banner="agent - type a goal, `memory`, :help, or :quit",
    prompt="goal> ",
    scope="terminal",
    finalize=False,
    timeout_s=600.0,
)


@RECEPTOR.command(help="run the agent on a goal")
def run(goal: str, max_steps: int = MAX_STEPS):
    """ONE capability-routed TASK goes out; the nodes chain the rest."""
    return task_input(goal, max_steps)


@RECEPTOR.command("memory", local=True, help="what the agent remembers")
def memory():
    """local=True: answered right here, nothing crosses the wire."""
    return memory_summary(agent_memory.ENGRAM)


@RECEPTOR.on_signal(SignalType.TOOL_CALL)
async def show_tool(sig):
    """The progress channel - observation only, nothing here changes the trace."""
    print(f"  . {(sig.payload or {}).get('tool', '?')}")


@RECEPTOR.on_result
def render(sig):
    """FINAL -> what the terminal prints: the report, then the artefacts."""
    result = result_of(sig)
    steps = result.get("steps") or []
    lines = [(result.get("report") or "").rstrip(), "",
             f"--- {len(steps)} step(s), source: {result.get('source', '?')}"]
    if result.get("report_path"):
        lines.append(f"report : {result['report_path']}")
    if result.get("code_path"):
        lines.append(f"code   : {result['code_path']}")
    return "\n".join(lines)

AXON = Axon.huggingface(
    neuron_id="planner",
    capabilities=["planner"],
    endpoint="https://router.huggingface.co",
    model="meta-llama/Llama-3.1-8B-Instruct",
    use_chat_api=True,
    max_new_tokens=1024,
    temperature=0.2,
    recognize=False,
    api_key="dummy",
)


@AXON.before_task
async def situate(input):
    """Cache check + progress recall + clock grounding, then the prompt."""
    tid, _ = ambient_trace()
    d = AXON.dendrite

    # A previously remembered answer short-circuits the whole run.
    cached = await d.recall(engram_id=ENGRAM_ID,
                            query={"merge_key": f"answer:{input['goal']}"},
                            deadline_ms=2000)
    if cached.hits:
        _PENDING[tid] = {**input, "cached": cached.hits[0].entry["content"]}
        return {"messages": [{"role": "user", "content": 'Reply with {}'}]}

    # Progress is not carried by any node - it is recalled from the engram.
    done = await d.recall(engram_id=ENGRAM_ID,
                          query={"tag": input["tag"], "top_k": 50},
                          deadline_ms=2000)
    notes = sum(1 for h in done.hits if "research" in h.entry.get("tags", []))
    coded = any("code" in h.entry.get("tags", []) for h in done.hits)

    # The clock only *grounds* the prompt - it is not load-bearing, so a
    # dead Effector degrades the answer instead of ending the run. A tool
    # error rides its TOOL_RESULT and never terminates the TASK by itself;
    # what would end it is raising here, in the hook. Compare research and
    # coding, which DO raise: without a search result or a written file
    # there is nothing for those neurons to be right about.
    clock = await d.call_tool(effector_id="clock-effector",
                              tool="get_current_time",
                              args={"timezone": "UTC"}, deadline_ms=30_000)
    if clock.error:
        print(f"! clock unavailable ({clock.error}) - planning without it")
        now = "unknown"
    else:
        out = clock.result or {}
        now = out.get("response") or str(out.get("result"))
    _PENDING[tid] = {**input, "now": now, "notes": notes, "code_written": coded}

    state = (f"research notes: {notes}; "
             f"code written: {'yes' if coded else 'no'}")
    return {"messages": [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": (
            f"Goal: {input['goal']}\nUTC time: {now}\n"
            f"Step {input['step']}/{input['max_steps']}\n"
            f"Progress so far -> {state}\n\nNext subtask?"
        )},
    ]}


@AXON.detects_output
def decide(raw):
    """Parse the model's JSON; force finish past max_steps; echo the chain."""
    tid, _ = ambient_trace()
    st = _PENDING.pop(tid, {})
    chain = {k: st.get(k) for k in _CHAIN_KEYS}
    if st.get("cached"):
        return {"route": "finish", "task": "", "cached": st["cached"], **chain}

    decision = _first_json(raw.get("response")) or {}
    route = (decision.get("route") or "").lower()
    task = (decision.get("task") or "").strip()
    if st.get("step", 1) > st.get("max_steps", 0):
        route, task = "finish", ""
    elif route not in {"research", "coding", "finish"}:
        route, task = (
            ("research", st.get("goal", "")) if not st.get("notes")
            else ("coding", st.get("goal", "")) if not st.get("code_written")
            else ("finish", "")
        )
    elif (route == "research" and st.get("notes")
          and not st.get("code_written")
          and st.get("step", 1) >= st.get("max_steps", 0)):
        # Last step and still no code: a small model loves to keep
        # researching - spend the final step writing the script instead.
        route, task = "coding", st.get("goal", "")
    return {"route": route, "task": task, **chain}

@AXON.host.on_agent_output(neuron="planner")
async def chain(sig):
    """The planner node's chain behaviour - the hosting Dendrite picks the
    decision up and creates the next TASK, or concludes the trace."""
    node = AXON.dendrite
    d = sig.payload.get("output", {})
    route, goal, tag = d.get("route"), d.get("goal"), d.get("tag")

    if route in ("research", "coding"):
        steps = (d.get("steps") or []) + [{"route": route,
                                           "task": d.get("task", "")}]
        await node.dispatch_task(
            capabilities=[route],
            input={"task": d.get("task", ""), "goal": goal, "tag": tag,
                   "step": d.get("step", 1),
                   "max_steps": d.get("max_steps", 0),
                   "steps": steps, "sources": d.get("sources") or []},
            trace_id=sig.trace_id, parent_id=sig.id,
        )
        return

    # finish - straight from the cache ...
    if d.get("cached"):
        await node.emit_final(
            trace_id=sig.trace_id, parent_id=sig.id, neuron="planner",
            result={"report": d["cached"], "source": "memory",
                    "steps": [], "trace_id": tag},
        )
        return

    # ... or assembled from this run's engram entries.
    got = await node.recall(engram_id=ENGRAM_ID,
                            query={"tag": tag, "top_k": 50},
                            deadline_ms=2000,
                            trace_id=sig.trace_id, parent_id=sig.id)
    notes = [h.entry["content"] for h in got.hits
             if "research" in h.entry.get("tags", [])]
    code = next((h.entry["content"].removeprefix("[code]\n")
                 for h in got.hits if "code" in h.entry.get("tags", [])),
                None)
    code_path = f"{OUT_DIR}/solution.py" if code else None

    # The report is the ANSWER, not the artefact: research prose in full,
    # and for the code a pointer to where the coding Neuron actually saved
    # it. The script is already on disk next to answer.md, so inlining it
    # here would only duplicate it - and a chat turn wants the finding plus
    # a path, not a screenful of Python.
    parts = [f"# {goal}", f"_generated {d.get('now')}_", ""]
    for n, note in enumerate(notes, 1):
        parts += [f"## Research note {n}", note, ""]
    if code:
        parts += ["## Generated solution",
                  f"Saved to `{ROOT / code_path}`",
                  f"({len(code.splitlines())} lines, written by the coding "
                  f"Neuron through the files Effector.)", ""]
    sources = d.get("sources") or []
    if sources:
        parts += ["## Sources", *[f"- {url}" for url in sources], ""]
    report = "\n".join(parts)
    report_path = f"{OUT_DIR}/answer.md"

    await node.imprint(engram_id=ENGRAM_ID, op="upsert",
                       merge_key=f"answer:{goal}",
                       entry={"content": report, "tags": ["answer"]},
                       meta={"path": report_path},
                       await_ack=True, deadline_ms=2000,
                       trace_id=sig.trace_id, parent_id=sig.id)
    await node.emit_final(
        trace_id=sig.trace_id, parent_id=sig.id, neuron="planner",
        result={"report": report, "code_path": code_path,
                "report_path": report_path, "source": "web",
                "steps": d.get("steps") or [],
                "sources": d.get("sources") or [], "trace_id": tag},
    )

@AXON.before_task
async def gather(input):
    """Gather web context via MY tools, shaped into the prompt."""
    tid, _ = ambient_trace()
    d = AXON.dendrite
    task = input["task"]

    search_out = await d.call_tool(effector_id="websearch-effector",
                                   tool="search",
                                   args={"query": task, "max_results": 5},
                                   deadline_ms=60_000)
    if search_out.error:
        raise RuntimeError(f"websearch error: {search_out.error}")
    search = search_out.result or {}
    ctx = search.get("response", "")[:1500]
    # Full page text is a BONUS on top of the search snippets, so an
    # unreachable page is skipped rather than fatal - plenty of sites answer
    # a server-side fetch with 403, a paywall, or a robots.txt refusal, and
    # losing the whole run to the luck of the first search hit is not a
    # research strategy. Walk the hits until one comes back; if none does,
    # the snippets alone still ground the note.
    #
    # The search itself DOES raise: with neither snippets nor a page there
    # is nothing for this Neuron to be right about.
    sources = []
    for url in _urls(search.get("response")):
        page_out = await d.call_tool(effector_id="fetch-effector",
                                     tool="fetch",
                                     args={"url": url, "max_length": 2500},
                                     deadline_ms=60_000)
        if page_out.error:
            print(f"! skipping {url} ({page_out.error})")
            continue
        page = page_out.result or {}
        ctx += f"\n\n[page {url}]\n{page.get('response', '')[:2000]}"
        sources.append(url)
        break
    if not sources:
        ctx += "\n\n[no page could be fetched - snippets only]"

    _PENDING[tid] = {**input, "found": sources}
    return {"messages": [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"Subtask: {task}\n\nWeb context:\n{ctx}"},
    ]}


@AXON.detects_output
async def note_and_imprint(raw):
    """Shape the note, remember it (dendrite.imprint), echo the chain."""
    tid, _ = ambient_trace()
    st = _PENDING.pop(tid, {})
    note = (raw.get("response") or "").strip()
    await AXON.dendrite.imprint(
        engram_id=ENGRAM_ID, op="append",
        entry={"content": note, "tags": ["research", st.get("tag", "")]},
        await_ack=True, deadline_ms=2000,
    )
    return {"note": note,
            "goal": st.get("goal"), "tag": st.get("tag"),
            "step": st.get("step"), "max_steps": st.get("max_steps"),
            "steps": st.get("steps") or [],
            "sources": (st.get("sources") or []) + (st.get("found") or [])}


@AXON.host.on_agent_output(neuron="research")
async def chain(sig):
    """The research node's chain behaviour - hand back to the planner."""
    out = sig.payload.get("output", {})
    await AXON.dendrite.dispatch_task(
        capabilities=["planner"],
        input={"goal": out.get("goal"), "tag": out.get("tag"),
               "step": (out.get("step") or 0) + 1,
               "max_steps": out.get("max_steps") or 0,
               "steps": out.get("steps") or [],
               "sources": out.get("sources") or []},
        trace_id=sig.trace_id, parent_id=sig.id,
    )

from cosmonapse import Neuron

from helpers import MCPEffector

EFFECTOR = MCPEffector(
    effector_id="websearch-effector", effector_kind="websearch",
    mcp=Neuron(source="mcp", command="uvx", args=["duckduckgo-mcp-server"]),
)

from cosmonapse import InMemoryEngram

from config import ROOT

ENGRAM_ID = "agent-memory"

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


@ENGRAM.host.on_imprint_signal
async def persist_answers(sig):
    """Mirror "answer" imprints carrying a path to disk (files Effector).

    ``ENGRAM.dendrite`` is the node this Engram was attached to - the
    memory-side twin of ``AXON.dendrite`` - so the handler can call a tool
    without brain.py handing it a reference.
    """
    entry = sig.payload.get("entry") or {}
    path = (sig.meta or {}).get("path")
    if not (path and "answer" in (entry.get("tags") or [])):
        return
    out = await ENGRAM.dendrite.call_tool(
        effector_id="files-effector", tool="write_file",
        args={"path": str(ROOT / path), "content": entry.get("content", "")},
        deadline_ms=30_000)
    if out.error:
        raise RuntimeError(f"answer mirror failed: {out.error}")

from __future__ import annotations

import asyncio
import contextlib
import os
import signal

from cosmonapse import (Dendrite, MemoryRegistryStore, MemorySynapse,
                        connect_synapse, run_brain)

from config import NAMESPACE
from effector import clock, fetch, files, websearch
from engram import agent_memory
from neurons.model import coding, planner, research
from receptors import chat, terminal

# The agents' hooks AND their chain handlers are declared in their modules
# (@AXON.before_task / @AXON.detects_output / @AXON.host.on_agent_output),
# so an agent node is one line here regardless of how much it does.
AGENT_AXONS = [planner.AXON, research.AXON, coding.AXON]


def build_memory(synapse) -> Dendrite:
    """The Engram that remembers - services RECALL and IMPRINT.

    role="worker" and yet it calls the files Effector: TOOL_CALLs are not
    role-gated, only TASK and STOP are. Nothing needs orchestrator rights
    to use a tool.
    """
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="memory-node", role="worker")
    node.attach_engram(agent_memory.ENGRAM)   # @ENGRAM.host.on_* replayed here
    return node


def build_agents(synapse) -> list[Dendrite]:
    """One node per agent Neuron - planner, research, coding.

    role="orchestrator" because each one's chain handler dispatches the next
    TASK. The Neurons themselves never decide what work exists; the hosting
    Dendrite does, off the output Signal.
    """
    nodes = []
    for axon in AGENT_AXONS:
        node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                        dendrite_id=f"{axon.neuron_id}-node",
                        role="orchestrator")
        node.attach_axon(axon)                # @AXON.host.on_* on announce
        nodes.append(node)
    return nodes


def build_web(synapse) -> Dendrite:
    """websearch + fetch on ONE node - two Effectors, one attachment point.

    The Dendrite routes each TOOL_CALL to the right Effector by directed
    id/kind on its own; no host-decorator filtering needed. A Dendrite
    hosting several components is always available - it just costs the
    independence the other builders keep.
    """
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="web-node", role="worker")
    for effector in (websearch.EFFECTOR, fetch.EFFECTOR):
        node.attach_effector(effector)
    return node


def build_tools(synapse) -> list[Dendrite]:
    """The clock and filesystem Effectors, a node each.

    Effectors need no registration beyond this: ``attach_effector`` alone
    makes a Dendrite service that tool's TOOL_CALLs.
    """
    nodes = []
    for effector in (clock.EFFECTOR, files.EFFECTOR):
        node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                        dendrite_id=f"{effector.effector_kind}-node",
                        role="worker")
        node.attach_effector(effector)
        nodes.append(node)
    return nodes


def build_terminal(synapse) -> Dendrite:
    """The Receptor that listens, on the one node that may dispatch.

    Orchestrator role, and not by preference: attach_receptor and dispatch
    both refuse a role="worker" Dendrite, because a Receptor's whole job is
    to originate TASKs. That is why interfaces get their own node.

    registry_store is what makes capability routing resolve from this side -
    the run TASK names ["planner"], not a neuron id.
    """
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="terminal-node", heartbeat_s=0,
                    registry_store=MemoryRegistryStore())
    node.attach_receptor(terminal.RECEPTOR)
    return node


def build_chat(synapse) -> Dendrite:
    """A second interface onto the same brain - a served chat page.

    Its own node, for the same reason the terminal has one. Two interfaces,
    two nodes, one brain - and the planner never learns which edge asked,
    because both send the identical TASK.

    Needs FastAPI + uvicorn: pip install 'cosmonapse[receptor]'
    """
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="chat-node", heartbeat_s=0,
                    registry_store=MemoryRegistryStore())
    node.attach_receptor(chat.RECEPTOR)
    return node

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

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py "Research X, then write a Python CLI that does Y"

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

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py "Research X, then write a Python CLI that does Y"
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py "Research X, then write a Python CLI that does Y"