feat(comms): publish message_posted on the event fabric (RIG-3107)

docs/designs/infra/runtime/compass-managed-delivery-cutover/design.md

@@ -274,10 +274,10 @@ resolve the tenant from ctx with the bootstrap fallback.
   and `CommsSubject(tenant string, kind EventKind) (string, error)` +
   `fabric.KindMessagePosted` (PR3);
   `store.TenantFromContext(ctx context.Context) (store.TenantID, bool)`
-  (`go/internal/store/context.go:23`). Produces: a new exported
-  `func (s *Store) ResolveTenant(ctx context.Context) TenantID` (promoting the
-  unexported `resolveTenant`, `go/internal/store/tenant.go:54-58`, so comms
-  gets the same set-or-bootstrap-fallback semantics the store's writes use);
+  (`go/internal/store/context.go:23`). Produces: comms resolves the tenant
+  through the exported `func (s *Store) EffectiveTenant(ctx context.Context)
+  TenantID` (which RIG-3108 added after this record froze, with the same
+  set-or-bootstrap-fallback semantics as the unexported `resolveTenant`);
   `comms.NewComms` gains a `fabric fabric.EventFabric` parameter (nil-safe:
   nil ⇒ bus-only, so unit tests and any not-yet-wired assembly keep working);
   `publishMessagePosted(ctx, m)` extended with the fabric publish + the
@@ -448,7 +448,7 @@ exists anymore) and re-derive the no-loss argument from JetStream durability.
 
 ## Tasks
 
-- [ ] T1: fabric publish in `publishMessagePosted` + `Store.ResolveTenant` +
+- [ ] T1: fabric publish in `publishMessagePosted` + `Store.EffectiveTenant` +
       failure counter (tests a–d)
 - [ ] T2: consumer trigger cutover — `SubscribeKind` in, bus tail out, per
       OQ-1/OQ-2/OQ-3 rulings; fabric serial-callback contract doc + no-overlap

go/internal/auth/classify_exhaustive_test.go

@@ -13,6 +13,7 @@ import (
 	"google.golang.org/protobuf/reflect/protoregistry"
 
 	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
+	compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1"
 )
 
 // procedurePath reconstructs the connect procedure path for a method descriptor.
@@ -36,6 +37,17 @@ func gatedFileDescriptors() []protoreflect.FileDescriptor {
 	}
 }
 
+// ungatedFileDescriptors are compass.v1 service files never mounted behind
+// AdminGate: Runner, agent-socket and guest-vsock surfaces with their own authz.
+// Importing them keeps the registry guard below independent of the link set.
+func ungatedFileDescriptors() []protoreflect.FileDescriptor {
+	return []protoreflect.FileDescriptor{
+		compassv1internal.File_compass_v1_runner_proto,
+		compassv1internal.File_compass_v1_agent_gateway_proto,
+		compassv1internal.File_compass_v1_guest_control_proto,
+	}
+}
+
 // TestClassifyProcedureCoversEveryGeneratedProcedure fails if any generated
 // CompassService or CommsService procedure is not explicitly classified by
 // classifyProcedure. This is the build-time gate the doc comment promises: a new
@@ -127,6 +139,12 @@ func TestClassificationGateCoversEveryRegisteredCompassService(t *testing.T) {
 			covered[services.Get(si).FullName()] = true
 		}
 	}
+	for _, file := range ungatedFileDescriptors() {
+		services := file.Services()
+		for si := range services.Len() {
+			covered[services.Get(si).FullName()] = true
+		}
+	}
 	if len(packages) == 0 {
 		t.Fatal("gatedFileDescriptors is empty — the classification gate covers nothing")
 	}
@@ -139,7 +157,7 @@ func TestClassificationGateCoversEveryRegisteredCompassService(t *testing.T) {
 				svc := services.Get(si)
 				checked++
 				if !covered[svc.FullName()] {
-					t.Errorf("service %q is registered in proto package %q but its file is not in gatedFileDescriptors — add its File_..._proto to the slice so classifyProcedure's exhaustiveness gate covers its RPCs (otherwise they silently fail-closed to adminOnly on the network door)", svc.FullName(), pkg)
+					t.Errorf("service %q is registered in proto package %q but its file is in neither gatedFileDescriptors nor ungatedFileDescriptors — add it to the gated slice so classifyProcedure's exhaustiveness gate covers its RPCs (otherwise they silently fail-closed to adminOnly on the network door), or to the ungated slice if it is never mounted behind AdminGate", svc.FullName(), pkg)
 				}
 			}
 			return true

