feat(comms): OpenDM handler + AsAccount adapter + relay dispatch + spawn auto-open (RIG-2964)

go/internal/comms/agent_caller.go

@@ -338,6 +338,27 @@ func (c *Comms) CreateChannelGroupAsAccount(
 	return resp.Msg, nil
 }
 
+// OpenDMAsAccount executes one agent-initiated OpenDM as account, mirroring
+// UpdateChannelMembersAsAccount: WithActor + the shared OpenDM handler path, so
+// the peer resolve, the same-owner authz, the reserved-DM-group upsert, and the
+// post-commit ChannelChanged fan-out are identical to a human caller's. An
+// unknown, cross-owner, or self peer collapses to the same code a human gets. The
+// request names the peer by handle, so there is no home-channel defaulting here.
+func (c *Comms) OpenDMAsAccount(
+	ctx context.Context,
+	account store.AccountID,
+	req *compassv1.OpenDMRequest,
+) (*compassv1.OpenDMResponse, error) {
+	if account == "" {
+		return nil, errNoActor
+	}
+	resp, err := c.OpenDM(WithActor(ctx, account), connect.NewRequest(req))
+	if err != nil {
+		return nil, err
+	}
+	return resp.Msg, nil
+}
+
 // CommitAgentPost commits one relayed MessagePosted frame as a durable comms row
 // under account. It builds a PostMessageRequest from the frame's blocks and
 // delegates to PostAsAccount, so this is the SAME PostMessage handler path a

go/internal/comms/comms.go

@@ -637,18 +637,72 @@ func (c *Comms) UpdatePinnedBoard(
 }
 
 // OpenDM resolves-or-creates the two-party DM channel between the caller and a
-// peer, addressed by handle (RIG-2962). T1 lands the contract (proto + regen)
-// proto-first; the real handler — caller/peer resolve, same-owner authz, the
-// reserved-DM-group upsert, and the post-commit ChannelChanged emit — is the
-// T3 leg (compass-agent-peer-dm design.md T3), which replaces this stub. Until
-// then it returns CodeUnimplemented so *Comms satisfies the generated
-// CommsServiceHandler (asserted with no Unimplemented embed) without pretending
-// to serve a surface whose store legs (T2 dm.go) do not exist yet.
+// peer, addressed by handle (RIG-2962 T3, design.md T3:745-762). The caller is
+// the actor on the connection; the peer is resolved owner-namespaced (resolve.go
+// AgentByHandle), and both must share the caller's owner. Unknown, cross-owner,
+// and self-handle-that-resolves-to-the-caller all collapse oracle-safe: an
+// unknown OR cross-owner handle is the byte-identical merged NOT_FOUND naming the
+// submitted handle (a foreign peer's existence is never leaked), and a self-DM is
+// CodeInvalidArgument. The name is the deterministic sorted-handle pair, so
+// open(a,b) and open(b,a) resolve the same channel; the whole open runs in one
+// store tx (lock → ensure group → upsert) and a create fans a best-effort
+// post-commit ChannelChanged (a resume emits nothing).
 func (c *Comms) OpenDM(
-	_ context.Context,
-	_ *connect.Request[compassv1.OpenDMRequest],
+	ctx context.Context,
+	req *connect.Request[compassv1.OpenDMRequest],
 ) (*connect.Response[compassv1.OpenDMResponse], error) {
-	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("comms: OpenDM not implemented until RIG-2962 T3"))
+	caller := c.actorFromContext(ctx)
+
+	// Resolve the peer owner-namespaced (bare → the caller's own owner). An
+	// unknown, wrong-owner, or non-agent handle is the merged NOT_FOUND naming
+	// the submitted handle.
+	peer, err := c.resolveAgentAccount(ctx, caller, req.Msg.GetPeerHandle())
+	if err != nil {
+		return nil, edgeError(err)
+	}
+
+	// Self-DM guard: a handle that resolves to the caller itself is not a peer.
+	if peer.ID == caller {
+		return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("comms: cannot open a DM with yourself"))
+	}
+
+	// Same-owner authz. resolveAgentAccount for a BARE handle already resolves in
+	// the caller's own owner namespace, so a bare peer is same-owner by
+	// construction; the check bites an owner-QUALIFIED handle naming another
+	// owner's agent. A cross-owner peer is byte-identical to an unknown one
+	// (oracle-safe, mirroring ReparentAgent's remap at comms.go:321-323) — the
+	// merged NOT_FOUND naming the submitted handle, never leaking the peer's
+	// existence.
+	owner, err := c.store.ResolveOwner(ctx, caller)
+	if err != nil {
+		return nil, edgeError(err)
+	}
+	if peer.Agent.OwnerUserID != owner {
+		return nil, edgeError(notFoundHandle(store.ErrNotFound, req.Msg.GetPeerHandle()))
+	}
+
+	// The deterministic name is keyed on the two HANDLES: the caller's own handle
+	// and the peer's, sorted lexicographically.
+	callerAcc, err := c.store.GetAccount(ctx, caller)
+	if err != nil {
+		return nil, edgeError(err)
+	}
+	name := dmChannelName(callerAcc.Handle, peer.Handle)
+
+	channelID, created, err := c.openDMTx(ctx, owner, name, []store.AccountID{caller, peer.ID})
+	if err != nil {
+		return nil, edgeError(err)
+	}
+
+	ch, err := c.store.GetChannel(ctx, channelID)
+	if err != nil {
+		return nil, edgeError(err)
+	}
+	c.emitDMCreated(ch, created)
+	return connect.NewResponse(&compassv1.OpenDMResponse{
+		Channel: channelToWire(ch),
+		Created: created,
+	}), nil
 }
 
 // applyBoardOp maps the request's op oneof to its store call: a plain pin

