feat(p2p): add peer-to-peer task coordination and mailbox system

- Make SKILL.md platform-agnostic (remove Windows-only path rules)
- Add 14-channel quick-reference table with required fields
- Create references/channels.md with detailed per-channel config
- Inherit model from parent config when not explicitly specified
- Consolidate duplicate file reads in _patch_config
- Add email channel consent_granted field documentation
- Fix auto_reply_enabled default value (true, not false)
- Add troubleshooting section to SKILL.md

nanobot/agent/context.py

@@ -39,6 +39,7 @@ class ContextBuilder:
         skill_names: list[str] | None = None,
         channel: str | None = None,
         session_summary: str | None = None,
+        session_key: str | None = None,
     ) -> str:
         """Build the system prompt from identity, bootstrap files, memory, and skills."""
         parts = [self._get_identity(channel=channel)]
@@ -73,8 +74,29 @@ class ContextBuilder:
         if session_summary:
             parts.append(f"[Archived Context Summary]\n\n{session_summary}")
 
+        # Inject P2P collaboration hint for task-scoped sessions
+        if session_key and session_key.startswith("task:"):
+            parts.append(self._p2p_collaboration_hint())
+
         return "\n\n---\n\n".join(parts)
 
+    @staticmethod
+    def _p2p_collaboration_hint() -> str:
+        return (
+            "# Multi-Agent Collaboration\n\n"
+            "You are part of a decentralized agent network. You can:\n"
+            "- Use `broadcast_task` to announce subtasks and collect BIDs\n"
+            "- Use `dispatch_task` to assign tasks to specific agents\n"
+            "- Use `poll_task_result` to check task status\n"
+            "- Use `report_user` to deliver final results to the user\n"
+            "- Use `finalize_task` to terminate tasks\n\n"
+            "Rules:\n"
+            "- Never block waiting for results. Dispatch and continue.\n"
+            "- If a task times out, decide whether to retry, failover, or report partial.\n"
+            "- Respect the user's INTERRUPT messages — they have highest priority.\n"
+            "- You are currently in a task-scoped session; focus on the delegated task."
+        )
+
     def _get_identity(self, channel: str | None = None) -> str:
         """Get the core identity section."""
         workspace_path = str(self.workspace.expanduser().resolve())
@@ -154,6 +176,7 @@ class ContextBuilder:
         sender_id: str | None = None,
         session_summary: str | None = None,
         session_metadata: Mapping[str, Any] | None = None,
+        session_key: str | None = None,
     ) -> list[dict[str, Any]]:
         """Build the complete message list for an LLM call."""
         extra = goal_state_runtime_lines(session_metadata)
@@ -175,7 +198,7 @@ class ContextBuilder:
         else:
             merged = user_content + [{"type": "text", "text": runtime_ctx}]
         messages = [
-            {"role": "system", "content": self.build_system_prompt(skill_names, channel=channel, session_summary=session_summary)},
+            {"role": "system", "content": self.build_system_prompt(skill_names, channel=channel, session_summary=session_summary, session_key=session_key)},
             *history,
         ]
         if messages[-1].get("role") == current_role:

nanobot/agent/loop.py

@@ -24,6 +24,14 @@ from nanobot.agent.runner import _MAX_INJECTIONS_PER_TURN, AgentRunner, AgentRun
 from nanobot.agent.subagent import SubagentManager
 from nanobot.agent.tools.file_state import FileStateStore, bind_file_states, reset_file_states
 from nanobot.agent.tools.message import MessageTool
+from nanobot.agent.tools.p2p import (
+    BroadcastTaskTool,
+    CheckAggregationTool,
+    DispatchTaskTool,
+    FinalizeTaskTool,
+    PollTaskResultTool,
+    ReportUserTool,
+)
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.self import MyTool
 from nanobot.bus.events import InboundMessage, OutboundMessage
@@ -185,6 +193,7 @@ class AgentLoop:
         model_preset: str | None = None,
         preset_snapshot_loader: preset_helpers.PresetSnapshotLoader | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
+        p2p_shell: Any | None = None,
     ):
         from nanobot.config.schema import ToolsConfig
 
@@ -192,6 +201,7 @@ class AgentLoop:
         defaults = AgentDefaults()
         self.bus = bus
         self.channels_config = channels_config
+        self.p2p_shell = p2p_shell
         self.provider = provider
         self._provider_snapshot_loader = provider_snapshot_loader
         self._preset_snapshot_loader = preset_snapshot_loader
@@ -463,6 +473,22 @@ class AgentLoop:
             )
             registered.append("my")
 
+        # Register P2P tools if enabled
+        if self.p2p_shell:
+            self.tools.register(DispatchTaskTool(shell=self.p2p_shell))
+            self.tools.register(PollTaskResultTool(shell=self.p2p_shell))
+            self.tools.register(BroadcastTaskTool(shell=self.p2p_shell))
+            self.tools.register(CheckAggregationTool(shell=self.p2p_shell))
+            self.tools.register(
+                ReportUserTool(
+                    send_callback=self.bus.publish_outbound,
+                    default_channel=getattr(self.channels_config, "default_channel", ""),
+                    default_chat_id=getattr(self.channels_config, "default_chat_id", ""),
+                )
+            )
+            self.tools.register(FinalizeTaskTool(shell=self.p2p_shell, session_manager=self.sessions))
+            registered.append("p2p")
+
         logger.info("Registered {} tools: {}", len(registered), registered)
 
     async def _connect_mcp(self) -> None:

nanobot/agent/tools/p2p.py