go/internal/auth/interceptor_pgtest_test.go

@@ -74,7 +74,7 @@ func TestBearerInterceptorSetsCommsActorNotAdminFallback(t *testing.T) {
 	// diverts attribution to the real caller.
 	commsBus := events.NewBus[*compassv1.SubscribeCommsResponse]()
 	t.Cleanup(commsBus.Close)
-	commsSvc := comms.NewComms(st, commsBus, admin)
+	commsSvc := comms.NewComms(st, commsBus, nil, admin)
 
 	// Drive CreateChannelGroup through the bearer interceptor: the interceptor
 	// resolves the member token and threads the caller into the ctx it hands the

go/internal/comms/agent_caller_pgtest_test.go

@@ -160,7 +160,7 @@ func TestPostAsAccountEmptyAccountFailsClosedNoAdminWrite(t *testing.T) {
 	if err != nil {
 		t.Fatalf("BootstrapAdmin: %v", err)
 	}
-	svc := NewComms(st, bus, admin.ID)
+	svc := NewComms(st, bus, nil, admin.ID)
 	ctx := context.Background()
 
 	// A channel the admin founds (so admin is a member) — the write target that

go/internal/comms/comms.go

@@ -18,14 +18,18 @@ package comms
 import (
 	"context"
 	"errors"
+	"log/slog"
 
 	"connectrpc.com/connect"
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/RigelBuild/compass/go/events"
 	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
 	"github.com/RigelBuild/compass/go/gen/compass/v1/compassv1connect"
+	"github.com/RigelBuild/compass/go/internal/fabric"
 	"github.com/RigelBuild/compass/go/internal/store"
 )
 
@@ -37,13 +41,17 @@ import (
 // edge stamps Seq/AtUnixMs/InstanceEpoch onto a copy.
 type commsBus = *events.Bus[*compassv1.SubscribeCommsResponse]
 
+const instrumentationScope = "github.com/RigelBuild/compass/go/internal/comms"
+
 // Comms implements compassv1connect.CommsServiceHandler over the store and the
 // comms event bus. Cheap to share by pointer; the store and bus are each safe
 // for concurrent use, and Comms holds no mutable state of its own — the store is
 // the source of truth, so there is no in-memory account/channel/group state.
 type Comms struct {
-	store *store.Store
-	bus   commsBus
+	store                 *store.Store
+	bus                   commsBus
+	fabric                fabric.EventFabric
+	fabricPublishFailures metric.Int64Counter
 	// adminID attributes every RPC on the local-socket door (the door has no
 	// interceptor yet). The T3 interceptor overrides this per-request by setting
 	// a caller on the context; adminID is the fallback when none is set.
@@ -59,9 +67,19 @@ type Comms struct {
 
 // NewComms constructs the CommsService handler over store and bus. adminID is the
 // bootstrap-admin account the local-socket door attributes callers to until the
-// T3 interceptor sets a real identity (design.md:1219-1222).
-func NewComms(st *store.Store, bus commsBus, adminID store.AccountID) *Comms {
-	return &Comms{store: st, bus: bus, adminID: adminID}
+// T3 interceptor sets a real identity (design.md:1219-1222). A nil fab keeps
+// message_posted on the bus only.
+func NewComms(st *store.Store, bus commsBus, fab fabric.EventFabric, adminID store.AccountID) *Comms {
+	// A metric miss must never fail construction, matching the delivery counter.
+	failures, err := otel.Meter(instrumentationScope).Int64Counter(
+		"compass.delivery.fabric_publish_failures",
+		metric.WithDescription("Count of message_posted fabric publishes that failed after commit."),
+	)
+	if err != nil {
+		slog.Warn("comms: failed to create fabric publish failure counter; metric disabled", "err", err)
+		failures = nil
+	}
+	return &Comms{store: st, bus: bus, fabric: fab, fabricPublishFailures: failures, adminID: adminID}
 }
 
 // SetPresenceSource wires the in-memory presence enum source GetRoster joins

go/internal/comms/comms_test.go

@@ -59,7 +59,7 @@ func newHandler(t *testing.T) (*Comms, *store.Store) {
 	if err != nil {
 		t.Fatalf("BootstrapAdmin: %v", err)
 	}
-	return NewComms(st, bus, admin.ID), st
+	return NewComms(st, bus, nil, admin.ID), st
 }
 
 func TestCreateChannelEmitsChannelChanged(t *testing.T) {