go/internal/comms/dm.go

@@ -0,0 +1,76 @@
+package comms
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/RigelBuild/compass/go/internal/store"
+)
+
+// dmChannelName derives the deterministic sorted-handle pair name for a peer DM.
+// Handles are sorted lexicographically so open(a,b) and open(b,a) resolve the
+// same channel (store dm_pgtest_test.go proves reversed member order → same name).
+//
+// The separator is `:`, a byte the handle grammar (store handle.go handleRE
+// `^[a-z0-9][a-z0-9._-]*$`) excludes — so no handle can contain it and the split
+// is unambiguous. A hyphen delimiter would NOT be injective: handles may contain
+// `--` (the grammar permits consecutive hyphens), so `dm--a--b--c` is ambiguous
+// between the pairs {a, b--c} and {a--b, c}, which would resolve two distinct
+// logical DMs onto ONE channel and cross-add members (a same-owner private-DM
+// confidentiality break). `:` cannot appear in a handle, so `dm:<lo>:<hi>` maps
+// each unordered pair to exactly one name.
+func dmChannelName(h1, h2 string) string {
+	lo, hi := h1, h2
+	if lo > hi {
+		lo, hi = hi, lo
+	}
+	return "dm:" + lo + ":" + hi
+}
+
+// openDMTx runs the whole peer-DM open for owner in ONE store transaction — the
+// same shape the store's openDM test helper (dm_pgtest_test.go:27-49) and
+// EnsureCoordinationChannel (coordination.go:179-195) use: take the per-owner DM
+// advisory lock, ensure the owner's reserved __dm__ group, then upsert the
+// deterministic-name channel for the two agent parties. Returns the resolved
+// channel id and whether it was created this call (a resume returns false). The
+// lock serializes every open under owner's DM namespace, so the group-ensure and
+// the channel-upsert cannot race a concurrent first-open into two groups or two
+// channels.
+func (c *Comms) openDMTx(ctx context.Context, owner store.AccountID, name string, members []store.AccountID) (store.ChannelID, bool, error) {
+	var (
+		channelID store.ChannelID
+		created   bool
+	)
+	if err := c.store.WithTx(ctx, func(tx pgx.Tx) error {
+		if err := store.LockOwnerDMTx(ctx, tx, owner); err != nil {
+			return err
+		}
+		gid, err := c.store.EnsureOwnerDMGroupTx(ctx, tx, owner)
+		if err != nil {
+			return err
+		}
+		channelID, created, err = c.store.UpsertDMChannelTx(ctx, tx, store.DMChannelSpec{
+			GroupID: gid,
+			Name:    name,
+			Members: members,
+		})
+		return err
+	}); err != nil {
+		return "", false, err
+	}
+	return channelID, created, nil
+}
+
+// emitDMCreated fans a best-effort ChannelChanged after a DM open COMMITTED and
+// only when the channel was created this call — the coordination hook's
+// post-commit emit posture (coordination.go:158-171): a resume (created=false)
+// is a no-op event-wise (nothing changed). It takes the channel the caller
+// already read for the response, so the create path does not re-read the same row
+// a second time. NEVER call before the commit.
+func (c *Comms) emitDMCreated(ch store.Channel, created bool) {
+	if !created {
+		return
+	}
+	c.publishChannelChanged(ch, nil)
+}