Conversation logging and compaction + Teams + Tasks

README.md

@@ -331,7 +331,7 @@ broadcast = await node.broadcast(
 )
 ```
 
-A node with an `@node.ask` handler will ACK the conversation, run the handler, and reply with the result. Nodes without a handler fail quietly from the caller's point of view, just like any other broadcast recipient that cannot help.
+A node with an `@node.ask` handler will ACK the conversation, then run the handler **in the background** and deliver the result later as a broadcast reply. The origin's RPC returns immediately with `{"accepted": True, "deferred": True}`, so a handler that wraps a slow LLM can take minutes without holding a socket open. If the handler raises, the node emits a `conversation.error` event instead of a reply. Nodes without a handler fail quietly from the caller's point of view, just like any other broadcast recipient that cannot help.
 
 The CLI wraps this flow:
 
@@ -414,6 +414,70 @@ Why this is useful:
 
 ---
 
+## 🗜️  Conversation memory: durability, sync, and compaction
+
+Conversation events live in a pluggable log. The default is in-memory; pass a SQLite log to survive restarts:
+
+```python
+from synapse_p2p import Node, SqliteConversationLog
+
+node = Node(
+    name="architect",
+    conversation_log=SqliteConversationLog("architect-conversations.db"),
+)
+```
+
+A late joiner (or a restarted node) can pull a conversation it missed from any peer:
+
+```python
+added = await node.sync_conversation(peer, conversation_id)
+```
+
+Long conversations compact automatically. When a conversation grows past `conversation_max_events`, the node folds older events into a single `summary` event, keeping the opening message and the most recent `conversation_keep_recent` events verbatim:
+
+```python
+node = Node(
+    name="architect",
+    conversation_max_events=100,
+    conversation_keep_recent=25,
+)
+
+
+@node.summarizer
+async def summarize(events: list[ConversationEvent]) -> str:
+    transcript = "\n".join(f"{e.peer.name} [{e.kind}]: {e.payload}" for e in events)
+    return await my_llm_summarize(transcript)  # bring your own model
+```
+
+Without a custom summarizer, Synapse uses a naive extractive digest. Compaction is **local**: each node compresses its own copy of the shared log, and gossip cannot resurrect compacted events. You can also compact on demand with `await node.compact_conversation(conversation_id)` and observe it with `@node.on("conversation.compacted")`.
+
+---
+
+## 👷  Teams: an optional task layer
+
+`synapse_p2p.teams` layers a small task vocabulary on top of conversation events — nothing in the substrate is special-cased for it. A `Team` offers work; `Worker`s claim tasks matching their capabilities; the team grants each task to the first claimant, so exactly one worker runs it.
+
+```python
+from synapse_p2p.teams import Assignment, Team, Worker
+
+# architect process
+team = Team(node)
+task = await team.offer("implement the parser", spec={"file": "parser.py"}, requires=["python"])
+result = await team.wait(task, timeout=600)
+
+# coder process
+worker = Worker(coder_node)
+
+@worker.task
+async def implement(assignment: Assignment) -> dict:
+    await assignment.progress("starting")
+    return {"diff": await my_agent.run(assignment.title, assignment.spec)}
+```
+
+Each task is one shared conversation (`task.offer` → `task.claim` → `task.grant` → `task.progress` → `task.done` / `task.failed`), so every peer can watch the work happen, late joiners can sync it, and long task threads compact like any other conversation. See [`examples/coding_team`](./examples/coding_team) for an architect on one model overseeing coders on another.
+
+---
+
 ## 🌅  Periodic tasks
 
 Nodes can wake up on a schedule: every few seconds, every weekday morning, or when the sun rises.
@@ -602,6 +666,7 @@ See [`examples/`](./examples). Each example folder has its own README.
 | [`pydantic_ai_team`](./examples/pydantic_ai_team) | Pydantic AI agents behind Synapse nodes. |
 | [`periodic_tasks`](./examples/periodic_tasks) | Interval, cron, sunrise, and sunset jobs in a garden-caretaker node. |
 | [`stock_trading_team`](./examples/stock_trading_team) | Analyst/news/trader swarm with a dumb paper exchange API and market-hours periodic scans. |
+| [`coding_team`](./examples/coding_team) | An architect and coder agents on different models, coordinated through the `synapse_p2p.teams` task layer. |
 
 The agent examples use `synapse.ask`, opt-in ACKs, shared conversation replies, and advertised `agent-card` artifacts. The stock example shows a periodic job that checks market hours before asking the swarm, so agent/model work only happens when the paper market is open.
 
@@ -621,6 +686,8 @@ Built-in endpoints:
 | `_synapse.heartbeat` | update peer liveness |
 | `_synapse.broadcast.reply` | reply to a broadcast nonce |
 | `_synapse.conversation.event` | gossip a shared conversation event |
+| `_synapse.conversation.sync` | serve a conversation's events to a late joiner |
+| `_synapse.conversation.list` | list locally known conversation ids |
 | `_synapse.artifacts` | list advertised artifacts |
 | `_synapse.artifact.get` | fetch one advertised artifact |
 | `_node.info` | name, role, description, capabilities |
@@ -633,6 +700,8 @@ Wire format:
 1. 4-byte unsigned big-endian payload length
 2. MsgPack payload bytes
 
+Frames up to 4 MiB are accepted by default (`Node(max_upload_size=...)`, `Client(max_download_size=...)`), so results can carry real payloads like diffs and documents.
+
 Request:
 
 ```python
