Install & run

Clone the examples repo, install, and run from inside the example's folder.

terminal

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install cosmonapse

$ cd cosmonapse-examples/04-pathway
$ python brain.py shapes                    # all three, written out by hand
$ python brain.py "ship feature X"          # wait, via the Receptor
$ python brain.py --stream "ship feature X" # iterate, via the Receptor

1) wait    -> {'plan': ['step-1', 'step-2'], 'goal': 'ship feature X'}
2) on      -> {'plan': ['step-1', 'step-2'], 'goal': 'ship feature Y'}
3) iterate -> AGENT_OUTPUT

Layout

One component per file, one Dendrite per node.

The skeleton cosmo init scaffolds. Modules under neurons/, engram/, effector/ and receptors/ declare behaviour; brain.py owns deployment - which node hosts what.

04-pathway

04-pathway/
  neurons/planner.py     a plain async function answering "plan"
  receptors/terminal.py  plan (the Receptor's default) + shapes (by hand)
  brain.py               planner-node + terminal-node

01 · The worker

One Axon serves every shape.

The shapes are all caller-side. The planner answers a TASK the same way however the caller chooses to consume the reply.

neurons/planner.py

from cosmonapse import Axon


async def planner(input, context):
    return {"plan": ["step-1", "step-2"], "goal": input.get("goal")}


AXON = Axon(neuron_id="planner", neuron_fn=planner, capabilities=["plan"])

02 · Through a Receptor

wait is the default, --stream iterates.

The Receptor routes by capability, so the terminal never names the planner. python brain.py "..." is request/reply, --stream walks every Signal on the trace, and --send fires and forgets.

receptors/terminal.py

RECEPTOR = CliReceptor(
    capabilities=["plan"],
    input_key="goal",
    prog="pathway",
    description="One Pathway, three ways to consume it.",
    timeout_s=5.0,
)


@RECEPTOR.command(help="plan a goal (routed by capability)")
def plan(goal: str):
    return {"goal": goal}

03 · By hand

await pw.wait() · @pw.on(...) · async for sig in pw.

dispatch_and_wait blocks until the first terminal Signal, closes the Pathway, and returns the Signal.
dispatch_and_subscribe returns the live Pathway; @pw.on(SignalType.X) callbacks fire for this trace only, not the whole namespace.
async for sig in pw yields every Signal on the trace as it arrives.

receptors/terminal.py

@RECEPTOR.command("shapes", local=True, help="run wait / on / iterate by hand")
async def shapes():
    orch = RECEPTOR.dendrite
    lines = []

    # 1. await pw.wait() - block until the first terminal Signal, then close.
    sig = await orch.dispatch_and_wait(
        capabilities=["plan"], input={"goal": "ship feature X"}, timeout_s=5.0,
    )
    lines.append(f"1) wait    -> {sig.payload['output']}")

    # 2. @pw.on(SignalType.X) - callbacks scoped to THIS trace only.
    pw = await orch.dispatch_and_subscribe(
        capabilities=["plan"], input={"goal": "ship feature Y"},
    )
    done = asyncio.Event()

    @pw.on(SignalType.AGENT_OUTPUT)
    async def on_output(s):
        lines.append(f"2) on      -> {s.payload['output']}")
        done.set()

    await asyncio.wait_for(done.wait(), timeout=5.0)
    await pw.close()

    # 3. async for sig in pw - walk every Signal on the trace as it arrives.
    async with await orch.dispatch(
        capabilities=["plan"], input={"goal": "ship feature Z"},
    ) as pw:
        async for s in pw:
            lines.append(f"3) iterate -> {s.type.value}")
            if s.type is SignalType.AGENT_OUTPUT:
                break

    return "\n".join(lines)

04 · Two more tools

scope="terminal" and observe_pathway.

scope="terminal" delivers only FINAL, ERROR, CLARIFICATION and PERMISSION: the caller wakes for a conclusion or a decision, while intermediate work is handled peer to peer. observe_pathway opens a Pathway in observer role on a trace someone else started, without emitting a TASK.

shapes.py

pw = await orch.dispatch(capabilities=["plan"], input={"goal": "..."},
                         scope="terminal", finalize=False)
sig = await pw.wait(timeout_s=60.0)          # FINAL / ERROR / CLARIFICATION / PERMISSION

pw = await monitor.observe_pathway(trace_id="trc_01J...")
assert pw.role == "observer"
async for sig in pw:
    print(sig.type.value)

05 · The brain

Two nodes.

brain.py

import asyncio
import contextlib
import signal

from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain

from config import NAMESPACE, SYNAPSE_URL
from neurons import planner
from receptors import terminal


def build_planner(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="planner-node", role="worker")
    node.attach_axon(planner.AXON)
    return node


def build_terminal(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="terminal-node", heartbeat_s=0)
    node.attach_receptor(terminal.RECEPTOR)
    return node

Over a real synapse

Same code, different transport.

With SYNAPSE_URL unset the brain runs on an in-process bus. Point it at a synapse and nothing in the modules changes. Prism attaches to any shared synapse and animates every Signal as it crosses.

terminal

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

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

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

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

http://127.0.0.1:7071 · -n demo

Prism renders every Signal on the bus as it fires - REGISTER, TASK, AGENT_OUTPUT, FINAL.

→

Pathway - three shapes.