From b12595442e714b525159911ead95f52b43b824e2 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 30 Aug 2026 17:34:31 -0400 Subject: [PATCH 1/3] feat(comms): OpenDM handler + AsAccount adapter + relay dispatch + spawn auto-open (RIG-2964) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Peer-DM task T3: wire the `OpenDM` RPC end to end and auto-open the manager↔peer DM at spawn. Builds on T1 (proto, #738) and T2 (store `dm.go`, #746), both merged. ### What changed - **`Comms.OpenDM` handler** (`internal/comms/comms.go`) — replaces the T1 `CodeUnimplemented` stub. Resolves the peer owner-namespaced (`resolveAgentAccount`), guards self-DM (`invalid_argument`), enforces same-owner authz collapsed oracle-safe to the merged `not_found` (a cross-owner peer is byte-identical to an unknown one, mirroring `ReparentAgent`), derives the deterministic sorted-handle name, and runs the whole open in one store tx. - **`internal/comms/dm.go`** (new) — `dmChannelName` (sorted-handle `dm----`), `openDMTx` (one-tx `LockOwnerDMTx` → `EnsureOwnerDMGroupTx` → `UpsertDMChannelTx`, mirroring the store's `openDM` test helper and `EnsureCoordinationChannel`), and `emitDMCreated` (best-effort post-commit `ChannelChanged`, create-only). - **`OpenDMAsAccount` adapter** (`internal/comms/agent_caller.go`) — the agent-tool entry, mirroring `UpdateChannelMembersAsAccount` (`errNoActor` guard + `WithActor` + the shared handler path). - **Relay dispatch** — `CommsCaller` gains `OpenDMAsAccount`; `executeCall` gains the `open_dm` arm; `fakeCommsCaller` gains the recorder. - **Spawn auto-open (R8)** (`server/lifecycle.go`) — `lifecycleService` gains a narrow `dmOpener` seam; `SpawnAsAccount` sets `DmChannelName` from one post-spawn site covering fresh/resume/idempotent paths (OpenDM is resolve-or-create, so a re-spawn returns the same DM). A post-spawn open failure is logged and returns an empty `dm_channel_name` — never a spawn rollback (the DM is recoverable next turn via `comms_open_dm`). ### Tests - `internal/comms/dm_open_pgtest_test.go` (new): same-owner create (kind=DM ∧ mandatory ∧ both parties ∧ deterministic name); reopen resumes the same channel in either handle order; unknown / cross-owner → `not_found`; self → `invalid_argument`; empty account → `errNoActor`; `ChannelChanged` on create; `AsAccount` resolve-or-create. - `internal/runnerhub/relay_open_dm_test.go` (new): the `open_dm` arm forwards under the bound account, result oneof + `call_id` round-trip, tool error rendered in-band. - `server/lifecycle_pgtest_test.go`: spawn returns a live `dm_channel_name` whose channel is a real DM; idempotent re-spawn returns the same name. All green against a real Postgres; build + vet + gofmt + golangci-lint (0 issues) clean. Spec-impact: none. Refs RIG-2964 Co-authored-by: Matt Wilkinson --- go/internal/comms/agent_caller.go | 21 ++ go/internal/comms/comms.go | 74 ++++++- go/internal/comms/dm.go | 74 +++++++ go/internal/comms/dm_open_pgtest_test.go | 196 +++++++++++++++++++ go/internal/runnerhub/helpers_test.go | 14 ++ go/internal/runnerhub/hub.go | 7 +- go/internal/runnerhub/relay_comms.go | 10 +- go/internal/runnerhub/relay_open_dm_test.go | 91 +++++++++ go/server/lifecycle.go | 73 ++++++- go/server/lifecycle_e2e_pgtest_test.go | 2 +- go/server/lifecycle_pgtest_test.go | 105 +++++++++- go/server/lifecycle_test.go | 2 +- go/server/lifecycle_wake_pgtest_test.go | 2 +- go/server/offline_mention_e2e_pgtest_test.go | 2 +- go/server/sinks.go | 6 +- 15 files changed, 650 insertions(+), 29 deletions(-) create mode 100644 go/internal/comms/dm.go create mode 100644 go/internal/comms/dm_open_pgtest_test.go create mode 100644 go/internal/runnerhub/relay_open_dm_test.go diff --git a/go/internal/comms/agent_caller.go b/go/internal/comms/agent_caller.go index 8b053b7fb..74ae65ed6 100644 --- a/go/internal/comms/agent_caller.go +++ b/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 diff --git a/go/internal/comms/comms.go b/go/internal/comms/comms.go index 02ad2dfcf..5005ead2c 100644 --- a/go/internal/comms/comms.go +++ b/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(ctx, channelID, 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 diff --git a/go/internal/comms/dm.go b/go/internal/comms/dm.go new file mode 100644 index 000000000..2ef28b5f5 --- /dev/null +++ b/go/internal/comms/dm.go @@ -0,0 +1,74 @@ +package comms + +import ( + "context" + "log/slog" + + "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). +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), and a re-read failure is logged and +// swallowed, never propagated, since the channel already committed and the event +// self-heals on the next open. NEVER call before the commit. +func (c *Comms) emitDMCreated(ctx context.Context, channelID store.ChannelID, created bool) { + if !created { + return + } + ch, err := c.store.GetChannel(ctx, channelID) + if err != nil { + slog.WarnContext(ctx, "comms: post-commit DM channel read for event failed; self-heals on next open", + "channel_id", string(channelID), "error", err.Error()) + return + } + c.publishChannelChanged(ch, nil) +} diff --git a/go/internal/comms/dm_open_pgtest_test.go b/go/internal/comms/dm_open_pgtest_test.go new file mode 100644 index 000000000..2229bbd2c --- /dev/null +++ b/go/internal/comms/dm_open_pgtest_test.go @@ -0,0 +1,196 @@ +//go:build pgtest + +package comms + +// The Comms.OpenDM handler + OpenDMAsAccount adapter (RIG-2962 T3, design.md +// T3:745-773): resolve-or-create the two-party peer DM addressed by handle, with +// same-owner authz, the deterministic sorted-handle name, and the post-commit +// ChannelChanged emit on create. Driven in-process via connect.NewRequest + +// WithActor (and the AsAccount adapter) against a real store and bus, mirroring +// comms_test.go / org_mgmt_pgtest_test.go. context.Background() is the test root +// (test-root ctx exemption). + +import ( + "context" + "testing" + + "connectrpc.com/connect" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" +) + +// TestOpenDMSameOwnerCreatesDMChannel: an agent opens a DM with a same-owner +// peer → the returned channel is a real DM (kind=DM, mandatory subscription, +// both parties members), created=true, and the deterministic sorted-handle name +// (dm----) is used. +func TestOpenDMSameOwnerCreatesDMChannel(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + bob := mustAgent(t, st, owner.ID, "bob") + + resp, err := svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "bob"})) + if err != nil { + t.Fatalf("OpenDM = %v, want success", err) + } + if !resp.Msg.GetCreated() { + t.Fatalf("created = false, want true on a first open") + } + ch := resp.Msg.GetChannel() + if ch.GetName() != "dm--alice--bob" { + t.Fatalf("channel name = %q, want dm--alice--bob (deterministic sorted-handle)", ch.GetName()) + } + if ch.GetKind() != compassv1.ChannelKind_CHANNEL_KIND_DM { + t.Fatalf("channel kind = %v, want CHANNEL_KIND_DM", ch.GetKind()) + } + if !ch.GetMandatorySubscription() { + t.Fatalf("mandatory_subscription = false, want true (born-mandatory DM)") + } + if !containsString(ch.GetMemberAccountIds(), string(alice.ID)) || !containsString(ch.GetMemberAccountIds(), string(bob.ID)) { + t.Fatalf("members = %v, want both alice %s and bob %s", ch.GetMemberAccountIds(), alice.ID, bob.ID) + } +} + +// TestOpenDMReopenResumesSameChannel: a second open of the same pair — in EITHER +// handle order — resumes the SAME channel (created=false, same id), proving the +// deterministic name is order-independent and the upsert resolves the existing row. +func TestOpenDMReopenResumesSameChannel(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + bob := mustAgent(t, st, owner.ID, "bob") + + first, err := svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "bob"})) + if err != nil { + t.Fatalf("OpenDM(alice->bob) = %v, want success", err) + } + if !first.Msg.GetCreated() { + t.Fatalf("first open created = false, want true") + } + + // Reverse order (bob opens with alice): same deterministic name, so it must + // resume the same channel, not create a second. + second, err := svc.OpenDM(WithActor(ctx, bob.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "alice"})) + if err != nil { + t.Fatalf("OpenDM(bob->alice) = %v, want success", err) + } + if second.Msg.GetCreated() { + t.Fatalf("reopen created = true, want false (resume)") + } + if second.Msg.GetChannel().GetId() != first.Msg.GetChannel().GetId() { + t.Fatalf("reopen channel id = %q, want the first open's %q", second.Msg.GetChannel().GetId(), first.Msg.GetChannel().GetId()) + } +} + +// TestOpenDMUnknownHandleIsNotFound: an unknown peer handle collapses to +// CodeNotFound naming the submitted handle — the oracle-safe resolve miss. +func TestOpenDMUnknownHandleIsNotFound(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + + _, err := svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "ghost"})) + connectCodeIs(t, err, connect.CodeNotFound, "OpenDM(unknown handle)") +} + +// TestOpenDMCrossOwnerIsIndistinguishableNotFound: an owner-qualified handle +// naming ANOTHER owner's agent collapses to the SAME CodeNotFound an unknown +// handle gets — the cross-owner authz is byte-identical to unknown, so a foreign +// peer's existence is never leaked (design.md T3:746-748). +func TestOpenDMCrossOwnerIsIndistinguishableNotFound(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + other := mustUser(t, st, "other") + mustAgent(t, st, other.ID, "foreign") + + // alice names other's agent by an owner-qualified handle: it resolves, but + // the same-owner check remaps it to NOT_FOUND naming the submitted handle. + _, err := svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "other/foreign"})) + connectCodeIs(t, err, connect.CodeNotFound, "OpenDM(cross-owner owner-qualified handle)") +} + +// TestOpenDMSelfIsInvalidArgument: a handle that resolves to the caller itself is +// not a peer — CodeInvalidArgument. +func TestOpenDMSelfIsInvalidArgument(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + + _, err := svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "alice"})) + connectCodeIs(t, err, connect.CodeInvalidArgument, "OpenDM(self handle)") +} + +// TestOpenDMAsAccountEmptyAccountIsNoActor: an empty account short-circuits to +// errNoActor (CodeInvalidArgument) before any handler work — the fail-closed +// guard mirroring the other AsAccount adapters. +func TestOpenDMAsAccountEmptyAccountIsNoActor(t *testing.T) { + svc, _ := newHandler(t) + _, err := svc.OpenDMAsAccount(context.Background(), "", &compassv1.OpenDMRequest{PeerHandle: "bob"}) + connectCodeIs(t, err, connect.CodeInvalidArgument, "OpenDMAsAccount(empty account)") +} + +// TestOpenDMEmitsChannelChangedOnCreate: a create fans a post-commit +// ChannelChanged carrying the new DM channel to a member's stream; a resume emits +// nothing new. Driven over the real stream (newStreamHarness), subscribing before +// the mutation. The caller (alice) is a member of the DM, so it drains the event. +func TestOpenDMEmitsChannelChangedOnCreate(t *testing.T) { + h := newStreamHarness(t) + ctx := context.Background() + owner := mustUser(t, h.store, "owner") + alice := mustAgent(t, h.store, owner.ID, "alice") + mustAgent(t, h.store, owner.ID, "bob") + + events := firstEventAfterBoundary(t, h, alice.ID, &compassv1.SubscribeCommsRequest{SinceSeq: 0}) + + resp, err := h.svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "bob"})) + if err != nil { + t.Fatalf("OpenDM: %v", err) + } + wantID := resp.Msg.GetChannel().GetId() + + got := awaitFirst(t, events) + cc := got.GetChannelChanged() + if cc == nil { + t.Fatalf("event payload = %T, want ChannelChanged", got.GetPayload()) + } + if cc.GetChannel().GetId() != wantID { + t.Fatalf("ChannelChanged id = %q, want the created DM %q", cc.GetChannel().GetId(), wantID) + } + if cc.GetChannel().GetKind() != compassv1.ChannelKind_CHANNEL_KIND_DM { + t.Fatalf("ChannelChanged kind = %v, want CHANNEL_KIND_DM", cc.GetChannel().GetKind()) + } +} + +// TestOpenDMAsAccountResolvesOrCreates: the agent-tool adapter runs the same +// handler path under the bound account — a first call creates, a second resumes +// the same channel — parity with the direct handler. +func TestOpenDMAsAccountResolvesOrCreates(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + alice := mustAgent(t, st, owner.ID, "alice") + mustAgent(t, st, owner.ID, "bob") + + first, err := svc.OpenDMAsAccount(ctx, alice.ID, &compassv1.OpenDMRequest{PeerHandle: "bob"}) + if err != nil { + t.Fatalf("OpenDMAsAccount(first) = %v, want success", err) + } + if !first.GetCreated() { + t.Fatalf("first created = false, want true") + } + second, err := svc.OpenDMAsAccount(ctx, alice.ID, &compassv1.OpenDMRequest{PeerHandle: "bob"}) + if err != nil { + t.Fatalf("OpenDMAsAccount(second) = %v, want success", err) + } + if second.GetCreated() { + t.Fatalf("second created = true, want false (resume)") + } + if second.GetChannel().GetId() != first.GetChannel().GetId() { + t.Fatalf("resume id = %q, want first %q", second.GetChannel().GetId(), first.GetChannel().GetId()) + } +} diff --git a/go/internal/runnerhub/helpers_test.go b/go/internal/runnerhub/helpers_test.go index a4fbedeff..cfbdeed34 100644 --- a/go/internal/runnerhub/helpers_test.go +++ b/go/internal/runnerhub/helpers_test.go @@ -186,6 +186,7 @@ type commsCall struct { createChannel *compassv1.CreateChannelRequest updateMembers *compassv1.UpdateChannelMembersRequest createChannelGroup *compassv1.CreateChannelGroupRequest + openDM *compassv1.OpenDMRequest } // fakeCommsCaller is a hand-written CommsCaller: it records every call (account @@ -223,6 +224,9 @@ type fakeCommsCaller struct { createChannelGroupResp *compassv1.CreateChannelGroupResponse createChannelGroupErr error + + openDMResp *compassv1.OpenDMResponse + openDMErr error } func (f *fakeCommsCaller) PostAsAccount(_ context.Context, account store.AccountID, req *compassv1.PostMessageRequest) (*compassv1.PostMessageResponse, error) { @@ -328,6 +332,16 @@ func (f *fakeCommsCaller) CreateChannelGroupAsAccount(_ context.Context, account return f.createChannelGroupResp, nil } +func (f *fakeCommsCaller) OpenDMAsAccount(_ context.Context, account store.AccountID, req *compassv1.OpenDMRequest) (*compassv1.OpenDMResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, commsCall{account: account, openDM: req}) + if f.openDMErr != nil { + return nil, f.openDMErr + } + return f.openDMResp, nil +} + func (f *fakeCommsCaller) snapshot() []commsCall { f.mu.Lock() defer f.mu.Unlock() diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index b20a81a26..92c47c920 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -234,7 +234,7 @@ type SessionTailSink interface { // is the safe Runner->Server leg: the account is resolved Server-side from the // hub's own binding, never asserted by the Runner (transport design Decision #3 // / OQ-2, comms-tools design T2). -type CommsCaller interface { +type CommsCaller interface { //nolint:interfacebloat // one method per agent-comms arm (post/list/roster/set_status/pin/create_channel/update_members/create_channel_group/open_dm) — a dispatch seam, not a bloated abstraction; splitting it would fragment the single relay dispatch in executeCall // PostAsAccount posts under account with an id-typed channel container — the // internal id-holder path (relay transcript, seeds). PostAsAccountByName is // the agent-tool path: it resolves the request's channel NAME to an id first @@ -258,6 +258,11 @@ type CommsCaller interface { CreateChannelAsAccount(ctx context.Context, account store.AccountID, req *compassv1.CreateChannelRequest) (*compassv1.CreateChannelResponse, error) UpdateChannelMembersAsAccount(ctx context.Context, account store.AccountID, req *compassv1.UpdateChannelMembersRequest) (*compassv1.UpdateChannelMembersResponse, error) CreateChannelGroupAsAccount(ctx context.Context, account store.AccountID, req *compassv1.CreateChannelGroupRequest) (*compassv1.CreateChannelGroupResponse, error) + // OpenDMAsAccount resolves-or-creates the two-party peer DM between account + // and the request's peer handle (RIG-2962 T3), same-owner authz enforced + // Server-side. The request names the peer by handle, so there is no + // home-channel defaulting. + OpenDMAsAccount(ctx context.Context, account store.AccountID, req *compassv1.OpenDMRequest) (*compassv1.OpenDMResponse, error) } // Hub is the Server-side seam: enrollment registry + command router + the diff --git a/go/internal/runnerhub/relay_comms.go b/go/internal/runnerhub/relay_comms.go index 2a8e772f1..36d212152 100644 --- a/go/internal/runnerhub/relay_comms.go +++ b/go/internal/runnerhub/relay_comms.go @@ -483,10 +483,18 @@ func (h *Hub) executeCall( return &compassv1internal.CommsCallResult{ Result: &compassv1internal.CommsCallResult_CreateChannelGroup{CreateChannelGroup: resp}, }, nil + case *compassv1internal.CommsCallRequest_OpenDm: + resp, err := h.comms.OpenDMAsAccount(ctx, account, c.OpenDm) + if err != nil { + return nil, err + } + return &compassv1internal.CommsCallResult{ + Result: &compassv1internal.CommsCallResult_OpenDm{OpenDm: resp}, + }, nil default: return nil, connect.NewError( connect.CodeInvalidArgument, - errors.New("runnerhub: comms call has no recognized variant set (post/list/roster/set_status/pin/create_channel/update_members/create_channel_group)"), + errors.New("runnerhub: comms call has no recognized variant set (post/list/roster/set_status/pin/create_channel/update_members/create_channel_group/open_dm)"), ) } } diff --git a/go/internal/runnerhub/relay_open_dm_test.go b/go/internal/runnerhub/relay_open_dm_test.go new file mode 100644 index 000000000..dd60dfd3b --- /dev/null +++ b/go/internal/runnerhub/relay_open_dm_test.go @@ -0,0 +1,91 @@ +//go:build unix + +package runnerhub + +// The peer-DM relay arm (RIG-2962 T3): RelayCommsCall dispatches an open_dm call +// to OpenDMAsAccount under the bound account, wrapping the open_dm result oneof +// with call_id round-tripped. A tool error on the arm is rendered in-band as a +// CommsCallError, never a transport teardown. Driven through the fakeCommsCaller, +// no store. context.Background() is the test root (test-root ctx exemption). + +import ( + "context" + "errors" + "testing" + + "connectrpc.com/connect" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" +) + +// relayOpenDM builds a RelayCommsCallRequest carrying an open_dm variant. +func relayOpenDM(sessionID, callID string, req *compassv1.OpenDMRequest) *compassv1internal.RelayCommsCallRequest { + return &compassv1internal.RelayCommsCallRequest{ + SessionId: sessionID, + Call: &compassv1internal.CommsCallRequest{ + CallId: callID, + Call: &compassv1internal.CommsCallRequest_OpenDm{OpenDm: req}, + }, + } +} + +// TestRelayCommsCallOpenDMArmForwardsUnderBoundAccount: an open_dm call forwards +// the request under the bound account and wraps the open_dm result oneof, with +// call_id round-tripped. +func TestRelayCommsCallOpenDMArmForwardsUnderBoundAccount(t *testing.T) { + hub, comms := newHubWithComms() + comms.openDMResp = &compassv1.OpenDMResponse{Channel: &compassv1.Channel{Id: "ch-dm"}, Created: true} + bindLiveSession(hub) + + req := &compassv1.OpenDMRequest{PeerHandle: "peer"} + resp, err := hub.RelayCommsCall(context.Background(), relayOpenDM("sess-1", "tc-dm", req)) + if err != nil { + t.Fatalf("RelayCommsCall(open_dm) = %v, want success", err) + } + calls := comms.snapshot() + if len(calls) != 1 { + t.Fatalf("caller invoked %d times, want 1", len(calls)) + } + if calls[0].account != "acct-agent" { + t.Fatalf("open_dm attributed to %q, want bound acct-agent", calls[0].account) + } + if calls[0].openDM != req { + t.Fatalf("caller received a different OpenDMRequest than relayed") + } + if resp.GetResult().GetOpenDm() != comms.openDMResp { + t.Fatalf("result oneof = %T, want the caller's open_dm response", resp.GetResult().GetResult()) + } + if resp.GetResult().GetOpenDm().GetCreated() != true { + t.Fatalf("open_dm created = false, want the caller's created=true round-tripped") + } + if got := resp.GetResult().GetCallId(); got != "tc-dm" { + t.Fatalf("response call_id = %q, want tc-dm", got) + } +} + +// TestRelayCommsCallOpenDMToolErrorIsInBandNotStreamError: a tool-level failure +// on the open_dm arm is rendered IN-BAND as a CommsCallError, not as a Connect +// stream error — the "tool failure != transport teardown" invariant on the +// peer-DM arm (mirrors the org-management arms). not_found is the oracle-safe +// collapse an unknown/cross-owner peer gets. +func TestRelayCommsCallOpenDMToolErrorIsInBandNotStreamError(t *testing.T) { + hub, comms := newHubWithComms() + comms.openDMErr = connect.NewError(connect.CodeNotFound, errors.New("handle \"peer\" not found")) + bindLiveSession(hub) + + resp, err := hub.RelayCommsCall(context.Background(), relayOpenDM("sess-1", "tc-dm-err", &compassv1.OpenDMRequest{PeerHandle: "peer"})) + if err != nil { + t.Fatalf("RelayCommsCall returned a stream error %v, want in-band tool error", err) + } + toolErr := resp.GetResult().GetError() + if toolErr == nil { + t.Fatal("response has no in-band CommsCallError, want the tool failure rendered in-band") + } + if toolErr.GetCode() != "not_found" { + t.Fatalf("in-band error code = %q, want not_found (the oracle-safe collapse token)", toolErr.GetCode()) + } + if got := resp.GetResult().GetCallId(); got != "tc-dm-err" { + t.Fatalf("response call_id = %q, want tc-dm-err", got) + } +} diff --git a/go/server/lifecycle.go b/go/server/lifecycle.go index 01cfd6030..a89e2a1e1 100644 --- a/go/server/lifecycle.go +++ b/go/server/lifecycle.go @@ -42,13 +42,26 @@ import ( "github.com/RigelBuild/compass/go/internal/store" ) +// dmOpener opens the manager<->peer DM at spawn time (R8). A narrow seam so +// lifecycleService does not pull the whole Comms handler in — satisfied by +// *comms.Comms via OpenDMAsAccount. It takes the PUBLIC compassv1 OpenDM types +// (the agent-caller adapter's signature). +type dmOpener interface { + OpenDMAsAccount(ctx context.Context, account store.AccountID, req *compassv1.OpenDMRequest) (*compassv1.OpenDMResponse, error) +} + // lifecycleService is the LifecycleCaller implementation. It holds the store (of // record for accounts + placements) and the hub (Provision/Start/Stop/Remove // relays to the owning Runner) — the same two dependencies the provision/start -// handlers use, wired here as the agent-initiated door. +// handlers use, wired here as the agent-initiated door — plus the dmOpener seam +// that auto-opens the manager<->peer DM after a spawn (R8). type lifecycleService struct { store *store.Store hub *runnerhub.Hub + // dm opens the manager<->new-peer DM at spawn time (R8). Nil for instances + // that never spawn (the waker, the store-free self-despawn test), in which + // case autoOpenSpawnDM returns an empty name. + dm dmOpener // wakeGroup coalesces concurrent WakeAgent calls for the SAME agent onto one // start (RIG-1641 T3 cost control, §Decisions OQ-2): a burst of messages at // one offline agent produces exactly one resume/Start, not a start-storm. A @@ -56,11 +69,12 @@ type lifecycleService struct { wakeGroup singleflight.Group } -// newLifecycleService constructs the lifecycle caller over the store and hub. -// Wired at serve assembly with hub.SetLifecycleCaller after both exist, breaking -// the hub<->lifecycleService construction cycle (serve.go). -func newLifecycleService(st *store.Store, hub *runnerhub.Hub) *lifecycleService { - return &lifecycleService{store: st, hub: hub} +// newLifecycleService constructs the lifecycle caller over the store, hub, and +// the DM-opener seam. Wired at serve assembly with hub.SetLifecycleCaller after +// all three exist, breaking the hub<->lifecycleService construction cycle +// (serve.go). +func newLifecycleService(st *store.Store, hub *runnerhub.Hub, dm dmOpener) *lifecycleService { + return &lifecycleService{store: st, hub: hub, dm: dm} } // Compile-time proof lifecycleService satisfies the seam the hub delegates into. @@ -216,14 +230,31 @@ func (l *lifecycleService) SpawnAsAccount( // form a cycle — the cycle check lives only on the mutable ReparentAgent. ParentAgentID: caller, }) + var resp *compassv1internal.SpawnPeerResponse switch { case err == nil: - return l.provisionAndStart(ctx, created.ID, created.Agent.Persona, created.Agent.Role, req) + resp, err = l.provisionAndStart(ctx, created.ID, created.Agent.Persona, created.Agent.Role, req) case errors.Is(err, store.ErrConflict): - return l.resumeOrReject(ctx, callerOwner, req) + resp, err = l.resumeOrReject(ctx, callerOwner, req) default: return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("creating agent: %w", err)) } + if err != nil { + return nil, err + } + + // Spawn auto-open (R8, design.md T3:755-762): after the spawn chain succeeds, + // open the manager<->new-peer DM and set DmChannelName on the response. This + // covers the fresh, resume, AND idempotent already-placed paths from one site + // — OpenDM is resolve-or-create, so every path yields the SAME name + // idempotently (a re-spawn returns the same DM). An open FAILURE post-spawn + // is logged and returned with an EMPTY dm_channel_name, NEVER a spawn + // rollback: the DM is recoverable next turn via comms_open_dm, the spawned + // peer is not. + if resp != nil && resp.GetAgentAccountId() != "" { + resp.DmChannelName = l.autoOpenSpawnDM(ctx, caller, store.AccountID(resp.GetAgentAccountId())) + } + return resp, nil } // DespawnAsAccount tears down a peer's compute (container + session), NOT its @@ -311,6 +342,32 @@ func (l *lifecycleService) DespawnAsAccount( return &compassv1internal.DespawnPeerResponse{}, nil } +// autoOpenSpawnDM opens the manager<->new-peer DM (R8) and returns its channel +// name, or "" on any failure. caller is the MANAGER (the spawn's caller); peerID +// is the created/resolved peer's account id. OpenDM addresses the peer by HANDLE, +// so the peer's handle is resolved from its account. On ANY error — a nil opener +// (an unwired test), a handle resolve miss, or the open itself — it LOGS and +// returns an empty name rather than failing the spawn: the DM is recoverable next +// turn via comms_open_dm, never a spawn rollback (design.md T3:755-762). +func (l *lifecycleService) autoOpenSpawnDM(ctx context.Context, caller, peerID store.AccountID) string { + if l.dm == nil { + return "" + } + peer, err := l.store.GetAccount(ctx, peerID) + if err != nil { + slog.WarnContext(ctx, "spawn: auto-open manager<->peer DM failed; recoverable via comms_open_dm", + "error", err.Error(), "peer", string(peerID)) + return "" + } + resp, err := l.dm.OpenDMAsAccount(ctx, caller, &compassv1.OpenDMRequest{PeerHandle: peer.Handle}) + if err != nil { + slog.WarnContext(ctx, "spawn: auto-open manager<->peer DM failed; recoverable via comms_open_dm", + "error", err.Error(), "peer", string(peerID)) + return "" + } + return resp.GetChannel().GetName() +} + // resumeOrReject handles a spawn whose handle is already taken in the CALLER'S // OWN owner namespace (CreateAgent conflicted on the per-owner agent index). An // already-placed agent is an idempotent success returning the existing diff --git a/go/server/lifecycle_e2e_pgtest_test.go b/go/server/lifecycle_e2e_pgtest_test.go index 2543b2f88..a4b346ad5 100644 --- a/go/server/lifecycle_e2e_pgtest_test.go +++ b/go/server/lifecycle_e2e_pgtest_test.go @@ -537,7 +537,7 @@ func newE2EWire(t *testing.T) *e2eWire { // Wire the lifecycleService as the hub's LifecycleCaller — the serve.go:250 // pattern. Only package server can construct it (unexported) and set it. - lc := newLifecycleService(st, hub) + lc := newLifecycleService(st, hub, commsSvc) hub.SetLifecycleCaller(lc) // Mount the RunnerService door on an h2c server, accepting one Runner token. diff --git a/go/server/lifecycle_pgtest_test.go b/go/server/lifecycle_pgtest_test.go index 3a27a0ed3..e0ff7563d 100644 --- a/go/server/lifecycle_pgtest_test.go +++ b/go/server/lifecycle_pgtest_test.go @@ -30,6 +30,7 @@ import ( "github.com/RigelBuild/compass/go/events" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/comms" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -44,8 +45,8 @@ type lifecycleFixture struct { } // newLifecycleFixture builds the placement fixture, constructs the lifecycleService -// over its real store + hub, and resolves the fixture agent's owner for the -// ownership assertions. +// over its real store + hub + a real comms handler (the R8 spawn-DM opener), and +// resolves the fixture agent's owner for the ownership assertions. func newLifecycleFixture(t *testing.T) lifecycleFixture { t.Helper() pf := newPlacementFixture(t) @@ -55,9 +56,15 @@ func newLifecycleFixture(t *testing.T) lifecycleFixture { if err != nil { t.Fatalf("AgentOwner(fixture agent) = %v", err) } + // A real comms handler is the R8 spawn-DM opener: SpawnAsAccount opens the + // manager<->peer DM through it. Its own bus (the socket-door admin fallback + // is unused — OpenDMAsAccount always sets the caller explicitly). + commsBus := events.NewBus[*compassv1.SubscribeCommsResponse]() + t.Cleanup(commsBus.Close) + commsSvc := comms.NewComms(pf.store, commsBus, owner) return lifecycleFixture{ placementFixture: pf, - lc: newLifecycleService(pf.store, pf.hub), + lc: newLifecycleService(pf.store, pf.hub, commsSvc), ownerAdmin: owner, } } @@ -628,3 +635,95 @@ func TestSpawnHandleCollidesWithSystemAccountIsAlreadyExists(t *testing.T) { t.Fatalf("system-handle-collision spawn err = %v, want wrapping errHandleTaken", err) } } + +// TestSpawnAutoOpensManagerPeerDM pins R8 (design.md T3:755-762): after the +// spawn chain succeeds, SpawnAsAccount opens the manager<->new-peer DM and +// returns a live dm_channel_name whose channel satisfies the full DM invariant +// set — kind=DM, mandatory subscription, and BOTH the manager (caller) and the +// new peer as members. +// +// Mutation: dropping the autoOpenSpawnDM call (or its assignment to +// resp.DmChannelName) leaves dm_channel_name empty, reddening the "live name" +// assertion; opening under the wrong caller drops the manager from the member set. +func TestSpawnAutoOpensManagerPeerDM(t *testing.T) { + f := newLifecycleFixture(t) + ctx := context.Background() + + resp, err := f.lc.SpawnAsAccount(ctx, f.agentID, &compassv1internal.SpawnPeerRequest{ + Handle: "peer-dm", + DisplayName: "Peer DM", + ClientRequestId: "spawn-dm-1", + }) + if err != nil { + t.Fatalf("SpawnAsAccount = %v, want success", err) + } + dmName := resp.GetDmChannelName() + if dmName == "" { + t.Fatal("dm_channel_name = empty, want a live manager<->peer DM name (R8 auto-open)") + } + peerID := store.AccountID(resp.GetAgentAccountId()) + + // Read the DM back as the manager (a member) and assert the full invariant. + ch, err := f.store.ChannelByNameForViewer(ctx, f.agentID, dmName) + if err != nil { + t.Fatalf("ChannelByNameForViewer(manager, %q) = %v, want the DM", dmName, err) + } + if ch.Kind != store.ChannelKindDM { + t.Fatalf("DM kind = %d, want ChannelKindDM", ch.Kind) + } + if !ch.Policy.MandatorySubscription { + t.Fatalf("DM mandatory_subscription = false, want true (born-mandatory)") + } + if !containsAccountID(ch.MemberAccountIDs, f.agentID) { + t.Fatalf("DM members = %v, want the manager (caller) %s", ch.MemberAccountIDs, f.agentID) + } + if !containsAccountID(ch.MemberAccountIDs, peerID) { + t.Fatalf("DM members = %v, want the new peer %s", ch.MemberAccountIDs, peerID) + } +} + +// TestSpawnIdempotentReturnsSameDMName pins the R8 idempotency (design.md +// T3:773): a re-spawn of the same handle by the same owner (the already-placed +// idempotent path) returns the SAME dm_channel_name — OpenDM is resolve-or-create +// so every spawn path yields the same deterministic name. +// +// Mutation: computing the name from a non-deterministic source (or opening a new +// channel per spawn) reddens the "same name" assertion. +func TestSpawnIdempotentReturnsSameDMName(t *testing.T) { + f := newLifecycleFixture(t) + ctx := context.Background() + + first, err := f.lc.SpawnAsAccount(ctx, f.agentID, &compassv1internal.SpawnPeerRequest{ + Handle: "peer-dm-idem", + ClientRequestId: "spawn-dm-idem-1", + }) + if err != nil { + t.Fatalf("SpawnAsAccount(first) = %v, want success", err) + } + if first.GetDmChannelName() == "" { + t.Fatal("first spawn dm_channel_name = empty, want a live name") + } + + // A second spawn of the same handle+owner is the idempotent already-placed + // path: it returns the same peer and MUST return the same DM name. + second, err := f.lc.SpawnAsAccount(ctx, f.agentID, &compassv1internal.SpawnPeerRequest{ + Handle: "peer-dm-idem", + ClientRequestId: "spawn-dm-idem-2", + }) + if err != nil { + t.Fatalf("SpawnAsAccount(second) = %v, want success", err) + } + if second.GetDmChannelName() != first.GetDmChannelName() { + t.Fatalf("re-spawn dm_channel_name = %q, want the first spawn's %q (idempotent)", second.GetDmChannelName(), first.GetDmChannelName()) + } +} + +// containsAccountID reports whether ids contains want. +func containsAccountID(ids []store.AccountID, want store.AccountID) bool { + for _, id := range ids { + if id == want { + return true + } + } + return false +} diff --git a/go/server/lifecycle_test.go b/go/server/lifecycle_test.go index d027a0d15..babe38aad 100644 --- a/go/server/lifecycle_test.go +++ b/go/server/lifecycle_test.go @@ -28,7 +28,7 @@ import ( // read) reddens this — a nil-store AgentOwner call panics rather than returning // the clean invalid_argument. func TestDespawnSelfIsRefusedBeforeAnyStoreCall(t *testing.T) { - lc := newLifecycleService(nil, nil) // nil store + hub: any store/hub call would panic + lc := newLifecycleService(nil, nil, nil) // nil store + hub + dm: any store/hub/dm call would panic const self = store.AccountID("agent-self") _, err := lc.DespawnAsAccount(context.Background(), self, &compassv1internal.DespawnPeerRequest{AgentHandle: string(self)}) diff --git a/go/server/lifecycle_wake_pgtest_test.go b/go/server/lifecycle_wake_pgtest_test.go index cf02bb869..b3f73b858 100644 --- a/go/server/lifecycle_wake_pgtest_test.go +++ b/go/server/lifecycle_wake_pgtest_test.go @@ -35,7 +35,7 @@ import ( func newWakeFixture(t *testing.T) (placementFixture, *lifecycleService) { t.Helper() pf := newPlacementFixture(t) - return pf, newLifecycleService(pf.store, pf.hub) + return pf, newLifecycleService(pf.store, pf.hub, nil) } // TestWakeAgentLiveIsNoOp pins the not-live pre-check: an agent with a LIVE diff --git a/go/server/offline_mention_e2e_pgtest_test.go b/go/server/offline_mention_e2e_pgtest_test.go index a1d171fe0..f0da2cda5 100644 --- a/go/server/offline_mention_e2e_pgtest_test.go +++ b/go/server/offline_mention_e2e_pgtest_test.go @@ -127,7 +127,7 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire { // The production delivery wire (sinks.go:142-155), assembled inline with the // REAL resume-based waker (newLifecycleService), not a fake. c := delivery.NewConsumer(commsBus, st, hub, hub, slog.New(slog.DiscardHandler)) - c.SetAgentWaker(newLifecycleService(st, hub)) + c.SetAgentWaker(newLifecycleService(st, hub, nil)) hub.SetSettleSink(c) hub.SetSessionStartSink(c) hub.SetDeliveryStore(st) diff --git a/go/server/sinks.go b/go/server/sinks.go index 119ad2086..fa279ea5b 100644 --- a/go/server/sinks.go +++ b/go/server/sinks.go @@ -94,7 +94,7 @@ var _ runnerhub.ForgeCaller = (*forgeService)(nil) // lifecycle T3-a, RelayBoardCall — the board caller executes against the store + // the issue projection). Called once at assembly before any RPC is served. func wireHubServiceCycles(hub *runnerhub.Hub, commsSvc *comms.Comms, st *store.Store, issueBrd *board.IssueProjection) { - hub.SetLifecycleCaller(newLifecycleService(st, hub)) + hub.SetLifecycleCaller(newLifecycleService(st, hub, commsSvc)) hub.SetBoardCaller(newBoardService(st, issueBrd)) // The roster read (RIG-1721 T2) joins the hub's in-memory presence enum; the // hub in turn reads it from the T8 presence projection wired at @@ -137,7 +137,9 @@ func startDeliveryConsumer(gctx context.Context, g *errgroup.Group, commsBus *ev // WakeAgent, so a second instance's group IS the wake's coalescer — sharing // the LifecycleCaller instance would buy nothing and couple two unrelated call // sites. Same (st, hub) inputs, so it runs the identical resume/start chain. - c.SetAgentWaker(newLifecycleService(st, hub)) + // nil dmOpener: the waker only drives WakeAgent, never SpawnAsAccount, so it + // never opens a DM. + c.SetAgentWaker(newLifecycleService(st, hub, nil)) hub.SetSettleSink(c) hub.SetSessionStartSink(c) hub.SetSessionReapSink(c) From 0f5eaeb8e13ee752a3436ef565be2fd83e094529 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 30 Aug 2026 18:40:36 -0400 Subject: [PATCH 2/3] fix(comms): injective DM name (colon sep) + R8 rollback-safety & emit tests (RIG-2964) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the T3 review (PR #757): - **medium (security):** `dmChannelName` was not injective — the `--` delimiter collides when a handle itself contains `--` (the handle grammar permits consecutive hyphens), so pair {a, b--c} and {a--b, c} both derived `dm--a--b--c` and the second open resumed onto the first's channel, cross-adding members (a same-owner private-DM confidentiality break). Switch the separator to `:`, a byte the handle grammar excludes, so `dm::` is injective. New pgtest opens both pairs and asserts DISTINCT channels + uncorrupted membership (RED-verified against the old delimiter). - **medium (test):** add the R8 headline-safety test — a post-spawn DM-open FAILURE (and a nil opener) never rolls back the spawn: peer placed, `dm_channel_name` empty. - **low (test):** assert a DM RESUME emits no second ChannelChanged (the created-only guard), canary-gated. - **low (perf):** `emitDMCreated` takes the already-read channel instead of re-reading it. Refs RIG-2964 Co-authored-by: Matt Wilkinson --- go/internal/comms/comms.go | 2 +- go/internal/comms/dm.go | 26 ++++--- go/internal/comms/dm_open_pgtest_test.go | 93 +++++++++++++++++++++++- go/server/lifecycle_pgtest_test.go | 60 +++++++++++++++ 4 files changed, 165 insertions(+), 16 deletions(-) diff --git a/go/internal/comms/comms.go b/go/internal/comms/comms.go index 5005ead2c..041b075ea 100644 --- a/go/internal/comms/comms.go +++ b/go/internal/comms/comms.go @@ -698,7 +698,7 @@ func (c *Comms) OpenDM( if err != nil { return nil, edgeError(err) } - c.emitDMCreated(ctx, channelID, created) + c.emitDMCreated(ch, created) return connect.NewResponse(&compassv1.OpenDMResponse{ Channel: channelToWire(ch), Created: created, diff --git a/go/internal/comms/dm.go b/go/internal/comms/dm.go index 2ef28b5f5..beecd6411 100644 --- a/go/internal/comms/dm.go +++ b/go/internal/comms/dm.go @@ -2,7 +2,6 @@ package comms import ( "context" - "log/slog" "github.com/jackc/pgx/v5" @@ -12,12 +11,21 @@ import ( // 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::` 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 + return "dm:" + lo + ":" + hi } // openDMTx runs the whole peer-DM open for owner in ONE store transaction — the @@ -57,18 +65,12 @@ func (c *Comms) openDMTx(ctx context.Context, owner store.AccountID, name string // 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), and a re-read failure is logged and -// swallowed, never propagated, since the channel already committed and the event -// self-heals on the next open. NEVER call before the commit. -func (c *Comms) emitDMCreated(ctx context.Context, channelID store.ChannelID, created bool) { +// 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 } - ch, err := c.store.GetChannel(ctx, channelID) - if err != nil { - slog.WarnContext(ctx, "comms: post-commit DM channel read for event failed; self-heals on next open", - "channel_id", string(channelID), "error", err.Error()) - return - } c.publishChannelChanged(ch, nil) } diff --git a/go/internal/comms/dm_open_pgtest_test.go b/go/internal/comms/dm_open_pgtest_test.go index 2229bbd2c..032c90138 100644 --- a/go/internal/comms/dm_open_pgtest_test.go +++ b/go/internal/comms/dm_open_pgtest_test.go @@ -21,7 +21,7 @@ import ( // TestOpenDMSameOwnerCreatesDMChannel: an agent opens a DM with a same-owner // peer → the returned channel is a real DM (kind=DM, mandatory subscription, // both parties members), created=true, and the deterministic sorted-handle name -// (dm----) is used. +// (dm::) is used. func TestOpenDMSameOwnerCreatesDMChannel(t *testing.T) { svc, st := newHandler(t) ctx := context.Background() @@ -37,8 +37,8 @@ func TestOpenDMSameOwnerCreatesDMChannel(t *testing.T) { t.Fatalf("created = false, want true on a first open") } ch := resp.Msg.GetChannel() - if ch.GetName() != "dm--alice--bob" { - t.Fatalf("channel name = %q, want dm--alice--bob (deterministic sorted-handle)", ch.GetName()) + if ch.GetName() != "dm:alice:bob" { + t.Fatalf("channel name = %q, want dm:alice:bob (deterministic sorted-handle)", ch.GetName()) } if ch.GetKind() != compassv1.ChannelKind_CHANNEL_KIND_DM { t.Fatalf("channel kind = %v, want CHANNEL_KIND_DM", ch.GetKind()) @@ -194,3 +194,90 @@ func TestOpenDMAsAccountResolvesOrCreates(t *testing.T) { t.Fatalf("resume id = %q, want first %q", second.GetChannel().GetId(), first.GetChannel().GetId()) } } + +// TestOpenDMDoubleHyphenHandlesResolveDistinctChannels pins the name-injectivity +// fix: with a `-`-delimited name, pair {a, b--c} and pair {a--b, c} would both +// derive the byte-identical `dm--a--b--c` and the second open would RESUME onto +// the first's channel, cross-adding members (a same-owner private-DM +// confidentiality break). The `:` separator is excluded from the handle grammar, +// so the two pairs map to distinct names (dm:a:b--c vs dm:a--b:c) and must +// resolve to DISTINCT channels. RED if dmChannelName reverts to a `-` delimiter. +func TestOpenDMDoubleHyphenHandlesResolveDistinctChannels(t *testing.T) { + svc, st := newHandler(t) + ctx := context.Background() + owner := mustUser(t, st, "owner") + a := mustAgent(t, st, owner.ID, "a") + bc := mustAgent(t, st, owner.ID, "b--c") + ab := mustAgent(t, st, owner.ID, "a--b") + c := mustAgent(t, st, owner.ID, "c") + + // Pair {a, b--c}. + first, err := svc.OpenDM(WithActor(ctx, a.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "b--c"})) + if err != nil { + t.Fatalf("OpenDM(a->b--c) = %v, want success", err) + } + // Pair {a--b, c} — a DIFFERENT unordered pair. Must mint its OWN channel, not + // resume the first pair's. + second, err := svc.OpenDM(WithActor(ctx, ab.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "c"})) + if err != nil { + t.Fatalf("OpenDM(a--b->c) = %v, want success", err) + } + if !second.Msg.GetCreated() { + t.Fatalf("second pair created = false, want true (a distinct pair must mint its own DM, not resume)") + } + if second.Msg.GetChannel().GetId() == first.Msg.GetChannel().GetId() { + t.Fatalf("the two distinct pairs collided onto one channel %q — dmChannelName is not injective", first.Msg.GetChannel().GetId()) + } + // And the first DM's membership is uncorrupted: exactly a + b--c (+ their + // owner), never the second pair's ab or c. + m := first.Msg.GetChannel().GetMemberAccountIds() + if !containsString(m, string(a.ID)) || !containsString(m, string(bc.ID)) { + t.Fatalf("first DM members = %v, want its own pair a=%s and b--c=%s", m, a.ID, bc.ID) + } + if containsString(m, string(ab.ID)) || containsString(m, string(c.ID)) { + t.Fatalf("first DM members = %v, leaked the second pair's parties (ab=%s c=%s)", m, ab.ID, c.ID) + } +} + +// TestOpenDMResumeEmitsNoChannelChanged pins the paired half of the emit +// contract (emitDMCreated's created-only guard): a create fans a ChannelChanged, +// but a RESUME emits nothing new. Event-gated via a canary — a globally-visible +// AccountChanged published AFTER the resume; draining the caller's replay up to +// the canary must surface NO ChannelChanged for the DM (in-order per-subscriber +// delivery guarantees a resume event, if any, would precede the canary). RED if +// the created-only guard is dropped (a spurious per-resume re-publish). +func TestOpenDMResumeEmitsNoChannelChanged(t *testing.T) { + h := newStreamHarness(t) + ctx := context.Background() + owner := mustUser(t, h.store, "owner") + alice := mustAgent(t, h.store, owner.ID, "alice") + mustAgent(t, h.store, owner.ID, "bob") + + // Create the DM first (drains the create event out of the way). + created, err := h.svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "bob"})) + if err != nil { + t.Fatalf("OpenDM(create): %v", err) + } + dmID := created.Msg.GetChannel().GetId() + + // Resume the SAME pair — must emit nothing. + if _, err := h.svc.OpenDM(WithActor(ctx, alice.ID), connect.NewRequest(&compassv1.OpenDMRequest{PeerHandle: "bob"})); err != nil { + t.Fatalf("OpenDM(resume): %v", err) + } + + // Canary published last; drain alice's replay up to it and assert no + // ChannelChanged for the DM rode along (the create event is in the replay too, + // so scope the check to events AFTER the create — any SECOND ChannelChanged + // for dmID is the spurious resume emit). + canary := mkCanary(t, h, "canary") + evts := drainReplayAsActor(t, h, alice.ID, canary) + var dmChannelChanges int + for _, e := range evts { + if cc := e.GetChannelChanged(); cc != nil && cc.GetChannel().GetId() == dmID { + dmChannelChanges++ + } + } + if dmChannelChanges != 1 { + t.Fatalf("ChannelChanged events for the DM = %d, want exactly 1 (the create only; a resume must emit nothing)", dmChannelChanges) + } +} diff --git a/go/server/lifecycle_pgtest_test.go b/go/server/lifecycle_pgtest_test.go index e0ff7563d..6ca3e64cd 100644 --- a/go/server/lifecycle_pgtest_test.go +++ b/go/server/lifecycle_pgtest_test.go @@ -718,6 +718,66 @@ func TestSpawnIdempotentReturnsSameDMName(t *testing.T) { } } +// errDMOpener is a dmOpener that always fails — the R8 failure-injection seam. +type errDMOpener struct{} + +func (errDMOpener) OpenDMAsAccount(context.Context, store.AccountID, *compassv1.OpenDMRequest) (*compassv1.OpenDMResponse, error) { + return nil, errors.New("dm open failed") +} + +// TestSpawnDMOpenFailureNeverRollsBackSpawn pins R8's headline safety guarantee +// (design.md T3:758-761): a post-spawn DM-open FAILURE is logged and returns an +// EMPTY dm_channel_name — it NEVER rolls back the spawn. The spawned peer is not +// recoverable, but the DM is (next turn via comms_open_dm), so an open failure +// must not un-place the peer. Two failure modes, both must succeed the spawn: +// - a failing opener (errDMOpener) → peer placed, dm_channel_name empty; +// - a nil opener (an instance that never wired one) → same. +// +// Mutation: propagating autoOpenSpawnDM's error out of SpawnAsAccount (instead of +// swallowing it to an empty name) turns a recoverable DM miss into a spawn +// rollback — this test reddens (err != nil, and the peer is gone). +func TestSpawnDMOpenFailureNeverRollsBackSpawn(t *testing.T) { + t.Run("failing opener", func(t *testing.T) { + pf := newPlacementFixture(t) + pf.runner.forget() + lc := newLifecycleService(pf.store, pf.hub, errDMOpener{}) + + resp, err := lc.SpawnAsAccount(context.Background(), pf.agentID, &compassv1internal.SpawnPeerRequest{ + Handle: "peer-dm-fail", + ClientRequestId: "spawn-dm-fail-1", + }) + if err != nil { + t.Fatalf("SpawnAsAccount = %v, want success despite the DM-open failure (never a rollback)", err) + } + if resp.GetAgentAccountId() == "" { + t.Fatal("agent_account_id = empty, want the placed peer (the spawn must not roll back on a DM-open failure)") + } + if resp.GetDmChannelName() != "" { + t.Fatalf("dm_channel_name = %q, want empty (a failed open returns no name)", resp.GetDmChannelName()) + } + }) + + t.Run("nil opener", func(t *testing.T) { + pf := newPlacementFixture(t) + pf.runner.forget() + lc := newLifecycleService(pf.store, pf.hub, nil) + + resp, err := lc.SpawnAsAccount(context.Background(), pf.agentID, &compassv1internal.SpawnPeerRequest{ + Handle: "peer-dm-nil", + ClientRequestId: "spawn-dm-nil-1", + }) + if err != nil { + t.Fatalf("SpawnAsAccount = %v, want success with a nil DM opener", err) + } + if resp.GetAgentAccountId() == "" { + t.Fatal("agent_account_id = empty, want the placed peer") + } + if resp.GetDmChannelName() != "" { + t.Fatalf("dm_channel_name = %q, want empty (nil opener degrades to no name)", resp.GetDmChannelName()) + } + }) +} + // containsAccountID reports whether ids contains want. func containsAccountID(ids []store.AccountID, want store.AccountID) bool { for _, id := range ids { From b510774a10d0e6919b5ddd0dbd39f5ef66b79401 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 30 Aug 2026 19:05:18 -0400 Subject: [PATCH 3/3] docs(comms): correct stale newLifecycleService signature + resume-emit test comments (RIG-2964) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review lows: three test doc-comments illustrated the pre-PR 2-arg newLifecycleService signature (now 3-arg with the dmOpener seam), and the resume-emit test comment described a 'scope after create' the assertion does not do (it counts all ChannelChanged for the DM and requires exactly 1). Comment-only — no behavior change. Refs RIG-2964 Co-authored-by: Matt Wilkinson --- go/internal/comms/dm_open_pgtest_test.go | 8 ++++---- go/server/lifecycle_pgtest_test.go | 2 +- go/server/lifecycle_wake_pgtest_test.go | 2 +- go/server/offline_mention_e2e_pgtest_test.go | 2 +- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/go/internal/comms/dm_open_pgtest_test.go b/go/internal/comms/dm_open_pgtest_test.go index 032c90138..baaeef6eb 100644 --- a/go/internal/comms/dm_open_pgtest_test.go +++ b/go/internal/comms/dm_open_pgtest_test.go @@ -265,10 +265,10 @@ func TestOpenDMResumeEmitsNoChannelChanged(t *testing.T) { t.Fatalf("OpenDM(resume): %v", err) } - // Canary published last; drain alice's replay up to it and assert no - // ChannelChanged for the DM rode along (the create event is in the replay too, - // so scope the check to events AFTER the create — any SECOND ChannelChanged - // for dmID is the spurious resume emit). + // Canary published last; drain alice's replay up to it and count every + // ChannelChanged for the DM across the whole replay. The create contributes + // exactly 1, so a total >1 is a spurious per-resume re-publish (in-order + // per-subscriber delivery guarantees a resume event, if any, precedes the canary). canary := mkCanary(t, h, "canary") evts := drainReplayAsActor(t, h, alice.ID, canary) var dmChannelChanges int diff --git a/go/server/lifecycle_pgtest_test.go b/go/server/lifecycle_pgtest_test.go index 6ca3e64cd..c94c90d6e 100644 --- a/go/server/lifecycle_pgtest_test.go +++ b/go/server/lifecycle_pgtest_test.go @@ -4,7 +4,7 @@ package server // The lifecycleService orchestration seam, against a real Postgres AND a real // Runner door. lifecycleService is an INTERNAL seam (runnerhub.LifecycleCaller), -// not a wire RPC, so these drive newLifecycleService(f.hub, f.store) DIRECTLY +// not a wire RPC, so these drive newLifecycleService(f.store, f.hub, dm) DIRECTLY // under a resolved caller AccountID — the same way the hub's RelayLifecycleCall // delegates into it — rather than through the connect client. The RemoveAgentWorkspace // operator door IS a wire RPC, so that one test drives it through f.client. diff --git a/go/server/lifecycle_wake_pgtest_test.go b/go/server/lifecycle_wake_pgtest_test.go index b3f73b858..fa86fbeca 100644 --- a/go/server/lifecycle_wake_pgtest_test.go +++ b/go/server/lifecycle_wake_pgtest_test.go @@ -7,7 +7,7 @@ package server // records every relayed command, so "a Start was pushed" / "no Start was pushed" // / "the resume body rode the internal envelope" are observed wire facts, not // mock expectations). WakeAgent is an INTERNAL seam (delivery.AgentWaker), not a -// wire RPC, so these drive newLifecycleService(store, hub).WakeAgent directly +// wire RPC, so these drive newLifecycleService(store, hub, nil).WakeAgent directly // under a resolved agent AccountID — the same way the delivery consumer's wake // seam calls it — rather than through the connect client. // diff --git a/go/server/offline_mention_e2e_pgtest_test.go b/go/server/offline_mention_e2e_pgtest_test.go index f0da2cda5..bafaee36f 100644 --- a/go/server/offline_mention_e2e_pgtest_test.go +++ b/go/server/offline_mention_e2e_pgtest_test.go @@ -15,7 +15,7 @@ package server // The wire the test stands up, inline, is production's: // // c := delivery.NewConsumer(commsBus, st, hub, hub, log) -// c.SetAgentWaker(newLifecycleService(st, hub)) // the REAL resume waker +// c.SetAgentWaker(newLifecycleService(st, hub, nil)) // the REAL resume waker // hub.SetSettleSink(c); hub.SetSessionStartSink(c); hub.SetDeliveryStore(st) // go c.Run(ctx) //