@@ -690,7 +759,7 @@ logger.enable("synapse_p2p")
 
 Synapse does **not** implement planning, memory, consensus, auth policy, NAT traversal, hosted registries, or UX.
 
-Those belong above Synapse.
+Those belong above Synapse. (Two exceptions are on the roadmap because they belong *in* the substrate: node identity + signed gossip for untrusted networks, and a relay mode so seeds can bridge peers that cannot dial each other.)
 
 Synapse is the substrate:
 

examples/README.md

@@ -58,5 +58,6 @@ Most agent examples publish an `agent-card` artifact and use shared conversation
 | [`pydantic_ai_team`](./pydantic_ai_team) | Pydantic AI agents behind Synapse nodes. |
 | [`periodic_tasks`](./periodic_tasks) | Interval, cron, and solar jobs in a garden-caretaker node. |
 | [`stock_trading_team`](./stock_trading_team) | Analyst/news/trader swarm with a dumb paper exchange API and market-hours periodic scans. |
+| [`coding_team`](./coding_team) | An architect and coder agents on different models: task offers, claims, grants, progress, and compacted task conversations via `synapse_p2p.teams`. |
 
 Each folder has its own README with exact run commands.

examples/coding_team/README.md

@@ -0,0 +1,53 @@
+# Coding team: an architect and coders on different models
+
+A minimal heterogeneous agent team built on `synapse_p2p.teams`:
+
+- **architect** — oversees the work. Offers tasks to the swarm, grants each to
+  the first capable claimant, collects results, and reviews them with its own
+  model (e.g. Claude).
+- **coder-1 / coder-2** — claim tasks whose `requires` match their advertised
+  capabilities, implement them with their own model (e.g. GPT), and narrate
+  progress into the shared task conversation.
+
+Every task is one shared conversation (`conversation_id == task id`) that all
+peers gossip: offer → claim → grant → progress → done. The architect compacts
+long task threads automatically (`conversation_max_events=30`), folding old
+progress chatter into a `summary` event.
+
+## Run it (offline)
+
+Without model env vars the agents use pydantic-ai's `TestModel`, so the whole
+flow runs with no API keys. From the repo root, in three terminals:
+
+```bash
+CODER_NAME=coder-1 uv run python -m examples.coding_team.coder
+CODER_NAME=coder-2 uv run python -m examples.coding_team.coder
+uv run python -m examples.coding_team.architect
+```
+
+Watch the swarm from a fourth terminal:
+
+```bash
+uv run sn watch team.electron.network
+```
+
+## Run it with real models
+
+```bash
+export ANTHROPIC_API_KEY=... OPENAI_API_KEY=...
+ARCHITECT_MODEL=anthropic:claude-fable-5 uv run python -m examples.coding_team.architect
+CODER_MODEL=openai:gpt-5.5 CODER_NAME=coder-1 uv run python -m examples.coding_team.coder
+```
+
+Any pydantic-ai model string works; the swarm doesn't care what model sits
+behind a node. To push a coder's reasoning effort up, configure the agent's
+`model_settings` in `common.py` (e.g. OpenAI's `reasoning_effort`).
+
+## What to look at
+
+- `synapse_p2p/teams.py` — the whole task vocabulary is ~200 lines of
+  conversation events. Nothing here is special-cased in the substrate.
+- Deferred results: coders return immediately at the RPC layer and deliver
+  results as conversation events, so a task can take as long as the model needs.
+- Late joiners can catch up on a task thread with
+  `node.sync_conversation(peer, task_id)`.

examples/coding_team/architect.py

@@ -0,0 +1,79 @@
+"""The architect node: offers tasks to the swarm, reviews what comes back.
+
+The architect and the coders can run different models entirely — the swarm
+only sees offers, claims, progress, and results. Each task is one shared
+conversation, and the architect compacts long task threads automatically.
+"""
+
+import asyncio
+
+from examples.coding_team.common import ARCHITECT_MODEL, SWARM, make_agent, run_agent
+from synapse_p2p import ConversationEvent, Node
+from synapse_p2p.teams import Team
+
+brain, test_model = make_agent(
+    ARCHITECT_MODEL,
+    "You are the software architect overseeing a team of coder agents. "
+    "Review their submitted implementations and give a short verdict.",
+    "Architect: both implementations look solid; ship it.",
+)
+
+node = Node(
+    name="architect",
+    role="architect",
+    swarm=SWARM,
+    capabilities=["architecture", "review", "coordination"],
+    mdns=True,
+    # Keep task conversations small: fold old progress chatter into summaries.
+    conversation_max_events=30,
+    conversation_keep_recent=10,
+)
+
+team = Team(node)
+
+FEATURE = "Add a `sn swarm top`-style live dashboard to the CLI"
+SUBTASKS = [
+    ("Implement the dashboard rendering loop", {"module": "synapse_p2p/cli.py"}),
+    ("Add tests for the dashboard event stream", {"module": "synapse_p2p/tests/"}),
+]
+
+
+@node.on("conversation.task.progress")
+async def on_progress(event: ConversationEvent) -> None:
+    print(f"  progress from {event.peer.name}: {event.payload.get('message')}")
+
+
+async def main() -> None:
+    await node.start()
+    await node.join(wait=1)
+    print(f"architect online, {len(node.peers)} peer(s) known")
+
+    tasks = []
+    for title, spec in SUBTASKS:
+        task = await team.offer(title, spec=spec, requires=["python", "implementation"])
+        print(f"offered: {title} ({task.id})")
+        tasks.append(task)
+
+    results = []
+    for task in tasks:
+        result = await team.wait(task, timeout=600)
+        assignee = task.assignee.name if task.assignee else "?"
+        print(f"done: {task.title} by {assignee}")
+        results.append(result)
+
+    verdict = await run_agent(
+        brain,
+        test_model,
+        f"Feature: {FEATURE}\nSubmissions: {results}\nGive your review verdict.",
+    )
+    print(f"\narchitect verdict:\n{verdict}")
+
+    for task in tasks:
+        kinds = [event.kind for event in node.conversation(task.id)]
+        print(f"conversation {task.id}: {kinds}")
+
+    await node.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/coding_team/coder.py

@@ -0,0 +1,67 @@
+"""A coder node: claims tasks it can do, implements them, narrates progress.
+
+Run several of these (they race to claim, the architect grants one):
+
+    CODER_NAME=coder-1 uv run python -m examples.coding_team.coder
+    CODER_NAME=coder-2 uv run python -m examples.coding_team.coder
+"""
+
+import asyncio
+import os
+
+from examples.coding_team.common import CODER_MODEL, SWARM, make_agent, run_agent
+from synapse_p2p import Node
+from synapse_p2p.teams import Assignment, Worker
+
+brain, test_model = make_agent(
+    CODER_MODEL,
+    "You are a senior implementation engineer on a distributed agent team. "
+    "You receive one task with a spec. Reply with the implementation: code, "
+    "tests, and a one-paragraph note for the reviewing architect.",
+    "Coder: implemented the task with a small patch, added tests, all passing.",
+)
+
+node = Node(
+    name=os.getenv("CODER_NAME", "coder-1"),
+    role="implementation",
+    swarm=SWARM,
+    capabilities=["python", "implementation", "tests"],
+    mdns=True,
+)
+
+node.artifact(
+    "agent-card",
+    {
+        "name": node.name,
+        "role": node.role,
+        "model": CODER_MODEL or "test-model",
+        "capabilities": ["python", "implementation", "tests"],
+        "description": "Claims implementation tasks and returns patches with tests.",
+    },
+    mime_type="application/vnd.synapse.agent-card+json",
+)
+
+worker = Worker(node)
+
+
+@worker.task
+async def implement(assignment: Assignment) -> dict:
+    await assignment.progress(f"{node.name} starting", title=assignment.title)
+    prompt = f"Task: {assignment.title}\nSpec: {assignment.spec}"
+    output = await run_agent(brain, test_model, prompt)
+    await assignment.progress(f"{node.name} finished, submitting result")
+    return {"implementation": output, "by": node.name}
+
+
+async def main() -> None:
+    await node.start()
+    try:
+        await node.join()
+        print(f"{node.name} online at {node.address}:{node.port}, waiting for offers")
+        await asyncio.Event().wait()
+    finally:
+        await node.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/coding_team/common.py

@@ -0,0 +1,30 @@
+import os
+from contextlib import nullcontext
+from typing import Any
+
+from pydantic_ai import Agent
+from pydantic_ai.models.test import TestModel
+
+SWARM = "team.electron.network"
+
+# Set these to real models to run the demo live, e.g.
+#   ARCHITECT_MODEL=anthropic:claude-fable-5 CODER_MODEL=openai:gpt-5.5
+# Unset, the demo runs offline against pydantic-ai's TestModel.
+ARCHITECT_MODEL = os.getenv("ARCHITECT_MODEL")
+CODER_MODEL = os.getenv("CODER_MODEL")
+
+
+def make_agent(
+    model: str | None, instructions: str, fallback: str
+) -> tuple[Agent, TestModel | None]:
+    if model:
+        return Agent(model, instructions=instructions), None
+    test_model = TestModel(custom_output_text=fallback)
+    return Agent(test_model, instructions=instructions), test_model
+
+
+async def run_agent(agent: Agent, test_model: TestModel | None, prompt: str) -> Any:
+    context = agent.override(model=test_model) if test_model is not None else nullcontext()
+    with context:
+        result = await agent.run(prompt)
+    return result.output

synapse_p2p/__init__.py

@@ -20,6 +20,12 @@
 """
 
 from synapse_p2p.client import Client
+from synapse_p2p.conversations import (
+    BaseConversationLog,
+    MemoryConversationLog,
+    SqliteConversationLog,
+    default_summarizer,
+)
 from synapse_p2p.messages import RemoteProcedureCall, RPCError, RPCRequest, RPCResponse
 from synapse_p2p.node import Capability, Node
 from synapse_p2p.schedules import CronSchedule, IntervalSchedule, SolarSchedule, cron, every, solar
@@ -40,13 +46,15 @@
 __all__ = [
     "AdvertisedArtifact",
     "Capability",
+    "BaseConversationLog",
     "BaseRPCSerializer",
     "Broadcast",
     "BroadcastReply",
     "Client",
     "Connection",
     "ConversationEvent",
     "CronSchedule",
+    "MemoryConversationLog",
     "MessagePackRPCSerializer",
     "Node",
     "NodeKind",
@@ -56,9 +64,11 @@
     "RPCResponse",
     "RemoteProcedureCall",
     "ServedArtifact",
+    "SqliteConversationLog",
     "IntervalSchedule",
     "SolarSchedule",
     "cron",
+    "default_summarizer",
     "every",
     "solar",
     "__logo__",