Pathway \u2014 three shapes, one primitive.

# worker.py - one Axon that streams PLAN / THOUGHT_DELTA / AGENT_OUTPUT
import asyncio
from cosmonapse import Axon, Dendrite, SignalType, connect_synapse

async def planner(input, context):
    # Inside a real Neuron you would stream chunks via emit_thought_delta.
    # Here we just return a final answer so the example stays self-contained.
    return {"plan": ["step-1", "step-2"]}

async def main():
    synapse = await connect_synapse("cosmo://127.0.0.1:7070")
    worker = Dendrite(synapse=synapse, namespace="demo", role="worker")
    worker.attach_axon(Axon(
        neuron_id="planner", neuron_fn=planner,
        capabilities=["plan"],
    ))
    async with worker:
        await asyncio.sleep(float("inf"))

asyncio.run(main())

# Shape 1 - sequential / request-reply.
# Block until the first terminal Signal arrives, then close the Pathway.
from cosmonapse import Dendrite, connect_synapse

async def main():
    synapse = await connect_synapse("cosmo://127.0.0.1:7070")
    orch = Dendrite(synapse=synapse, namespace="demo")

    async with orch:
        sig = await orch.dispatch_and_wait(
            capabilities=["plan"],
            input={"goal": "ship feature X"},
            timeout_s=5.0,
        )
        print(sig.payload["output"])
    await synapse.close()

# Shape 2 - reactive / trace-scoped callbacks.
# Pathway.on(SignalType.X) registers a handler that fires for each matching
# Signal on this trace - not the whole namespace.
from cosmonapse import Dendrite, SignalType, connect_synapse

async def main():
    synapse = await connect_synapse("cosmo://127.0.0.1:7070")
    orch = Dendrite(synapse=synapse, namespace="demo")

    async with orch:
        pw = await orch.dispatch_and_subscribe(
            capabilities=["plan"], input={"goal": "..."},
        )

        @pw.on(SignalType.THOUGHT_DELTA)
        async def stream(s): print(s.payload["delta"], end="")

        @pw.on(SignalType.PLAN)
        async def on_plan(s): print("\nPLAN:", s.payload["steps"])

        @pw.on(SignalType.AGENT_OUTPUT)
        async def done(s):
            print("DONE:", s.payload["output"])
            await pw.close()

        await asyncio.sleep(2.0)   # give callbacks time to fire

# Shape 3 - streaming iteration.
# Walk every Signal on the trace as it arrives, until you break or it closes.
from cosmonapse import Dendrite, SignalType, connect_synapse

async def main():
    synapse = await connect_synapse("cosmo://127.0.0.1:7070")
    orch = Dendrite(synapse=synapse, namespace="demo")

    async with orch:
        async with await orch.dispatch(
            capabilities=["plan"], input={"goal": "..."},
        ) as pw:
            async for sig in pw:
                print(sig.type, sig.payload)
                if sig.type is SignalType.AGENT_OUTPUT:
                    break

# scope="terminal" - decentralised orchestration.
# The Cortex only wakes for FINAL / ERROR / CLARIFICATION. Intermediate
# AGENT_OUTPUT / PLAN / TOOL_CALL fly past on the bus and are handled
# peer-to-peer by other orchestrators in the same trace.
async with orch:
    pw = await orch.dispatch(
        capabilities=["plan"],
        input={"goal": "..."},
        scope="terminal",
    )
    sig = await pw.wait(timeout_s=60.0)   # only FINAL / ERROR / CLARIFICATION resolves this

# observe_pathway - watch a trace another peer started.
# No TASK is emitted; the Pathway just attaches to inbound Signals
# matching the given trace_id.
async with monitor_dendrite:
    pw = await monitor_dendrite.observe_pathway(trace_id="trc_01J...")
    assert pw.role == "observer"
    async for sig in pw:
        log.info("observed", type=sig.type.value, neuron=sig.neuron)