diff --git a/Makefile b/Makefile index e7ed7847..ba1bb57b 100644 --- a/Makefile +++ b/Makefile @@ -9,6 +9,7 @@ build: go build -v -o "$(OUT_DIR)/sam-node" ./cmd/sam-node go build -v -o "$(OUT_DIR)/sam-hub" ./cmd/sam-hub go build -v -o "$(OUT_DIR)/mcp-client" ./cmd/mcp-client + go build -v -o "$(OUT_DIR)/sam-catalog" ./cmd/sam-catalog .PHONY: proto proto: diff --git a/api/sam.pb.go b/api/sam.pb.go index 28ef1be7..c1b4f328 100644 --- a/api/sam.pb.go +++ b/api/sam.pb.go @@ -41,6 +41,7 @@ const ( ServiceType_SERVICE_TYPE_UNSPECIFIED ServiceType = 0 ServiceType_SERVICE_TYPE_MCP ServiceType = 1 ServiceType_SERVICE_TYPE_INFERENCE ServiceType = 2 + ServiceType_SERVICE_TYPE_CATALOG ServiceType = 3 ) // Enum value maps for ServiceType. @@ -49,11 +50,13 @@ var ( 0: "SERVICE_TYPE_UNSPECIFIED", 1: "SERVICE_TYPE_MCP", 2: "SERVICE_TYPE_INFERENCE", + 3: "SERVICE_TYPE_CATALOG", } ServiceType_value = map[string]int32{ "SERVICE_TYPE_UNSPECIFIED": 0, "SERVICE_TYPE_MCP": 1, "SERVICE_TYPE_INFERENCE": 2, + "SERVICE_TYPE_CATALOG": 3, } ) @@ -498,6 +501,98 @@ func (x *ServiceInfo) GetDescription() string { return "" } +type ServiceAnnounce struct { + state protoimpl.MessageState `protogen:"open.v1"` + Type ServiceType `protobuf:"varint,1,opt,name=type,proto3,enum=sam.v1.ServiceType" json:"type,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + PeerId string `protobuf:"bytes,3,opt,name=peer_id,json=peerId,proto3" json:"peer_id,omitempty"` + Addrs []string `protobuf:"bytes,4,rep,name=addrs,proto3" json:"addrs,omitempty"` + Timestamp int64 `protobuf:"varint,5,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + TtlMs int64 `protobuf:"varint,6,opt,name=ttl_ms,json=ttlMs,proto3" json:"ttl_ms,omitempty"` + Signature []byte `protobuf:"bytes,7,opt,name=signature,proto3" json:"signature,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ServiceAnnounce) Reset() { + *x = ServiceAnnounce{} + mi := &file_api_sam_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ServiceAnnounce) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ServiceAnnounce) ProtoMessage() {} + +func (x *ServiceAnnounce) ProtoReflect() protoreflect.Message { + mi := &file_api_sam_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ServiceAnnounce.ProtoReflect.Descriptor instead. +func (*ServiceAnnounce) Descriptor() ([]byte, []int) { + return file_api_sam_proto_rawDescGZIP(), []int{6} +} + +func (x *ServiceAnnounce) GetType() ServiceType { + if x != nil { + return x.Type + } + return ServiceType_SERVICE_TYPE_UNSPECIFIED +} + +func (x *ServiceAnnounce) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *ServiceAnnounce) GetPeerId() string { + if x != nil { + return x.PeerId + } + return "" +} + +func (x *ServiceAnnounce) GetAddrs() []string { + if x != nil { + return x.Addrs + } + return nil +} + +func (x *ServiceAnnounce) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + +func (x *ServiceAnnounce) GetTtlMs() int64 { + if x != nil { + return x.TtlMs + } + return 0 +} + +func (x *ServiceAnnounce) GetSignature() []byte { + if x != nil { + return x.Signature + } + return nil +} + type CommandBackend struct { state protoimpl.MessageState `protogen:"open.v1"` Command []string `protobuf:"bytes,1,rep,name=command,proto3" json:"command,omitempty"` @@ -508,7 +603,7 @@ type CommandBackend struct { func (x *CommandBackend) Reset() { *x = CommandBackend{} - mi := &file_api_sam_proto_msgTypes[6] + mi := &file_api_sam_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -520,7 +615,7 @@ func (x *CommandBackend) String() string { func (*CommandBackend) ProtoMessage() {} func (x *CommandBackend) ProtoReflect() protoreflect.Message { - mi := &file_api_sam_proto_msgTypes[6] + mi := &file_api_sam_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -533,7 +628,7 @@ func (x *CommandBackend) ProtoReflect() protoreflect.Message { // Deprecated: Use CommandBackend.ProtoReflect.Descriptor instead. func (*CommandBackend) Descriptor() ([]byte, []int) { - return file_api_sam_proto_rawDescGZIP(), []int{6} + return file_api_sam_proto_rawDescGZIP(), []int{7} } func (x *CommandBackend) GetCommand() []string { @@ -564,7 +659,7 @@ type RegisterServiceRequest struct { func (x *RegisterServiceRequest) Reset() { *x = RegisterServiceRequest{} - mi := &file_api_sam_proto_msgTypes[7] + mi := &file_api_sam_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -576,7 +671,7 @@ func (x *RegisterServiceRequest) String() string { func (*RegisterServiceRequest) ProtoMessage() {} func (x *RegisterServiceRequest) ProtoReflect() protoreflect.Message { - mi := &file_api_sam_proto_msgTypes[7] + mi := &file_api_sam_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -589,7 +684,7 @@ func (x *RegisterServiceRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RegisterServiceRequest.ProtoReflect.Descriptor instead. func (*RegisterServiceRequest) Descriptor() ([]byte, []int) { - return file_api_sam_proto_rawDescGZIP(), []int{7} + return file_api_sam_proto_rawDescGZIP(), []int{8} } func (x *RegisterServiceRequest) GetService() *ServiceInfo { @@ -652,7 +747,7 @@ type DiscoveredProvider struct { func (x *DiscoveredProvider) Reset() { *x = DiscoveredProvider{} - mi := &file_api_sam_proto_msgTypes[8] + mi := &file_api_sam_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -664,7 +759,7 @@ func (x *DiscoveredProvider) String() string { func (*DiscoveredProvider) ProtoMessage() {} func (x *DiscoveredProvider) ProtoReflect() protoreflect.Message { - mi := &file_api_sam_proto_msgTypes[8] + mi := &file_api_sam_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -677,7 +772,7 @@ func (x *DiscoveredProvider) ProtoReflect() protoreflect.Message { // Deprecated: Use DiscoveredProvider.ProtoReflect.Descriptor instead. func (*DiscoveredProvider) Descriptor() ([]byte, []int) { - return file_api_sam_proto_rawDescGZIP(), []int{8} + return file_api_sam_proto_rawDescGZIP(), []int{9} } func (x *DiscoveredProvider) GetPeerId() string { @@ -720,7 +815,7 @@ type HubInfoResponse struct { func (x *HubInfoResponse) Reset() { *x = HubInfoResponse{} - mi := &file_api_sam_proto_msgTypes[9] + mi := &file_api_sam_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -732,7 +827,7 @@ func (x *HubInfoResponse) String() string { func (*HubInfoResponse) ProtoMessage() {} func (x *HubInfoResponse) ProtoReflect() protoreflect.Message { - mi := &file_api_sam_proto_msgTypes[9] + mi := &file_api_sam_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -745,7 +840,7 @@ func (x *HubInfoResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use HubInfoResponse.ProtoReflect.Descriptor instead. func (*HubInfoResponse) Descriptor() ([]byte, []int) { - return file_api_sam_proto_rawDescGZIP(), []int{9} + return file_api_sam_proto_rawDescGZIP(), []int{10} } func (x *HubInfoResponse) GetOidcIssuer() string { @@ -811,7 +906,15 @@ const file_api_sam_proto_rawDesc = "" + "\vServiceInfo\x12'\n" + "\x04type\x18\x01 \x01(\x0e2\x13.sam.v1.ServiceTypeR\x04type\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12 \n" + - "\vdescription\x18\x03 \x01(\tR\vdescription\"\x95\x01\n" + + "\vdescription\x18\x03 \x01(\tR\vdescription\"\xd0\x01\n" + + "\x0fServiceAnnounce\x12'\n" + + "\x04type\x18\x01 \x01(\x0e2\x13.sam.v1.ServiceTypeR\x04type\x12\x12\n" + + "\x04name\x18\x02 \x01(\tR\x04name\x12\x17\n" + + "\apeer_id\x18\x03 \x01(\tR\x06peerId\x12\x14\n" + + "\x05addrs\x18\x04 \x03(\tR\x05addrs\x12\x1c\n" + + "\ttimestamp\x18\x05 \x01(\x03R\ttimestamp\x12\x15\n" + + "\x06ttl_ms\x18\x06 \x01(\x03R\x05ttlMs\x12\x1c\n" + + "\tsignature\x18\a \x01(\fR\tsignature\"\x95\x01\n" + "\x0eCommandBackend\x12\x18\n" + "\acommand\x18\x01 \x03(\tR\acommand\x121\n" + "\x03env\x18\x02 \x03(\v2\x1f.sam.v1.CommandBackend.EnvEntryR\x03env\x1a6\n" + @@ -834,11 +937,12 @@ const file_api_sam_proto_rawDesc = "" + "oidcIssuer\x12\x1b\n" + "\tclient_id\x18\x02 \x01(\tR\bclientId\x12\x1a\n" + "\baudience\x18\x03 \x01(\tR\baudience\x12#\n" + - "\rhub_addresses\x18\x04 \x03(\tR\fhubAddresses*]\n" + + "\rhub_addresses\x18\x04 \x03(\tR\fhubAddresses*w\n" + "\vServiceType\x12\x1c\n" + "\x18SERVICE_TYPE_UNSPECIFIED\x10\x00\x12\x14\n" + "\x10SERVICE_TYPE_MCP\x10\x01\x12\x1a\n" + - "\x16SERVICE_TYPE_INFERENCE\x10\x02B\x1bZ\x19github.com/google/sam/apib\x06proto3" + "\x16SERVICE_TYPE_INFERENCE\x10\x02\x12\x18\n" + + "\x14SERVICE_TYPE_CATALOG\x10\x03B\x1bZ\x19github.com/google/sam/apib\x06proto3" var ( file_api_sam_proto_rawDescOnce sync.Once @@ -853,7 +957,7 @@ func file_api_sam_proto_rawDescGZIP() []byte { } var file_api_sam_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_api_sam_proto_msgTypes = make([]protoimpl.MessageInfo, 11) +var file_api_sam_proto_msgTypes = make([]protoimpl.MessageInfo, 12) var file_api_sam_proto_goTypes = []any{ (ServiceType)(0), // 0: sam.v1.ServiceType (MeshEvent_Type)(0), // 1: sam.v1.MeshEvent.Type @@ -863,23 +967,25 @@ var file_api_sam_proto_goTypes = []any{ (*EnrollRequest)(nil), // 5: sam.v1.EnrollRequest (*EnrollResponse)(nil), // 6: sam.v1.EnrollResponse (*ServiceInfo)(nil), // 7: sam.v1.ServiceInfo - (*CommandBackend)(nil), // 8: sam.v1.CommandBackend - (*RegisterServiceRequest)(nil), // 9: sam.v1.RegisterServiceRequest - (*DiscoveredProvider)(nil), // 10: sam.v1.DiscoveredProvider - (*HubInfoResponse)(nil), // 11: sam.v1.HubInfoResponse - nil, // 12: sam.v1.CommandBackend.EnvEntry + (*ServiceAnnounce)(nil), // 8: sam.v1.ServiceAnnounce + (*CommandBackend)(nil), // 9: sam.v1.CommandBackend + (*RegisterServiceRequest)(nil), // 10: sam.v1.RegisterServiceRequest + (*DiscoveredProvider)(nil), // 11: sam.v1.DiscoveredProvider + (*HubInfoResponse)(nil), // 12: sam.v1.HubInfoResponse + nil, // 13: sam.v1.CommandBackend.EnvEntry } var file_api_sam_proto_depIdxs = []int32{ 1, // 0: sam.v1.MeshEvent.type:type_name -> sam.v1.MeshEvent.Type 0, // 1: sam.v1.ServiceInfo.type:type_name -> sam.v1.ServiceType - 12, // 2: sam.v1.CommandBackend.env:type_name -> sam.v1.CommandBackend.EnvEntry - 7, // 3: sam.v1.RegisterServiceRequest.service:type_name -> sam.v1.ServiceInfo - 8, // 4: sam.v1.RegisterServiceRequest.command:type_name -> sam.v1.CommandBackend - 5, // [5:5] is the sub-list for method output_type - 5, // [5:5] is the sub-list for method input_type - 5, // [5:5] is the sub-list for extension type_name - 5, // [5:5] is the sub-list for extension extendee - 0, // [0:5] is the sub-list for field type_name + 0, // 2: sam.v1.ServiceAnnounce.type:type_name -> sam.v1.ServiceType + 13, // 3: sam.v1.CommandBackend.env:type_name -> sam.v1.CommandBackend.EnvEntry + 7, // 4: sam.v1.RegisterServiceRequest.service:type_name -> sam.v1.ServiceInfo + 9, // 5: sam.v1.RegisterServiceRequest.command:type_name -> sam.v1.CommandBackend + 6, // [6:6] is the sub-list for method output_type + 6, // [6:6] is the sub-list for method input_type + 6, // [6:6] is the sub-list for extension type_name + 6, // [6:6] is the sub-list for extension extendee + 0, // [0:6] is the sub-list for field type_name } func init() { file_api_sam_proto_init() } @@ -887,7 +993,7 @@ func file_api_sam_proto_init() { if File_api_sam_proto != nil { return } - file_api_sam_proto_msgTypes[7].OneofWrappers = []any{ + file_api_sam_proto_msgTypes[8].OneofWrappers = []any{ (*RegisterServiceRequest_TargetUrl)(nil), (*RegisterServiceRequest_Command)(nil), } @@ -897,7 +1003,7 @@ func file_api_sam_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_api_sam_proto_rawDesc), len(file_api_sam_proto_rawDesc)), NumEnums: 2, - NumMessages: 11, + NumMessages: 12, NumExtensions: 0, NumServices: 0, }, diff --git a/api/sam.proto b/api/sam.proto index 8aa775a1..56f4867b 100644 --- a/api/sam.proto +++ b/api/sam.proto @@ -59,6 +59,7 @@ enum ServiceType { SERVICE_TYPE_UNSPECIFIED = 0; SERVICE_TYPE_MCP = 1; SERVICE_TYPE_INFERENCE = 2; + SERVICE_TYPE_CATALOG = 3; } message ServiceInfo { @@ -67,6 +68,16 @@ message ServiceInfo { string description = 3; } +message ServiceAnnounce { + ServiceType type = 1; + string name = 2; + string peer_id = 3; + repeated string addrs = 4; + int64 timestamp = 5; + int64 ttl_ms = 6; + bytes signature = 7; +} + message CommandBackend { repeated string command = 1; map env = 2; diff --git a/api/types.go b/api/types.go index dd4b7780..c344c623 100644 --- a/api/types.go +++ b/api/types.go @@ -26,6 +26,7 @@ const EnrollProtocolID protocol.ID = "/sam/enroll/1.0.0" const MCPProtocolID protocol.ID = "/sam/mcp/1.0.0" const GossipEvents = "/sam/mesh/events/v1" const GossipHubSync = "/sam/hub/sync/v1" +const GossipServiceAnnounce = "/sam/service/announce/v1" const AuthProtocolID protocol.ID = "/sam/auth/1.0.0" // CatalogTarget is the special target service name used to retrieve tool catalogs from remote nodes. @@ -101,6 +102,7 @@ func ParseServiceTarget(target string) (svcType, svcName string) { const ( ServiceTypeStringMCP = "mcp" ServiceTypeStringInference = "inference" + ServiceTypeStringCatalog = "catalog" ) // InferenceServicePrefix is the conventional prefix used for LLM gateway inference services. @@ -113,6 +115,8 @@ func ParseServiceType(s string) (ServiceType, error) { return ServiceType_SERVICE_TYPE_MCP, nil case ServiceTypeStringInference: return ServiceType_SERVICE_TYPE_INFERENCE, nil + case ServiceTypeStringCatalog: + return ServiceType_SERVICE_TYPE_CATALOG, nil default: return ServiceType_SERVICE_TYPE_UNSPECIFIED, fmt.Errorf("invalid service type: %s", s) } @@ -125,6 +129,8 @@ func ServiceTypeToString(t ServiceType) (string, error) { return ServiceTypeStringMCP, nil case ServiceType_SERVICE_TYPE_INFERENCE: return ServiceTypeStringInference, nil + case ServiceType_SERVICE_TYPE_CATALOG: + return ServiceTypeStringCatalog, nil default: return "", fmt.Errorf("invalid or unspecified service type") } diff --git a/cmd/sam-catalog/ingest.go b/cmd/sam-catalog/ingest.go new file mode 100644 index 00000000..3f38e855 --- /dev/null +++ b/cmd/sam-catalog/ingest.go @@ -0,0 +1,197 @@ +// 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 main + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "time" + + "github.com/google/sam/api" + "github.com/google/sam/internal/announce" + "github.com/google/sam/internal/catalog" + "google.golang.org/protobuf/encoding/protojson" +) + +type nodeClient struct { + baseURL string + token string + hc *http.Client +} + +func newNodeClient(baseURL, token string) *nodeClient { + return &nodeClient{baseURL: baseURL, token: token, hc: &http.Client{}} +} + +// serviceTypeStr maps ServiceType to the short query-param string. +func serviceTypeStr(t api.ServiceType) (string, error) { + switch t { + case api.ServiceType_SERVICE_TYPE_MCP: + return "mcp", nil + case api.ServiceType_SERVICE_TYPE_INFERENCE: + return "inference", nil + case api.ServiceType_SERVICE_TYPE_CATALOG: + return "catalog", nil + default: + return "", fmt.Errorf("unsupported service type: %v", t) + } +} + +// bootstrap fetches discovered providers per type and upserts synthetic announces. +// Per-type failures are logged and skipped; remaining types are still processed. +func (c *nodeClient) bootstrap(ctx context.Context, store *catalog.Store, types []api.ServiceType) error { + for _, t := range types { + typeStr, err := serviceTypeStr(t) + if err != nil { + log.Printf("bootstrap: skip unknown type %v: %v", t, err) + continue + } + url := c.baseURL + "/sam/service/discover?type=" + typeStr + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + log.Printf("bootstrap: build request for %s: %v", typeStr, err) + continue + } + req.Header.Set("Authorization", "Bearer "+c.token) + resp, err := c.hc.Do(req) + if err != nil { + log.Printf("bootstrap: fetch %s: %v", typeStr, err) + continue + } + if resp.StatusCode != http.StatusOK { + // surface auth/server errors instead of silently skipping + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + _ = resp.Body.Close() + log.Printf("bootstrap: %s returned status %d: %s", typeStr, resp.StatusCode, strings.TrimSpace(string(snippet))) + continue + } + var providers []*api.DiscoveredProvider + if err := json.NewDecoder(resp.Body).Decode(&providers); err != nil { + _ = resp.Body.Close() + log.Printf("bootstrap: decode providers for %s: %v", typeStr, err) + continue + } + _ = resp.Body.Close() + for _, p := range providers { + // No Addrs: discover doesn't return provider dial addrs. Bootstrap-only + // entries carry empty Addrs until a live announce refreshes them (~1 + // reprovide cycle). Harmless today: consumers route by peer id via the + // egress proxy, not Entry.Addrs. Fill from peerstore/DHT only if a + // consumer ever dials Addrs directly. + ann := &api.ServiceAnnounce{ + Type: t, + Name: p.SrvName, + PeerId: p.PeerId, + TtlMs: announce.TTL.Milliseconds(), + } + store.Upsert(ann, time.Now()) + } + } + return nil +} + +// tail streams SSE announces from the node and upserts each into the store. +// It reconnects with a fixed backoff (on any exit, error or clean EOF) until ctx is done. +func (c *nodeClient) tail(ctx context.Context, store *catalog.Store) error { + const backoff = 2 * time.Second + for { + if ctx.Err() != nil { + return ctx.Err() + } + if err := c.readSSEStream(ctx, store); err != nil && ctx.Err() == nil { + log.Printf("tail: stream error: %v; reconnecting in %s", err, backoff) + } + // Always back off before reconnecting, whether stream ended cleanly or not. + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + } +} + +// readSSEStream opens one SSE connection and processes events until EOF or ctx done. +func (c *nodeClient) readSSEStream(ctx context.Context, store *catalog.Store) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/sam/service/announce/stream", nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+c.token) + req.Header.Set("Accept", "text/event-stream") + resp, err := c.hc.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + // surface auth/server errors; defer closes the body + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("announce stream returned status %d: %s", resp.StatusCode, strings.TrimSpace(string(snippet))) + } + + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "data: ") { + continue // skip event:/blank lines + } + payload := strings.TrimPrefix(line, "data: ") + var ann api.ServiceAnnounce + if err := protojson.Unmarshal([]byte(payload), &ann); err != nil { + continue + } + store.Upsert(&ann, time.Now()) + } + return scanner.Err() +} + +// runSweeper calls store.Sweep on a ticker until ctx is done. +func runSweeper(ctx context.Context, store *catalog.Store, every time.Duration) { + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case now := <-ticker.C: + store.Sweep(now) + } + } +} + +// runRewalk periodically calls bootstrap to reconcile entries missed via gossip. +func runRewalk(ctx context.Context, nc *nodeClient, store *catalog.Store, every time.Duration) { + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := nc.bootstrap(ctx, store, []api.ServiceType{ + api.ServiceType_SERVICE_TYPE_MCP, + api.ServiceType_SERVICE_TYPE_INFERENCE, + }); err != nil && ctx.Err() == nil { + log.Printf("rewalk: %v", err) + } + } + } +} diff --git a/cmd/sam-catalog/ingest_test.go b/cmd/sam-catalog/ingest_test.go new file mode 100644 index 00000000..781615dc --- /dev/null +++ b/cmd/sam-catalog/ingest_test.go @@ -0,0 +1,155 @@ +// 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 main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/sam/api" + "github.com/google/sam/internal/announce" + "github.com/google/sam/internal/catalog" + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" + "google.golang.org/protobuf/encoding/protojson" +) + +func newTestIdentity(t *testing.T) (crypto.PrivKey, peer.ID) { + t.Helper() + priv, pub, err := crypto.GenerateKeyPair(crypto.Ed25519, -1) + if err != nil { + t.Fatalf("GenerateKeyPair: %v", err) + } + pid, err := peer.IDFromPublicKey(pub) + if err != nil { + t.Fatalf("IDFromPublicKey: %v", err) + } + return priv, pid +} + +func buildSSEEvent(t *testing.T, serviceType api.ServiceType, name string) string { + t.Helper() + priv, pid := newTestIdentity(t) + info := &api.ServiceInfo{Type: serviceType, Name: name} + a := announce.Build(info, pid, nil, time.Now(), announce.TTL) + if err := announce.Sign(priv, a); err != nil { + t.Fatalf("sign: %v", err) + } + data, err := protojson.Marshal(a) + if err != nil { + t.Fatalf("protojson marshal: %v", err) + } + return fmt.Sprintf("data: %s\n\n", data) +} + +// TestTailPopulatesStoreFromSSE verifies that tail ingests two SSE announce events into the store. +func TestTailPopulatesStoreFromSSE(t *testing.T) { + ev1 := buildSSEEvent(t, api.ServiceType_SERVICE_TYPE_MCP, "tool-a") + ev2 := buildSSEEvent(t, api.ServiceType_SERVICE_TYPE_INFERENCE, "model-b") + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-token" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + _, _ = fmt.Fprint(w, ev1) + _, _ = fmt.Fprint(w, ev2) + if flusher != nil { + flusher.Flush() + } + <-r.Context().Done() + })) + defer srv.Close() + + store := catalog.New() + client := newNodeClient(srv.URL, "test-token") + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + go func() { _ = client.tail(ctx, store) }() + + // Wait up to 2s for 2 entries to land. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if len(store.List(api.ServiceType_SERVICE_TYPE_UNSPECIFIED, "")) == 2 { + break + } + time.Sleep(20 * time.Millisecond) + } + + cancel() + + entries := store.List(api.ServiceType_SERVICE_TYPE_UNSPECIFIED, "") + if got := len(entries); got != 2 { + t.Fatalf("expected 2 entries in store, got %d", got) + } +} + +// TestBootstrapUpsertsSyntheticAnnounces verifies bootstrap queries per type and upserts entries. +func TestBootstrapUpsertsSyntheticAnnounces(t *testing.T) { + _, pid1 := newTestIdentity(t) + _, pid2 := newTestIdentity(t) + + providers := map[string][]*api.DiscoveredProvider{ + "mcp": { + {PeerId: pid1.String(), SrvName: "tool-x"}, + }, + "inference": { + {PeerId: pid2.String(), SrvName: "model-y"}, + }, + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-token" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + typeParam := r.URL.Query().Get("type") + list, ok := providers[typeParam] + if !ok { + http.Error(w, "unknown type", http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + // Mirror the node: json.NewEncoder uses struct json tags (snake_case). + _ = json.NewEncoder(w).Encode(list) + })) + defer srv.Close() + + store := catalog.New() + client := newNodeClient(srv.URL, "test-token") + + ctx := context.Background() + if err := client.bootstrap(ctx, store, []api.ServiceType{ + api.ServiceType_SERVICE_TYPE_MCP, + api.ServiceType_SERVICE_TYPE_INFERENCE, + }); err != nil { + t.Fatalf("bootstrap: %v", err) + } + + entries := store.List(api.ServiceType_SERVICE_TYPE_UNSPECIFIED, "") + if got := len(entries); got != 2 { + t.Fatalf("expected 2 entries after bootstrap, got %d", got) + } +} diff --git a/cmd/sam-catalog/main.go b/cmd/sam-catalog/main.go new file mode 100644 index 00000000..0e8e29c1 --- /dev/null +++ b/cmd/sam-catalog/main.go @@ -0,0 +1,123 @@ +// 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 main + +import ( + "context" + "fmt" + "log" + "net" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/google/sam/api" + "github.com/google/sam/internal/catalog" + "github.com/spf13/cobra" +) + +var ( + nodeURL string + nodeToken string + bindAddr string + ownURL string + rewalkInterval time.Duration + sweepInterval time.Duration +) + +func main() { + rootCmd := &cobra.Command{ + Use: "sam-catalog", + Short: "SAM service catalog — aggregates and exposes service discoveries via MCP", + RunE: run, + } + + rootCmd.Flags().StringVar(&nodeURL, "node-url", "", "sam-node sidecar base URL (e.g. http://127.0.0.1:8080)") + rootCmd.Flags().StringVar(&nodeToken, "node-token", "", "Bearer token for sam-node sidecar API") + rootCmd.Flags().StringVar(&bindAddr, "bind-addr", "127.0.0.1:0", "Catalog MCP listen address") + rootCmd.Flags().StringVar(&ownURL, "own-url", "", "URL the node proxies to this catalog (default: http://)") + rootCmd.Flags().DurationVar(&rewalkInterval, "rewalk-interval", 3*time.Minute, "Interval between full bootstrap re-walks") + rootCmd.Flags().DurationVar(&sweepInterval, "sweep-interval", 1*time.Minute, "Interval between TTL sweeps") + + _ = rootCmd.MarkFlagRequired("node-url") + _ = rootCmd.MarkFlagRequired("node-token") + + if err := rootCmd.Execute(); err != nil { + os.Exit(1) + } +} + +func run(cmd *cobra.Command, _ []string) error { + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + store := catalog.New() + + // Start MCP HTTP server. + ln, err := net.Listen("tcp", bindAddr) + if err != nil { + return fmt.Errorf("listen %s: %w", bindAddr, err) + } + actualAddr := ln.Addr().String() + log.Printf("catalog MCP on %s", actualAddr) + + srv := &http.Server{ + Handler: newCatalogMCPHandler(store), + ReadHeaderTimeout: 5 * time.Second, + } + go func() { + if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed { + log.Printf("catalog HTTP server error: %v", err) + } + }() + + if ownURL == "" { + ownURL = "http://" + actualAddr + } + + nc := newNodeClient(nodeURL, nodeToken) + + // Subscribe before snapshot to avoid missing announces during bootstrap. + go func() { + if err := nc.tail(ctx, store); err != nil && ctx.Err() == nil { + log.Printf("tail exited: %v", err) + } + }() + + // Initial snapshot. + if err := nc.bootstrap(ctx, store, []api.ServiceType{ + api.ServiceType_SERVICE_TYPE_MCP, + api.ServiceType_SERVICE_TYPE_INFERENCE, + }); err != nil { + log.Printf("bootstrap warning: %v", err) + } + + // Register this catalog as a service so others can discover it. + if err := registerSelf(ctx, nodeURL, nodeToken, ownURL); err != nil { + log.Printf("registerSelf warning: %v", err) + } + + go runSweeper(ctx, store, sweepInterval) + go runRewalk(ctx, nc, store, rewalkInterval) + + <-ctx.Done() + + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + _ = srv.Shutdown(shutdownCtx) + return nil +} diff --git a/cmd/sam-catalog/server.go b/cmd/sam-catalog/server.go new file mode 100644 index 00000000..ac121c33 --- /dev/null +++ b/cmd/sam-catalog/server.go @@ -0,0 +1,121 @@ +// 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 main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + + "github.com/google/sam/api" + "github.com/google/sam/internal/catalog" + "github.com/modelcontextprotocol/go-sdk/mcp" + "google.golang.org/protobuf/encoding/protojson" +) + +// serviceTypeByShortName maps the short string used in query params to ServiceType. +var serviceTypeByShortName = map[string]api.ServiceType{ + "mcp": api.ServiceType_SERVICE_TYPE_MCP, + "inference": api.ServiceType_SERVICE_TYPE_INFERENCE, + "catalog": api.ServiceType_SERVICE_TYPE_CATALOG, +} + +// QueryCatalogParams defines parameters for the query_catalog tool. +type QueryCatalogParams struct { + Type string `json:"type,omitempty"` + Name string `json:"name,omitempty"` +} + +// handleQueryCatalog implements the query_catalog MCP tool. +func handleQueryCatalog(store *catalog.Store) func(context.Context, *mcp.CallToolRequest, QueryCatalogParams) (*mcp.CallToolResult, any, error) { + return func(_ context.Context, _ *mcp.CallToolRequest, params QueryCatalogParams) (*mcp.CallToolResult, any, error) { + typeFilter := api.ServiceType_SERVICE_TYPE_UNSPECIFIED + if params.Type != "" { + t, ok := serviceTypeByShortName[strings.ToLower(params.Type)] + if !ok { + return nil, nil, fmt.Errorf("unknown service type: %s", params.Type) + } + typeFilter = t + } + entries := store.List(typeFilter, params.Name) + data, err := json.Marshal(entries) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, + }, nil, nil + } +} + +// newCatalogMCPHandler returns an HTTP handler exposing the query_catalog MCP tool. +func newCatalogMCPHandler(store *catalog.Store) http.Handler { + server := mcp.NewServer(&mcp.Implementation{ + Name: "sam-catalog-mcp", + Version: "0.1.0", + }, nil) + + mcp.AddTool(server, &mcp.Tool{ + Name: "query_catalog", + Description: "Query the service catalog. Filter by type (mcp/inference/catalog) and/or name.", + }, handleQueryCatalog(store)) + + streamableHandler := mcp.NewStreamableHTTPHandler(func(_ *http.Request) *mcp.Server { + return server + }, nil) + + mux := http.NewServeMux() + mux.Handle("/mcp", streamableHandler) + return mux +} + +// registerSelf POSTs a SERVICE_TYPE_CATALOG registration to the sam-node sidecar. +func registerSelf(ctx context.Context, nodeBaseURL, token, ownMCPURL string) error { + reqProto := &api.RegisterServiceRequest{ + Service: &api.ServiceInfo{ + Type: api.ServiceType_SERVICE_TYPE_CATALOG, + Name: "catalog", + Description: "service catalog", + }, + Backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: ownMCPURL}, + } + body, err := protojson.Marshal(reqProto) + if err != nil { + return fmt.Errorf("marshal: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, nodeBaseURL+"/sam/service/register", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return fmt.Errorf("post: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("register failed: %s %s", resp.Status, respBody) + } + return nil +} diff --git a/cmd/sam-catalog/server_test.go b/cmd/sam-catalog/server_test.go new file mode 100644 index 00000000..73dc52f6 --- /dev/null +++ b/cmd/sam-catalog/server_test.go @@ -0,0 +1,86 @@ +// 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 main + +import ( + "context" + "encoding/json" + "net/http/httptest" + "testing" + "time" + + "github.com/google/sam/api" + "github.com/google/sam/internal/catalog" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +func TestQueryCatalogTool(t *testing.T) { + store := catalog.New() + // Pre-seed one entry. + store.Upsert(&api.ServiceAnnounce{ + Type: api.ServiceType_SERVICE_TYPE_MCP, + Name: "tool-a", + PeerId: "peer123", + TtlMs: 60000, + }, time.Now()) + + ts := httptest.NewServer(newCatalogMCPHandler(store)) + defer ts.Close() + + ctx := context.Background() + client := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "0.1.0"}, nil) + session, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: ts.URL + "/mcp"}, nil) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer func() { + if err := session.Close(); err != nil { + t.Logf("close session: %v", err) + } + }() + + result, err := session.CallTool(ctx, &mcp.CallToolParams{ + Name: "query_catalog", + Arguments: map[string]any{"type": "mcp"}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + + var text string + for _, c := range result.Content { + if tc, ok := c.(*mcp.TextContent); ok { + text = tc.Text + break + } + } + if text == "" { + t.Fatal("empty text content in response") + } + + var entries []catalog.Entry + if err := json.Unmarshal([]byte(text), &entries); err != nil { + t.Fatalf("unmarshal entries: %v", err) + } + if len(entries) != 1 { + t.Fatalf("expected 1 entry, got %d: %s", len(entries), text) + } + if entries[0].Name != "tool-a" { + t.Errorf("expected name=tool-a, got %s", entries[0].Name) + } + if entries[0].PeerID != "peer123" { + t.Errorf("expected peerID=peer123, got %s", entries[0].PeerID) + } +} diff --git a/cmd/sam-node/announce.go b/cmd/sam-node/announce.go new file mode 100644 index 00000000..e9ef8c7e --- /dev/null +++ b/cmd/sam-node/announce.go @@ -0,0 +1,80 @@ +// 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 main + +import ( + "context" + "time" + + "github.com/google/sam/api" + "github.com/google/sam/internal/announce" + pubsub "github.com/libp2p/go-libp2p-pubsub" + "google.golang.org/protobuf/proto" +) + +// joinTopic returns the cached topic, joining and caching it on first use. +func (n *SamNode) joinTopic(name string) (*pubsub.Topic, error) { + n.mu.Lock() + defer n.mu.Unlock() + if t, ok := n.topics[name]; ok { + return t, nil + } + t, err := n.PubSub.Join(name) + if err != nil { + return nil, err + } + n.topics[name] = t + return t, nil +} + +// announceServices publishes a signed ServiceAnnounce for every local service. +func (n *SamNode) announceServices(ctx context.Context) { + if n.PubSub == nil || n.Host == nil || n.services == nil { + return + } + priv := n.Host.Peerstore().PrivKey(n.Host.ID()) + if priv == nil { + logger.Warnf("[Announce] no private key for self; skipping announce") + return + } + + addrs := make([]string, 0, len(n.Host.Addrs())) + for _, a := range n.Host.Addrs() { + addrs = append(addrs, a.String()) + } + + topic, err := n.joinTopic(api.GossipServiceAnnounce) + if err != nil { + logger.Warnf("[Announce] join topic: %v", err) + return + } + + now := time.Now() + for _, info := range n.services.List(api.ServiceType_SERVICE_TYPE_UNSPECIFIED) { + ann := announce.Build(info, n.Host.ID(), addrs, now, announce.TTL) + if err := announce.Sign(priv, ann); err != nil { + logger.Warnf("[Announce] sign %s: %v", info.Name, err) + continue + } + data, err := proto.Marshal(ann) + if err != nil { + logger.Warnf("[Announce] marshal %s: %v", info.Name, err) + continue + } + if err := topic.Publish(ctx, data); err != nil { + logger.Warnf("[Announce] publish %s: %v", info.Name, err) + } + } +} diff --git a/cmd/sam-node/announce_test.go b/cmd/sam-node/announce_test.go new file mode 100644 index 00000000..547d9767 --- /dev/null +++ b/cmd/sam-node/announce_test.go @@ -0,0 +1,32 @@ +// 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 main + +import ( + "context" + "testing" +) + +func TestAnnounceServicesNoPubSubIsSafe(t *testing.T) { + // A zero-value node has no PubSub/Host; announceServices must not panic, + // just log and return. Guards the nil paths. + n := &SamNode{} + defer func() { + if r := recover(); r != nil { + t.Fatalf("announceServices panicked: %v", r) + } + }() + n.announceServices(context.Background()) +} diff --git a/cmd/sam-node/catalog_client.go b/cmd/sam-node/catalog_client.go new file mode 100644 index 00000000..3eb68f50 --- /dev/null +++ b/cmd/sam-node/catalog_client.go @@ -0,0 +1,154 @@ +// 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 main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/google/sam/api" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// catalogEntry mirrors internal/catalog.Entry's JSON (default Go field names). +type catalogEntry struct { + Type api.ServiceType + Name string + PeerID string +} + +// catalogEntriesToProviders parses the catalog's JSON result into providers. +func catalogEntriesToProviders(n *SamNode, raw, typeStr string) ([]*api.DiscoveredProvider, error) { + var entries []catalogEntry + if err := json.Unmarshal([]byte(raw), &entries); err != nil { + return nil, fmt.Errorf("catalog result unmarshal: %w", err) + } + out := []*api.DiscoveredProvider{} + for _, e := range entries { + // Skip self; guard nil Host so the mapper works in unit tests. + if n.Host != nil && e.PeerID == n.Host.ID().String() { + continue + } + pid, err := peer.Decode(e.PeerID) + if err != nil { + logger.Warnf("[Catalog] skipping entry with bad peer id %q: %v", e.PeerID, err) + continue + } + out = append(out, &api.DiscoveredProvider{ + PeerId: e.PeerID, + LocalProxyUrl: n.localProxyURL(pid, typeStr, e.Name), + SrvName: e.Name, + }) + } + return out, nil +} + +// findCatalogProvider returns a remembered catalog peer, discovering one if needed. +func (n *SamNode) findCatalogProvider(ctx context.Context, forceRefresh bool) (peer.ID, bool) { + n.catalogMu.Lock() + defer n.catalogMu.Unlock() + if n.catalogPeer != "" && !forceRefresh { + return n.catalogPeer, true + } + infos, err := n.FindProvidersByType(ctx, api.ServiceType_SERVICE_TYPE_CATALOG) + if err != nil { + logger.Warnf("[Catalog] provider lookup failed: %v", err) + return "", false + } + for _, p := range infos { + if p.ID == n.Host.ID() { + continue + } + n.catalogPeer = p.ID + return p.ID, true + } + n.catalogPeer = "" + return "", false +} + +// catalogEndpoint is a resolved catalog location: either a directly-reachable HTTP +// URL (a catalog this node hosts) or a mesh peer. Exactly one field is set. +type catalogEndpoint struct { + url string // non-empty -> reach over HTTP MCP + peer peer.ID // valid -> reach over libp2p +} + +// resolveCatalogEndpoints returns the catalogs to try, in priority order. On the first +// pass the hosted URL (if any) is prepended so it is reached directly over HTTP; on a +// forceRefresh pass only the mesh peer is re-resolved, avoiding a redundant hosted dial. +func (n *SamNode) resolveCatalogEndpoints(ctx context.Context, forceRefresh bool) []catalogEndpoint { + var eps []catalogEndpoint + if !forceRefresh { + if url, ok := n.services.hostedCatalogURL(); ok { + eps = append(eps, catalogEndpoint{url: url}) + } + } + if p, ok := n.findCatalogProvider(ctx, forceRefresh); ok { + eps = append(eps, catalogEndpoint{peer: p}) + } + return eps +} + +// callCatalog invokes query_catalog on the endpoint over whichever transport it implies. +func (n *SamNode) callCatalog(ctx context.Context, ep catalogEndpoint, args map[string]string) (*mcp.CallToolResult, error) { + if ep.url != "" { + client := mcp.NewClient(&mcp.Implementation{Name: "sam-node-catalog-client", Version: "0.1.0"}, nil) + session, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: strings.TrimRight(ep.url, "/") + "/mcp"}, nil) + if err != nil { + return nil, fmt.Errorf("connect %s: %w", ep.url, err) + } + defer func() { _ = session.Close() }() + return session.CallTool(ctx, &mcp.CallToolParams{Name: "query_catalog", Arguments: args}) + } + return n.callMCPToolOnce(ctx, ep.peer, "catalog.query_catalog", args) +} + +// queryCatalog asks a catalog to resolve the request over whichever transport its +// endpoint implies. It walks the resolved endpoints in priority order; a transport +// error moves on to the next, while a reachable catalog's empty/bad answer is +// authoritative. A second pass rediscovers the mesh peer; exhaustion falls back to DHT. +func (n *SamNode) queryCatalog(ctx context.Context, typeStr, serviceName string) ([]*api.DiscoveredProvider, bool) { + args := map[string]string{"type": typeStr} + if serviceName != "" { + args["name"] = serviceName + } + for attempt := 0; attempt < 2; attempt++ { + for _, ep := range n.resolveCatalogEndpoints(ctx, attempt > 0) { + res, err := n.callCatalog(ctx, ep, args) + if err != nil { + logger.Warnf("[Catalog] query failed: %v", err) + continue // transport failure: try the next endpoint + } + if res == nil || len(res.Content) == 0 { + return nil, false + } + text, ok := res.Content[0].(*mcp.TextContent) + if !ok { + return nil, false + } + providers, err := catalogEntriesToProviders(n, text.Text, typeStr) + if err != nil || len(providers) == 0 { + return nil, false + } + logger.Infof("[Catalog] resolved %d providers via catalog", len(providers)) + return providers, true + } + } + logger.Debugf("[Catalog] no usable catalog; falling back to DHT discovery") + return nil, false +} diff --git a/cmd/sam-node/catalog_client_test.go b/cmd/sam-node/catalog_client_test.go new file mode 100644 index 00000000..dd9a62ae --- /dev/null +++ b/cmd/sam-node/catalog_client_test.go @@ -0,0 +1,124 @@ +// 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 main + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +func TestCatalogEntriesToProviders(t *testing.T) { + priv, pub, _ := crypto.GenerateKeyPair(crypto.Ed25519, -1) + _ = priv + pid, _ := peer.IDFromPublicKey(pub) + idStr := pid.String() + + n := &SamNode{BoundHTTPAddr: "127.0.0.1:9999"} + raw := fmt.Sprintf(`[{"Type":1,"Name":"github-tools","PeerID":%q,"Addrs":["/ip4/1.2.3.4/tcp/1"],"Expiry":"2030-01-01T00:00:00Z"}]`, idStr) + got, err := catalogEntriesToProviders(n, raw, "mcp") + if err != nil { + t.Fatalf("parse: %v", err) + } + if len(got) != 1 { + t.Fatalf("want 1 provider, got %d", len(got)) + } + if got[0].PeerId != idStr { + t.Fatalf("PeerId: want %q, got %q", idStr, got[0].PeerId) + } + if !strings.Contains(got[0].LocalProxyUrl, idStr) { + t.Fatalf("LocalProxyUrl %q does not contain peer id %q", got[0].LocalProxyUrl, idStr) + } + if got[0].SrvName != "github-tools" { + t.Fatalf("SrvName: want %q, got %q", "github-tools", got[0].SrvName) + } +} + +func TestCatalogEntriesToProvidersBadPeerSkipped(t *testing.T) { + n := &SamNode{BoundHTTPAddr: "127.0.0.1:9999"} + raw := `[{"Type":1,"Name":"bad-svc","PeerID":"not-a-peer-id"}]` + got, err := catalogEntriesToProviders(n, raw, "mcp") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 0 { + t.Fatalf("want 0 providers for bad peer id, got %d", len(got)) + } +} + +func TestCatalogEntriesToProvidersEmpty(t *testing.T) { + n := &SamNode{BoundHTTPAddr: "127.0.0.1:9999"} + got, err := catalogEntriesToProviders(n, `[]`, "mcp") + if err != nil { + t.Fatalf("parse: %v", err) + } + if len(got) != 0 { + t.Fatalf("want 0, got %d", len(got)) + } +} + +func TestCallCatalog_HTTPEndpoint(t *testing.T) { + _, pub, _ := crypto.GenerateKeyPair(crypto.Ed25519, -1) + pid, _ := peer.IDFromPublicKey(pub) + idStr := pid.String() + + cannedJSON := fmt.Sprintf(`[{"Type":1,"Name":"github-tools","PeerID":%q}]`, idStr) + + server := mcp.NewServer(&mcp.Implementation{Name: "test-catalog", Version: "0.1.0"}, nil) + mcp.AddTool(server, &mcp.Tool{Name: "query_catalog"}, func(_ context.Context, _ *mcp.CallToolRequest, _ map[string]any) (*mcp.CallToolResult, any, error) { + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: cannedJSON}}}, nil, nil + }) + streamableHandler := mcp.NewStreamableHTTPHandler(func(_ *http.Request) *mcp.Server { + return server + }, nil) + + mux := http.NewServeMux() + mux.Handle("/mcp", streamableHandler) + ts := httptest.NewServer(mux) + defer ts.Close() + + n := &SamNode{BoundHTTPAddr: "127.0.0.1:9999"} + res, err := n.callCatalog(context.Background(), catalogEndpoint{url: ts.URL}, map[string]string{"type": "mcp", "name": "github-tools"}) + if err != nil { + t.Fatalf("callCatalog: %v", err) + } + text, ok := res.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("content type: got %T, want *mcp.TextContent", res.Content[0]) + } + providers, err := catalogEntriesToProviders(n, text.Text, "mcp") + if err != nil { + t.Fatalf("map providers: %v", err) + } + if len(providers) != 1 { + t.Fatalf("want 1 provider, got %d", len(providers)) + } + if providers[0].PeerId != idStr { + t.Errorf("PeerId = %q, want %q", providers[0].PeerId, idStr) + } + if !strings.Contains(providers[0].LocalProxyUrl, idStr) { + t.Errorf("LocalProxyUrl %q does not contain %q", providers[0].LocalProxyUrl, idStr) + } + if providers[0].SrvName != "github-tools" { + t.Errorf("SrvName = %q, want github-tools", providers[0].SrvName) + } +} diff --git a/cmd/sam-node/node.go b/cmd/sam-node/node.go index 7173e42a..a4668f61 100644 --- a/cmd/sam-node/node.go +++ b/cmd/sam-node/node.go @@ -120,6 +120,7 @@ type SamNode struct { peerLastEventTime map[string]int64 receivedMsgs map[string][]string topics map[string]*pubsub.Topic + subscribedTopics map[string]bool mu sync.Mutex LocalPolicy *NodeConfigComplete revokedPeers *lru.Cache[string, int64] @@ -137,6 +138,8 @@ type SamNode struct { currentRelays []peer.AddrInfo reprovideTrigger chan struct{} BiscuitTimeout time.Duration + catalogPeer peer.ID + catalogMu sync.Mutex } // UpdateRelays updates the current relays used by AutoRelay. @@ -221,6 +224,7 @@ func NewSamNode(ctx context.Context, cfg SamNodeConfig) (*SamNode, error) { peerLastEventTime: make(map[string]int64), receivedMsgs: make(map[string][]string), topics: make(map[string]*pubsub.Topic), + subscribedTopics: map[string]bool{}, authenticatedHubs: make(map[peer.ID]bool), LocalPolicy: cfg.NodeConfig, AllowLoopback: cfg.AllowLoopback, @@ -504,6 +508,7 @@ func (n *SamNode) startReprovideLoop(ctx context.Context, interval time.Duration return case <-time.After(5 * time.Second): n.services.ReprovideAll(ctx) + n.announceServices(ctx) } for { @@ -512,8 +517,10 @@ func (n *SamNode) startReprovideLoop(ctx context.Context, interval time.Duration return case <-ticker.C: n.services.ReprovideAll(ctx) + n.announceServices(ctx) case <-n.reprovideTrigger: n.services.ReprovideAll(ctx) + n.announceServices(ctx) } } }() @@ -1013,23 +1020,27 @@ func (n *SamNode) verifyEvent(event *api.MeshEvent) bool { func (n *SamNode) subscribeToTopic(ctx context.Context, topicName string) error { n.mu.Lock() - defer n.mu.Unlock() - - if _, ok := n.topics[topicName]; ok { + if n.subscribedTopics[topicName] { + n.mu.Unlock() return nil } - - topic, err := n.PubSub.Join(topicName) - if err != nil { - return err + topic, ok := n.topics[topicName] + if !ok { + t, err := n.PubSub.Join(topicName) + if err != nil { + n.mu.Unlock() + return err + } + n.topics[topicName] = t + topic = t } - sub, err := topic.Subscribe() if err != nil { + n.mu.Unlock() return err } - - n.topics[topicName] = topic + n.subscribedTopics[topicName] = true + n.mu.Unlock() logger.Infof("[PubSub] Started subscription background loop for topic: %s", topicName) go func() { @@ -1323,6 +1334,9 @@ func (n *SamNode) DiscoverRemoteServices(ctx context.Context, serviceType api.Se if err != nil { return nil, err } + if providers, ok := n.queryCatalog(ctx, typeStr, serviceName); ok { + return providers, nil + } if serviceName == "" { return n.discoverServicesByType(ctx, serviceType, typeStr) } @@ -1342,6 +1356,17 @@ func (n *SamNode) DiscoverRemoteServicesStream(ctx context.Context, serviceType go func() { defer close(out) + if providers, ok := n.queryCatalog(ctx, typeStr, serviceName); ok { + for _, dp := range providers { + select { + case <-ctx.Done(): + return + case out <- dp: + } + } + return + } + if serviceName != "" { peers, err := n.FindProvidersByName(ctx, serviceType, serviceName) if err != nil { diff --git a/cmd/sam-node/service.go b/cmd/sam-node/service.go index c1e996cf..d47ff4ae 100644 --- a/cmd/sam-node/service.go +++ b/cmd/sam-node/service.go @@ -36,6 +36,8 @@ type Service interface { Init(ctx context.Context) error Handler() http.Handler Teardown() error + // TargetURL returns the URL backend if this service is URL-backed. + TargetURL() (string, bool) } // baseService is the shared embeddable. Holds the fields and default @@ -73,6 +75,14 @@ func newReverseProxyHandler(targetURL string) (http.Handler, error) { func (b *baseService) Info() *api.ServiceInfo { return b.info } func (b *baseService) Handler() http.Handler { return b.handler } +// TargetURL returns the URL backend if this service is URL-backed. +func (b *baseService) TargetURL() (string, bool) { + if t, ok := b.backend.(*api.RegisterServiceRequest_TargetUrl); ok { + return t.TargetUrl, true + } + return "", false +} + // Init builds the ingress handler for the backend. URL -> reverse-proxy, // Command -> StdioBridge. MCPService extends this; it does not replace it. func (b *baseService) Init(ctx context.Context) error { @@ -117,6 +127,8 @@ func NewServiceFromRequest(req *api.RegisterServiceRequest) (Service, error) { return &MCPService{baseService: baseService{info: info, backend: req.Backend}}, nil case api.ServiceType_SERVICE_TYPE_INFERENCE: return &InferenceService{baseService: baseService{info: info, backend: req.Backend}}, nil + case api.ServiceType_SERVICE_TYPE_CATALOG: + return &MCPService{baseService: baseService{info: info, backend: req.Backend}}, nil default: return nil, fmt.Errorf("unspecified or unsupported service type: %v", info.Type) } @@ -152,7 +164,6 @@ func buildRegisterRequest(sCfg api.ServiceConfig) (*api.RegisterServiceRequest, return req, nil } - // serviceKeyToCID hashes "sam:service[:part]..." into a DHT rendezvous CID. func serviceKeyToCID(parts ...string) (cid.Cid, error) { srvKey := strings.Join(append([]string{"sam:service"}, parts...), ":") diff --git a/cmd/sam-node/service_catalog_host_test.go b/cmd/sam-node/service_catalog_host_test.go new file mode 100644 index 00000000..c14d61f5 --- /dev/null +++ b/cmd/sam-node/service_catalog_host_test.go @@ -0,0 +1,35 @@ +// 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 main + +import ( + "testing" + + "github.com/google/sam/api" +) + +func TestNewServiceFromRequestCatalog(t *testing.T) { + req := &api.RegisterServiceRequest{ + Service: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_CATALOG, Name: "catalog"}, + Backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: "http://127.0.0.1:9"}, + } + svc, err := NewServiceFromRequest(req) + if err != nil { + t.Fatalf("CATALOG should be hostable: %v", err) + } + if _, ok := svc.(*MCPService); !ok { + t.Fatalf("CATALOG should be served via MCPService, got %T", svc) + } +} diff --git a/cmd/sam-node/service_catalog_type_test.go b/cmd/sam-node/service_catalog_type_test.go new file mode 100644 index 00000000..4f03a238 --- /dev/null +++ b/cmd/sam-node/service_catalog_type_test.go @@ -0,0 +1,54 @@ +// 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 main + +import ( + "testing" + + "github.com/google/sam/api" +) + +func TestParseAndStringifyCatalogType(t *testing.T) { + got, err := api.ParseServiceType("catalog") + if err != nil { + t.Fatalf("ParseServiceType(catalog): %v", err) + } + if got != api.ServiceType_SERVICE_TYPE_CATALOG { + t.Fatalf("got %v, want SERVICE_TYPE_CATALOG", got) + } + s, err := api.ServiceTypeToString(api.ServiceType_SERVICE_TYPE_CATALOG) + if err != nil { + t.Fatalf("ServiceTypeToString(CATALOG): %v", err) + } + if s != "catalog" { + t.Fatalf("got %q, want \"catalog\"", s) + } +} + +func TestServiceAnnounceTypeExists(t *testing.T) { + // Compile-time check that the generated struct + fields exist. + a := &api.ServiceAnnounce{ + Type: api.ServiceType_SERVICE_TYPE_MCP, + Name: "x", + PeerId: "p", + Addrs: []string{"/ip4/127.0.0.1/tcp/1"}, + Timestamp: 1, + TtlMs: 2, + Signature: []byte{0x1}, + } + if a.Name != "x" { + t.Fatal("unexpected") + } +} diff --git a/cmd/sam-node/service_registry.go b/cmd/sam-node/service_registry.go index f7685579..f1d6fd5f 100644 --- a/cmd/sam-node/service_registry.go +++ b/cmd/sam-node/service_registry.go @@ -135,6 +135,18 @@ func (r *ServiceRegistry) insertService(svc Service) { r.services[svc.Info().Name] = svc } +// hostedCatalogURL returns the target URL of a CATALOG service this node hosts, if any. +func (r *ServiceRegistry) hostedCatalogURL() (string, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + for _, svc := range r.services { + if svc.Info().Type == api.ServiceType_SERVICE_TYPE_CATALOG { + return svc.TargetURL() + } + } + return "", false +} + // TeardownAll calls Teardown on every registered service and clears the // map. Per-service errors are logged; iteration continues. func (r *ServiceRegistry) TeardownAll() { diff --git a/cmd/sam-node/service_registry_test.go b/cmd/sam-node/service_registry_test.go index 13ffbaca..44ecfd1e 100644 --- a/cmd/sam-node/service_registry_test.go +++ b/cmd/sam-node/service_registry_test.go @@ -52,7 +52,8 @@ func (f *fakeService) Init(ctx context.Context) error { f.initCalls++ return f.initErr } -func (f *fakeService) Handler() http.Handler { return f.handler } +func (f *fakeService) Handler() http.Handler { return f.handler } +func (f *fakeService) TargetURL() (string, bool) { return "", false } func (f *fakeService) Teardown() error { f.teardownCalls++ return nil @@ -145,6 +146,31 @@ func TestServiceRegistry_ListFiltersByType(t *testing.T) { } } +func TestHostedCatalogURL_WithCatalogService(t *testing.T) { + r := newServiceRegistryForTest(&fakeDHT{}) + svc := &MCPService{baseService: baseService{ + info: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_CATALOG, Name: "catalog"}, + backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: "http://127.0.0.1:9100"}, + }} + r.insertService(svc) + url, ok := r.hostedCatalogURL() + if !ok { + t.Fatal("hostedCatalogURL: want true when CATALOG service present") + } + if url != "http://127.0.0.1:9100" { + t.Errorf("hostedCatalogURL = %q, want http://127.0.0.1:9100", url) + } +} + +func TestHostedCatalogURL_NoCatalogService(t *testing.T) { + r := newServiceRegistryForTest(&fakeDHT{}) + r.insertService(newFakeSvc("mcp-svc", api.ServiceType_SERVICE_TYPE_MCP)) + _, ok := r.hostedCatalogURL() + if ok { + t.Fatal("hostedCatalogURL: want false when no CATALOG service registered") + } +} + func TestServiceRegistry_TeardownAllContinuesOnError(t *testing.T) { r := newServiceRegistryForTest(&fakeDHT{}) a := newFakeSvc("a", api.ServiceType_SERVICE_TYPE_MCP) diff --git a/cmd/sam-node/service_test.go b/cmd/sam-node/service_test.go index 6a26c1be..81637ab8 100644 --- a/cmd/sam-node/service_test.go +++ b/cmd/sam-node/service_test.go @@ -36,6 +36,7 @@ func (t *testService) Info() *api.ServiceInfo { return t.info } func (t *testService) Init(_ context.Context) error { return nil } func (t *testService) Handler() http.Handler { return t.handler } func (t *testService) Teardown() error { return nil } +func (t *testService) TargetURL() (string, bool) { return "", false } func TestBaseService_InitURLBackend_BuildsReverseProxy(t *testing.T) { upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -191,3 +192,30 @@ func TestBuildRegisterRequest_InvalidType(t *testing.T) { t.Fatal("expected error for invalid type, got nil") } } + +func TestBaseServiceTargetURL_URLBacked(t *testing.T) { + b := &baseService{ + info: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_MCP, Name: "svc"}, + backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: "http://localhost:8080"}, + } + got, ok := b.TargetURL() + if !ok { + t.Fatal("TargetURL: want true for URL-backed service") + } + if got != "http://localhost:8080" { + t.Errorf("TargetURL = %q, want http://localhost:8080", got) + } +} + +func TestBaseServiceTargetURL_CommandBacked(t *testing.T) { + b := &baseService{ + info: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_MCP, Name: "svc"}, + backend: &api.RegisterServiceRequest_Command{ + Command: &api.CommandBackend{Command: []string{"/bin/cat"}}, + }, + } + got, ok := b.TargetURL() + if ok { + t.Fatalf("TargetURL: want false for command-backed service, got url=%q", got) + } +} diff --git a/cmd/sam-node/sidecar.go b/cmd/sam-node/sidecar.go index af7a70ec..bf54d771 100644 --- a/cmd/sam-node/sidecar.go +++ b/cmd/sam-node/sidecar.go @@ -29,10 +29,12 @@ import ( "time" "github.com/google/sam/api" + "github.com/google/sam/internal/announce" libp2phttp "github.com/libp2p/go-libp2p-http" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" ) func startSidecarServer(node *SamNode, addr, token, certFile, keyFile, caFile string) error { @@ -52,6 +54,9 @@ func startSidecarServer(node *SamNode, addr, token, certFile, keyFile, caFile st mux.Handle("/sam/service/discover", withAuth(token, withMeshConnection(node, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { handleDiscoverService(node, w, r) })))) + mux.Handle("/sam/service/announce/stream", withMeshConnection(node, withAuth(token, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handleAnnounceStream(node, w, r) + })))) // Mount Egress Proxy mux.Handle("/sam/", withAuth(token, withMeshConnection(node, createEgressProxy(node)))) @@ -388,6 +393,57 @@ func handleDiscoverService(node *SamNode, w http.ResponseWriter, r *http.Request } } +func handleAnnounceStream(node *SamNode, w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + topic, err := node.joinTopic(api.GossipServiceAnnounce) + if err != nil { + http.Error(w, fmt.Sprintf("join topic: %v", err), http.StatusInternalServerError) + return + } + sub, err := topic.Subscribe() + if err != nil { + http.Error(w, fmt.Sprintf("subscribe: %v", err), http.StatusInternalServerError) + return + } + defer sub.Cancel() + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Streaming unsupported", http.StatusInternalServerError) + return + } + + ctx := r.Context() + for { + msg, err := sub.Next(ctx) + if err != nil { + return // ctx cancelled or sub closed + } + var ann api.ServiceAnnounce + if err := proto.Unmarshal(msg.Data, &ann); err != nil { + continue + } + ok, err := announce.Verify(&ann) + if err != nil || !ok { + continue // drop forged/invalid + } + data, err := protojson.Marshal(&ann) + if err != nil { + continue + } + if _, err := fmt.Fprintf(w, "data: %s\n\n", data); err != nil { + return + } + flusher.Flush() + } +} + func createEgressProxy(node *SamNode) http.Handler { transport := libp2phttp.NewTransport(node.Host) diff --git a/internal/announce/announce.go b/internal/announce/announce.go new file mode 100644 index 00000000..8cc95bbb --- /dev/null +++ b/internal/announce/announce.go @@ -0,0 +1,74 @@ +// 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 announce + +import ( + "time" + + "github.com/google/sam/api" + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" + "google.golang.org/protobuf/proto" +) + +// TTL is the per-entry lifetime (~3x the 5-min reprovide tick). +const TTL = 15 * time.Minute + +// Build assembles an unsigned announce for a local service. +func Build(info *api.ServiceInfo, peerID peer.ID, addrs []string, now time.Time, ttl time.Duration) *api.ServiceAnnounce { + return &api.ServiceAnnounce{ + Type: info.Type, + Name: info.Name, + PeerId: peerID.String(), + Addrs: addrs, + Timestamp: now.UnixMilli(), + TtlMs: ttl.Milliseconds(), + } +} + +// Sign signs the announce over its signature-cleared marshalling. +func Sign(priv crypto.PrivKey, a *api.ServiceAnnounce) error { + a.Signature = nil + data, err := proto.Marshal(a) + if err != nil { + return err + } + sig, err := priv.Sign(data) + if err != nil { + return err + } + a.Signature = sig + return nil +} + +// Verify checks the signature against the key derived from PeerId. +func Verify(a *api.ServiceAnnounce) (bool, error) { + pid, err := peer.Decode(a.PeerId) + if err != nil { + return false, err + } + pub, err := pid.ExtractPublicKey() + if err != nil { + return false, err + } + sig := a.Signature + a.Signature = nil + data, err := proto.Marshal(a) + a.Signature = sig // restore + if err != nil { + return false, err + } + return pub.Verify(data, sig) +} diff --git a/internal/announce/announce_test.go b/internal/announce/announce_test.go new file mode 100644 index 00000000..ec1dedda --- /dev/null +++ b/internal/announce/announce_test.go @@ -0,0 +1,84 @@ +// 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 announce + +import ( + "testing" + "time" + + "github.com/google/sam/api" + "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/peer" +) + +func newTestIdentity(t *testing.T) (crypto.PrivKey, peer.ID) { + t.Helper() + priv, pub, err := crypto.GenerateKeyPair(crypto.Ed25519, -1) + if err != nil { + t.Fatalf("GenerateKeyPair: %v", err) + } + pid, err := peer.IDFromPublicKey(pub) + if err != nil { + t.Fatalf("IDFromPublicKey: %v", err) + } + return priv, pid +} + +func TestServiceAnnounceSignVerifyRoundTrip(t *testing.T) { + priv, pid := newTestIdentity(t) + info := &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_MCP, Name: "github-tools"} + a := Build(info, pid, []string{"/ip4/127.0.0.1/tcp/4001"}, time.UnixMilli(1_700_000_000_000), TTL) + if a.PeerId != pid.String() { + t.Fatalf("peer id not set: %q", a.PeerId) + } + if err := Sign(priv, a); err != nil { + t.Fatalf("sign: %v", err) + } + ok, err := Verify(a) + if err != nil { + t.Fatalf("verify: %v", err) + } + if !ok { + t.Fatal("expected signature to verify") + } +} + +func TestServiceAnnounceTamperFails(t *testing.T) { + priv, pid := newTestIdentity(t) + info := &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_MCP, Name: "a"} + a := Build(info, pid, nil, time.UnixMilli(1), time.Minute) + if err := Sign(priv, a); err != nil { + t.Fatalf("sign: %v", err) + } + a.Name = "b" // tamper after signing + ok, _ := Verify(a) + if ok { + t.Fatal("expected verification to fail after tamper") + } +} + +func TestServiceAnnounceWrongPeerFails(t *testing.T) { + priv, _ := newTestIdentity(t) + _, otherPID := newTestIdentity(t) + info := &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_MCP, Name: "a"} + a := Build(info, otherPID, nil, time.UnixMilli(1), time.Minute) // claims other peer + if err := Sign(priv, a); err != nil { + t.Fatalf("sign: %v", err) + } + ok, _ := Verify(a) // verifies against otherPID's key, not priv + if ok { + t.Fatal("expected verification to fail for mismatched peer id") + } +} diff --git a/internal/catalog/store.go b/internal/catalog/store.go new file mode 100644 index 00000000..8c49e815 --- /dev/null +++ b/internal/catalog/store.go @@ -0,0 +1,82 @@ +// 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 catalog + +import ( + "sync" + "time" + + "github.com/google/sam/api" +) + +type Entry struct { + Type api.ServiceType + Name string + PeerID string + Addrs []string + Expiry time.Time +} + +type Store struct { + mu sync.RWMutex + entries map[string]Entry +} + +func New() *Store { return &Store{entries: map[string]Entry{}} } + +func key(t api.ServiceType, name, peerID string) string { + return t.String() + "|" + name + "|" + peerID +} + +// Upsert inserts or refreshes an entry, setting Expiry from the announce TTL. +func (s *Store) Upsert(a *api.ServiceAnnounce, now time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + s.entries[key(a.Type, a.Name, a.PeerId)] = Entry{ + Type: a.Type, Name: a.Name, PeerID: a.PeerId, Addrs: a.Addrs, + Expiry: now.Add(time.Duration(a.TtlMs) * time.Millisecond), + } +} + +// List returns entries matching the optional type and name filters. +func (s *Store) List(typeFilter api.ServiceType, name string) []Entry { + s.mu.RLock() + defer s.mu.RUnlock() + out := []Entry{} + for _, e := range s.entries { + if typeFilter != api.ServiceType_SERVICE_TYPE_UNSPECIFIED && e.Type != typeFilter { + continue + } + if name != "" && e.Name != name { + continue + } + out = append(out, e) + } + return out +} + +// Sweep removes expired entries and returns the count removed. +func (s *Store) Sweep(now time.Time) int { + s.mu.Lock() + defer s.mu.Unlock() + n := 0 + for k, e := range s.entries { + if now.After(e.Expiry) { + delete(s.entries, k) + n++ + } + } + return n +} diff --git a/internal/catalog/store_test.go b/internal/catalog/store_test.go new file mode 100644 index 00000000..981ee388 --- /dev/null +++ b/internal/catalog/store_test.go @@ -0,0 +1,68 @@ +// 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 catalog + +import ( + "testing" + "time" + + "github.com/google/sam/api" +) + +func ann(name string, ttlMs int64, ts int64) *api.ServiceAnnounce { + return &api.ServiceAnnounce{ + Type: api.ServiceType_SERVICE_TYPE_MCP, Name: name, PeerId: "p-" + name, + Addrs: []string{"/ip4/127.0.0.1/tcp/1"}, Timestamp: ts, TtlMs: ttlMs, + } +} + +func TestUpsertAndList(t *testing.T) { + s := New() + now := time.UnixMilli(1000) + s.Upsert(ann("a", 60000, 1000), now) + s.Upsert(ann("b", 60000, 1000), now) + if got := s.List(api.ServiceType_SERVICE_TYPE_MCP, ""); len(got) != 2 { + t.Fatalf("want 2, got %d", len(got)) + } + if got := s.List(api.ServiceType_SERVICE_TYPE_MCP, "a"); len(got) != 1 || got[0].Name != "a" { + t.Fatalf("name filter failed: %+v", got) + } + if got := s.List(api.ServiceType_SERVICE_TYPE_INFERENCE, ""); len(got) != 0 { + t.Fatalf("type filter failed: %+v", got) + } +} + +func TestUpsertRefreshesExpiry(t *testing.T) { + s := New() + s.Upsert(ann("a", 60000, 1000), time.UnixMilli(1000)) // expiry 61000 + s.Upsert(ann("a", 60000, 1000), time.UnixMilli(5000)) // refresh -> 65000 + if n := s.Sweep(time.UnixMilli(62000)); n != 0 { + t.Fatalf("entry should have been refreshed, swept %d", n) + } + if got := s.List(api.ServiceType_SERVICE_TYPE_MCP, ""); len(got) != 1 { + t.Fatalf("want 1 after refresh, got %d", len(got)) + } +} + +func TestSweepEvictsExpired(t *testing.T) { + s := New() + s.Upsert(ann("a", 1000, 1000), time.UnixMilli(1000)) // expiry 2000 + if n := s.Sweep(time.UnixMilli(3000)); n != 1 { + t.Fatalf("want 1 swept, got %d", n) + } + if got := s.List(api.ServiceType_SERVICE_TYPE_UNSPECIFIED, ""); len(got) != 0 { + t.Fatalf("want empty after sweep, got %d", len(got)) + } +} diff --git a/tests/e2e/catalog.bats b/tests/e2e/catalog.bats new file mode 100644 index 00000000..87dca4dd --- /dev/null +++ b/tests/e2e/catalog.bats @@ -0,0 +1,192 @@ +#!/usr/bin/env bats +# +# E2E for the service catalog. Topology (self-hosted catalog): +# node-2 hosts the "calculator" MCP service (calc-mcp backend) +# node-1 is the consumer AND hosts the catalog (sam-catalog registers to it) +# sam-catalog connects to node-1, ingests via discover + announce stream +# +# Three cases share provision_catalog_mesh: (1) services are retrieved via the +# catalog, (2) a service added on the fly is picked up, (3) discovery falls back +# to the DHT when the catalog is gone. Requires docker (run with +# `make test-e2e` / `make e2e-test WHAT=catalog`). + +load "lib/container_mesh.bash" + +CALC_MCP_IMAGE="sam-calc-mcp:local" + +build_calc_mcp_image() { + if ! docker image inspect "${CALC_MCP_IMAGE}" >/dev/null 2>&1; then + docker build -t "${CALC_MCP_IMAGE}" \ + -f tests/e2e/docker/calc-mcp/Dockerfile \ + tests/e2e/docker/calc-mcp >/dev/null + fi +} + +start_calc_mcp() { + local name="${MESH_PREFIX}-calc-mcp" + docker run -d \ + --name "${name}" \ + --network "${MESH_NETWORK}" \ + --network-alias calc-mcp \ + "${CALC_MCP_IMAGE}" >/dev/null + MESH_CONTAINERS+=("${name}") + mesh_wait_for_log "${name}" "Uvicorn running on" 20 +} + +# start_catalog runs sam-catalog pointed at node-1, reachable at sam-catalog:9090. +start_catalog() { + local name="${MESH_PREFIX}-sam-catalog" + docker run -d \ + --name "${name}" \ + --network "${MESH_NETWORK}" \ + --network-alias sam-catalog \ + "${MESH_RUNTIME_IMAGE}" \ + /usr/local/bin/sam-catalog \ + --node-url "http://sam-node-1:8080" \ + --node-token "secret-token" \ + --bind-addr "0.0.0.0:9090" \ + --own-url "http://sam-catalog:9090" \ + --rewalk-interval 5s \ + --sweep-interval 5s >/dev/null + MESH_CONTAINERS+=("${name}") + mesh_wait_for_log "${name}" "catalog MCP on" 20 +} + +# register_service registers an MCP service on a node via the +# sidecar REST endpoint (Zero Trust: bearer token required). +register_service() { + local node_alias="$1" + local svc_name="$2" + docker run --rm --network "${MESH_NETWORK}" python:3.12 python3 -c " +import json, urllib.request +data = {'service': {'type': 'SERVICE_TYPE_MCP', 'name': '${svc_name}', 'description': 'on-the-fly ${svc_name}'}, 'targetUrl': 'http://calc-mcp:7777/mcp'} +req = urllib.request.Request('http://${node_alias}:8080/sam/service/register', data=json.dumps(data).encode(), headers={'Authorization': 'Bearer secret-token', 'Content-Type': 'application/json'}) +print(urllib.request.urlopen(req).read().decode()) +" +} + +# poll_catalog_has : query the catalog's own MCP until is ingested. +poll_catalog_has() { + local name="$1" + local timeout_secs="${2:-60}" + local i out cnt + for ((i=0; i/dev/null | tail -n 1) + # catalog returns []catalog.Entry (Go default JSON field names: .Name). + cnt=$(echo "$out" | jq --arg n "$name" '[.[] | select(.Name==$n)] | length' 2>/dev/null || echo 0) + if [[ "${cnt:-0}" -ge 1 ]]; then return 0; fi + sleep 1 + done + return 1 +} + +# poll_discover_has : call discover_remote_services +# on a node until it returns . Echoes the last discovery JSON on success. +poll_discover_has() { + local node_idx="$1" + local name="$2" + local timeout_secs="${3:-60}" + local i out cnt + for ((i=0; i/dev/null | tail -n 1) + # discover returns []DiscoveredProvider (snake_case JSON: .srv_name). + cnt=$(echo "$out" | jq --arg n "$name" '[.[] | select(.srv_name==$n)] | length' 2>/dev/null || echo 0) + if [[ "${cnt:-0}" -ge 1 ]]; then echo "$out"; return 0; fi + sleep 1 + done + return 1 +} + +# provision_catalog_mesh builds the shared topology used by every case: hub + +# node-2 hosting "calculator" + consumer node-1 hosting the catalog, connected, +# with the catalog confirmed to have ingested "calculator". Exports NODE2_PEER_ID. +provision_catalog_mesh() { + run mesh_start_mock_oidc + [[ "$status" -eq 0 ]] + mesh_start_hub + + # --- Provider node-2 hosting "calculator" --- + echo "[$(date +%T)] Starting calc-mcp backend" + start_calc_mcp + + echo "[$(date +%T)] Starting Node 2 with calculator service" + run mesh_start_node 2 "--log-level debug" "tests/e2e/docker/calc-mcp/sam-node-config.yaml" + [[ "$status" -eq 0 ]] + mesh_wait_for_log "${MESH_PREFIX}-node-2" "SAM Node Online" 60 + mesh_wait_for_mcp_ready 2 20 + + NODE2_PEER_ID=$(docker logs "${MESH_PREFIX}-node-2" 2>&1 | grep "PeerID:" | head -n 1 | awk '{print $2}' | tr -d '\r') + + # --- Consumer node-1 (will also host the catalog) --- + echo "[$(date +%T)] Starting Node 1" + run mesh_start_node 1 "--log-level debug" + [[ "$status" -eq 0 ]] + mesh_wait_for_log "${MESH_PREFIX}-node-1" "SAM Node Online" 60 + mesh_wait_for_mcp_ready 1 20 + + echo "[$(date +%T)] Connecting Node 1 to Node 2" + local node2_addr="/dns4/sam-node-2/tcp/5002/p2p/${NODE2_PEER_ID}" + run docker run --rm --network "${MESH_NETWORK}" "${MESH_RUNTIME_IMAGE}" mcp-client \ + -url "http://sam-node-1:8080/mcp" -tool "connect_peer" -args "{\"peer_addr\":\"${node2_addr}\"}" + [[ "$status" -eq 0 ]] + mesh_wait_for_peer_connection 1 "${NODE2_PEER_ID}" 20 + + # --- Start the catalog (registers itself on node-1) and wait until it ingested calculator --- + echo "[$(date +%T)] Starting sam-catalog" + start_catalog + + run poll_catalog_has "calculator" 60 + [[ "$status" -eq 0 ]] # catalog ingested calculator (bootstrap + announce) +} + +setup() { + export BATS_TEST_TIMEOUT=300 + mesh_setup_env + build_calc_mcp_image +} + +teardown() { + mesh_cleanup_env +} + +@test "catalog: discovery is served via the catalog" { + provision_catalog_mesh + + run poll_discover_has 1 "calculator" 60 + echo "discover output: $output" + [[ "$status" -eq 0 ]] # node-1 discovery returns calculator + echo "$output" | jq -e --arg pid "${NODE2_PEER_ID}" '[.[] | select(.srv_name=="calculator" and .peer_id==$pid)] | length >= 1' >/dev/null || return 1 + + # Prove it came VIA the catalog (not the DHT fan-out). Explicit `|| return 1` + # so this is a hard assertion regardless of the bats set-e behavior. + docker logs "${MESH_PREFIX}-node-1" 2>&1 | grep -q "\[Catalog\] resolved" || return 1 +} + +@test "catalog: a service added on the fly is picked up" { + provision_catalog_mesh + + echo "[$(date +%T)] Registering 'weather' on node-2" + run register_service "sam-node-2" "weather" + [[ "$status" -eq 0 ]] + + run poll_catalog_has "weather" 60 + [[ "$status" -eq 0 ]] # catalog ingested the new service dynamically (announce/re-walk) + + run poll_discover_has 1 "weather" 60 + echo "discover output: $output" + [[ "$status" -eq 0 ]] # node-1 now resolves the on-the-fly service via the catalog +} + +@test "catalog: discovery falls back to the DHT when the catalog is gone" { + provision_catalog_mesh + + echo "[$(date +%T)] Stopping catalog; discovery must fall back to the DHT" + docker rm -f "${MESH_PREFIX}-sam-catalog" >/dev/null 2>&1 || true + + # With the catalog process gone, the hosted-URL query fails -> fall back to DHT. + run poll_discover_has 1 "calculator" 60 + echo "discover output (fallback): $output" + [[ "$status" -eq 0 ]] # discovery still works via the DHT path +} diff --git a/tests/integration/catalog_e2e_test.go b/tests/integration/catalog_e2e_test.go new file mode 100644 index 00000000..0bb4ca24 --- /dev/null +++ b/tests/integration/catalog_e2e_test.go @@ -0,0 +1,165 @@ +// 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 integration_test + +import ( + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// startCatalog spawns the sam-catalog binary and returns the process and log path. +func startCatalog(t *testing.T, catalogBin, nodeURL, token, homeDir string) (string, *exec.Cmd) { + t.Helper() + logPath := filepath.Join(homeDir, "catalog.log") + logFile, err := os.Create(logPath) + if err != nil { + t.Fatalf("create catalog log: %v", err) + } + + cmd := exec.Command(catalogBin, + "--node-url", nodeURL, + "--node-token", token, + "--bind-addr", "127.0.0.1:0", + "--rewalk-interval", "30s", + "--sweep-interval", "60s", + ) + cmd.Stdout = logFile + cmd.Stderr = logFile + + if err := cmd.Start(); err != nil { + t.Fatalf("start sam-catalog: %v", err) + } + + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = logFile.Close() + }) + + return logPath, cmd +} + +// waitForCatalogAddr reads the catalog log until "catalog MCP on " appears. +func waitForCatalogAddr(t *testing.T, logPath string) string { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + data, _ := os.ReadFile(logPath) + for _, line := range strings.Split(string(data), "\n") { + if strings.Contains(line, "catalog MCP on ") { + parts := strings.Split(line, "catalog MCP on ") + if len(parts) > 1 { + return strings.TrimSpace(parts[1]) + } + } + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("timeout waiting for catalog MCP addr in %s", logPath) + return "" +} + +// TestCatalogEndToEnd verifies the full catalog flow: +// - Node A announces a static MCP service (github-tools). +// - sam-catalog ingests it via bootstrap + SSE tail. +// - query_catalog returns github-tools with node A's peer ID. +func TestCatalogEndToEnd(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + + nodeBin := buildBinary(t, "./cmd/sam-node") + catalogBin := buildBinary(t, "./cmd/sam-catalog") + _, hubAddr := startMockLibp2pHub(t) + + // Node A home dir with static MCP service declared. + homeA := t.TempDir() + cfgDir := filepath.Join(homeA, ".config", "sam") + if err := os.MkdirAll(cfgDir, 0o755); err != nil { + t.Fatal(err) + } + nodeConfig := "version: v1\nservices:\n - type: mcp\n name: github-tools\n description: test\n target_url: http://127.0.0.1:65535\n" + cfgPath := filepath.Join(cfgDir, "node.yaml") + if err := os.WriteFile(cfgPath, []byte(nodeConfig), 0o644); err != nil { + t.Fatal(err) + } + + startBackgroundNode(t, nodeBin, hubAddr, homeA, + "--listen", "/ip4/127.0.0.1/tcp/0", + "--config", cfgPath, + ) + logA := filepath.Join(homeA, "node.log") + + // Resolve actual sidecar address (used by both MCP and sidecar API). + nodeAAddr := waitForMCPAddr(t, logA) + nodeURL := "http://" + nodeAAddr + + // Wait for sidecar API to be ready. + waitForAPI(t, nodeAAddr) + + // Capture node A's peer ID for assertion. + peerInfoA := waitForPeerInfoInLog(t, logA) + // peerInfoA is "/p2p/"; extract the peer ID. + peerIDP2P := peerInfoA[strings.LastIndex(peerInfoA, "/p2p/")+len("/p2p/"):] + + // Start sam-catalog pointed at node A. + homeCat := t.TempDir() + catLogPath, _ := startCatalog(t, catalogBin, nodeURL, "test-token", homeCat) + catalogAddr := waitForCatalogAddr(t, catLogPath) + + // Poll query_catalog until github-tools appears (deadline ~30s). + // Node A fires its first announce ~5s after start. + type catalogEntry struct { + Name string `json:"Name"` + PeerID string `json:"PeerID"` + } + + deadline := time.Now().Add(35 * time.Second) + var found bool + var lastResult string + for time.Now().Before(deadline) { + out := callMCP(t, catalogAddr, "query_catalog", map[string]any{"type": "mcp"}) + lastResult = out + if strings.Contains(out, "github-tools") { + found = true + break + } + time.Sleep(1 * time.Second) + } + + if !found { + t.Fatalf("query_catalog never returned github-tools within deadline; last result: %s", lastResult) + } + + // Assert the entry's PeerID matches node A. + var entries []catalogEntry + if err := json.Unmarshal([]byte(lastResult), &entries); err != nil { + t.Fatalf("failed to parse catalog entries: %v (raw: %s)", err, lastResult) + } + var matched bool + for _, e := range entries { + if e.Name == "github-tools" && e.PeerID == peerIDP2P { + matched = true + break + } + } + if !matched { + t.Fatalf("github-tools found but PeerID mismatch: want %s, entries: %s", peerIDP2P, lastResult) + } +} diff --git a/tests/integration/catalog_resolve_test.go b/tests/integration/catalog_resolve_test.go new file mode 100644 index 00000000..47e65dc7 --- /dev/null +++ b/tests/integration/catalog_resolve_test.go @@ -0,0 +1,218 @@ +// 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 integration_test + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// discoveredProvider mirrors api.DiscoveredProvider JSON for assertion. +type discoveredProvider struct { + PeerID string `json:"peer_id"` + SrvName string `json:"srv_name"` +} + +// TestDiscoveryUsesCatalog asserts that a consumer node resolves +// discover_remote_services via the catalog when sam-catalog is running. +func TestDiscoveryUsesCatalog(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + // Harness limitation, not a code gap: this exercises the biscuit-authed + // node->node->catalog libp2p MCP path, but startMockLibp2pHub issues a stub + // 15-byte HubPublicKey + "mock-biscuit-token" (minimal_helpers_test.go), which + // middleware.go rejects (needs a 32-byte ed25519 key + valid biscuit). The + // existing DHT fan-out discovery shares this gap. Body kept for a real-biscuit + // harness; verify the catalog path against a real hub/mesh until then. + t.Skip("blocked by mock-hub stub biscuit; requires real ed25519 biscuit issuance") + + nodeBin := buildBinary(t, "./cmd/sam-node") + catalogBin := buildBinary(t, "./cmd/sam-catalog") + _, hubAddr := startMockLibp2pHub(t) + + // Provider node P with static github-tools MCP service. + homeP := t.TempDir() + cfgDir := filepath.Join(homeP, ".config", "sam") + if err := os.MkdirAll(cfgDir, 0o755); err != nil { + t.Fatal(err) + } + nodeConfig := "version: v1\nservices:\n - type: mcp\n name: github-tools\n description: test\n target_url: http://127.0.0.1:65535\n" + cfgPath := filepath.Join(cfgDir, "node.yaml") + if err := os.WriteFile(cfgPath, []byte(nodeConfig), 0o644); err != nil { + t.Fatal(err) + } + + startBackgroundNode(t, nodeBin, hubAddr, homeP, + "--listen", "/ip4/127.0.0.1/tcp/0", + "--config", cfgPath, + ) + logP := filepath.Join(homeP, "node.log") + nodeAddrP := waitForMCPAddr(t, logP) + waitForAPI(t, nodeAddrP) + + // Extract P's peer ID from log for assertion. + peerInfoP := waitForPeerInfoInLog(t, logP) + peerIDP := peerInfoP[strings.LastIndex(peerInfoP, "/p2p/")+len("/p2p/"):] + + // Start sam-catalog pointed at node P (it registers itself as CATALOG type). + homeCat := t.TempDir() + catLogPath, _ := startCatalog(t, catalogBin, "http://"+nodeAddrP, "test-token", homeCat) + catalogAddr := waitForCatalogAddr(t, catLogPath) + + // Wait until catalog has ingested github-tools. + deadline := time.Now().Add(35 * time.Second) + var catalogReady bool + for time.Now().Before(deadline) { + out := callMCP(t, catalogAddr, "query_catalog", map[string]any{"type": "mcp"}) + if strings.Contains(out, "github-tools") { + catalogReady = true + break + } + time.Sleep(1 * time.Second) + } + if !catalogReady { + t.Fatal("sam-catalog never ingested github-tools within deadline") + } + + // Consumer node C — no static services. + homeC := t.TempDir() + startBackgroundNode(t, nodeBin, hubAddr, homeC, + "--listen", "/ip4/127.0.0.1/tcp/0", + ) + logC := filepath.Join(homeC, "node.log") + mcpC := waitForMCPAddr(t, logC) + waitForAPI(t, mcpC) + + // Connect C to P so FindProvidersByType(CATALOG) can resolve via DHT. + callMCP(t, mcpC, "connect_peer", map[string]any{"peer_addr": peerInfoP}) + + // Poll C's discover_remote_services until github-tools appears (catalog path). + // type=mcp without name triggers queryCatalog first, then falls back to DHT fan-out. + deadline = time.Now().Add(30 * time.Second) + var lastResult string + var found bool + for time.Now().Before(deadline) { + out := callMCP(t, mcpC, "discover_remote_services", map[string]any{"type": "mcp"}) + lastResult = out + if strings.Contains(out, "github-tools") { + found = true + break + } + time.Sleep(1 * time.Second) + } + if !found { + t.Fatalf("discover_remote_services never returned github-tools; last: %s", lastResult) + } + + // Assert the returned provider's peer_id matches P. + var providers []discoveredProvider + if err := json.Unmarshal([]byte(lastResult), &providers); err != nil { + t.Fatalf("parse discover_remote_services result: %v (raw: %s)", err, lastResult) + } + var matched bool + for _, p := range providers { + if p.SrvName == "github-tools" && p.PeerID == peerIDP { + matched = true + break + } + } + if !matched { + t.Fatalf("github-tools found but peer_id mismatch: want %s, result: %s", peerIDP, lastResult) + } +} + +// TestDiscoveryFallsBackWithoutCatalog asserts that discover_remote_services +// finds github-tools via DHT when no sam-catalog is running. +func TestDiscoveryFallsBackWithoutCatalog(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + + nodeBin := buildBinary(t, "./cmd/sam-node") + _, hubAddr := startMockLibp2pHub(t) + + // Provider node P with static github-tools. + homeP := t.TempDir() + cfgDir := filepath.Join(homeP, ".config", "sam") + if err := os.MkdirAll(cfgDir, 0o755); err != nil { + t.Fatal(err) + } + nodeConfig := "version: v1\nservices:\n - type: mcp\n name: github-tools\n description: test\n target_url: http://127.0.0.1:65535\n" + cfgPath := filepath.Join(cfgDir, "node.yaml") + if err := os.WriteFile(cfgPath, []byte(nodeConfig), 0o644); err != nil { + t.Fatal(err) + } + + startBackgroundNode(t, nodeBin, hubAddr, homeP, + "--listen", "/ip4/127.0.0.1/tcp/0", + "--config", cfgPath, + ) + logP := filepath.Join(homeP, "node.log") + nodeAddrP := waitForMCPAddr(t, logP) + waitForAPI(t, nodeAddrP) + peerInfoP := waitForPeerInfoInLog(t, logP) + peerIDP := peerInfoP[strings.LastIndex(peerInfoP, "/p2p/")+len("/p2p/"):] + + // Consumer node C — no catalog present anywhere. + homeC := t.TempDir() + startBackgroundNode(t, nodeBin, hubAddr, homeC, + "--listen", "/ip4/127.0.0.1/tcp/0", + ) + logC := filepath.Join(homeC, "node.log") + mcpC := waitForMCPAddr(t, logC) + waitForAPI(t, mcpC) + + // Connect C to P so they share the DHT. + callMCP(t, mcpC, "connect_peer", map[string]any{"peer_addr": peerInfoP}) + + // Poll discover_remote_services via DHT name-lookup (no catalog, pure DHT path). + // name=github-tools → discoverServicesByName → DHT CID lookup, no biscuit auth required. + deadline := time.Now().Add(25 * time.Second) + var lastResult string + var found bool + for time.Now().Before(deadline) { + out := callMCP(t, mcpC, "discover_remote_services", map[string]any{"type": "mcp", "name": "github-tools"}) + lastResult = out + if strings.Contains(out, "github-tools") { + found = true + break + } + time.Sleep(1 * time.Second) + } + if !found { + t.Fatalf("DHT fallback: discover_remote_services never returned github-tools; last: %s", lastResult) + } + + // Assert provider's peer_id matches P. + var providers []discoveredProvider + if err := json.Unmarshal([]byte(lastResult), &providers); err != nil { + t.Fatalf("parse discover_remote_services result: %v (raw: %s)", err, lastResult) + } + var matched bool + for _, p := range providers { + if p.SrvName == "github-tools" && p.PeerID == peerIDP { + matched = true + break + } + } + if !matched { + t.Fatalf("DHT fallback: github-tools found but peer_id mismatch: want %s, result: %s", peerIDP, lastResult) + } +} diff --git a/tests/integration/service_announce_test.go b/tests/integration/service_announce_test.go new file mode 100644 index 00000000..165f542e --- /dev/null +++ b/tests/integration/service_announce_test.go @@ -0,0 +1,74 @@ +// 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 integration_test + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// TestServiceAnnouncePropagates verifies that a signed ServiceAnnounce published +// by node A (which hosts a static MCP service) is received by node B over gossip. +func TestServiceAnnouncePropagates(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + nodeBin := buildBinary(t, "./cmd/sam-node") + _, hubAddr := startMockLibp2pHub(t) + + // Node A home with a static MCP service declared in its node config. + homeA := t.TempDir() + cfgDir := filepath.Join(homeA, ".config", "sam") + if err := os.MkdirAll(cfgDir, 0o755); err != nil { + t.Fatal(err) + } + nodeConfig := "version: v1\nservices:\n - type: mcp\n name: github-tools\n description: test\n target_url: http://127.0.0.1:65535\n" + cfgPath := filepath.Join(cfgDir, "node.yaml") + if err := os.WriteFile(cfgPath, []byte(nodeConfig), 0o644); err != nil { + t.Fatal(err) + } + + startBackgroundNode(t, nodeBin, hubAddr, homeA, + "--listen", "/ip4/127.0.0.1/tcp/0", + "--config", cfgPath, + ) + logA := filepath.Join(homeA, "node.log") + addrA := waitForPeerInfoInLog(t, logA) + + // Node B subscribes to the announce topic and waits for A's message. + homeB := t.TempDir() + startBackgroundNode(t, nodeBin, hubAddr, homeB, "--listen", "/ip4/127.0.0.1/tcp/0") + logB := filepath.Join(homeB, "node.log") + mcpB := waitForMCPAddr(t, logB) + + // B connects to A so they share a gossip mesh. + callMCP(t, mcpB, "connect_peer", map[string]any{"peer_addr": addrA}) + callMCP(t, mcpB, "subscribe_topic", map[string]any{"topic": "/sam/service/announce/v1"}) + + // A's first announce fires ~5s after start (initial reprovide delay). Poll up to 25s. + deadline := time.Now().Add(25 * time.Second) + for time.Now().Before(deadline) { + out := callMCP(t, mcpB, "poll_messages", map[string]any{"topic": "/sam/service/announce/v1"}) + // poll_messages returns "Messages on topic : []" when empty. + if !strings.HasSuffix(strings.TrimSpace(out), ": []") && strings.TrimSpace(out) != "" { + return // received at least one announce + } + time.Sleep(1 * time.Second) + } + t.Fatal("node B never received a service announce from node A") +}