feat: add MCP live collab sessions

cmd/clawdchan/cmd_setup.go

@@ -514,10 +514,11 @@ func promptString(prompt, defaultVal string) string {
 // body of both files matches; change them together.
 const clawdchanGuideMarkdown = "# ClawdChan agent guide\n\n" +
 	"You have the ClawdChan MCP tools (`clawdchan_*`). The surface is\n" +
-	"peer-centric and deliberately small: four tools cover everything —\n" +
+	"peer-centric and deliberately small: four core tools cover one-shot work —\n" +
 	"`clawdchan_toolkit`, `clawdchan_pair`, `clawdchan_message`,\n" +
-	"`clawdchan_inbox`. Thread IDs never surface. This file is your\n" +
-	"operator manual — how to act, not what the tools do.\n\n" +
+	"`clawdchan_inbox`. MCP clients also have first-class live-collab\n" +
+	"session tools named `clawdchan_collab_*`. Thread IDs never surface.\n" +
+	"This file is your operator manual — how to act, not what the tools do.\n\n" +
 	"## First action every session\n\n" +
 	"Call `clawdchan_toolkit`. It returns `self`, the list of paired\n" +
 	"`peers` with per-peer stats, and a `setup.user_message`. If\n" +
@@ -548,23 +549,24 @@ const clawdchanGuideMarkdown = "# ClawdChan agent guide\n\n" +
 	"  On **live**, delegate to a Task sub-agent. Do NOT run the loop on\n" +
 	"  your own turn; it freezes the user. Brief the sub-agent:\n\n" +
 	"  > You own a live ClawdChan collaboration with peer_id `<hex>`\n" +
-	"  > about `<problem>`. First action is a **liveness probe**:\n" +
-	"  > `clawdchan_message(peer, text=\"<one-line 'live on <topic>?'\n" +
-	"  > check>\", intent='ask', collab=true)` →\n" +
-	"  > `clawdchan_inbox(peer_id=<hex>, wait_seconds=15)`. If the poll\n" +
-	"  > returns `new: 0`, exit with \"peer not live on this\" — do not\n" +
-	"  > keep sending. If it returns fresh envelopes from the peer,\n" +
-	"  > record the `next_cursor` and enter the loop:\n" +
-	"  > `clawdchan_message(peer, text, intent='ask', collab=true)` →\n" +
-	"  > `clawdchan_inbox(peer_id=<hex>, wait_seconds=<T>,\n" +
-	"  > after_cursor=<last next_cursor>)` → integrate → respond. Size\n" +
-	"  > `<T>` to the work: ~10s for quick clarifications, 30–60s for\n" +
-	"  > design-level turns — not a fixed 10s. Converge on `<definition\n" +
-	"  > of done>`. Stop after `<N>` rounds, 2–3 consecutive empty polls\n" +
-	"  > (\"peer went silent\"), or any error. Return a summary: what was\n" +
-	"  > agreed, open questions, closing message. Do not ask the user\n" +
-	"  > anything. Always set `collab=true` on outbound — that tags the\n" +
-	"  > envelope so the peer knows a sub-agent is waiting.\n\n" +
+	"  > about `<problem>`. Pick a stable `owner_id` for yourself, then\n" +
+	"  > call `clawdchan_collab_start(peer_id=<hex>, topic=<topic>,\n" +
+	"  > definition_of_done=<done>, max_rounds=<N>, owner_id=<owner>)`.\n" +
+	"  > Send the liveness probe with\n" +
+	"  > `clawdchan_collab_send(session_id, text=\"<one-line live on\n" +
+	"  > <topic>? check>\", intent=\"ask\", owner_id=<owner>)`, then\n" +
+	"  > `clawdchan_collab_await(session_id, wait_seconds=15,\n" +
+	"  > heartbeat=true, owner_id=<owner>)`. If it returns `new: 0`,\n" +
+	"  > close with `status=\"timed_out\"` and summary \"peer not live on\n" +
+	"  > this\" — do not keep sending. If it returns fresh peer envelopes,\n" +
+	"  > enter the loop: integrate → `clawdchan_collab_send` →\n" +
+	"  > `clawdchan_collab_await`. Size waits to the work: ~10s for quick\n" +
+	"  > clarifications, 30–60s for design-level turns. Converge on\n" +
+	"  > `<definition of done>`. Stop after `<N>` rounds, 2–3 consecutive\n" +
+	"  > empty polls (\"peer went silent\"), a lease error, or any tool\n" +
+	"  > error. Close with `clawdchan_collab_close(session_id, status,\n" +
+	"  > summary, close_reason)` and return the same summary to the main\n" +
+	"  > agent. Do not ask the user anything.\n\n" +
 	"  Free the main turn. Tell the user the loop is running; you'll\n" +
 	"  surface the result when it converges or the probe fails. If the\n" +
 	"  probe reports \"not live\", tell the user and offer a one-shot\n" +
@@ -576,7 +578,9 @@ const clawdchanGuideMarkdown = "# ClawdChan agent guide\n\n" +
 	"> X's agent is waiting live: *\"<preview>\"*. Engage live (I'll spawn\n" +
 	"> my own sub-agent) or handle at your pace?\n\n" +
 	"Live → spawn a Task sub-agent with the same loop shape, skipping\n" +
-	"the probe (the peer already opened the channel).\n" +
+	"the probe (the peer already opened the channel). The sub-agent should\n" +
+	"start or resume a `clawdchan_collab_*` session, then send its reply\n" +
+	"through `clawdchan_collab_send`.\n" +
 	"Paced → reply once with `clawdchan_message` (no `collab=true`); the\n" +
 	"sender's sub-agent detects the slower cadence and closes cleanly.\n\n" +
 	"**ask_human is not yours to answer.** Items in\n" +

core/node/collab.go

@@ -0,0 +1,136 @@
+package node
+
+import (
+	"context"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/agents-first/clawdchan/core/envelope"
+	"github.com/agents-first/clawdchan/core/identity"
+	"github.com/agents-first/clawdchan/core/store"
+)
+
+const (
+	CollabStatusActive    = "active"
+	CollabStatusWaiting   = "waiting"
+	CollabStatusConverged = "converged"
+	CollabStatusTimedOut  = "timed_out"
+	CollabStatusCancelled = "cancelled"
+	CollabStatusClosed    = "closed"
+)
+
+var ErrCollabLeaseHeld = errors.New("collab session lease held by another owner")
+
+type CollabCreateOptions struct {
+	PeerID           identity.NodeID
+	ThreadID         envelope.ThreadID
+	Topic            string
+	LastCursor       string
+	MaxRounds        int
+	DefinitionOfDone string
+	OwnerID          string
+	LeaseDuration    time.Duration
+}
+
+func (n *Node) CreateCollabSession(ctx context.Context, opts CollabCreateOptions) (store.CollabSession, error) {
+	if opts.OwnerID == "" {
+		opts.OwnerID = newCollabID("owner")
+	}
+	now := time.Now().UnixMilli()
+	leaseExpires := now
+	if opts.LeaseDuration > 0 {
+		leaseExpires = now + opts.LeaseDuration.Milliseconds()
+	}
+	cs := store.CollabSession{
+		SessionID:        newCollabID("collab"),
+		PeerID:           opts.PeerID,
+		ThreadID:         opts.ThreadID,
+		Topic:            opts.Topic,
+		Status:           CollabStatusActive,
+		LastCursor:       opts.LastCursor,
+		MaxRounds:        opts.MaxRounds,
+		DefinitionOfDone: opts.DefinitionOfDone,
+		OwnerID:          opts.OwnerID,
+		HeartbeatMs:      now,
+		LeaseExpiresMs:   leaseExpires,
+		CreatedMs:        now,
+		UpdatedMs:        now,
+		LastActivityMs:   now,
+	}
+	if err := n.store.CreateCollabSession(ctx, cs); err != nil {
+		return store.CollabSession{}, err
+	}
+	return cs, nil
+}
+
+func (n *Node) GetCollabSession(ctx context.Context, sessionID string) (store.CollabSession, error) {
+	return n.store.GetCollabSession(ctx, sessionID)
+}
+
+func (n *Node) ListCollabSessions(ctx context.Context, activeOnly bool) ([]store.CollabSession, error) {
+	return n.store.ListCollabSessions(ctx, activeOnly)
+}
+
+func (n *Node) HeartbeatCollabSession(ctx context.Context, sessionID, ownerID string, leaseDuration time.Duration) (store.CollabSession, error) {
+	cs, err := n.store.GetCollabSession(ctx, sessionID)
+	if err != nil {
+		return store.CollabSession{}, err
+	}
+	if ownerID == "" {
+		ownerID = cs.OwnerID
+	}
+	now := time.Now().UnixMilli()
+	if cs.OwnerID != "" && cs.OwnerID != ownerID && cs.LeaseExpiresMs > now {
+		return store.CollabSession{}, fmt.Errorf("%w: owner=%s lease_expires_ms=%d", ErrCollabLeaseHeld, cs.OwnerID, cs.LeaseExpiresMs)
+	}
+	cs.OwnerID = ownerID
+	cs.HeartbeatMs = now
+	cs.LeaseExpiresMs = now
+	if leaseDuration > 0 {
+		cs.LeaseExpiresMs = now + leaseDuration.Milliseconds()
+	}
+	cs.UpdatedMs = now
+	if err := n.store.UpdateCollabSession(ctx, cs); err != nil {
+		return store.CollabSession{}, err
+	}
+	return cs, nil
+}
+
+func (n *Node) UpdateCollabSession(ctx context.Context, cs store.CollabSession) error {
+	cs.UpdatedMs = time.Now().UnixMilli()
+	return n.store.UpdateCollabSession(ctx, cs)
+}
+
+func (n *Node) UpdateCollabCursor(ctx context.Context, sessionID, cursor, status string, activity bool) (store.CollabSession, error) {
+	cs, err := n.store.GetCollabSession(ctx, sessionID)
+	if err != nil {
+		return store.CollabSession{}, err
+	}
+	now := time.Now().UnixMilli()
+	cs.LastCursor = cursor
+	if status != "" {
+		cs.Status = status
+	}
+	cs.UpdatedMs = now
+	if activity {
+		cs.LastActivityMs = now
+	}
+	if err := n.store.UpdateCollabSession(ctx, cs); err != nil {
+		return store.CollabSession{}, err
+	}
+	return cs, nil
+}
+
+func (n *Node) CloseCollabSession(ctx context.Context, sessionID, status, summary, reason string) error {
+	if status == "" {
+		status = CollabStatusClosed
+	}
+	return n.store.CloseCollabSession(ctx, sessionID, status, summary, reason, time.Now().UnixMilli())
+}
+
+func newCollabID(prefix string) string {
+	id := newULID()
+	return prefix + "-" + hex.EncodeToString(id[:])
+}

core/node/node_test.go

@@ -2,14 +2,17 @@ package node_test
 
 import (
 	"context"
+	"errors"
 	"net/http/httptest"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/agents-first/clawdchan/core/envelope"
+	"github.com/agents-first/clawdchan/core/identity"
 	"github.com/agents-first/clawdchan/core/node"
+	"github.com/agents-first/clawdchan/core/pairing"
 	"github.com/agents-first/clawdchan/core/surface"
 	"github.com/agents-first/clawdchan/internal/relayserver"
 )
@@ -139,6 +142,46 @@ saySuccess:
 	t.Fatal("bob's human was never notified")
 }
 
