Bidding.
Clone the examples repo, install, and run from inside the example's folder.
$ git clone https://github.com/Cosmonapse/cosmonapse-examples $ pip install cosmonapse $ cd cosmonapse-examples/10-bidding $ python brain.py offer "... a long article ..." # lowest_cost $ python brain.py offer "... a long article ..." --select highest_confidence $ python brain.py offer "... a long article ..." --select first_bid
[lowest_cost] winner: summarizer-a ... a long article ....... [highest_confidence] winner: summarizer-b ... a long article .......
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.
10-bidding/ neurons/summarizer.py make_axon(id, cost, eta_ms, confidence) + its bidding policy receptors/terminal.py offer - dispatch_offer, then wait brain.py a node per bidder + terminal-node
Routing covers the homogeneous case.
Capability routing trusts the broker to deliver a TASK once within a queue group of identical workers. When workers differ in cost, latency or quality, the producer should choose. Bidding is that choice as a protocol: the offer is broadcast, bids come back, one worker is awarded, everyone else is declined.
The policy lives next to the Neuron.
@axon.host.on_task_offer is a deferred host decorator: it is applied to whichever Dendrite hosts the Axon, and registering it replaces that node's automatic bidder. bid() bypasses the orchestrator role guard, because bidding is a worker announcing it can take work, not orchestration.
from cosmonapse import Axon async def summarize(input, context): return {"summary": input["text"][:80] + "..."} def make_axon(neuron_id: str, *, cost: float, eta_ms: int, confidence: float) -> Axon: axon = Axon(neuron_id=neuron_id, neuron_fn=summarize, capabilities=["summarize", "english"]) @axon.host.on_task_offer(capability="summarize") async def respond(offer): # bid() bypasses the orchestrator role guard - bidding is how a # worker says "I can take this", not orchestration. await axon.dendrite.bid( offer, neuron=neuron_id, cost=cost, # estimated USD eta_ms=eta_ms, # estimated latency confidence=confidence, # self-assessment 0..1 ) return axon BIDDERS = { "summarizer-a": dict(cost=0.002, eta_ms=300, confidence=0.82), "summarizer-b": dict(cost=0.005, eta_ms=120, confidence=0.97), }
dispatch_offer returns a Pathway.
Bids are collected for deadline_ms, then select picks: first_bid ends the window at the first bid, lowest_cost and highest_confidence drain it. The awarded worker's node turns TASK_AWARDED into an internal TASK, and its AGENT_OUTPUT arrives on the Pathway.
from cosmonapse import CliReceptor RECEPTOR = CliReceptor( prog="bidding", description="TASK_OFFER -> BID -> TASK_AWARDED -> AGENT_OUTPUT.", timeout_s=5.0, ) @RECEPTOR.command(local=True, default=True, help="offer a summarize task to bidders") async def offer(text: str, select: str = "lowest_cost"): pw = await RECEPTOR.dendrite.dispatch_offer( input={"text": text}, capabilities=["summarize"], deadline_ms=250, # collect BIDs for this window select=select, # lowest_cost | highest_confidence | first_bid ) sig = await pw.wait(timeout_s=5.0) winner = sig.directed.id if sig.directed else "?" return f"[{select}] winner: {winner}\n{sig.payload['output']['summary']}"
terminal-node --[TASK_OFFER]------------------> broadcast
summarizer-a --[BID cost=0.002 conf=0.82]----> terminal-node
summarizer-b --[BID cost=0.005 conf=0.97]----> terminal-node
(deadline_ms passes, select picks)
terminal-node --[TASK_AWARDED summarizer-a]---> bus
terminal-node --[TASK_DECLINED summarizer-b]--> bus
summarizer-a --[AGENT_OUTPUT]----------------> the PathwayTwo bidders, one terminal.
import asyncio import contextlib import signal from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain from config import NAMESPACE, SYNAPSE_URL from neurons import summarizer from receptors import terminal def build_bidder(synapse, neuron_id: str) -> Dendrite: node = Dendrite(synapse=synapse, namespace=NAMESPACE, dendrite_id=f"{neuron_id}-node", role="worker") node.attach_axon(summarizer.make_axon(neuron_id, **summarizer.BIDDERS[neuron_id])) 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
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 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 offer "..." # 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 offer "..." $ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py offer "..."