Members ask their routers for the directory and listen as gossip leaves

api/agentmesh.proto

@@ -176,26 +176,6 @@ message DiscoveredProvider {
   string srv_description = 4;
 }
 
-// ServiceAnnounce is gossiped by a node on per-key topics (see
-// DiscoveryTopic) while those topics have subscribers. It is a routing
-// hint signed at the pubsub layer by the announcing peer: consumers use
-// it for freshness and load awareness, never for authorization.
-message ServiceAnnounce {
-  string peer_id = 1;
-  ServiceType type = 2;
-  string service_name = 3;
-  // Routing keys served by this service: model IDs for inference,
-  // tool names for MCP.
-  repeated string keys = 4;
-  // Operator-declared labels (e.g. "region"). Operator claims always
-  // take precedence over runtime-derived values.
-  map<string, string> labels = 5;
-  // Runtime load hints; zero values mean unknown.
-  uint32 active_requests = 6;
-  double latency_ewma_ms = 7;
-  google.protobuf.Timestamp announce_time = 8;
-}
-
 message ControlPlaneInfoResponse {
   string oidc_issuer = 1;
   string client_id = 2;
@@ -227,9 +207,9 @@ message ControlPlaneInfoResponse {
 // once per mesh on the control plane (agentmesh-control-plane
 // --dht-replication, --dht-reprovide-interval) and published on /info.
 message DirectoryParameters {
-  // How many routers hold each record a member publishes and answer each
-  // lookup it makes: the routers closest to the key. A member holds a
-  // session with each of them besides the routers it is attached to.
+  // How many routers each record a member announces is placed on, the
+  // closest to its key, besides the router that took the announcement.
+  // Routers read it; a member asks the routers it holds.
   int32 replication = 1;
   // How often a member re-announces each service it publishes.
   google.protobuf.Duration reprovide_interval = 2;
@@ -240,6 +220,57 @@ message DirectoryParameters {
   google.protobuf.Duration record_ttl = 3;
 }
 
+// DirectoryRequest is what a member asks a router it is attached to over
+// /mesh/directory/1.0.0: one request per stream, answered with one
+// DirectoryResponse. The member keeps no DHT of its own; the router runs
+// the lookup across the routers' DHT and answers from it. A router sends
+// the same message to another router to place a member's record on it.
+message DirectoryRequest {
+  oneof request {
+    DirectoryProvide provide = 1;
+    DirectoryFindProviders find_providers = 2;
+    DirectoryFindPeer find_peer = 3;
+  }
+}
+
+// DirectoryProvide announces the sender as a provider of key, reached at
+// addrs. A key is a sha256 multihash: of "mesh:service:<type>[:<name>]" for
+// a service, of "mesh:peer:<peer_id>" for the member itself. provider is set
+// only between routers, by the router that holds the member, and refused
+// from anyone else: a member can only announce itself.
+message DirectoryProvide {
+  bytes key = 1;
+  repeated string addrs = 2;
+  string provider = 3;
+}
+
+// DirectoryFindProviders asks for up to limit providers of key.
+message DirectoryFindProviders {
+  bytes key = 1;
+  uint32 limit = 2;
+}
+
+// DirectoryFindPeer asks where peer_id is reached.
+message DirectoryFindPeer {
+  string peer_id = 1;
+}
+
+// DirectoryResponse answers one DirectoryRequest. providers answers
+// find_providers; peer answers find_peer, unset when nobody knows the
+// peer; error is set when the router could not run the request.
+message DirectoryResponse {
+  repeated PeerRecord providers = 1;
+  PeerRecord peer = 2;
+  string error = 3;
+}
+
+// PeerRecord is a peer and the addresses it is reached at, relayed ones
+// included, each ending in its peer ID.
+message PeerRecord {
+  string peer_id = 1;
+  repeated string addrs = 2;
+}
+
 // RouterInfo is one router as the control plane lists it on /info.
 message RouterInfo {
   string peer_id = 1;

api/datalog.go

@@ -319,8 +319,7 @@ const (
 	// attested with region="us-east-1" carries label("region", "us-east-1"),
 	// satisfying `check if label("region", "us-east-1")` only — composition
 	// across values is left entirely to the operator (attest as many labels
-	// as needed). Distinct from the unauthenticated gossip routing hint
-	// carried in ServiceAnnounce.labels.
+	// as needed).
 	// Contains: biscuit.String(key), biscuit.String(value)
 	// Example Datalog: check if label("region", "us-east-1")
 	FactLabel = "label"

api/directory_keys.go

@@ -0,0 +1,73 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package api
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"strings"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
+)
+
+// DirectoryProtocolID is the libp2p protocol on which a member asks a
+// router it is attached to for the directory: one DirectoryRequest per
+// stream, one DirectoryResponse back. Routers place records on each other
+// over it as well.
+const DirectoryProtocolID protocol.ID = "/mesh/directory/1.0.0"
+
+// DirectoryKeyBytes is the length of a directory key: a sha256 multihash.
+const DirectoryKeyBytes = 34
+
+// DirectoryMaxAddrs bounds the addresses one record carries.
+const DirectoryMaxAddrs = 32
+
+// directoryKey is the sha256 multihash of a name: the two-byte multihash
+// header (sha2-256, 32 bytes) and the digest. go-libp2p-kad-dht keys
+// provider records by this multihash, so a record a member announces here
+// is the record a router's DHT holds.
+func directoryKey(name string) []byte {
+	sum := sha256.Sum256([]byte(name))
+	return append([]byte{0x12, 0x20}, sum[:]...)
+}
+
+// ServiceKey is the directory key of a service type, or of one named
+// service of that type: "mesh:service:<type>" or "mesh:service:<type>:<name>".
+func ServiceKey(t ServiceType, name string) ([]byte, error) {
+	typeStr, err := ServiceTypeToString(t)
+	if err != nil {
+		return nil, err
+	}
+	parts := []string{"mesh:service", typeStr}
+	if name != "" {
+		parts = append(parts, name)
+	}
+	return directoryKey(strings.Join(parts, ":")), nil
+}
+
+// PeerKey is the directory key under which a member announces where it is
+// reached: "mesh:peer:<peer_id>". A find_peer is a lookup of this key.
+func PeerKey(p peer.ID) []byte {
+	return directoryKey("mesh:peer:" + p.String())
+}
+
+// ValidateDirectoryKey rejects anything but a sha256 multihash.
+func ValidateDirectoryKey(key []byte) error {
+	if len(key) != DirectoryKeyBytes || key[0] != 0x12 || key[1] != 0x20 {
+		return fmt.Errorf("directory key must be a sha256 multihash of %d bytes, got %d", DirectoryKeyBytes, len(key))
+	}
+	return nil
+}

api/directory_keys_test.go

@@ -0,0 +1,70 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package api
+
+import (
+	"bytes"
+	"encoding/hex"
+	"testing"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+// The keys are the sha256 multihash of a fixed name, so a record a member
+// announces in any language lands where the others look it up. The vectors
+// here are shared with the SDK tests.
+func TestDirectoryKeys(t *testing.T) {
+	mcp, err := ServiceKey(ServiceType_SERVICE_TYPE_MCP, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	calc, err := ServiceKey(ServiceType_SERVICE_TYPE_MCP, "calculator")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Equal(mcp, calc) {
+		t.Fatal("a service type and one of its services share a key")
+	}
+	for name, key := range map[string][]byte{"mcp": mcp, "mcp:calculator": calc} {
+		if err := ValidateDirectoryKey(key); err != nil {
+			t.Errorf("%s: %v", name, err)
+		}
+	}
+	// sha256("mesh:service:mcp:calculator") behind the multihash header.
+	const want = "1220" + "e99b7ec759ced5276520910d75b1f6fe3a5baf5fe46655ac34957139e04a0859"
+	if got := hex.EncodeToString(calc); got != want {
+		t.Fatalf("calculator key = %s, want %s", got, want)
+	}
+
+	if _, err := ServiceKey(ServiceType_SERVICE_TYPE_UNSPECIFIED, "x"); err == nil {
+		t.Error("an unspecified service type got a key")
+	}
+
+	p, err := peer.Decode("12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := ValidateDirectoryKey(PeerKey(p)); err != nil {
+		t.Error(err)
+	}
+	if bytes.Equal(PeerKey(p), mcp) {
+		t.Error("a peer key collides with a service key")
+	}
+	for _, bad := range [][]byte{nil, make([]byte, 33), append([]byte{0x11, 0x20}, make([]byte, 32)...)} {
+		if err := ValidateDirectoryKey(bad); err == nil {
+			t.Errorf("key %x accepted", bad)
+		}
+	}
+}

api/network.go

@@ -15,8 +15,6 @@
 package api
 
 import (
-	"crypto/sha256"
-	"encoding/hex"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -56,11 +54,6 @@ const (
 	// GossipControlPlaneSync is the GossipSub topic used by the control plane to sync cluster state.
 	GossipControlPlaneSync = "/mesh/control-plane/sync/v1"
 
-	// DiscoveryTopicPrefix is the GossipSub topic namespace for interest-scoped
-	// service announcements (ServiceAnnounce messages). Full topics are built
-	// with DiscoveryTopic; the version segment allows wire evolution.
-	DiscoveryTopicPrefix = "/mesh/discovery/v1"
-
 	// DefaultAudience is the default audience string used in OIDC token validation.
 	DefaultAudience = "agentmesh-audience"
 )
@@ -388,23 +381,6 @@ func ServiceTypeToString(t ServiceType) (string, error) {
 	}
 }
 
-// DiscoveryTopic returns the GossipSub topic for announcements about one
-// routing key (a model ID for inference, a tool name for MCP). Keys are
-// hashed so topic names stay bounded; consumers match exact keys from the
-// ServiceAnnounce payload, so hash collisions only merge announcement
-// streams, never routing decisions.
-func DiscoveryTopic(t ServiceType, key string) (string, error) {
-	typeStr, err := ServiceTypeToString(t)
-	if err != nil {
-		return "", err
-	}
-	if key == "" {
-		return "", fmt.Errorf("discovery topic key cannot be empty")
-	}
-	sum := sha256.Sum256([]byte(key))
-	return fmt.Sprintf("%s/%s/%s", DiscoveryTopicPrefix, typeStr, hex.EncodeToString(sum[:8])), nil
-}
-
 // ============================================================================
 // Parsing & Routing Utilities
 // ============================================================================

api/validation.go

@@ -20,49 +20,6 @@ import (
 	"strings"
 )
 
-// Caps for ServiceAnnounce fields: announcements are unsolicited gossip, so
-// receivers bound every dimension before processing.
-const (
-	MaxAnnounceKeys      = 64
-	MaxAnnounceLabels    = 16
-	MaxAnnounceStringLen = 256
-)
-
-// ValidateServiceAnnounce bounds and sanity-checks a gossiped announcement.
-// Origin authenticity and freshness are the receiver's responsibility.
-func ValidateServiceAnnounce(a *ServiceAnnounce) error {
-	if a == nil {
-		return fmt.Errorf("announce is nil")
-	}
-	if a.GetPeerId() == "" || len(a.GetPeerId()) > MaxAnnounceStringLen {
-		return fmt.Errorf("invalid peer_id")
-	}
-	if _, err := ServiceTypeToString(a.GetType()); err != nil {
-		return fmt.Errorf("invalid type: %w", err)
-	}
-	if a.GetServiceName() == "" || len(a.GetServiceName()) > MaxAnnounceStringLen {
-		return fmt.Errorf("invalid service_name")
-	}
-	if len(a.GetKeys()) == 0 || len(a.GetKeys()) > MaxAnnounceKeys {
-		return fmt.Errorf("keys count %d out of range [1, %d]", len(a.GetKeys()), MaxAnnounceKeys)
-	}
-	for _, k := range a.GetKeys() {
-		if k == "" || len(k) > MaxAnnounceStringLen {
-			return fmt.Errorf("invalid key %q", k)
-		}
-	}
-	if len(a.GetLabels()) > MaxAnnounceLabels {
-		return fmt.Errorf("labels count %d exceeds %d", len(a.GetLabels()), MaxAnnounceLabels)
-	}
-	if err := ValidateLabels(a.GetLabels()); err != nil {
-		return fmt.Errorf("invalid labels: %w", err)
-	}
-	if a.GetAnnounceTime() == nil || !a.GetAnnounceTime().IsValid() || a.GetAnnounceTime().AsTime().Unix() <= 0 {
-		return fmt.Errorf("missing or invalid announce_time")
-	}
-	return nil
-}
-
 // ValidateServiceFormat ensures the service string follows the explicit URI format.
 func ValidateServiceFormat(svc string) error {
 	if svc == "*" {

api/validation_test.go

@@ -16,8 +16,6 @@ package api
 
 import (
 	"testing"
-
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func TestValidateServiceFormat(t *testing.T) {
@@ -83,32 +81,3 @@ func TestValidateTargetFormat(t *testing.T) {
 		})
 	}
 }
-
-func TestValidateServiceAnnounce(t *testing.T) {
-	validAnnounce := func() *ServiceAnnounce {
-		return &ServiceAnnounce{
-			PeerId:       "12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB",
-			Type:         ServiceType_SERVICE_TYPE_MCP,
-			ServiceName:  "weather",
-			Keys:         []string{"get_weather"},
-			Labels:       map[string]string{"env": "prod"},
-			AnnounceTime: timestamppb.Now(),
-		}
-	}
-	if err := ValidateServiceAnnounce(validAnnounce()); err != nil {
-		t.Fatalf("ValidateServiceAnnounce(valid) = %v", err)
-	}
-
-	epochAnnounce := validAnnounce()
-	epochAnnounce.AnnounceTime.Seconds = 0
-	epochAnnounce.AnnounceTime.Nanos = 0
-	if err := ValidateServiceAnnounce(epochAnnounce); err == nil {
-		t.Fatal("expected epoch AnnounceTime to be rejected")
-	}
-
-	badLabelAnnounce := validAnnounce()
-	badLabelAnnounce.Labels = map[string]string{"bad key!": "val"}
-	if err := ValidateServiceAnnounce(badLabelAnnounce); err == nil {
-		t.Fatal("expected invalid label key in ServiceAnnounce to be rejected")
-	}
-}