+func TestCollabHeartbeatLeaseOwnership(t *testing.T) {
+	ctx := context.Background()
+	n := mkNode(t, "ws://127.0.0.1:1", "alice", nil)
+	peerIdentity, _ := identity.Generate()
+	peer := pairing.Peer{
+		NodeID:     peerIdentity.SigningPublic,
+		KexPub:     peerIdentity.KexPublic,
+		Alias:      "bob",
+		Trust:      pairing.TrustPaired,
+		PairedAtMs: time.Now().UnixMilli(),
+	}
+	if err := n.Store().UpsertPeer(ctx, peer); err != nil {
+		t.Fatal(err)
+	}
+	threadID, err := n.OpenThread(ctx, peer.NodeID, "lease")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cs, err := n.CreateCollabSession(ctx, node.CollabCreateOptions{
+		PeerID:        peer.NodeID,
+		ThreadID:      threadID,
+		OwnerID:       "owner-a",
+		LeaseDuration: 50 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := n.HeartbeatCollabSession(ctx, cs.SessionID, "owner-b", time.Second); !errors.Is(err, node.ErrCollabLeaseHeld) {
+		t.Fatalf("expected live lease rejection, got %v", err)
+	}
+	time.Sleep(70 * time.Millisecond)
+	claimed, err := n.HeartbeatCollabSession(ctx, cs.SessionID, "owner-b", time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if claimed.OwnerID != "owner-b" {
+		t.Fatalf("expected owner-b to claim expired lease, got %q", claimed.OwnerID)
+	}
+}
+
 func TestAskHumanRoundTrip(t *testing.T) {
 	relay := spinRelay(t)
 	alice := mkNode(t, relay, "alice", nil)