@@ -0,0 +1,328 @@
+"""P2P tools for inter-agent task dispatch and coordination."""
+
+from __future__ import annotations
+
+from typing import Any, Awaitable, Callable
+
+from nanobot.agent.tools.base import Tool
+from nanobot.bus.events import OutboundMessage
+
+
+class DispatchTaskTool(Tool):
+    """Asynchronously dispatch a task to another agent. Non-blocking."""
+
+    def __init__(self, shell: "P2PShell"):
+        self._shell = shell
+
+    @property
+    def name(self) -> str:
+        return "dispatch_task"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Dispatch a task to a specific target agent. Returns immediately with a receipt. "
+            "The target agent will process the task independently. Use poll_task_result later to check completion. "
+            "Do NOT block waiting for results."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "to": {"type": "string", "description": "Target agent ID"},
+                "task_description": {"type": "string", "description": "Clear description of the task"},
+                "parent_task_id": {"type": "string", "description": "Parent task ID for ancestry tracking"},
+                "deadline_seconds": {"type": "integer", "default": 300, "description": "Task deadline in seconds"},
+                "allow_redelegation": {"type": "boolean", "default": True, "description": "Whether the target may re-delegate"},
+            },
+            "required": ["to", "task_description"],
+        }
+
+    async def execute(
+        self,
+        to: str,
+        task_description: str,
+        parent_task_id: str | None = None,
+        deadline_seconds: int = 300,
+        allow_redelegation: bool = True,
+        **kwargs: Any,
+    ) -> str:
+        result = self._shell.dispatch(
+            to=to,
+            parent_task_id=parent_task_id,
+            description=task_description,
+            deadline_seconds=deadline_seconds,
+            allow_redelegation=allow_redelegation,
+        )
+        if result.get("status") == "rejected":
+            return f"Error: dispatch rejected — {result.get('reason', 'unknown')}"
+        if result.get("status") == "circuit_open":
+            failover = result.get("failover_to")
+            return f"Error: circuit open for {to}. Failover candidate: {failover or 'none'}"
+        return (
+            f"Dispatched to {to}. Task ID: {result.get('task_id')}. "
+            f"Depth: {result.get('depth', 0)}."
+        )
+
+
+class PollTaskResultTool(Tool):
+    """Poll the status of a previously dispatched task."""
+
+    def __init__(self, shell: "P2PShell"):
+        self._shell = shell
+
+    @property
+    def name(self) -> str:
+        return "poll_task_result"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Check the current status of a task you previously dispatched. "
+            "Returns completed, pending, timeout, failed, or not_found. "
+            "Call this proactively — do not wait for automatic notifications."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "task_id": {"type": "string", "description": "Task ID returned by dispatch_task"},
+            },
+            "required": ["task_id"],
+        }
+
+    async def execute(self, task_id: str, **kwargs: Any) -> str:
+        result = self._shell.poll(task_id)
+        status = result.get("status")
+        if status == "not_found":
+            return f"Task {task_id} not found."
+        if status == "pending":
+            return f"Task {task_id} is pending (elapsed {result.get('elapsed', '?')}s)."
+        if status == "timeout":
+            return f"Task {task_id} timed out after {result.get('elapsed', '?')}s."
+        if status in ("completed", "failed", "aborted"):
+            from_agent = result.get("from", "unknown")
+            content = result.get("result", "")
+            preview = content[:500] + "..." if len(content) > 500 else content
+            return f"Task {task_id} is {status} (from {from_agent}).\n\n{preview}"
+        return f"Task {task_id} status: {status}"
+
+
+class BroadcastTaskTool(Tool):
+    """Broadcast subtasks to discover capable agents."""
+
+    def __init__(self, shell: "P2PShell"):
+        self._shell = shell
+
+    @property
+    def name(self) -> str:
+        return "broadcast_task"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Announce subtasks to the agent network to collect BIDs. "
+            "Returns immediately. Use check_aggregation later to see which agents responded. "
+            "Each subtask should include a capability hint for matching."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "task_id": {"type": "string", "description": "Your task identifier"},
+                "subtasks": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "subtask_id": {"type": "string"},
+                            "description": {"type": "string"},
+                            "capability": {"type": "string", "description": "Required capability, e.g. 'web_search'"},
+                            "budget_seconds": {"type": "integer", "default": 300},
+                        },
+                        "required": ["subtask_id", "description", "capability"],
+                    },
+                },
+                "aggregation_timeout": {"type": "integer", "default": 30, "description": "Seconds to wait for BIDs"},
+            },
+            "required": ["task_id", "subtasks"],
+        }
+
+    async def execute(
+        self,
+        task_id: str,
+        subtasks: list[dict[str, Any]],
+        aggregation_timeout: int = 30,
+        **kwargs: Any,
+    ) -> str:
+        result = self._shell.broadcast(task_id, subtasks, aggregation_timeout)
+        invited = result.get("invited", 0)
+        return f"Broadcast opened for {task_id}. Invited {invited} agent(s). Use check_aggregation to collect BIDs."
+
+
+class CheckAggregationTool(Tool):
+    """Check the status of a broadcast aggregation window."""
+
+    def __init__(self, shell: "P2PShell"):
+        self._shell = shell
+
+    @property
+    def name(self) -> str:
+        return "check_aggregation"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Check whether a previously broadcast task has collected enough BIDs or timed out. "
+            "Returns the list of responding agents and their bids, or a pending status with counts."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "task_id": {"type": "string", "description": "Task ID used in broadcast_task"},
+            },
+            "required": ["task_id"],
+        }
+
+    async def execute(self, task_id: str, **kwargs: Any) -> str:
+        result = self._shell.check_aggregation(task_id)
+        status = result.get("status")
+        if status == "no_window":
+            return f"No broadcast window found for {task_id}."
+        if status == "pending":
+            received = result.get("received", 0)
+            expected = result.get("expected", "?")
+            remaining = result.get("seconds_remaining", 0)
+            return (
+                f"Aggregation pending for {task_id}: "
+                f"{received}/{expected} received, {remaining}s remaining."
+            )
+        if status == "closed":
+            entries = result.get("entries", [])
+            lines = [f"Aggregation closed for {task_id} ({result.get('reason', '')}):", ""]
+            for e in entries:
+                agent = e.get("from", "unknown")
+                sub = e.get("subtask_id", "")
+                lines.append(f"- {agent} bid for {sub}")
+            return "\n".join(lines)
+        return f"Unknown aggregation status for {task_id}: {status}"
+
+
+class ReportUserTool(Tool):
+    """Deliver a final answer to the user."""
+
+    def __init__(
+        self,
+        send_callback: Callable[[OutboundMessage], Awaitable[None]] | None = None,
+        default_channel: str = "",
+        default_chat_id: str = "",
+    ):
+        self._send_callback = send_callback
+        self._default_channel = default_channel
+        self._default_chat_id = default_chat_id
+
+    @property
+    def name(self) -> str:
+        return "report_user"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Report the final answer to the user. Use this when you have gathered enough results. "
+            "Status 'partial' means some subtasks are incomplete — list them in pending_items."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "final_answer": {"type": "string", "description": "Complete answer for the user"},
+                "status": {"type": "string", "enum": ["success", "partial", "failed"]},
+                "pending_items": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Incomplete items when status is partial",
+                },
+                "task_summary": {"type": "string", "description": "Optional brief summary"},
+            },
+            "required": ["final_answer", "status"],
+        }
+
+    async def execute(
+        self,
+        final_answer: str,
+        status: str,
+        pending_items: list[str] | None = None,
+        task_summary: str = "",
+        **kwargs: Any,
+    ) -> str:
+        if not self._send_callback:
+            return "Error: report_user not configured (no send callback)"
+
+        parts = [final_answer]
+        if pending_items:
+            parts.append(f"\n\nPending items:\n" + "\n".join(f"- {i}" for i in pending_items))
+        if task_summary:
+            parts.append(f"\n\nSummary: {task_summary}")
+
+        content = "\n".join(parts)
+        msg = OutboundMessage(
+            channel=self._default_channel,
+            chat_id=self._default_chat_id,
+            content=content,
+        )
+        await self._send_callback(msg)
+        return f"Reported to user (status={status})."
+
+
+class FinalizeTaskTool(Tool):
+    """Force-finalize a task and close its sessions."""
+
+    def __init__(self, shell: "P2PShell", session_manager: "SessionManager | None" = None):
+        self._shell = shell
+        self._session_manager = session_manager
+
+    @property
+    def name(self) -> str:
+        return "finalize_task"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Terminate a task and all its subtasks. Use when the user says 'stop', "
+            "or when a task is fundamentally blocked. outcome can be completed, failed, or aborted."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "task_id": {"type": "string"},
+                "outcome": {"type": "string", "enum": ["completed", "failed", "aborted"]},
+                "reason": {"type": "string", "description": "Why the task was finalized"},
+            },
+            "required": ["task_id", "outcome"],
+        }
+
+    async def execute(
+        self,
+        task_id: str,
+        outcome: str,
+        reason: str = "",
+        **kwargs: Any,
+    ) -> str:
+        self._shell.finalize(task_id, outcome, reason)
+        if self._session_manager:
+            self._session_manager.finalize_task_session(task_id)
+        return f"Task {task_id} finalized with outcome={outcome}."

nanobot/cli/commands.py

@@ -75,6 +75,7 @@ class SafeFileHistory(FileHistory):
 from nanobot.cli.stream import StreamRenderer, ThinkingSpinner
 from nanobot.config.paths import get_workspace_path, is_default_workspace
 from nanobot.config.schema import Config
