Integrating an Engram.

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

$ cd cosmonapse-examples/06-engram-integration
$ python brain.py twice "what is the meaning of life?"
$ python brain.py                     # REPL - ask the same question twice

first call  computed -> Answer to 'what is the meaning of life?': 42
second call    cache -> Answer to 'what is the meaning of life?': 42

06-engram-integration/
  engram/context.py      InMemoryEngram(engram_id="ctx")
  neurons/researcher.py  recall, compute, imprint
  receptors/terminal.py  ask, and twice (compute, then recall)
  brain.py               memory-node + researcher-node + terminal-node

from cosmonapse import InMemoryEngram

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

from cosmonapse import Axon, EngramBinding


async def researcher(input, context, *, recall, imprint):
    question = input["question"]
    key = f"q:{question}"

    # 1. Look in shared memory for a prior answer to this exact question.
    prior = await recall("ctx", query={"merge_key": key})
    if prior.hits:
        cached = prior.hits[0].entry["content"]["answer"]
        return {"answer": cached, "source": "cache"}

    # 2. Compute a "fresh" answer (stubbed for the demo).
    answer = f"Answer to {question!r}: 42"

    # 3. Write it back so the next call hits the cache. merge_key dedupes by
    #    question, so repeated imprints upsert a single entry.
    await imprint(
        "ctx",
        op="upsert",
        entry={"content": {"question": question, "answer": answer},
               "tags": ["qa"]},
        merge_key=key,
        await_ack=True,
        deadline_ms=500,
    )
    return {"answer": answer, "source": "computed"}


# The binding maps a local name ("ctx") to the wire engram_id
# (directed_id), so the Neuron addresses memory by a stable local name -
# ops can repoint the backend without editing Neuron code.
AXON = Axon(
    neuron_id="researcher",
    neuron_fn=researcher,
    capabilities=["research"],
    engrams=[EngramBinding(name="ctx", directed_id="ctx")],
)

from cosmonapse import CliReceptor, SignalType

RECEPTOR = CliReceptor(
    neuron="researcher",
    input_key="question",
    prog="researcher",
    description="A Neuron with shared memory over the bus.",
    timeout_s=5.0,
)


@RECEPTOR.command(help="ask the researcher a question")
def ask(question: str):
    return {"question": question}


@RECEPTOR.command("twice", local=True, help="ask once to compute, once to recall")
async def twice(question: str):
    first = await RECEPTOR.ask({"question": question})
    second = await RECEPTOR.ask({"question": question})
    return f"first call  {first}\nsecond call {second}"


@RECEPTOR.on_result
def render(sig):
    if sig.type is SignalType.ERROR:
        return f"ERROR {sig.payload.get('message')}"
    out = sig.payload["output"]
    return f"{out['source']:>8s} -> {out['answer']}"

import asyncio
import contextlib
import signal

from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain

from config import NAMESPACE, SYNAPSE_URL
from engram import context
from neurons import researcher
from receptors import terminal


def build_memory(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="memory-node", role="worker")
    node.attach_engram(context.ENGRAM)
    return node


def build_researcher(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="researcher-node", role="worker")
    node.attach_axon(researcher.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

# 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 twice "what is the meaning of life?"

# 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 twice "what is the meaning of life?"
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py twice "what is the meaning of life?"