+from nanobot.p2p.shell import P2PShell
 from nanobot.utils.helpers import sync_workspace_templates
 from nanobot.utils.restart import (
     consume_restart_notice_from_env,
@@ -92,6 +93,17 @@ app = typer.Typer(
 console = Console()
 EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
 
+
+def _resolve_p2p(config: Config) -> P2PShell | None:
+    """Resolve P2P config and create the stateless P2P shell."""
+    mb_cfg = config.mailbox
+    if not mb_cfg.enabled:
+        return None
+    return P2PShell(
+        agent_id=mb_cfg.agent_id,
+        mailboxes_root=mb_cfg.mailboxes_root,
+    )
+
 # ---------------------------------------------------------------------------
 # CLI input: prompt_toolkit for editing, paste, history, and display
 # ---------------------------------------------------------------------------
@@ -581,10 +593,13 @@ def serve(
     sync_workspace_templates(runtime_config.workspace_path)
     bus = MessageBus()
     session_manager = SessionManager(runtime_config.workspace_path)
+    p2p_shell = _resolve_p2p(runtime_config)
+
     try:
         agent_loop = AgentLoop.from_config(
             runtime_config, bus,
             session_manager=session_manager,
+            p2p_shell=p2p_shell,
             image_generation_provider_configs={
                 "openrouter": runtime_config.providers.openrouter,
                 "aihubmix": runtime_config.providers.aihubmix,
@@ -690,6 +705,8 @@ def _run_gateway(
     cron_store_path = config.workspace_path / "cron" / "jobs.json"
     cron = CronService(cron_store_path)
 
+    p2p_shell = _resolve_p2p(config)
+
     # Create agent with cron service
     agent = AgentLoop.from_config(
         config, bus,
@@ -709,6 +726,7 @@ def _run_gateway(
             preset,
         ),
         provider_signature=provider_snapshot.signature,
+        p2p_shell=p2p_shell,
     )
 
     from nanobot.agent.loop import UNIFIED_SESSION_KEY
@@ -921,6 +939,8 @@ def _run_gateway(
         interval_s=hb_cfg.interval_s,
         enabled=hb_cfg.enabled,
         timezone=config.agents.defaults.timezone,
+        p2p_shell=p2p_shell,
+        bus=bus,
     )
 
     if channels.enabled_channels:
@@ -1083,6 +1103,8 @@ def agent(
     cron_store_path = config.workspace_path / "cron" / "jobs.json"
     cron = CronService(cron_store_path)
 
+    p2p_shell = _resolve_p2p(config)
+
     if logs:
         logger.enable("nanobot")
     else:
@@ -1092,6 +1114,7 @@ def agent(
         agent_loop = AgentLoop.from_config(
             config, bus,
             cron_service=cron,
+            p2p_shell=p2p_shell,
         )
     except ValueError as exc:
         console.print(f"[red]Error: {exc}[/red]")
@@ -1289,6 +1312,7 @@ def agent(
                         console.print("\nGoodbye!")
                         break
             finally:
+                pass
                 agent_loop.stop()
                 outbound_task.cancel()
                 await asyncio.gather(bus_task, outbound_task, return_exceptions=True)

nanobot/config/schema.py

@@ -282,6 +282,19 @@ class ToolsConfig(Base):
     ssrf_whitelist: list[str] = Field(default_factory=list)  # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale)
 
 
+class P2PConfig(Base):
+    """P2P collaboration network configuration."""
+
+    enabled: bool = False
+    agent_id: str = ""
+    description: str = ""
+    capabilities: list[str] = Field(default_factory=list)
+    allow_from: list[str] = Field(default_factory=lambda: ["*"])
+    max_concurrent_tasks: int = 3
+    poll_interval: float = 5.0
+    mailboxes_root: str = "~/.nanobot/mailboxes"
+
+
 class Config(BaseSettings):
     """Root configuration for nanobot."""
 
@@ -295,6 +308,7 @@ class Config(BaseSettings):
         default_factory=dict,
         validation_alias=AliasChoices("modelPresets", "model_presets"),
     )
+    mailbox: P2PConfig = Field(default_factory=P2PConfig)
 
     @model_validator(mode="after")
     def _validate_model_preset(self) -> "Config":

nanobot/heartbeat/service.py

@@ -60,6 +60,8 @@ class HeartbeatService:
         interval_s: int = 30 * 60,
         enabled: bool = True,
         timezone: str | None = None,
+        p2p_shell: Any | None = None,
+        bus: Any | None = None,
     ):
         self.workspace = workspace
         self.provider = provider
@@ -69,8 +71,11 @@ class HeartbeatService:
         self.interval_s = interval_s
         self.enabled = enabled
         self.timezone = timezone
+        self.p2p_shell = p2p_shell
+        self.bus = bus
         self._running = False
         self._task: asyncio.Task | None = None
+        self._last_inbox_scan: float = 0.0
 
     @property
     def heartbeat_file(self) -> Path:
@@ -185,6 +190,32 @@ class HeartbeatService:
         """Execute a single heartbeat tick."""
         from nanobot.utils.evaluator import evaluate_response
 
+        # --- P2P inbox scan ---
+        if self.p2p_shell and self.bus:
+            try:
+                new_msgs = self.p2p_shell.scan_new_inbox(since=self._last_inbox_scan)
+                if new_msgs:
+                    self._last_inbox_scan = time.time()
+                    from nanobot.bus.events import InboundMessage
+                    for msg in new_msgs:
+                        await self.bus.publish_inbound(
+                            InboundMessage(
+                                channel="p2p",
+                                sender_id=msg.get("from", "unknown"),
+                                chat_id=msg.get("task_id", ""),
+                                content=msg.get("payload", {}).get("description", ""),
+                                metadata={"p2p_msg": msg},
+                            )
+                        )
+                        logger.info(
+                            "Heartbeat: injected P2P task {} from {}",
+                            msg.get("task_id", ""),
+                            msg.get("from", "unknown"),
+                        )
+            except Exception:
+                logger.exception("Heartbeat P2P scan failed")
+
+        # --- Legacy heartbeat file check ---
         content = self._read_heartbeat_file()
         if not content:
             logger.debug("Heartbeat: HEARTBEAT.md missing or empty")

nanobot/p2p/__init__.py

@@ -0,0 +1,5 @@
+"""P2P inter-agent coordination layer."""
+
+from nanobot.p2p.shell import P2PShell
+
+__all__ = ["P2PShell"]

nanobot/p2p/shell.py

@@ -0,0 +1,426 @@
+"""P2P shell: filesystem-backed inter-agent coordination.
+
+All state is stored in the mailbox filesystem; this class is stateless.
+Restarting the gateway restores all task state by scanning files.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any, Literal
+
+from loguru import logger
+
+
+class P2PShell:
+    """Stateless P2P coordination shell backed by the mailbox filesystem."""
+
+    def __init__(self, agent_id: str, mailboxes_root: str):
+        self.agent_id = agent_id
+        self.root = Path(mailboxes_root).expanduser()
+        self.inbox = self.root / agent_id / "inbox"
+        self.processed = self.root / agent_id / "processed"
+        self.links_dir = self.root / "_links"
+        self.windows_dir = self.root / "_windows"
+
+        for d in (self.inbox, self.processed, self.links_dir, self.windows_dir):
+            d.mkdir(parents=True, exist_ok=True)
+
+    # ------------------------------------------------------------------
+    # Discovery
+    # ------------------------------------------------------------------
+
+    def discover(self, capability: str, top_k: int = 3) -> list[dict[str, Any]]:
+        """Read _registry.json and return candidates matching capability."""
+        registry = self._load_json(self.root / "_registry.json", default={})
+        candidates: list[dict[str, Any]] = []
+        for aid, info in registry.items():
+            if aid == self.agent_id:
+                continue
+            caps = info.get("capabilities", [])
+            if capability.lower() in " ".join(caps).lower():
+                candidates.append({"agent_id": aid, **info})
+        # Sort: idle first, then by current task load
+        candidates.sort(key=lambda x: (x.get("status") != "idle", x.get("current_tasks", 0)))
+        return candidates[:top_k]
+
+    def heartbeat(self, description: str, capabilities: list[str]) -> None:
+        """Write self state into the shared _registry.json."""
+        registry = self._load_json(self.root / "_registry.json", default={})
+        registry[self.agent_id] = {
+            "description": description,
+            "capabilities": capabilities,
+            "status": "idle",
+            "last_heartbeat": int(time.time()),
+            "endpoint": "",
+        }
+        self._atomic_write(self.root / "_registry.json", registry)
+
+    # ------------------------------------------------------------------
+    # Task dispatch
+    # ------------------------------------------------------------------
+
+    def dispatch(
+        self,
+        to: str,
+        parent_task_id: str | None,
+        description: str,
+        deadline_seconds: int = 300,
+        allow_redelegation: bool = True,
+    ) -> dict[str, Any]:
+        """Write a task into the target agent's inbox and return a receipt."""
+        task_id = (
+            f"{parent_task_id}.{int(time.time())}"
+            if parent_task_id
+            else f"root_{int(time.time())}"
+        )
+
+        depth = self._get_depth(parent_task_id) if parent_task_id else 0
+        if depth >= 3:
+            return {"status": "rejected", "reason": "max_depth_exceeded"}
+
+        if parent_task_id and self._is_ancestor(to, parent_task_id):
+            return {"status": "rejected", "reason": "ancestry_loop"}
+
+        if not self._circuit_allow(to):
+            failover = self._find_failover(to)
+            return {"status": "circuit_open", "failover_to": failover}
+
+        target_inbox = self.root / to / "inbox"
+        target_inbox.mkdir(parents=True, exist_ok=True)
+        if list(target_inbox.glob(f"task_{task_id}_from_{self.agent_id}_*.json")):
+            return {"status": "dispatched", "task_id": task_id, "note": "cached"}
+
+        ancestry = (
+            (self._get_ancestry(parent_task_id) + [self.agent_id])
+            if parent_task_id
+            else [self.agent_id]
+        )
+
+        msg: dict[str, Any] = {
+            "version": "p2p/v1",
+            "type": "task_dispatch",
+            "from": self.agent_id,
+            "to": to,
+            "task_id": task_id,
+            "ancestry": ancestry,
+            "depth": depth + 1,
+            "payload": {
+                "description": description,
+                "allow_redelegation": allow_redelegation,
+            },
+            "deadline": int(time.time()) + deadline_seconds,
+            "timestamp": int(time.time()),
+        }
+
+        path = target_inbox / f"task_{task_id}_from_{self.agent_id}_{os.urandom(4).hex()}.json"
+        self._atomic_write(path, msg)
+        logger.info("P2P dispatch: {} -> {} (task_id={})", self.agent_id, to, task_id)
+        return {"status": "dispatched", "task_id": task_id, "depth": depth + 1}
+
+    def poll(self, task_id: str) -> dict[str, Any]:
+        """Scan inbox/processed and return task status."""
+        # Check processed results first
+        results = list(self.processed.glob(f"result_{task_id}_from_*.json"))
+        if results:
+            data = self._load_json(results[0])
+            payload = data.get("payload", {})
+            return {
+                "status": payload.get("outcome", "completed"),
+                "result": payload.get("content", ""),
+                "from": data["from"],
+            }
+
+        # Check inbox for results (not yet moved to processed)
+        inbox_results = list(self.inbox.glob(f"result_{task_id}_from_*.json"))
+        if inbox_results:
+            data = self._load_json(inbox_results[0])
+            payload = data.get("payload", {})
+            return {
+                "status": payload.get("outcome", "completed"),
+                "result": payload.get("content", ""),
+                "from": data["from"],
+            }
+
+        # Check inbox for pending task dispatches
+        pending = list(self.inbox.glob(f"task_{task_id}_from_*.json"))
+        if pending:
+            data = self._load_json(pending[0])
+            deadline = data.get("deadline", 0)
+            elapsed = int(time.time() - data["timestamp"])
+            if time.time() > deadline:
+                return {"status": "timeout", "elapsed": elapsed}
+            return {"status": "pending", "elapsed": elapsed}
+
+        return {"status": "not_found"}
+
+    # ------------------------------------------------------------------
+    # Aggregation (broadcast + check)
+    # ------------------------------------------------------------------
+
+    def broadcast(
+        self,
+        task_id: str,
+        subtasks: list[dict[str, Any]],
+        aggregation_timeout: int = 30,
+    ) -> dict[str, Any]:
+        """Write bid requests to candidate agents and create a window descriptor."""
+        targets: list[tuple[str, str]] = []  # (subtask_id, agent_id)
+        for sub in subtasks:
+            caps = sub.get("capability", "")
+            found = self.discover(caps, top_k=3)
+            targets.extend([(sub["subtask_id"], a["agent_id"]) for a in found])
+
+        for subtask_id, target in targets:
+            msg: dict[str, Any] = {
+                "version": "p2p/v1",
+                "type": "bid_request",
+                "from": self.agent_id,
+                "to": target,
+                "task_id": task_id,
+                "subtask_id": subtask_id,
+                "payload": sub,
+                "deadline": int(time.time()) + aggregation_timeout,
+                "timestamp": int(time.time()),
+            }
+            target_inbox = self.root / target / "inbox"
+            target_inbox.mkdir(parents=True, exist_ok=True)
+            path = target_inbox / f"bid_{task_id}_{subtask_id}_from_{self.agent_id}.json"
+            self._atomic_write(path, msg)
+
+        window: dict[str, Any] = {
+            "task_id": task_id,
+            "mode": "bid",
+            "expected": len(targets),
+            "deadline": int(time.time()) + aggregation_timeout,
+            "created_at": int(time.time()),
+        }
+        self._atomic_write(self.windows_dir / f"{task_id}.json", window)
+        logger.info(
+            "P2P broadcast: {} invited {} agents for task_id={}",
+            self.agent_id,
+            len(targets),
+            task_id,
+        )
+        return {"status": "bidding_opened", "task_id": task_id, "invited": len(targets)}
+
+    def check_aggregation(self, task_id: str) -> dict[str, Any]:
+        """Lazily check aggregation status by scanning files."""
+        window_path = self.windows_dir / f"{task_id}.json"
+        if not window_path.exists():
+            return {"status": "no_window"}
+
+        window = self._load_json(window_path)
+        mode = window.get("mode", "bid")
+        deadline = window.get("deadline", 0)
+
+        pattern = f"{mode}_{task_id}_*_from_*.json"
+        entries: list[dict[str, Any]] = []
+        for f in self.inbox.glob(pattern):
+            data = self._load_json(f)
+            entries.append(
+                {
+                    "from": data.get("from", ""),
+                    "subtask_id": data.get("subtask_id", ""),
+                    "payload": data.get("payload", {}),
+                }
+            )
+
+        is_timeout = time.time() > deadline
+        is_full = window.get("expected") and len(entries) >= window["expected"]
+
+        if is_timeout or is_full:
+            self._atomic_write(
+                self.processed / f"window_{task_id}.json",
+                {**window, "closed_at": int(time.time()), "received": len(entries)},
+            )
+            window_path.unlink(missing_ok=True)
+            return {
+                "status": "closed",
+                "mode": mode,
+                "entries": entries,
+                "reason": "timeout" if is_timeout else "full",
+            }
+
+        return {
+            "status": "pending",
+            "received": len(entries),
+            "expected": window.get("expected"),
+            "seconds_remaining": max(0, deadline - int(time.time())),
+        }
+
+    # ------------------------------------------------------------------
+    # Result reporting
+    # ------------------------------------------------------------------
+
+    def report_result(
+        self,
+        to: str,
+        task_id: str,
+        outcome: Literal["completed", "failed", "aborted"],
+        content: str,
+        callback: dict[str, Any] | None = None,
+    ) -> None:
+        """Worker calls this to write a result into the manager's inbox."""
+        msg: dict[str, Any] = {
+            "version": "p2p/v1",
+            "type": "result",
+            "from": self.agent_id,
+            "to": to,
+            "task_id": task_id,
+            "payload": {"outcome": outcome, "content": content},
+            "timestamp": int(time.time()),
+        }
+        if callback:
+            msg["callback"] = callback
+        target_inbox = self.root / to / "inbox"
+        target_inbox.mkdir(parents=True, exist_ok=True)
+        path = target_inbox / f"result_{task_id}_from_{self.agent_id}_{os.urandom(4).hex()}.json"
+        self._atomic_write(path, msg)
+        logger.info("P2P result: {} -> {} (task_id={}, outcome={})", self.agent_id, to, task_id, outcome)
+
+    # ------------------------------------------------------------------
+    # Finalization
+    # ------------------------------------------------------------------
+
+    def finalize(self, task_id: str, outcome: str, reason: str = "") -> None:
+        """Move all task files from inbox to processed and mark outcome."""
+        for src in list(self.inbox.glob(f"*{task_id}*")):
+            data = self._load_json(src)
+            data.setdefault("payload", {})
+            data["payload"]["outcome"] = outcome
+            data["payload"]["reason"] = reason
+            dst = self.processed / src.name
+            self._atomic_write(dst, data)
+            src.unlink(missing_ok=True)
+        logger.info("P2P finalize: task_id={} outcome={}", task_id, outcome)
+
+    # ------------------------------------------------------------------
+    # Circuit breaker
+    # ------------------------------------------------------------------
+
+    def _circuit_allow(self, to: str) -> bool:
+        link = self._load_json(
+            self.links_dir / f"{to}.json",
+            default={"failures": 0, "last_failure": 0, "open": False},
+        )
+        if not link.get("open"):
+            return True
+        backoff = 300 * (2 ** max(0, link.get("failures", 0) - 3))
+        if time.time() - link.get("last_failure", 0) > backoff:
+            link["open"] = False
+            self._atomic_write(self.links_dir / f"{to}.json", link)
+            return True
+        return False
+
+    def record_failure(self, to: str) -> None:
+        link = self._load_json(
+            self.links_dir / f"{to}.json",
+            default={"failures": 0, "last_failure": 0, "open": False},
+        )
+        link["failures"] = link.get("failures", 0) + 1
+        link["last_failure"] = int(time.time())
+        if link["failures"] >= 3:
+            link["open"] = True
+        self._atomic_write(self.links_dir / f"{to}.json", link)
+
+    def record_success(self, to: str) -> None:
+        link = self._load_json(
+            self.links_dir / f"{to}.json",
+            default={"failures": 0, "last_failure": 0, "open": False},
+        )
+        link["failures"] = 0
+        link["open"] = False
+        self._atomic_write(self.links_dir / f"{to}.json", link)
+
+    # ------------------------------------------------------------------
+    # Inbox scanning (for HeartbeatService)
+    # ------------------------------------------------------------------
+
+    def scan_inbox(self) -> list[dict[str, Any]]:
+        """Return all task_dispatch messages currently in inbox."""
+        messages: list[dict[str, Any]] = []
+        for f in sorted(self.inbox.glob("task_*_from_*.json"), key=lambda p: p.stat().st_mtime):
+            data = self._load_json(f)
+            # Skip expired tasks
+            if time.time() > data.get("deadline", 0):
+                continue
+            data["_filename"] = f.name
+            messages.append(data)
+        return messages
+
+    def scan_new_inbox(self, since: float | None = None) -> list[dict[str, Any]]:
+        """Return inbox messages newer than the given timestamp."""
+        messages: list[dict[str, Any]] = []
+        for f in self.inbox.glob("task_*_from_*.json"):
+            mtime = f.stat().st_mtime
+            if since is not None and mtime <= since:
+                continue
+            data = self._load_json(f)
+            if time.time() > data.get("deadline", 0):
+                continue
+            data["_filename"] = f.name
+            data["_mtime"] = mtime
+            messages.append(data)
+        return sorted(messages, key=lambda x: x.get("_mtime", 0))
+
+    def mark_processed(self, filename: str) -> None:
+        """Move a single inbox file to processed."""
+        src = self.inbox / filename
+        if not src.exists():
+            return
+        dst = self.processed / filename
+        try:
+            import shutil
+            shutil.move(str(src), str(dst))
+        except Exception:
+            logger.warning("Failed to mark processed: {}", filename)
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def _load_json(self, path: Path, default: Any | None = None) -> Any:
+        if not path.exists():
+            return default if default is not None else {}
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+
+    def _atomic_write(self, path: Path, data: dict[str, Any]) -> None:
+        tmp = path.with_suffix(".tmp")
+        with open(tmp, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        tmp.rename(path)
+
+    def _get_depth(self, task_id: str) -> int:
+        return task_id.count(".")
+
+    def _is_ancestor(self, agent_id: str, parent_task_id: str) -> bool:
+        for f in list(self.processed.glob(f"*{parent_task_id}*")) + list(
+            self.inbox.glob(f"*{parent_task_id}*")
+        ):
+            data = self._load_json(f)
+            if agent_id in data.get("ancestry", []):
+                return True
+        return False
+
+    def _get_ancestry(self, task_id: str) -> list[str]:
+        for f in list(self.processed.glob(f"*{task_id}*")) + list(
+            self.inbox.glob(f"*{task_id}*")
+        ):
+            data = self._load_json(f)
+            return data.get("ancestry", [])
+        return []
+
+    def _find_failover(self, to: str) -> str | None:
+        registry = self._load_json(self.root / "_registry.json", default={})
+        target_caps = registry.get(to, {}).get("capabilities", [])
+        for aid, info in registry.items():
+            if aid == to:
+                continue
+            if any(c in info.get("capabilities", []) for c in target_caps):
+                return aid
+        return None

nanobot/session/manager.py

@@ -8,7 +8,7 @@ from contextlib import suppress
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 from loguru import logger
 
@@ -581,6 +581,36 @@ class SessionManager:
                 return self._session_payload(repaired)
             return None
 
+    def get_or_create_task_session(
+        self,
+        base_key: str,
+        task_id: str,
+        role: Literal["manager", "worker"] = "worker",
+    ) -> Session:
+        """Get or create an isolated session for a specific task.
+
+        Key format: task:{base_key}:{task_id}:{role}
+        Example: task:slack:C123:root_qml:manager
+        """
+        task_key = f"task:{base_key}:{task_id}:{role}"
+        return self.get_or_create(task_key)
+
+    def list_task_sessions(self, base_key: str) -> list[Session]:
+        """List all task-scoped sessions for a given base key."""
+        prefix = f"task:{base_key}:"
+        return [
+            session for key, session in self._cache.items()
+            if key.startswith(prefix)
+        ]
+
+    def finalize_task_session(self, task_id: str) -> None:
+        """Mark a task session as finalized (read-only) by setting metadata."""
+        prefix = f"task:"
+        for key, session in list(self._cache.items()):
+            if f":{task_id}:" in key and key.startswith(prefix):
+                session.metadata["finalized"] = True
+                self.save(session)
+
     def list_sessions(self) -> list[dict[str, Any]]:
         """
         List all sessions.

nanobot/skills/create-instance/SKILL.md

@@ -0,0 +1,64 @@
+---
+name: create-instance
+description: "Create a new nanobot instance with separate config and workspace. Use when the user wants to set up a new bot, create a new instance for a different channel, persona, or purpose. Triggers on: create instance, new bot, set up bot, add bot, create telegram/discord/feishu/slack/wechat/wecom/dingtalk/qq/email/matrix/msteams/whatsapp bot, multi-instance setup, inter-agent communication."
+---
+
+# Create Instance
+
+Set up a new nanobot instance with its own config and workspace.
+
+## Steps
+
+1. **Collect information** (ask one at a time if not already provided):
+   - **Instance name** (required): short identifier, e.g. `telegram-bot`, `work-slack`
+   - **Channel type** (required): see table below
+   - **Model** (optional): LLM model, defaults to current instance
+
+2. **Do NOT collect secrets** in the chat (API keys, bot tokens). API keys are automatically inherited from the current instance via `--inherit-config`. Channel-specific tokens must be filled in manually after creation.
+
+3. **Run the creation script**:
+
+```bash
+python <skill-dir>/scripts/create_instance.py --name <name> --channel <channel> --inherit-config <current-config>
+```
+
+- `<skill-dir>` — the directory containing this SKILL.md
+- `<current-config>` — current instance's config path, typically `~/.nanobot/config.json`
+- Optional: `--model <model>`, `--config-dir <path>`
+
+**Exec tool constraints:**
+- Use forward-slash paths (works on all platforms)
+- Do not wrap paths in quotes
+- Do not use `cd`; pass the full script path directly
+
+4. **Report results** to the user:
+   - Config and workspace paths (script outputs them)
+   - Required fields to fill in (script lists them)
+   - Start command: `nanobot gateway --config <config-path>`
+
+## Available Channels
+
+| Channel | Key | Required Fields |
+|---------|-----|-----------------|
+| Telegram | `telegram` | token |
+| Discord | `discord` | token |
+| Feishu / Lark | `feishu` | app_id, app_secret |
+| DingTalk | `dingtalk` | client_id, client_secret |
+| Slack | `slack` | bot_token, app_token |
+| WeCom | `wecom` | bot_id, secret |
+| WeChat OA | `weixin` | token |
+| WhatsApp | `whatsapp` | bridge_token |
+| QQ | `qq` | app_id, secret |
+| Email | `email` | imap_host, imap_username, imap_password, smtp_host, smtp_username, smtp_password, from_address |
+| Matrix | `matrix` | user_id, password or access_token |
+| MS Teams | `msteams` | app_id, app_password, tenant_id |
+| MoChat | `mochat` | claw_token |
+| WebSocket | `websocket` | token |
+
+For detailed channel configuration including optional fields, see `references/channels.md`.
+
+## Troubleshooting
+
+- **"Unknown channel"**: Channel name must match the Key column exactly. Run the script without arguments to see usage.
+- **"Config already exists"**: Use a different `--name` or `--config-dir` to create in a new location.
+- **Port conflicts**: The script auto-assigns free ports for gateway and API if defaults are in use.

nanobot/skills/create-instance/references/channels.md

@@ -0,0 +1,195 @@
+# Channel Configuration Reference
+
+Detailed configuration for each supported channel.
+
+## Field Types
+
+- **Required**: defaults to empty string `""`, must be filled in before the instance can start
+- **Optional**: has a sensible default, can be customized
+
+---
+
+## telegram
+
+**Required:**
+- `token` — Bot token from @BotFather
+
+**Notable optional:**
+- `proxy` — HTTP proxy URL
+- `group_policy` — `"open"` (all messages) or `"mention"` (default, only when @mentioned)
+- `streaming` — Enable streaming responses (default: true)
+- `reply_to_message` — Reply to the triggering message (default: false)
+- `react_emoji` — Emoji for "thinking" reaction (default: `"eyes"`)
+- `inline_keyboards` — Enable inline keyboard buttons (default: false)
+
+## discord
+
+**Required:**
+- `token` — Bot token from Discord Developer Portal
+
+**Notable optional:**
+- `allow_channels` — Restrict to specific channel IDs
+- `group_policy` — `"mention"` (default) or `"open"`
+- `streaming` — Enable streaming (default: true)
+- `proxy` — HTTP proxy URL
+- `intents` — Discord gateway intents (default: 37377)
+- `read_receipt_emoji` — Emoji for read receipt
+- `working_emoji` — Emoji for "working" indicator
+
+## feishu
+
+**Required:**
+- `app_id` — Feishu app ID
+- `app_secret` — Feishu app secret
+
+**Notable optional:**
+- `encrypt_key` — Event encryption key
+- `verification_token` — Event verification token
+- `domain` — `"feishu"` (default) or `"lark"`
+- `group_policy` — `"mention"` (default) or `"open"`
+- `streaming` — Enable streaming (default: true)
+
+## dingtalk
+
+**Required:**
+- `client_id` — DingTalk app client ID
+- `client_secret` — DingTalk app client secret
+
+**Notable optional:**
+- `allow_from` — Allowed user IDs
+
+## slack
+
+**Required:**
+- `bot_token` — Bot OAuth token (`xoxb-...`)
+- `app_token` — App-level token (`xapp-...`)
+
+**Notable optional:**
+- `mode` — `"socket"` (default, Socket Mode) or `"webhook"`
+- `reply_in_thread` — Reply in thread (default: true)
+- `react_emoji` — "thinking" emoji (default: `"eyes"`)
+- `done_emoji` — "done" emoji (default: `"white_check_mark"`)
+- `group_policy` — `"mention"` (default) or `"open"`
+- `dm.enabled` — Enable DM support
+- `dm.policy` — DM policy
+- `dm.allow_from` — Allowed DM users
+
+## wecom
+
+**Required:**
+- `bot_id` — WeCom bot ID
+- `secret` — WeCom bot secret
+
+**Notable optional:**
+- `allow_from` — Allowed users
+- `welcome_message` — Welcome message for new chats
+
+## weixin
+
+**Required:**
+- `token` — WeChat Official Account token
+
+**Notable optional:**
+- `base_url` — API base URL
+- `cdn_base_url` — CDN base URL
+- `state_dir` — State persistence directory
+- `poll_timeout` — Long polling timeout
+
+## whatsapp
+
+**Required:**
+- `bridge_token` — WhatsApp bridge token (auto-generated if absent)
+
+**Notable optional:**
+- `bridge_url` — Bridge WebSocket URL (default: `"ws://localhost:3001"`)
+- `group_policy` — `"open"` (default) or `"mention"`
+
+## qq
+
+**Required:**
+- `app_id` — QQ bot app ID
+- `secret` — QQ bot secret
+
+**Notable optional:**
+- `msg_format` — `"plain"` or `"markdown"`
+- `ack_message` — Acknowledgment message text
+- `media_dir` — Media file directory
+
+## email
+
+**Required:**
+- `imap_host` — IMAP server hostname
+- `imap_username` — IMAP login username
+- `imap_password` — IMAP login password
+- `smtp_host` — SMTP server hostname
+- `smtp_username` — SMTP login username
+- `smtp_password` — SMTP login password
+- `from_address` — Sender email address
+
+**Notable optional:**
+- `imap_port` — IMAP port (default: 993)
+- `smtp_port` — SMTP port (default: 587)
+- `imap_use_ssl` — Use SSL for IMAP (default: true)
+- `smtp_use_tls` — Use TLS for SMTP (default: true)
+- `poll_interval_seconds` — Polling interval (default: 30)
+- `mark_seen` — Mark emails as read (default: true)
+- `max_body_chars` — Max email body length (default: 12000)
+- `subject_prefix` — Reply subject prefix (default: `"Re: "`)
+- `verify_dkim` — Verify DKIM signatures (default: true)
+- `verify_spf` — Verify SPF records (default: true)
+- `allowed_attachment_types` — Allowed file extensions
+- `max_attachment_size` — Max attachment size in bytes
+- `consent_granted` — Must be set to `true` for the channel to start (default: false)
+- `auto_reply_enabled` — Enable auto-reply (default: true)
+
+## matrix
+
+**Required:**
+- `user_id` — Matrix user ID (e.g. `@bot:matrix.org`)
+- `password` or `access_token` — Login password OR access token
+
+**Notable optional:**
+- `homeserver` — Homeserver URL (default: `"https://matrix.org"`)
+- `device_id` — Device ID
+- `e2eeEnabled` — Enable end-to-end encryption (default: true)
+- `group_policy` — `"open"`, `"mention"`, or `"allowlist"`
+- `streaming` — Enable streaming (default: false)
+- `max_media_bytes` — Max media file size (default: 20MB)
+
+## msteams
+
+**Required:**
+- `app_id` — Azure AD app ID
+- `app_password` — Azure AD app password/secret
+- `tenant_id` — Azure AD tenant ID
+
+**Notable optional:**
+- `host` — Listen host (default: `"0.0.0.0"`)
+- `port` — Listen port (default: 3978)
+- `reply_in_thread` — Reply in thread (default: true)
+- `validate_inbound_auth` — Validate incoming auth (default: true)
+
+## mochat
+
+**Required:**
+- `claw_token` — MoChat Claw token
+
+**Notable optional:**
+- `base_url` — API base URL
+- `socket_url` — WebSocket URL
+- `refresh_interval_ms` — Refresh interval in ms
+- `watch_timeout_ms` — Watch timeout in ms
+
+## websocket
+
+Built-in WebSocket channel for programmatic access.
+
+**Required:**
+- `token` — Authentication token (enabled by default; set `websocket_requires_token: false` to disable)
+
+**Notable optional:**
+- `host` — Listen host (default: `"127.0.0.1"`)
+- `port` — Listen port (default: 8765)
+- `allow_from` — Allowed origins (default: `["*"]`)
+- `streaming` — Enable streaming (default: true)
+

nanobot/skills/create-instance/scripts/create_instance.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+"""Create a new nanobot instance with a dedicated config and workspace.
+
+Usage:
+    create_instance.py --name <name> --channel <channel> [--model <model>] [--config-dir <dir>]
+
+Examples:
+    create_instance.py --name telegram-bot --channel telegram
+    create_instance.py --name discord-bot --channel discord --model deepseek/deepseek-chat
+    create_instance.py --name my-bot --channel telegram --config-dir ~/.nanobot-custom
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import socket
+import sys
+from pathlib import Path
+
+
+def _validate_name(name: str) -> str:
+    """Normalize and validate instance name."""
+    name = name.strip().lower()
+    name = re.sub(r"[^a-z0-9-]", "-", name)
+    name = re.sub(r"-{2,}", "-", name)
+    name = name.strip("-")
+    if not name:
+        print("[ERROR] Instance name must contain at least one letter or digit.", file=sys.stderr)
+        sys.exit(1)
+    if len(name) > 64:
+        print(f"[ERROR] Instance name too long ({len(name)} chars, max 64).", file=sys.stderr)
+        sys.exit(1)
+    return name
+
+
+def _get_available_channels() -> list[str]:
+    """Get list of available channel names without importing channel classes."""
+    from nanobot.channels.registry import discover_channel_names
+
+    return discover_channel_names()
+
+
+def _run_onboard(config_path: Path, workspace: Path) -> None:
+    """Create skeleton config + workspace using nanobot's programmatic API."""
+    from nanobot.cli.commands import _onboard_plugins
+    from nanobot.config.loader import save_config, set_config_path
+    from nanobot.config.paths import get_workspace_path
+    from nanobot.config.schema import Config
+    from nanobot.utils.helpers import sync_workspace_templates
+
+    config = Config()
+    config.agents.defaults.workspace = str(workspace)
+    set_config_path(config_path)
+    save_config(config, config_path)
+    _onboard_plugins(config_path)
+
+    workspace_path = get_workspace_path(config.workspace_path)
+    if not workspace_path.exists():
+        workspace_path.mkdir(parents=True, exist_ok=True)
+    sync_workspace_templates(workspace_path)
+
+
+def _patch_config(
+    config_path: Path,
+    *,
+    channel: str,
+    workspace: Path,
+    model: str | None,
+    name: str | None = None,
+    inherit_config_path: Path | None = None,
+) -> dict:
+    """Patch the generated config: enable channel, set workspace, optionally set model."""
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+
+    # Inherit providers and model from current instance
+    if inherit_config_path and inherit_config_path.exists():
+        try:
+            src = json.loads(inherit_config_path.read_text(encoding="utf-8"))
+
+            # Inherit providers (API keys, api_base, etc.)
+            src_providers = src.get("providers", {})
+            if src_providers:
+                data.setdefault("providers", {})
+                for key, val in src_providers.items():
+                    if isinstance(val, dict) and val.get("apiKey"):
+                        data["providers"][key] = val
+
+            # Inherit model if not explicitly overridden
+            if not model:
+                parent_model = src.get("agents", {}).get("defaults", {}).get("model")
+                if parent_model:
+                    model = parent_model
+
+        except Exception as exc:
+            print(f"[WARN] Could not inherit from {inherit_config_path}: {exc}", file=sys.stderr)
+
+    # Set workspace and model
+    data.setdefault("agents", {}).setdefault("defaults", {})
+    data["agents"]["defaults"]["workspace"] = str(workspace)
+    if model:
+        data["agents"]["defaults"]["model"] = model
+
+    # Enable the target channel
+    channels = data.setdefault("channels", {})
+    if channel in channels and isinstance(channels[channel], dict):
+        channels[channel]["enabled"] = True
+    else:
+        channels[channel] = {"enabled": True}
+
+    # Auto-assign ports if defaults are already in use
+    _assign_free_ports(data)
+
+    # Validate with Pydantic, then save
+    from nanobot.config.schema import Config
+
+    Config.model_validate(data)
+    config_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+    return data
+
+
+def _is_port_in_use(port: int, host: str = "127.0.0.1") -> bool:
+    """Check if a port is already in use."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        try:
+            s.bind((host, port))
+            return False
+        except OSError:
+            return True
+
+
+def _find_free_port(start: int, host: str = "127.0.0.1", max_tries: int = 100) -> int:
+    """Find the first free port starting from `start`."""
+    for port in range(start, start + max_tries):
+        if not _is_port_in_use(port, host):
+            return port
+    # OS-level fallback: ask the kernel for an ephemeral port
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind((host, 0))
+        return s.getsockname()[1]
+
+
+def _assign_free_ports(data: dict) -> None:
+    """If default gateway or API ports are in use, assign free ones."""
+    from nanobot.config.schema import ApiConfig, GatewayConfig
+
+    defaults = [
+        ("gateway", GatewayConfig()),
+        ("api", ApiConfig()),
+    ]
+    for key, default_cfg in defaults:
+        section = data.setdefault(key, {})
+        port = section.get("port", default_cfg.port)
+        host = section.get("host", default_cfg.host)
+        if _is_port_in_use(port, host):
+            section["port"] = _find_free_port(port + 1, host)
+
+
+def _get_channel_required_fields(channel: str) -> list[str]:
+    """Inspect a channel's default config and list fields that are empty strings."""
+    try:
+        from nanobot.channels.registry import load_channel_class
+
+        cls = load_channel_class(channel)
+        default = cls.default_config()
+        return sorted(k for k, v in default.items() if isinstance(v, str) and v == "" and k != "enabled")
+    except Exception as exc:
+        print(f"[WARN] Could not inspect channel '{channel}' defaults: {exc}", file=sys.stderr)
+        return []
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Create a new nanobot instance.",
+    )
+    parser.add_argument("--name", required=True, help="Instance name (e.g. telegram-bot)")
+    parser.add_argument("--channel", required=True, help="Channel type (e.g. telegram, discord)")
+    parser.add_argument("--model", default=None, help="LLM model (default: same as current instance)")
+    parser.add_argument(
+        "--config-dir",
+        default=None,
+        help="Config directory (default: ~/.nanobot-{name})",
+    )
+    parser.add_argument(
+        "--inherit-config",
+        default=None,
+        help="Path to current instance's config.json to copy API keys from",
+    )
+    args = parser.parse_args()
+
+    # Validate name
+    name = _validate_name(args.name)
+
+    # Validate channel
+    available = _get_available_channels()
+    if args.channel not in available:
+        print(f"[ERROR] Unknown channel: {args.channel}", file=sys.stderr)
+        print(f"Available channels: {', '.join(sorted(available))}", file=sys.stderr)
+        sys.exit(1)
+
+    # Resolve paths
+    home = Path.home()
+    config_dir = Path(args.config_dir).expanduser().resolve() if args.config_dir else home / f".nanobot-{name}"
+    config_path = config_dir / "config.json"
+    workspace = config_dir / "workspace"
+
+    # Check for duplicate
+    if config_path.exists():
+        print(f"[ERROR] Config already exists at {config_path}", file=sys.stderr)
+        print("Delete it first or use a different --config-dir.", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"Creating instance '{name}'...")
+    print(f"  Config dir: {config_dir}")
+    print(f"  Workspace:  {workspace}")
+    print(f"  Channel:    {args.channel}")
+    if args.model:
+        print(f"  Model:      {args.model}")
+
+    # Run onboard
+    _run_onboard(config_path, workspace)
+
+    # Patch config
+    inherit_path = Path(args.inherit_config).expanduser().resolve() if args.inherit_config else None
+    _patch_config(
+        config_path,
+        channel=args.channel,
+        workspace=workspace,
+        model=args.model,
+        name=name,
+        inherit_config_path=inherit_path,
+    )
+
+    # Report
+    print(f"\n[OK] Instance '{name}' created successfully.")
+    print(f"  Config: {config_path}")
+    print(f"  Workspace: {workspace}")
+
+    # List fields the user needs to fill in
+    required_fields = _get_channel_required_fields(args.channel)
+    if required_fields:
+        print(f"\n[IMPORTANT] Edit {config_path} and fill in these fields:")
+        for field in required_fields:
+            print(f"  - channels.{args.channel}.{field}")
+
+    print(f"\nTo start the instance:")
+    print(f"  nanobot gateway --config {config_path}")
+
+
+if __name__ == "__main__":
+    main()

tests/skills/test_create_instance.py

@@ -0,0 +1,168 @@
+"""Tests for nanobot/skills/create-instance/scripts/create_instance.py."""
+
+from __future__ import annotations
+
+import json
+import socket
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).parent.parent.parent / "nanobot" / "skills" / "create-instance" / "scripts" / "create_instance.py"
+
+
+@pytest.fixture
+def tmp_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Point HOME at a temp dir so nanobot writes configs there."""
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.delenv("NANOBOT_CONFIG", raising=False)
+    return tmp_path
+
+
+def _run_script(*args: str, cwd: Path | None = None) -> subprocess.CompletedProcess:
+    """Run create_instance.py as a subprocess."""
+    return subprocess.run(
+        [sys.executable, str(SCRIPT), *args],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        cwd=cwd,
+    )
+
+
+class TestValidation:
+    """Argument validation tests."""
+
+    def test_missing_required_args_exits_with_error(self) -> None:
+        result = _run_script()
+        assert result.returncode != 0
+
+    def test_invalid_channel_exits_with_error(self, tmp_home: Path) -> None:
+        result = _run_script("--name", "test", "--channel", "nonexistent_channel")
+        assert result.returncode != 0
+        assert "nonexistent_channel" in result.stderr or "nonexistent_channel" in result.stdout
+
+
+class TestCreateInstance:
+    """End-to-end instance creation tests."""
+
+    def test_creates_config_and_workspace(self, tmp_home: Path) -> None:
+        config_dir = tmp_home / ".nanobot-test"
+        result = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+        )
+        assert result.returncode == 0, result.stderr
+
+        config_path = config_dir / "config.json"
+        assert config_path.exists(), f"Config not created at {config_path}"
+
+        workspace = config_dir / "workspace"
+        assert workspace.exists(), f"Workspace not created at {workspace}"
+
+    def test_config_has_channel_enabled(self, tmp_home: Path) -> None:
+        config_dir = tmp_home / ".nanobot-test"
+        result = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+        )
+        assert result.returncode == 0, result.stderr
+
+        data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
+        assert data["channels"]["telegram"]["enabled"] is True
+
+    def test_config_workspace_path_set(self, tmp_home: Path) -> None:
+        config_dir = tmp_home / ".nanobot-test"
+        result = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+        )
+        assert result.returncode == 0, result.stderr
+
+        data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
+        ws = data["agents"]["defaults"]["workspace"]
+        assert str(config_dir / "workspace") in ws or "workspace" in ws
+
+    def test_model_override(self, tmp_home: Path) -> None:
+        config_dir = tmp_home / ".nanobot-test"
+        result = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--model", "deepseek/deepseek-chat",
+            "--config-dir", str(config_dir),
+        )
+        assert result.returncode == 0, result.stderr
+
+        data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
+        assert data["agents"]["defaults"]["model"] == "deepseek/deepseek-chat"
+
+    def test_rejects_duplicate_instance(self, tmp_home: Path) -> None:
+        config_dir = tmp_home / ".nanobot-test"
+        result1 = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+        )
+        assert result1.returncode == 0
+
+        result2 = _run_script(
+            "--name", "test-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+        )
+        assert result2.returncode != 0
+
+    def test_port_reassigned_when_default_in_use(self, tmp_home: Path) -> None:
+        """When default gateway port is occupied, script should pick a different one."""
+        config_dir = tmp_home / ".nanobot-test"
+
+        # Bind to the default gateway port to simulate a running instance
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as blocker:
+            blocker.bind(("127.0.0.1", 18790))
+            blocker.listen(1)
+
+            result = _run_script(
+                "--name", "test-bot",
+                "--channel", "telegram",
+                "--config-dir", str(config_dir),
+            )
+            assert result.returncode == 0, result.stderr
+
+        data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
+        assert data["gateway"]["port"] != 18790
+
+    def test_inherits_api_key_from_current_instance(self, tmp_home: Path) -> None:
+        """API keys from --inherit-config should be copied to new instance."""
+        # Create a fake "current instance" config with an API key
+        src_dir = tmp_home / ".nanobot-current"
+        src_dir.mkdir()
+        src_config = src_dir / "config.json"
+        src_config.write_text(json.dumps({
+            "providers": {
+                "anthropic": {"apiKey": "sk-test-key-12345"},
+                "deepseek": {"apiKey": "dsk-another-key"},
+                "openai": {},  # no key, should not be copied
+            },
+        }), encoding="utf-8")
+
+        config_dir = tmp_home / ".nanobot-new"
+        result = _run_script(
+            "--name", "new-bot",
+            "--channel", "telegram",
+            "--config-dir", str(config_dir),
+            "--inherit-config", str(src_config),
+        )
+        assert result.returncode == 0, result.stderr
+
+        data = json.loads((config_dir / "config.json").read_text(encoding="utf-8"))
+        providers = data.get("providers", {})
+        assert providers.get("anthropic", {}).get("apiKey") == "sk-test-key-12345"
+        assert providers.get("deepseek", {}).get("apiKey") == "dsk-another-key"
+        # openai had no key, so it should not be in the new config's providers
+        assert providers.get("openai", {}).get("apiKey") is None