From eb33cd44f4eed9102e5c2ccd16979bb5459cbbc7 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 13:46:04 +0000 Subject: [PATCH 01/13] feat(node): ring buffer for typed node events --- internal/node/events.go | 72 ++++++++++++++++++++++++++++++ internal/node/events_test.go | 86 ++++++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 internal/node/events.go create mode 100644 internal/node/events_test.go diff --git a/internal/node/events.go b/internal/node/events.go new file mode 100644 index 00000000..2b178a26 --- /dev/null +++ b/internal/node/events.go @@ -0,0 +1,72 @@ +package node + +import ( + "sync" + "time" +) + +const nodeEventBufferSize = 1000 + +// NodeEvent is one verified observation recorded by the node. +type NodeEvent struct { + Seq uint64 `json:"seq"` + Timestamp int64 `json:"timestamp"` + Category string `json:"category"` + Type string `json:"type"` + PeerID string `json:"peer_id,omitempty"` + Message string `json:"message,omitempty"` +} + +type nodeEventBuffer struct { + mu sync.Mutex + entries []NodeEvent + nextSeq uint64 +} + +func newNodeEventBuffer() *nodeEventBuffer { + return &nodeEventBuffer{nextSeq: 1} +} + +var globalEventBuffer = newNodeEventBuffer() + +// RecordNodeEvent appends an event to the global node event buffer. +func RecordNodeEvent(category, eventType, peerID, message string) { + globalEventBuffer.record(category, eventType, peerID, message) +} + +func (b *nodeEventBuffer) record(category, eventType, peerID, message string) { + b.mu.Lock() + defer b.mu.Unlock() + + b.entries = append(b.entries, NodeEvent{ + Seq: b.nextSeq, + Timestamp: time.Now().UnixMilli(), + Category: category, + Type: eventType, + PeerID: peerID, + Message: message, + }) + b.nextSeq++ + if len(b.entries) > nodeEventBufferSize { + discarded := len(b.entries) - nodeEventBufferSize + // Zero the trimmed prefix so its strings don't linger until the next reallocation. + for i := range discarded { + b.entries[i] = NodeEvent{} + } + b.entries = b.entries[discarded:] + } +} + +// poll returns events with seq > sinceSeq, oldest first, plus the latest seq. +func (b *nodeEventBuffer) poll(sinceSeq uint64) ([]NodeEvent, uint64) { + b.mu.Lock() + defer b.mu.Unlock() + + events := []NodeEvent{} + for _, event := range b.entries { + if event.Seq > sinceSeq { + events = append(events, event) + } + } + return events, b.nextSeq - 1 +} diff --git a/internal/node/events_test.go b/internal/node/events_test.go new file mode 100644 index 00000000..fc812175 --- /dev/null +++ b/internal/node/events_test.go @@ -0,0 +1,86 @@ +package node + +import ( + "sync" + "testing" +) + +func TestNodeEventBufferRecordAndPoll(t *testing.T) { + buffer := newNodeEventBuffer() + buffer.record("security", "spoofing_attempt", "peer-a", "invalid signature") + buffer.record("mesh_event", "banned", "peer-b", "peer banned by hub") + + events, latestSeq := buffer.poll(0) + if latestSeq != 2 { + t.Fatalf("latestSeq = %d, want 2", latestSeq) + } + if len(events) != 2 { + t.Fatalf("len(events) = %d, want 2", len(events)) + } + if events[0].Seq != 1 || events[0].Type != "spoofing_attempt" || events[0].PeerID != "peer-a" { + t.Fatalf("unexpected first event: %+v", events[0]) + } + if events[0].Timestamp == 0 { + t.Fatal("timestamp not set") + } +} + +func TestNodeEventBufferCursor(t *testing.T) { + buffer := newNodeEventBuffer() + for i := 0; i < 5; i++ { + buffer.record("security", "stale_event", "peer-a", "stale") + } + + events, latestSeq := buffer.poll(3) + if latestSeq != 5 { + t.Fatalf("latestSeq = %d, want 5", latestSeq) + } + if len(events) != 2 || events[0].Seq != 4 || events[1].Seq != 5 { + t.Fatalf("unexpected events after cursor 3: %+v", events) + } + + // Re-reads are idempotent. + again, _ := buffer.poll(3) + if len(again) != 2 { + t.Fatalf("re-read returned %d events, want 2", len(again)) + } +} + +func TestNodeEventBufferWrap(t *testing.T) { + buffer := newNodeEventBuffer() + for i := 0; i < 1500; i++ { + buffer.record("security", "rate_limit_drop", "peer-a", "dropped") + } + + events, latestSeq := buffer.poll(0) + if latestSeq != 1500 { + t.Fatalf("latestSeq = %d, want 1500", latestSeq) + } + if len(events) != 1000 { + t.Fatalf("len(events) = %d, want 1000", len(events)) + } + if events[0].Seq != 501 || events[999].Seq != 1500 { + t.Fatalf("wrap kept wrong window: first=%d last=%d", events[0].Seq, events[999].Seq) + } +} + +func TestNodeEventBufferConcurrent(t *testing.T) { + buffer := newNodeEventBuffer() + var waitGroup sync.WaitGroup + for worker := 0; worker < 8; worker++ { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + for i := 0; i < 200; i++ { + buffer.record("security", "rate_limit_drop", "peer-a", "dropped") + buffer.poll(0) + } + }() + } + waitGroup.Wait() + + _, latestSeq := buffer.poll(0) + if latestSeq != 1600 { + t.Fatalf("latestSeq = %d, want 1600", latestSeq) + } +} From 0da3cf1e3b1919a4dfc6840e4f63e9a984768eeb Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 13:52:10 +0000 Subject: [PATCH 02/13] feat(node): poll_node_events MCP tool, self peer_id in get_mesh_info --- internal/node/mcp.go | 6 ++++++ internal/node/mcp_handlers.go | 18 ++++++++++++++++++ internal/node/mcp_handlers_test.go | 27 +++++++++++++++++++++++++++ 3 files changed, 51 insertions(+) diff --git a/internal/node/mcp.go b/internal/node/mcp.go index f2bd8d19..255a6427 100644 --- a/internal/node/mcp.go +++ b/internal/node/mcp.go @@ -153,6 +153,12 @@ func NewMCPServer(node *SamNode) *mcp.Server { Description: "Returns the last few lines of the node's log output.", }, node.handleGetRecentLogs) + // Add the poll_node_events tool. + mcp.AddTool(mcpServer, &mcp.Tool{ + Name: "poll_node_events", + Description: "Poll typed node events (mesh events, security events) with cursor semantics. Pass since_seq from the previous response; 0 returns everything buffered.", + }, node.handlePollNodeEvents) + return mcpServer } diff --git a/internal/node/mcp_handlers.go b/internal/node/mcp_handlers.go index 030ce60e..d8dc0dc3 100644 --- a/internal/node/mcp_handlers.go +++ b/internal/node/mcp_handlers.go @@ -204,6 +204,7 @@ func (n *SamNode) handleGetMeshInfo(ctx context.Context, req *mcp.CallToolReques dhtSize := n.DHT.RoutingTable().Size() resData := map[string]any{ + "peer_id": n.Host.ID().String(), "connected_peers": connectedPeers, "dht_size": dhtSize, "router_peer_id": n.RouterPeerID.String(), @@ -896,3 +897,20 @@ func (n *SamNode) handleGetRecentLogs(ctx context.Context, req *mcp.CallToolRequ Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, }, nil, nil } + +// PollNodeEventsParams defines parameters for the poll_node_events tool. +type PollNodeEventsParams struct { + SinceSeq uint64 `json:"since_seq,omitempty" jsonschema:"Return events with seq greater than this. 0 or omitted returns everything buffered."` +} + +// handlePollNodeEvents implements the poll_node_events tool. +func (n *SamNode) handlePollNodeEvents(ctx context.Context, req *mcp.CallToolRequest, params PollNodeEventsParams) (*mcp.CallToolResult, any, error) { + events, latestSeq := globalEventBuffer.poll(params.SinceSeq) + data, err := json.Marshal(map[string]any{"events": events, "latest_seq": latestSeq}) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, + }, nil, nil +} diff --git a/internal/node/mcp_handlers_test.go b/internal/node/mcp_handlers_test.go index 9b345725..fec3dfc1 100644 --- a/internal/node/mcp_handlers_test.go +++ b/internal/node/mcp_handlers_test.go @@ -1064,3 +1064,30 @@ func newPreDiscoverMCPHandler(t *testing.T, tools []*mcp.Tool) http.Handler { real.ServeHTTP(w, r) }) } + +func TestHandlePollNodeEvents(t *testing.T) { + globalEventBuffer = newNodeEventBuffer() + RecordNodeEvent("security", "spoofing_attempt", "peer-a", "invalid signature") + RecordNodeEvent("mesh_event", "banned", "peer-b", "peer banned by hub") + + node := &SamNode{} + result, _, err := node.handlePollNodeEvents(context.Background(), nil, PollNodeEventsParams{SinceSeq: 1}) + if err != nil { + t.Fatalf("handler error: %v", err) + } + + text := result.Content[0].(*mcp.TextContent).Text + var response struct { + Events []NodeEvent `json:"events"` + LatestSeq uint64 `json:"latest_seq"` + } + if err := json.Unmarshal([]byte(text), &response); err != nil { + t.Fatalf("bad JSON: %v", err) + } + if response.LatestSeq != 2 { + t.Fatalf("latest_seq = %d, want 2", response.LatestSeq) + } + if len(response.Events) != 1 || response.Events[0].Type != "banned" { + t.Fatalf("unexpected events: %+v", response.Events) + } +} From c1e16b6229cab55db0fdd7810551a11a844d8f7a Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 13:58:00 +0000 Subject: [PATCH 03/13] feat(node): record typed events at hub-event emit points --- internal/node/node.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/internal/node/node.go b/internal/node/node.go index c079da7a..2f54b35d 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -1123,6 +1123,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.rateLimiter.Allow(msg.ReceivedFrom.String()) { logger.Warnf("[Mesh Event] Rate limit exceeded for %s, dropping message", msg.ReceivedFrom) + RecordNodeEvent("security", "rate_limit_drop", msg.ReceivedFrom.String(), "rate limit exceeded, message dropped") continue } @@ -1140,6 +1141,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.verifyEvent(&event) { logger.Warnf("[Mesh Event] Potential spoofing attempt: invalid signature on event from %s", msg.ReceivedFrom) + RecordNodeEvent("security", "spoofing_attempt", msg.ReceivedFrom.String(), "invalid event signature") continue } @@ -1147,15 +1149,19 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { eventTime := time.UnixMilli(event.Timestamp) if time.Since(eventTime) > FreshnessThreshold || time.Until(eventTime) > FreshnessThreshold { logger.Warnf("[Mesh Event] Dropping stale or future event from %s (timestamp: %d)", msg.ReceivedFrom, event.Timestamp) + RecordNodeEvent("security", "stale_event", msg.ReceivedFrom.String(), "stale or future event dropped") continue } switch event.Type { case api.MeshEvent_BANNED: + RecordNodeEvent("mesh_event", "banned", event.PeerId, "peer banned by hub") n.handleBannedEvent(&event) case api.MeshEvent_KEY_ROTATION: + RecordNodeEvent("mesh_event", "key_rotation", event.PeerId, "hub key rotation") n.handleKeyRotationEvent(&event) case api.MeshEvent_POLICY_UPDATE: + RecordNodeEvent("mesh_event", "policy_update", event.PeerId, "mesh policy update received") logger.Infof("[Mesh Event] Received POLICY_UPDATE event from %s, triggering sync", msg.ReceivedFrom) go func() { maxJitter := n.config.PolicySyncJitter From 1aaf001e3bb08cdcbf4a0c7253e8ae33b3ec2632 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 14:03:48 +0000 Subject: [PATCH 04/13] feat(mesh-cop): alert model, channel interface, delivery fan-out --- agents/mesh-cop/mesh_cop.py | 114 +++++++++++++++++++++++++++++++ agents/mesh-cop/test_mesh_cop.py | 87 +++++++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 agents/mesh-cop/mesh_cop.py create mode 100644 agents/mesh-cop/test_mesh_cop.py diff --git a/agents/mesh-cop/mesh_cop.py b/agents/mesh-cop/mesh_cop.py new file mode 100644 index 00000000..dfb4ed08 --- /dev/null +++ b/agents/mesh-cop/mesh_cop.py @@ -0,0 +1,114 @@ +import abc +import asyncio +from dataclasses import dataclass +from typing import List + +import httpx + + +@dataclass +class CopConfig: + poll_interval: float + miss_threshold: int + min_peers: int + + +def load_config(env) -> CopConfig: + return CopConfig( + poll_interval=float(env.get("POLL_INTERVAL", "30")), + miss_threshold=int(env.get("MISS_THRESHOLD", "3")), + min_peers=int(env.get("MIN_PEERS", "1")), + ) + + +@dataclass +class Alert: + severity: str + title: str + description: str + peer_id: str = "" + service: str = "" + + +SEVERITY_EMOJI = {"critical": "🚨", "warning": "⚠ī¸", "info": "ℹī¸"} + + +def format_alert(alert: Alert, node_peer_id: str, timestamp: str) -> str: + lines = [f"{SEVERITY_EMOJI[alert.severity]} [{alert.severity.upper()}] {alert.title}"] + if alert.description: + lines.append(alert.description) + if alert.peer_id: + lines.append(f"peer: {alert.peer_id}") + if alert.service: + lines.append(f"service: {alert.service}") + lines.append(f"reported by {node_peer_id} at {timestamp}") + return "\n".join(lines) + + +class Channel(abc.ABC): + """Delivery backend. Implement send() and add one entry in build_channels().""" + + name = "channel" + + @abc.abstractmethod + async def send(self, message: str) -> None: ... + + +class SlackChannel(Channel): + name = "slack" + + def __init__(self, webhook_url: str): + self.webhook_url = webhook_url + + async def send(self, message: str) -> None: + async with httpx.AsyncClient() as http_client: + response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) + response.raise_for_status() + + +class TelegramChannel(Channel): + name = "telegram" + + def __init__(self, bot_token: str, chat_id: str): + self.bot_token = bot_token + self.chat_id = chat_id + + async def send(self, message: str) -> None: + url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" + async with httpx.AsyncClient() as http_client: + response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) + response.raise_for_status() + + +class StdoutChannel(Channel): + name = "stdout" + + async def send(self, message: str) -> None: + print(f"[ALERT] {message}", flush=True) + + +def build_channels(env) -> List[Channel]: + channels: List[Channel] = [] + if env.get("SLACK_WEBHOOK_URL"): + channels.append(SlackChannel(env["SLACK_WEBHOOK_URL"])) + if env.get("TELEGRAM_BOT_TOKEN") and env.get("TELEGRAM_CHAT_ID"): + channels.append(TelegramChannel(env["TELEGRAM_BOT_TOKEN"], env["TELEGRAM_CHAT_ID"])) + if not channels: + channels.append(StdoutChannel()) + return channels + + +async def deliver(channels: List[Channel], message: str) -> None: + async def deliver_to_channel(channel: Channel) -> None: + for attempt in range(3): + try: + await channel.send(message) + return + except Exception as error: + if attempt == 2: + print(f"[-] {channel.name} delivery failed after 3 attempts: {error}", flush=True) + else: + await asyncio.sleep(2**attempt) + + # Concurrent so one slow or failing channel's retries don't delay the others. + await asyncio.gather(*(deliver_to_channel(channel) for channel in channels)) diff --git a/agents/mesh-cop/test_mesh_cop.py b/agents/mesh-cop/test_mesh_cop.py new file mode 100644 index 00000000..ef7fef2e --- /dev/null +++ b/agents/mesh-cop/test_mesh_cop.py @@ -0,0 +1,87 @@ +import asyncio + +from mesh_cop import ( + Alert, + Channel, + SlackChannel, + StdoutChannel, + TelegramChannel, + build_channels, + deliver, + format_alert, + load_config, +) + + +def test_load_config_defaults(): + config = load_config({}) + assert config.poll_interval == 30.0 + assert config.miss_threshold == 3 + assert config.min_peers == 1 + + +def test_load_config_overrides(): + config = load_config({"POLL_INTERVAL": "5", "MISS_THRESHOLD": "2", "MIN_PEERS": "0"}) + assert config.poll_interval == 5.0 + assert config.miss_threshold == 2 + assert config.min_peers == 0 + + +def test_build_channels_stdout_fallback(): + channels = build_channels({}) + assert len(channels) == 1 + assert isinstance(channels[0], StdoutChannel) + + +def test_build_channels_slack_and_telegram(): + env = { + "SLACK_WEBHOOK_URL": "https://hooks.slack.example/x", + "TELEGRAM_BOT_TOKEN": "token", + "TELEGRAM_CHAT_ID": "42", + } + channels = build_channels(env) + assert {type(c) for c in channels} == {SlackChannel, TelegramChannel} + + +def test_build_channels_telegram_requires_chat_id(): + channels = build_channels({"TELEGRAM_BOT_TOKEN": "token"}) + assert len(channels) == 1 + assert isinstance(channels[0], StdoutChannel) + + +def test_format_alert(): + alert = Alert("critical", "node event: banned", "peer banned by hub", peer_id="12D3KooPeer") + message = format_alert(alert, "12D3KooSelf", "2026-08-06T10:00:00+00:00") + assert message.startswith("🚨 [CRITICAL] node event: banned") + assert "peer banned by hub" in message + assert "peer: 12D3KooPeer" in message + assert "reported by 12D3KooSelf at 2026-08-06T10:00:00+00:00" in message + + +class FlakyChannel(Channel): + name = "flaky" + + def __init__(self, failures): + self.failures = failures + self.attempts = 0 + self.delivered = [] + + async def send(self, message): + self.attempts += 1 + if self.attempts <= self.failures: + raise RuntimeError("boom") + self.delivered.append(message) + + +def test_deliver_retries_then_succeeds(): + channel = FlakyChannel(failures=2) + asyncio.run(deliver([channel], "hello")) + assert channel.delivered == ["hello"] + assert channel.attempts == 3 + + +def test_deliver_gives_up_without_raising(): + channel = FlakyChannel(failures=5) + asyncio.run(deliver([channel], "hello")) + assert channel.delivered == [] + assert channel.attempts == 3 From 855b25c483a568f68d44d0c528a30f140be3dfa6 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 14:11:51 +0000 Subject: [PATCH 05/13] feat(mesh-cop): pure-function detection engine --- agents/mesh-cop/mesh_cop.py | 103 ++++++++++++++++++++++++++++- agents/mesh-cop/test_mesh_cop.py | 108 +++++++++++++++++++++++++++++++ 2 files changed, 209 insertions(+), 2 deletions(-) diff --git a/agents/mesh-cop/mesh_cop.py b/agents/mesh-cop/mesh_cop.py index dfb4ed08..c81883d2 100644 --- a/agents/mesh-cop/mesh_cop.py +++ b/agents/mesh-cop/mesh_cop.py @@ -1,7 +1,7 @@ import abc import asyncio -from dataclasses import dataclass -from typing import List +from dataclasses import dataclass, field +from typing import Dict, List import httpx @@ -112,3 +112,102 @@ async def deliver_to_channel(channel: Channel) -> None: # Concurrent so one slow or failing channel's retries don't delay the others. await asyncio.gather(*(deliver_to_channel(channel) for channel in channels)) + + +SEVERITY_BY_EVENT_TYPE = { + "banned": "critical", + "spoofing_attempt": "critical", + "stale_event": "warning", + "rate_limit_drop": "warning", + "key_rotation": "warning", + "policy_update": "info", +} + + +@dataclass +class CopState: + cursor: int = 0 + baseline: set = field(default_factory=set) + miss_counts: dict = field(default_factory=dict) + partitioned: bool = False + first_snapshot: bool = True + + +def detect_partition(state: CopState, connected_peer_count: int, min_peers: int) -> List[Alert]: + alerts = [] + if connected_peer_count < min_peers: + if not state.partitioned: + alerts.append(Alert("critical", "node partitioned", + f"connected peers dropped to {connected_peer_count} (min {min_peers}); churn detection suspended")) + state.partitioned = True + elif state.partitioned: + alerts.append(Alert("info", "connectivity restored", + f"{connected_peer_count} peer(s) connected; churn baseline reset")) + state.partitioned = False + state.first_snapshot = True + state.miss_counts = {} + return alerts + + +def detect_node_events(state: CopState, node_events: List[dict], latest_seq: int) -> List[Alert]: + alerts = [] + rate_limit_drops_by_peer: Dict[str, int] = {} + for event in node_events: + event_type = event.get("type", "") + if event_type == "rate_limit_drop": + peer_id = event.get("peer_id", "") + rate_limit_drops_by_peer[peer_id] = rate_limit_drops_by_peer.get(peer_id, 0) + 1 + continue + severity = SEVERITY_BY_EVENT_TYPE.get(event_type, "info") + alerts.append(Alert(severity, f"node event: {event_type}", + event.get("message", ""), peer_id=event.get("peer_id", ""))) + for peer_id, count in rate_limit_drops_by_peer.items(): + alerts.append(Alert("warning", "node event: rate_limit_drop", + f"{count} message(s) dropped this cycle", peer_id=peer_id)) + if state.cursor > 0 and latest_seq - state.cursor > len(node_events): + lost = latest_seq - state.cursor - len(node_events) + alerts.append(Alert("warning", "node event buffer wrapped", + f"~{lost} node event(s) lost before this poll")) + state.cursor = latest_seq + return alerts + + +def detect_churn(state: CopState, services_snapshot: set, miss_threshold: int) -> List[Alert]: + if state.partitioned: + return [] + if state.first_snapshot: + state.baseline = set(services_snapshot) + state.miss_counts = {} + state.first_snapshot = False + return [] + + alerts = [] + for entry in sorted(services_snapshot - state.baseline): + service_type, service_name, peer_id = entry + alerts.append(Alert("info", "service appeared", f"{service_type}/{service_name} advertised on the mesh", + peer_id=peer_id, service=f"{service_type}/{service_name}")) + state.baseline.add(entry) + for entry in list(state.miss_counts): + if entry in services_snapshot: + del state.miss_counts[entry] + for entry in sorted(state.baseline - services_snapshot): + misses = state.miss_counts.get(entry, 0) + 1 + if misses >= miss_threshold: + service_type, service_name, peer_id = entry + alerts.append(Alert("warning", "service disappeared", + f"{service_type}/{service_name} missing for {misses} consecutive polls", + peer_id=peer_id, service=f"{service_type}/{service_name}")) + state.baseline.discard(entry) + state.miss_counts.pop(entry, None) + else: + state.miss_counts[entry] = misses + return alerts + + +def evaluate_cycle(state: CopState, connected_peer_count: int, node_events: List[dict], + latest_seq: int, services_snapshot: set, config: CopConfig): + alerts = [] + alerts.extend(detect_partition(state, connected_peer_count, config.min_peers)) + alerts.extend(detect_node_events(state, node_events, latest_seq)) + alerts.extend(detect_churn(state, services_snapshot, config.miss_threshold)) + return state, alerts diff --git a/agents/mesh-cop/test_mesh_cop.py b/agents/mesh-cop/test_mesh_cop.py index ef7fef2e..4b4a42d2 100644 --- a/agents/mesh-cop/test_mesh_cop.py +++ b/agents/mesh-cop/test_mesh_cop.py @@ -85,3 +85,111 @@ def test_deliver_gives_up_without_raising(): asyncio.run(deliver([channel], "hello")) assert channel.delivered == [] assert channel.attempts == 3 + + +from mesh_cop import CopConfig, CopState, evaluate_cycle + +CONFIG = CopConfig(poll_interval=30.0, miss_threshold=3, min_peers=1) +SERVICE_A = ("mcp", "service-a", "peer-1") +SERVICE_B = ("inference", "service-b", "peer-2") + + +def cycle(state, peers=2, events=None, latest_seq=None, snapshot=frozenset()): + events = events or [] + if latest_seq is None: + latest_seq = state.cursor + len(events) + return evaluate_cycle(state, peers, events, latest_seq, set(snapshot), CONFIG) + + +def test_first_snapshot_sets_baseline_without_alerts(): + state, alerts = cycle(CopState(), snapshot={SERVICE_A}) + assert alerts == [] + assert state.baseline == {SERVICE_A} + + +def test_service_appeared_alerts_immediately(): + state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot={SERVICE_A, SERVICE_B}) + assert [a.severity for a in alerts] == ["info"] + assert "appeared" in alerts[0].title + assert SERVICE_B in state.baseline + + +def test_service_disappeared_needs_consecutive_misses(): + state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + state, alerts = cycle(state, snapshot=set()) + assert [a.severity for a in alerts] == ["warning"] + assert "disappeared" in alerts[0].title + assert SERVICE_A not in state.baseline + + +def test_reappearing_service_resets_miss_count(): + state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(state, snapshot=set()) + state, _ = cycle(state, snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + + +def test_partition_alerts_once_and_suppresses_churn(): + state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, peers=0, snapshot=set()) + assert [a.severity for a in alerts] == ["critical"] + state, alerts = cycle(state, peers=0, snapshot=set()) + assert alerts == [] + assert state.baseline == {SERVICE_A} + assert state.miss_counts == {} + + +def test_partition_recovery_resets_baseline(): + state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(state, peers=0, snapshot=set()) + state, alerts = cycle(state, peers=2, snapshot={SERVICE_B}) + assert [a.severity for a in alerts] == ["info"] + assert "restored" in alerts[0].title + state, alerts = cycle(state, snapshot={SERVICE_B}) + assert alerts == [] + assert state.baseline == {SERVICE_B} + + +def test_node_event_severities(): + events = [ + {"seq": 1, "type": "banned", "peer_id": "p", "message": "m"}, + {"seq": 2, "type": "spoofing_attempt", "peer_id": "p", "message": "m"}, + {"seq": 3, "type": "policy_update", "peer_id": "", "message": "m"}, + {"seq": 4, "type": "key_rotation", "peer_id": "", "message": "m"}, + ] + state, alerts = cycle(CopState(cursor=0, first_snapshot=False), events=events, latest_seq=4) + assert [a.severity for a in alerts] == ["critical", "critical", "info", "warning"] + assert state.cursor == 4 + + +def test_rate_limit_drops_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "rate_limit_drop", "peer_id": "p1", "message": "m"}, + {"seq": 2, "type": "rate_limit_drop", "peer_id": "p1", "message": "m"}, + {"seq": 3, "type": "rate_limit_drop", "peer_id": "p2", "message": "m"}, + ] + _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=3) + assert len(alerts) == 2 + assert all(a.severity == "warning" for a in alerts) + descriptions = " | ".join(a.description for a in alerts) + assert "2 message(s)" in descriptions and "1 message(s)" in descriptions + + +def test_event_loss_detected_after_first_poll(): + state = CopState(cursor=5, first_snapshot=False) + state, alerts = cycle(state, events=[{"seq": 100, "type": "policy_update", "peer_id": "", "message": "m"}], latest_seq=100) + severities = [a.severity for a in alerts] + assert severities.count("warning") == 1 + assert any("buffer wrapped" in a.title for a in alerts) + assert state.cursor == 100 + + +def test_no_event_loss_alert_on_first_poll(): + _, alerts = cycle(CopState(first_snapshot=False), events=[], latest_seq=5000) + assert alerts == [] From da40018033fe1ff34f4f0059846cebfa0a2052e7 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 14:19:21 +0000 Subject: [PATCH 06/13] feat(mesh-cop): poll loop and inlined SamClient --- agents/mesh-cop/mesh_cop.py | 145 ++++++++++++++++++++++++++++++- agents/mesh-cop/test_mesh_cop.py | 36 ++++++++ 2 files changed, 180 insertions(+), 1 deletion(-) diff --git a/agents/mesh-cop/mesh_cop.py b/agents/mesh-cop/mesh_cop.py index c81883d2..a19b62fa 100644 --- a/agents/mesh-cop/mesh_cop.py +++ b/agents/mesh-cop/mesh_cop.py @@ -1,9 +1,76 @@ import abc import asyncio +import json +import os +import sys from dataclasses import dataclass, field -from typing import Dict, List +from datetime import datetime, timezone +from typing import Dict, List, Optional, Any import httpx +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client + + +class SamClient: + """Inlined SAM Client for self-contained execution.""" + def __init__(self, server_url: Optional[str] = None, token: Optional[str] = None): + if server_url is None: + server_url = os.environ.get("SAM_MCP_URL", "http://localhost:8080/mcp") + if token is None: + token = os.environ.get("SAM_API_TOKEN") + self.server_url = server_url + self.token = token + self.session: Optional[ClientSession] = None + self._sse_cm = None + self.lock = asyncio.Lock() + + async def connect(self): + headers = {"Accept": "application/json, text/event-stream"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + self._sse_cm = streamable_http_client(self.server_url, headers=headers) + read_stream, write_stream, _ = await self._sse_cm.__aenter__() + self.session = ClientSession(read_stream, write_stream) + await self.session.__aenter__() + await self.session.initialize() + + async def close(self): + if self.session: + await self.session.__aexit__(None, None, None) + if self._sse_cm: + await self._sse_cm.__aexit__(None, None, None) + self.session = None + self._sse_cm = None + + async def get_tools(self) -> List[Dict[str, Any]]: + if not self.session: + raise RuntimeError("Not connected") + async with self.lock: + resp = await self.session.list_tools() + return [t.model_dump() if hasattr(t, "model_dump") else t for t in resp.tools] + + async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: + if not self.session: + raise RuntimeError("Not connected") + async with self.lock: + resp = await self.session.call_tool(name, arguments) + return resp.model_dump() if hasattr(resp, "model_dump") else resp + + async def __aenter__(self): + for attempt in range(1, 13): + try: + await self.connect() + return self + except Exception as e: + if attempt == 12: + raise + print(f"[-] Failed to connect to SAM node (attempt {attempt}/12): {e}. Retrying in 5 seconds...") + await asyncio.sleep(5.0) + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.close() @dataclass @@ -211,3 +278,79 @@ def evaluate_cycle(state: CopState, connected_peer_count: int, node_events: List alerts.extend(detect_node_events(state, node_events, latest_seq)) alerts.extend(detect_churn(state, services_snapshot, config.miss_threshold)) return state, alerts + + +SERVICE_TYPES = ["mcp", "inference"] +DISCOVERY_PAGE_LIMIT = 200 + + +def parse_tool_json(result) -> Optional[Any]: + text = result.get("content", [{}])[0].get("text", "") + if not text: + return None + try: + return json.loads(text) + except json.JSONDecodeError: + return None + + +async def fetch_services(client) -> set: + snapshot = set() + for service_type in SERVICE_TYPES: + offset = 0 + while True: + result = await client.call_tool("discover_remote_services", + {"type": service_type, "limit": DISCOVERY_PAGE_LIMIT, "offset": offset}) + providers = parse_tool_json(result) or [] + for provider in providers: + snapshot.add((service_type, provider.get("srv_name", ""), provider.get("peer_id", ""))) + if len(providers) < DISCOVERY_PAGE_LIMIT: + break + offset += DISCOVERY_PAGE_LIMIT + return snapshot + + +async def run_cycle(client, state: CopState, channels: List[Channel], config: CopConfig, node_peer_id: str) -> CopState: + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + connected_peer_count = len(mesh_info.get("connected_peers") or []) + + events_response = parse_tool_json(await client.call_tool("poll_node_events", {"since_seq": state.cursor})) or {} + node_events = events_response.get("events") or [] + latest_seq = events_response.get("latest_seq", state.cursor) + + services_snapshot = await fetch_services(client) + + state, alerts = evaluate_cycle(state, connected_peer_count, node_events, latest_seq, services_snapshot, config) + + timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds") + for alert in alerts: + await deliver(channels, format_alert(alert, node_peer_id, timestamp)) + return state + + +async def run_mesh_cop(): + config = load_config(os.environ) + channels = build_channels(os.environ) + channel_names = ", ".join(channel.name for channel in channels) + print(f"[*] mesh-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " + f"min_peers={config.min_peers} channels=[{channel_names}]", flush=True) + + async with SamClient() as client: + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + node_peer_id = mesh_info.get("peer_id", "unknown") + print(f"[+] connected to local node {node_peer_id}", flush=True) + + state = CopState() + while True: + try: + state = await run_cycle(client, state, channels, config, node_peer_id) + except Exception as error: + print(f"[-] poll cycle failed: {error}", flush=True) + await asyncio.sleep(config.poll_interval) + + +if __name__ == "__main__": + try: + asyncio.run(run_mesh_cop()) + except KeyboardInterrupt: + sys.exit(0) diff --git a/agents/mesh-cop/test_mesh_cop.py b/agents/mesh-cop/test_mesh_cop.py index 4b4a42d2..4257cde2 100644 --- a/agents/mesh-cop/test_mesh_cop.py +++ b/agents/mesh-cop/test_mesh_cop.py @@ -1,4 +1,5 @@ import asyncio +import json from mesh_cop import ( Alert, @@ -193,3 +194,38 @@ def test_event_loss_detected_after_first_poll(): def test_no_event_loss_alert_on_first_poll(): _, alerts = cycle(CopState(first_snapshot=False), events=[], latest_seq=5000) assert alerts == [] + + +from mesh_cop import fetch_services, parse_tool_json + + +def test_parse_tool_json(): + result = {"content": [{"text": '{"events": [], "latest_seq": 7}'}]} + assert parse_tool_json(result) == {"events": [], "latest_seq": 7} + + +def test_parse_tool_json_empty_text(): + assert parse_tool_json({"content": [{"text": ""}]}) is None + + +class FakeSamClient: + def __init__(self, pages): + self.pages = pages + self.calls = [] + + async def call_tool(self, name, arguments): + self.calls.append((name, arguments)) + service_type = arguments["type"] + offset = arguments["offset"] + providers = self.pages.get(service_type, [])[offset:offset + arguments["limit"]] + return {"content": [{"text": json.dumps(providers)}]} + + +def test_fetch_services_paginates_and_builds_tuples(): + mcp_providers = [{"peer_id": f"peer-{i}", "srv_name": f"service-{i}"} for i in range(250)] + client = FakeSamClient({"mcp": mcp_providers, "inference": [{"peer_id": "peer-x", "srv_name": "vllm"}]}) + snapshot = asyncio.run(fetch_services(client)) + assert ("inference", "vllm", "peer-x") in snapshot + assert len(snapshot) == 251 + mcp_calls = [c for c in client.calls if c[1]["type"] == "mcp"] + assert len(mcp_calls) == 2 # 250 providers, limit 200 → two pages From b12c5c96872d9a6a5272af6979ebe983df9e41f1 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 20:08:52 +0000 Subject: [PATCH 07/13] refactor(mesh-cop): import SamClient from sam_mcp SDK, add connect retry --- agents/mesh-cop/mesh_cop.py | 83 ++++++++------------------------ agents/mesh-cop/test_mesh_cop.py | 17 +++++++ 2 files changed, 36 insertions(+), 64 deletions(-) diff --git a/agents/mesh-cop/mesh_cop.py b/agents/mesh-cop/mesh_cop.py index a19b62fa..dffe9281 100644 --- a/agents/mesh-cop/mesh_cop.py +++ b/agents/mesh-cop/mesh_cop.py @@ -8,69 +8,7 @@ from typing import Dict, List, Optional, Any import httpx -from mcp import ClientSession -from mcp.client.streamable_http import streamable_http_client - - -class SamClient: - """Inlined SAM Client for self-contained execution.""" - def __init__(self, server_url: Optional[str] = None, token: Optional[str] = None): - if server_url is None: - server_url = os.environ.get("SAM_MCP_URL", "http://localhost:8080/mcp") - if token is None: - token = os.environ.get("SAM_API_TOKEN") - self.server_url = server_url - self.token = token - self.session: Optional[ClientSession] = None - self._sse_cm = None - self.lock = asyncio.Lock() - - async def connect(self): - headers = {"Accept": "application/json, text/event-stream"} - if self.token: - headers["Authorization"] = f"Bearer {self.token}" - self._sse_cm = streamable_http_client(self.server_url, headers=headers) - read_stream, write_stream, _ = await self._sse_cm.__aenter__() - self.session = ClientSession(read_stream, write_stream) - await self.session.__aenter__() - await self.session.initialize() - - async def close(self): - if self.session: - await self.session.__aexit__(None, None, None) - if self._sse_cm: - await self._sse_cm.__aexit__(None, None, None) - self.session = None - self._sse_cm = None - - async def get_tools(self) -> List[Dict[str, Any]]: - if not self.session: - raise RuntimeError("Not connected") - async with self.lock: - resp = await self.session.list_tools() - return [t.model_dump() if hasattr(t, "model_dump") else t for t in resp.tools] - - async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: - if not self.session: - raise RuntimeError("Not connected") - async with self.lock: - resp = await self.session.call_tool(name, arguments) - return resp.model_dump() if hasattr(resp, "model_dump") else resp - - async def __aenter__(self): - for attempt in range(1, 13): - try: - await self.connect() - return self - except Exception as e: - if attempt == 12: - raise - print(f"[-] Failed to connect to SAM node (attempt {attempt}/12): {e}. Retrying in 5 seconds...") - await asyncio.sleep(5.0) - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - await self.close() +from sam_mcp.client import SamClient @dataclass @@ -328,6 +266,20 @@ async def run_cycle(client, state: CopState, channels: List[Channel], config: Co return state +async def connect_with_retry(retries: int = 12, delay: float = 5.0) -> SamClient: + for attempt in range(1, retries + 1): + client = SamClient() + try: + await client.connect() + return client + except Exception as error: + if attempt == retries: + raise + print(f"[-] Failed to connect to SAM node (attempt {attempt}/{retries}): {error}. " + f"Retrying in {delay} seconds...", flush=True) + await asyncio.sleep(delay) + + async def run_mesh_cop(): config = load_config(os.environ) channels = build_channels(os.environ) @@ -335,7 +287,8 @@ async def run_mesh_cop(): print(f"[*] mesh-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " f"min_peers={config.min_peers} channels=[{channel_names}]", flush=True) - async with SamClient() as client: + client = await connect_with_retry() + try: mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} node_peer_id = mesh_info.get("peer_id", "unknown") print(f"[+] connected to local node {node_peer_id}", flush=True) @@ -347,6 +300,8 @@ async def run_mesh_cop(): except Exception as error: print(f"[-] poll cycle failed: {error}", flush=True) await asyncio.sleep(config.poll_interval) + finally: + await client.close() if __name__ == "__main__": diff --git a/agents/mesh-cop/test_mesh_cop.py b/agents/mesh-cop/test_mesh_cop.py index 4257cde2..6f0acdcf 100644 --- a/agents/mesh-cop/test_mesh_cop.py +++ b/agents/mesh-cop/test_mesh_cop.py @@ -229,3 +229,20 @@ def test_fetch_services_paginates_and_builds_tuples(): assert len(snapshot) == 251 mcp_calls = [c for c in client.calls if c[1]["type"] == "mcp"] assert len(mcp_calls) == 2 # 250 providers, limit 200 → two pages + + +def test_connect_with_retry_returns_after_transient_failures(monkeypatch): + import mesh_cop as mesh_cop_module + + attempts = [] + + class FlakyConnectClient: + async def connect(self): + attempts.append(1) + if len(attempts) < 3: + raise RuntimeError("node not up yet") + + monkeypatch.setattr(mesh_cop_module, "SamClient", FlakyConnectClient) + client = asyncio.run(mesh_cop_module.connect_with_retry(retries=5, delay=0)) + assert isinstance(client, FlakyConnectClient) + assert len(attempts) == 3 From 34dbe671cf17bca5444fd6c2a7144afb1ed44897 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 6 Aug 2026 20:35:00 +0000 Subject: [PATCH 08/13] fix(mesh-cop): reconnect on session loss, sanitize alerts, aggregate security events, surface tool errors --- agents/mesh-cop/mesh_cop.py | 74 +++++++++++++--- agents/mesh-cop/test_mesh_cop.py | 148 ++++++++++++++++++++++++++++++- 2 files changed, 210 insertions(+), 12 deletions(-) diff --git a/agents/mesh-cop/mesh_cop.py b/agents/mesh-cop/mesh_cop.py index dffe9281..43e1bbe2 100644 --- a/agents/mesh-cop/mesh_cop.py +++ b/agents/mesh-cop/mesh_cop.py @@ -2,10 +2,11 @@ import asyncio import json import os +import re import sys from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import Dict, List, Optional, Any +from typing import Dict, List, Optional, Any, Tuple import httpx from sam_mcp.client import SamClient @@ -37,15 +38,26 @@ class Alert: SEVERITY_EMOJI = {"critical": "🚨", "warning": "⚠ī¸", "info": "ℹī¸"} +CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]+") + + +def sanitize_alert_text(text: str, limit: int = 200) -> str: + """Strips control chars (incl. newlines) so attacker-controlled strings can't forge alert lines.""" + cleaned = CONTROL_CHARS_RE.sub(" ", text).strip() + if len(cleaned) > limit: + cleaned = cleaned[:limit].rstrip() + "..." + return cleaned + def format_alert(alert: Alert, node_peer_id: str, timestamp: str) -> str: - lines = [f"{SEVERITY_EMOJI[alert.severity]} [{alert.severity.upper()}] {alert.title}"] + title = sanitize_alert_text(alert.title) + lines = [f"{SEVERITY_EMOJI[alert.severity]} [{alert.severity.upper()}] {title}"] if alert.description: - lines.append(alert.description) + lines.append(sanitize_alert_text(alert.description)) if alert.peer_id: - lines.append(f"peer: {alert.peer_id}") + lines.append(f"peer: {sanitize_alert_text(alert.peer_id)}") if alert.service: - lines.append(f"service: {alert.service}") + lines.append(f"service: {sanitize_alert_text(alert.service)}") lines.append(f"reported by {node_peer_id} at {timestamp}") return "\n".join(lines) @@ -128,6 +140,9 @@ async def deliver_to_channel(channel: Channel) -> None: "policy_update": "info", } +# Event types that arrive as attacker-controlled floods and get aggregated per peer per cycle. +AGGREGATED_EVENT_TYPES = ("rate_limit_drop", "spoofing_attempt", "stale_event") + @dataclass class CopState: @@ -156,19 +171,25 @@ def detect_partition(state: CopState, connected_peer_count: int, min_peers: int) def detect_node_events(state: CopState, node_events: List[dict], latest_seq: int) -> List[Alert]: alerts = [] - rate_limit_drops_by_peer: Dict[str, int] = {} + counts_by_type_and_peer: Dict[str, Dict[str, int]] = {event_type: {} for event_type in AGGREGATED_EVENT_TYPES} for event in node_events: event_type = event.get("type", "") - if event_type == "rate_limit_drop": + if event_type in counts_by_type_and_peer: peer_id = event.get("peer_id", "") - rate_limit_drops_by_peer[peer_id] = rate_limit_drops_by_peer.get(peer_id, 0) + 1 + counts_by_type_and_peer[event_type][peer_id] = counts_by_type_and_peer[event_type].get(peer_id, 0) + 1 continue severity = SEVERITY_BY_EVENT_TYPE.get(event_type, "info") alerts.append(Alert(severity, f"node event: {event_type}", event.get("message", ""), peer_id=event.get("peer_id", ""))) - for peer_id, count in rate_limit_drops_by_peer.items(): + for peer_id, count in counts_by_type_and_peer["rate_limit_drop"].items(): alerts.append(Alert("warning", "node event: rate_limit_drop", f"{count} message(s) dropped this cycle", peer_id=peer_id)) + for peer_id, count in counts_by_type_and_peer["spoofing_attempt"].items(): + alerts.append(Alert("critical", "node event: spoofing_attempt", + f"{count} spoofing attempt(s) (invalid event signature) this cycle", peer_id=peer_id)) + for peer_id, count in counts_by_type_and_peer["stale_event"].items(): + alerts.append(Alert("warning", "node event: stale_event", + f"{count} stale event(s) this cycle", peer_id=peer_id)) if state.cursor > 0 and latest_seq - state.cursor > len(node_events): lost = latest_seq - state.cursor - len(node_events) alerts.append(Alert("warning", "node event buffer wrapped", @@ -223,7 +244,14 @@ def evaluate_cycle(state: CopState, connected_peer_count: int, node_events: List def parse_tool_json(result) -> Optional[Any]: - text = result.get("content", [{}])[0].get("text", "") + if result.get("isError"): + content = result.get("content") or [] + detail = content[0].get("text", "") if content else repr(result) + raise RuntimeError(f"tool error: {detail}") + content = result.get("content") or [] + if not content: + return None + text = content[0].get("text", "") if not text: return None try: @@ -280,6 +308,22 @@ async def connect_with_retry(retries: int = 12, delay: float = 5.0) -> SamClient await asyncio.sleep(delay) +CONSECUTIVE_FAILURE_LIMIT = 3 + + +async def reconnect(client, state: CopState) -> Tuple[SamClient, str, CopState]: + """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" + try: + await client.close() + except Exception as error: + print(f"[-] error closing stale client: {error}", flush=True) + client = await connect_with_retry() + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + node_peer_id = mesh_info.get("peer_id", "unknown") + state.cursor = 0 + return client, node_peer_id, state + + async def run_mesh_cop(): config = load_config(os.environ) channels = build_channels(os.environ) @@ -294,11 +338,19 @@ async def run_mesh_cop(): print(f"[+] connected to local node {node_peer_id}", flush=True) state = CopState() + consecutive_failures = 0 while True: try: state = await run_cycle(client, state, channels, config, node_peer_id) + consecutive_failures = 0 except Exception as error: - print(f"[-] poll cycle failed: {error}", flush=True) + consecutive_failures += 1 + print(f"[-] poll cycle failed ({consecutive_failures}/{CONSECUTIVE_FAILURE_LIMIT}): {error}", flush=True) + if consecutive_failures >= CONSECUTIVE_FAILURE_LIMIT: + print("[-] too many consecutive failures, reconnecting to SAM node", flush=True) + client, node_peer_id, state = await reconnect(client, state) + consecutive_failures = 0 + print(f"[+] reconnected to local node {node_peer_id}", flush=True) await asyncio.sleep(config.poll_interval) finally: await client.close() diff --git a/agents/mesh-cop/test_mesh_cop.py b/agents/mesh-cop/test_mesh_cop.py index 6f0acdcf..ad58af3c 100644 --- a/agents/mesh-cop/test_mesh_cop.py +++ b/agents/mesh-cop/test_mesh_cop.py @@ -1,6 +1,8 @@ import asyncio import json +import pytest + from mesh_cop import ( Alert, Channel, @@ -11,6 +13,7 @@ deliver, format_alert, load_config, + sanitize_alert_text, ) @@ -165,7 +168,8 @@ def test_node_event_severities(): {"seq": 4, "type": "key_rotation", "peer_id": "", "message": "m"}, ] state, alerts = cycle(CopState(cursor=0, first_snapshot=False), events=events, latest_seq=4) - assert [a.severity for a in alerts] == ["critical", "critical", "info", "warning"] + # spoofing_attempt is aggregated per peer, so it surfaces after the one-alert-per-event types. + assert [a.severity for a in alerts] == ["critical", "info", "warning", "critical"] assert state.cursor == 4 @@ -182,6 +186,31 @@ def test_rate_limit_drops_aggregate_per_peer(): assert "2 message(s)" in descriptions and "1 message(s)" in descriptions +def test_spoofing_attempts_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + {"seq": 2, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + {"seq": 3, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + ] + _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=3) + assert len(alerts) == 1 + assert alerts[0].severity == "critical" + assert alerts[0].peer_id == "p1" + assert "3 spoofing attempt(s)" in alerts[0].description + + +def test_stale_events_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "stale_event", "peer_id": "p2", "message": "m"}, + {"seq": 2, "type": "stale_event", "peer_id": "p2", "message": "m"}, + ] + _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=2) + assert len(alerts) == 1 + assert alerts[0].severity == "warning" + assert alerts[0].peer_id == "p2" + assert "2 stale event(s)" in alerts[0].description + + def test_event_loss_detected_after_first_poll(): state = CopState(cursor=5, first_snapshot=False) state, alerts = cycle(state, events=[{"seq": 100, "type": "policy_update", "peer_id": "", "message": "m"}], latest_seq=100) @@ -196,6 +225,31 @@ def test_no_event_loss_alert_on_first_poll(): assert alerts == [] +def test_sanitize_alert_text_strips_control_chars_and_collapses(): + hostile = "svc\n🚨 [CRITICAL] fake alert\x07more" + sanitized = sanitize_alert_text(hostile) + assert "\n" not in sanitized + assert "\x07" not in sanitized + assert "svc" in sanitized and "fake alert" in sanitized + + +def test_sanitize_alert_text_truncates_with_ellipsis(): + truncated = sanitize_alert_text("a" * 500, limit=50) + assert len(truncated) == 53 + assert truncated.endswith("...") + + +def test_format_alert_sanitizes_hostile_service_name(): + hostile_service = "svc\n🚨 [CRITICAL] fake node event: banned" + alert = Alert("info", "service appeared", "desc", peer_id="peer-1", service=hostile_service) + message = format_alert(alert, "12D3KooSelf", "2026-08-06T10:00:00+00:00") + lines = message.split("\n") + assert len(lines) == 5 # no forged extra line from the injected newline + assert lines[0] == "ℹī¸ [INFO] service appeared" + assert "\n" not in lines[3] + assert "svc" in lines[3] and "CRITICAL" in lines[3] + + from mesh_cop import fetch_services, parse_tool_json @@ -208,6 +262,16 @@ def test_parse_tool_json_empty_text(): assert parse_tool_json({"content": [{"text": ""}]}) is None +def test_parse_tool_json_raises_on_tool_error(): + result = {"isError": True, "content": [{"text": "boom"}]} + with pytest.raises(RuntimeError): + parse_tool_json(result) + + +def test_parse_tool_json_empty_content_list_returns_none(): + assert parse_tool_json({"content": []}) is None + + class FakeSamClient: def __init__(self, pages): self.pages = pages @@ -246,3 +310,85 @@ async def connect(self): client = asyncio.run(mesh_cop_module.connect_with_retry(retries=5, delay=0)) assert isinstance(client, FlakyConnectClient) assert len(attempts) == 3 + + +def test_reconnect_swallows_close_errors_refetches_peer_id_and_resets_cursor(monkeypatch): + import mesh_cop as mesh_cop_module + + class BrokenCloseClient: + async def close(self): + raise RuntimeError("close boom") + + class NewClient: + async def call_tool(self, name, arguments): + return {"content": [{"text": json.dumps({"peer_id": "peer-new"})}]} + + async def fake_connect_with_retry(retries=12, delay=5.0): + return NewClient() + + monkeypatch.setattr(mesh_cop_module, "connect_with_retry", fake_connect_with_retry) + + state = CopState(cursor=42, first_snapshot=False) + new_client, node_peer_id, state = asyncio.run(mesh_cop_module.reconnect(BrokenCloseClient(), state)) + assert isinstance(new_client, NewClient) + assert node_peer_id == "peer-new" + assert state.cursor == 0 + + +class StopLoop(Exception): + """Sentinel used to break run_mesh_cop's infinite loop once the assertions are ready.""" + + +def test_run_mesh_cop_reconnects_after_three_consecutive_failures(monkeypatch): + import mesh_cop as mesh_cop_module + + class FailsAfterStartupClient: + def __init__(self): + self.calls = 0 + self.closed = False + + async def call_tool(self, name, arguments): + self.calls += 1 + if self.calls == 1: + return {"content": [{"text": json.dumps({"peer_id": "peer-old"})}]} + raise RuntimeError("mesh unreachable") + + async def close(self): + self.closed = True + + class ReconnectedClient: + def __init__(self): + self.calls = 0 + + async def call_tool(self, name, arguments): + self.calls += 1 + return {"content": [{"text": json.dumps({"peer_id": "peer-new"})}]} + + async def close(self): + pass + + old_client = FailsAfterStartupClient() + new_client = ReconnectedClient() + connect_calls = [] + + async def fake_connect_with_retry(retries=12, delay=5.0): + connect_calls.append(1) + return old_client if len(connect_calls) == 1 else new_client + + sleep_calls = [] + + async def fake_sleep(delay): + sleep_calls.append(delay) + if len(sleep_calls) >= 3: # 2 failed cycles + the reconnecting cycle + raise StopLoop() + + monkeypatch.setattr(mesh_cop_module, "connect_with_retry", fake_connect_with_retry) + monkeypatch.setattr(mesh_cop_module.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(mesh_cop_module.os, "environ", {}) + + with pytest.raises(StopLoop): + asyncio.run(mesh_cop_module.run_mesh_cop()) + + assert old_client.closed is True + assert len(connect_calls) == 2 + assert new_client.calls >= 1 From 66f1006176374028e400b594084b6c702b8eb368 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Fri, 7 Aug 2026 08:36:37 +0000 Subject: [PATCH 09/13] refactor(sam-cop): rename from mesh-cop, move to development/examples, add channel registry Move the agent to development/examples/sam-cop and rename Cop symbols to SamCop. Split the delivery backends into channels.py, where Channel subclasses self-register and declare required_env so build_channels picks them up without a registry edit. Node event categories and types are now exported constants instead of string literals at the call sites. --- development/examples/sam-cop/channels.py | 67 ++++++++++++++ .../examples/sam-cop/sam_cop.py | 88 ++++--------------- .../examples/sam-cop/test_sam_cop.py | 84 +++++++++++------- internal/node/events.go | 17 ++++ internal/node/node.go | 12 +-- 5 files changed, 161 insertions(+), 107 deletions(-) create mode 100644 development/examples/sam-cop/channels.py rename agents/mesh-cop/mesh_cop.py => development/examples/sam-cop/sam_cop.py (81%) rename agents/mesh-cop/test_mesh_cop.py => development/examples/sam-cop/test_sam_cop.py (82%) diff --git a/development/examples/sam-cop/channels.py b/development/examples/sam-cop/channels.py new file mode 100644 index 00000000..859f2d29 --- /dev/null +++ b/development/examples/sam-cop/channels.py @@ -0,0 +1,67 @@ +import abc +from typing import List + +import httpx + + +class Channel(abc.ABC): + """Delivery backend. Subclasses self-register; declare required_env and build_channels() picks them up.""" + + name = "channel" + required_env: tuple = () + registry: list = [] + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + Channel.registry.append(cls) + + @classmethod + def from_env(cls, env): + # __init__ parameters must line up with required_env order. + return cls(*(env[var] for var in cls.required_env)) + + @abc.abstractmethod + async def send(self, message: str) -> None: ... + + +class SlackChannel(Channel): + name = "slack" + required_env = ("SLACK_WEBHOOK_URL",) + + def __init__(self, webhook_url: str): + self.webhook_url = webhook_url + + async def send(self, message: str) -> None: + async with httpx.AsyncClient() as http_client: + response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) + response.raise_for_status() + + +class TelegramChannel(Channel): + name = "telegram" + required_env = ("TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID") + + def __init__(self, bot_token: str, chat_id: str): + self.bot_token = bot_token + self.chat_id = chat_id + + async def send(self, message: str) -> None: + url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" + async with httpx.AsyncClient() as http_client: + response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) + response.raise_for_status() + + +class StdoutChannel(Channel): + name = "stdout" + + async def send(self, message: str) -> None: + print(f"[ALERT] {message}", flush=True) + + +def build_channels(env) -> List[Channel]: + channels = [cls.from_env(env) for cls in Channel.registry + if cls.required_env and all(env.get(var) for var in cls.required_env)] + if not channels: + channels.append(StdoutChannel()) + return channels diff --git a/agents/mesh-cop/mesh_cop.py b/development/examples/sam-cop/sam_cop.py similarity index 81% rename from agents/mesh-cop/mesh_cop.py rename to development/examples/sam-cop/sam_cop.py index 43e1bbe2..2fdb1149 100644 --- a/agents/mesh-cop/mesh_cop.py +++ b/development/examples/sam-cop/sam_cop.py @@ -1,4 +1,3 @@ -import abc import asyncio import json import os @@ -8,19 +7,20 @@ from datetime import datetime, timezone from typing import Dict, List, Optional, Any, Tuple -import httpx from sam_mcp.client import SamClient +from channels import Channel, build_channels + @dataclass -class CopConfig: +class SamCopConfig: poll_interval: float miss_threshold: int min_peers: int -def load_config(env) -> CopConfig: - return CopConfig( +def load_config(env) -> SamCopConfig: + return SamCopConfig( poll_interval=float(env.get("POLL_INTERVAL", "30")), miss_threshold=int(env.get("MISS_THRESHOLD", "3")), min_peers=int(env.get("MIN_PEERS", "1")), @@ -62,59 +62,6 @@ def format_alert(alert: Alert, node_peer_id: str, timestamp: str) -> str: return "\n".join(lines) -class Channel(abc.ABC): - """Delivery backend. Implement send() and add one entry in build_channels().""" - - name = "channel" - - @abc.abstractmethod - async def send(self, message: str) -> None: ... - - -class SlackChannel(Channel): - name = "slack" - - def __init__(self, webhook_url: str): - self.webhook_url = webhook_url - - async def send(self, message: str) -> None: - async with httpx.AsyncClient() as http_client: - response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) - response.raise_for_status() - - -class TelegramChannel(Channel): - name = "telegram" - - def __init__(self, bot_token: str, chat_id: str): - self.bot_token = bot_token - self.chat_id = chat_id - - async def send(self, message: str) -> None: - url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" - async with httpx.AsyncClient() as http_client: - response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) - response.raise_for_status() - - -class StdoutChannel(Channel): - name = "stdout" - - async def send(self, message: str) -> None: - print(f"[ALERT] {message}", flush=True) - - -def build_channels(env) -> List[Channel]: - channels: List[Channel] = [] - if env.get("SLACK_WEBHOOK_URL"): - channels.append(SlackChannel(env["SLACK_WEBHOOK_URL"])) - if env.get("TELEGRAM_BOT_TOKEN") and env.get("TELEGRAM_CHAT_ID"): - channels.append(TelegramChannel(env["TELEGRAM_BOT_TOKEN"], env["TELEGRAM_CHAT_ID"])) - if not channels: - channels.append(StdoutChannel()) - return channels - - async def deliver(channels: List[Channel], message: str) -> None: async def deliver_to_channel(channel: Channel) -> None: for attempt in range(3): @@ -145,7 +92,7 @@ async def deliver_to_channel(channel: Channel) -> None: @dataclass -class CopState: +class SamCopState: cursor: int = 0 baseline: set = field(default_factory=set) miss_counts: dict = field(default_factory=dict) @@ -153,7 +100,7 @@ class CopState: first_snapshot: bool = True -def detect_partition(state: CopState, connected_peer_count: int, min_peers: int) -> List[Alert]: +def detect_partition(state: SamCopState, connected_peer_count: int, min_peers: int) -> List[Alert]: alerts = [] if connected_peer_count < min_peers: if not state.partitioned: @@ -169,8 +116,9 @@ def detect_partition(state: CopState, connected_peer_count: int, min_peers: int) return alerts -def detect_node_events(state: CopState, node_events: List[dict], latest_seq: int) -> List[Alert]: +def detect_node_events(state: SamCopState, node_events: List[dict], latest_seq: int) -> List[Alert]: alerts = [] + # Tally flood-prone event types per peer so N events collapse into one alert; other types alert one-to-one. counts_by_type_and_peer: Dict[str, Dict[str, int]] = {event_type: {} for event_type in AGGREGATED_EVENT_TYPES} for event in node_events: event_type = event.get("type", "") @@ -198,7 +146,7 @@ def detect_node_events(state: CopState, node_events: List[dict], latest_seq: int return alerts -def detect_churn(state: CopState, services_snapshot: set, miss_threshold: int) -> List[Alert]: +def detect_churn(state: SamCopState, services_snapshot: set, miss_threshold: int) -> List[Alert]: if state.partitioned: return [] if state.first_snapshot: @@ -230,8 +178,8 @@ def detect_churn(state: CopState, services_snapshot: set, miss_threshold: int) - return alerts -def evaluate_cycle(state: CopState, connected_peer_count: int, node_events: List[dict], - latest_seq: int, services_snapshot: set, config: CopConfig): +def evaluate_cycle(state: SamCopState, connected_peer_count: int, node_events: List[dict], + latest_seq: int, services_snapshot: set, config: SamCopConfig): alerts = [] alerts.extend(detect_partition(state, connected_peer_count, config.min_peers)) alerts.extend(detect_node_events(state, node_events, latest_seq)) @@ -276,7 +224,7 @@ async def fetch_services(client) -> set: return snapshot -async def run_cycle(client, state: CopState, channels: List[Channel], config: CopConfig, node_peer_id: str) -> CopState: +async def run_cycle(client, state: SamCopState, channels: List[Channel], config: SamCopConfig, node_peer_id: str) -> SamCopState: mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} connected_peer_count = len(mesh_info.get("connected_peers") or []) @@ -311,7 +259,7 @@ async def connect_with_retry(retries: int = 12, delay: float = 5.0) -> SamClient CONSECUTIVE_FAILURE_LIMIT = 3 -async def reconnect(client, state: CopState) -> Tuple[SamClient, str, CopState]: +async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" try: await client.close() @@ -324,11 +272,11 @@ async def reconnect(client, state: CopState) -> Tuple[SamClient, str, CopState]: return client, node_peer_id, state -async def run_mesh_cop(): +async def run_sam_cop(): config = load_config(os.environ) channels = build_channels(os.environ) channel_names = ", ".join(channel.name for channel in channels) - print(f"[*] mesh-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " + print(f"[*] sam-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " f"min_peers={config.min_peers} channels=[{channel_names}]", flush=True) client = await connect_with_retry() @@ -337,7 +285,7 @@ async def run_mesh_cop(): node_peer_id = mesh_info.get("peer_id", "unknown") print(f"[+] connected to local node {node_peer_id}", flush=True) - state = CopState() + state = SamCopState() consecutive_failures = 0 while True: try: @@ -358,6 +306,6 @@ async def run_mesh_cop(): if __name__ == "__main__": try: - asyncio.run(run_mesh_cop()) + asyncio.run(run_sam_cop()) except KeyboardInterrupt: sys.exit(0) diff --git a/agents/mesh-cop/test_mesh_cop.py b/development/examples/sam-cop/test_sam_cop.py similarity index 82% rename from agents/mesh-cop/test_mesh_cop.py rename to development/examples/sam-cop/test_sam_cop.py index ad58af3c..49c50121 100644 --- a/agents/mesh-cop/test_mesh_cop.py +++ b/development/examples/sam-cop/test_sam_cop.py @@ -3,13 +3,15 @@ import pytest -from mesh_cop import ( - Alert, +from channels import ( Channel, SlackChannel, StdoutChannel, TelegramChannel, build_channels, +) +from sam_cop import ( + Alert, deliver, format_alert, load_config, @@ -53,6 +55,26 @@ def test_build_channels_telegram_requires_chat_id(): assert isinstance(channels[0], StdoutChannel) +def test_build_channels_picks_up_registered_subclass(): + class EchoChannel(Channel): + name = "echo" + required_env = ("ECHO_TARGET",) + + def __init__(self, target): + self.target = target + + async def send(self, message): + pass + + try: + channels = build_channels({"ECHO_TARGET": "x"}) + assert len(channels) == 1 + assert isinstance(channels[0], EchoChannel) + assert channels[0].target == "x" + finally: + Channel.registry.remove(EchoChannel) + + def test_format_alert(): alert = Alert("critical", "node event: banned", "peer banned by hub", peer_id="12D3KooPeer") message = format_alert(alert, "12D3KooSelf", "2026-08-06T10:00:00+00:00") @@ -91,9 +113,9 @@ def test_deliver_gives_up_without_raising(): assert channel.attempts == 3 -from mesh_cop import CopConfig, CopState, evaluate_cycle +from sam_cop import SamCopConfig, SamCopState, evaluate_cycle -CONFIG = CopConfig(poll_interval=30.0, miss_threshold=3, min_peers=1) +CONFIG = SamCopConfig(poll_interval=30.0, miss_threshold=3, min_peers=1) SERVICE_A = ("mcp", "service-a", "peer-1") SERVICE_B = ("inference", "service-b", "peer-2") @@ -106,13 +128,13 @@ def cycle(state, peers=2, events=None, latest_seq=None, snapshot=frozenset()): def test_first_snapshot_sets_baseline_without_alerts(): - state, alerts = cycle(CopState(), snapshot={SERVICE_A}) + state, alerts = cycle(SamCopState(), snapshot={SERVICE_A}) assert alerts == [] assert state.baseline == {SERVICE_A} def test_service_appeared_alerts_immediately(): - state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) state, alerts = cycle(state, snapshot={SERVICE_A, SERVICE_B}) assert [a.severity for a in alerts] == ["info"] assert "appeared" in alerts[0].title @@ -120,7 +142,7 @@ def test_service_appeared_alerts_immediately(): def test_service_disappeared_needs_consecutive_misses(): - state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) state, alerts = cycle(state, snapshot=set()) assert alerts == [] state, alerts = cycle(state, snapshot=set()) @@ -132,7 +154,7 @@ def test_service_disappeared_needs_consecutive_misses(): def test_reappearing_service_resets_miss_count(): - state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) state, _ = cycle(state, snapshot=set()) state, _ = cycle(state, snapshot={SERVICE_A}) state, alerts = cycle(state, snapshot=set()) @@ -140,7 +162,7 @@ def test_reappearing_service_resets_miss_count(): def test_partition_alerts_once_and_suppresses_churn(): - state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) state, alerts = cycle(state, peers=0, snapshot=set()) assert [a.severity for a in alerts] == ["critical"] state, alerts = cycle(state, peers=0, snapshot=set()) @@ -150,7 +172,7 @@ def test_partition_alerts_once_and_suppresses_churn(): def test_partition_recovery_resets_baseline(): - state, _ = cycle(CopState(), snapshot={SERVICE_A}) + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) state, _ = cycle(state, peers=0, snapshot=set()) state, alerts = cycle(state, peers=2, snapshot={SERVICE_B}) assert [a.severity for a in alerts] == ["info"] @@ -167,7 +189,7 @@ def test_node_event_severities(): {"seq": 3, "type": "policy_update", "peer_id": "", "message": "m"}, {"seq": 4, "type": "key_rotation", "peer_id": "", "message": "m"}, ] - state, alerts = cycle(CopState(cursor=0, first_snapshot=False), events=events, latest_seq=4) + state, alerts = cycle(SamCopState(cursor=0, first_snapshot=False), events=events, latest_seq=4) # spoofing_attempt is aggregated per peer, so it surfaces after the one-alert-per-event types. assert [a.severity for a in alerts] == ["critical", "info", "warning", "critical"] assert state.cursor == 4 @@ -179,7 +201,7 @@ def test_rate_limit_drops_aggregate_per_peer(): {"seq": 2, "type": "rate_limit_drop", "peer_id": "p1", "message": "m"}, {"seq": 3, "type": "rate_limit_drop", "peer_id": "p2", "message": "m"}, ] - _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=3) + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=3) assert len(alerts) == 2 assert all(a.severity == "warning" for a in alerts) descriptions = " | ".join(a.description for a in alerts) @@ -192,7 +214,7 @@ def test_spoofing_attempts_aggregate_per_peer(): {"seq": 2, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, {"seq": 3, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, ] - _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=3) + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=3) assert len(alerts) == 1 assert alerts[0].severity == "critical" assert alerts[0].peer_id == "p1" @@ -204,7 +226,7 @@ def test_stale_events_aggregate_per_peer(): {"seq": 1, "type": "stale_event", "peer_id": "p2", "message": "m"}, {"seq": 2, "type": "stale_event", "peer_id": "p2", "message": "m"}, ] - _, alerts = cycle(CopState(first_snapshot=False), events=events, latest_seq=2) + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=2) assert len(alerts) == 1 assert alerts[0].severity == "warning" assert alerts[0].peer_id == "p2" @@ -212,7 +234,7 @@ def test_stale_events_aggregate_per_peer(): def test_event_loss_detected_after_first_poll(): - state = CopState(cursor=5, first_snapshot=False) + state = SamCopState(cursor=5, first_snapshot=False) state, alerts = cycle(state, events=[{"seq": 100, "type": "policy_update", "peer_id": "", "message": "m"}], latest_seq=100) severities = [a.severity for a in alerts] assert severities.count("warning") == 1 @@ -221,7 +243,7 @@ def test_event_loss_detected_after_first_poll(): def test_no_event_loss_alert_on_first_poll(): - _, alerts = cycle(CopState(first_snapshot=False), events=[], latest_seq=5000) + _, alerts = cycle(SamCopState(first_snapshot=False), events=[], latest_seq=5000) assert alerts == [] @@ -250,7 +272,7 @@ def test_format_alert_sanitizes_hostile_service_name(): assert "svc" in lines[3] and "CRITICAL" in lines[3] -from mesh_cop import fetch_services, parse_tool_json +from sam_cop import fetch_services, parse_tool_json def test_parse_tool_json(): @@ -296,7 +318,7 @@ def test_fetch_services_paginates_and_builds_tuples(): def test_connect_with_retry_returns_after_transient_failures(monkeypatch): - import mesh_cop as mesh_cop_module + import sam_cop as sam_cop_module attempts = [] @@ -306,14 +328,14 @@ async def connect(self): if len(attempts) < 3: raise RuntimeError("node not up yet") - monkeypatch.setattr(mesh_cop_module, "SamClient", FlakyConnectClient) - client = asyncio.run(mesh_cop_module.connect_with_retry(retries=5, delay=0)) + monkeypatch.setattr(sam_cop_module, "SamClient", FlakyConnectClient) + client = asyncio.run(sam_cop_module.connect_with_retry(retries=5, delay=0)) assert isinstance(client, FlakyConnectClient) assert len(attempts) == 3 def test_reconnect_swallows_close_errors_refetches_peer_id_and_resets_cursor(monkeypatch): - import mesh_cop as mesh_cop_module + import sam_cop as sam_cop_module class BrokenCloseClient: async def close(self): @@ -326,21 +348,21 @@ async def call_tool(self, name, arguments): async def fake_connect_with_retry(retries=12, delay=5.0): return NewClient() - monkeypatch.setattr(mesh_cop_module, "connect_with_retry", fake_connect_with_retry) + monkeypatch.setattr(sam_cop_module, "connect_with_retry", fake_connect_with_retry) - state = CopState(cursor=42, first_snapshot=False) - new_client, node_peer_id, state = asyncio.run(mesh_cop_module.reconnect(BrokenCloseClient(), state)) + state = SamCopState(cursor=42, first_snapshot=False) + new_client, node_peer_id, state = asyncio.run(sam_cop_module.reconnect(BrokenCloseClient(), state)) assert isinstance(new_client, NewClient) assert node_peer_id == "peer-new" assert state.cursor == 0 class StopLoop(Exception): - """Sentinel used to break run_mesh_cop's infinite loop once the assertions are ready.""" + """Sentinel used to break run_sam_cop's infinite loop once the assertions are ready.""" -def test_run_mesh_cop_reconnects_after_three_consecutive_failures(monkeypatch): - import mesh_cop as mesh_cop_module +def test_run_sam_cop_reconnects_after_three_consecutive_failures(monkeypatch): + import sam_cop as sam_cop_module class FailsAfterStartupClient: def __init__(self): @@ -382,12 +404,12 @@ async def fake_sleep(delay): if len(sleep_calls) >= 3: # 2 failed cycles + the reconnecting cycle raise StopLoop() - monkeypatch.setattr(mesh_cop_module, "connect_with_retry", fake_connect_with_retry) - monkeypatch.setattr(mesh_cop_module.asyncio, "sleep", fake_sleep) - monkeypatch.setattr(mesh_cop_module.os, "environ", {}) + monkeypatch.setattr(sam_cop_module, "connect_with_retry", fake_connect_with_retry) + monkeypatch.setattr(sam_cop_module.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(sam_cop_module.os, "environ", {}) with pytest.raises(StopLoop): - asyncio.run(mesh_cop_module.run_mesh_cop()) + asyncio.run(sam_cop_module.run_sam_cop()) assert old_client.closed is True assert len(connect_calls) == 2 diff --git a/internal/node/events.go b/internal/node/events.go index 2b178a26..cf357bb8 100644 --- a/internal/node/events.go +++ b/internal/node/events.go @@ -7,6 +7,22 @@ import ( const nodeEventBufferSize = 1000 +// Node event categories. +const ( + EventCategorySecurity = "security" + EventCategoryMesh = "mesh_event" +) + +// Node event types. +const ( + EventTypeRateLimitDrop = "rate_limit_drop" + EventTypeSpoofingAttempt = "spoofing_attempt" + EventTypeStaleEvent = "stale_event" + EventTypeBanned = "banned" + EventTypeKeyRotation = "key_rotation" + EventTypePolicyUpdate = "policy_update" +) + // NodeEvent is one verified observation recorded by the node. type NodeEvent struct { Seq uint64 `json:"seq"` @@ -24,6 +40,7 @@ type nodeEventBuffer struct { } func newNodeEventBuffer() *nodeEventBuffer { + // Seqs start at 1 so cursor 0 means "never polled": poll(0) returns everything and an empty buffer reports latest seq 0. return &nodeEventBuffer{nextSeq: 1} } diff --git a/internal/node/node.go b/internal/node/node.go index 2f54b35d..41502e26 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -1123,7 +1123,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.rateLimiter.Allow(msg.ReceivedFrom.String()) { logger.Warnf("[Mesh Event] Rate limit exceeded for %s, dropping message", msg.ReceivedFrom) - RecordNodeEvent("security", "rate_limit_drop", msg.ReceivedFrom.String(), "rate limit exceeded, message dropped") + RecordNodeEvent(EventCategorySecurity, EventTypeRateLimitDrop, msg.ReceivedFrom.String(), "rate limit exceeded, message dropped") continue } @@ -1141,7 +1141,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.verifyEvent(&event) { logger.Warnf("[Mesh Event] Potential spoofing attempt: invalid signature on event from %s", msg.ReceivedFrom) - RecordNodeEvent("security", "spoofing_attempt", msg.ReceivedFrom.String(), "invalid event signature") + RecordNodeEvent(EventCategorySecurity, EventTypeSpoofingAttempt, msg.ReceivedFrom.String(), "invalid event signature") continue } @@ -1149,19 +1149,19 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { eventTime := time.UnixMilli(event.Timestamp) if time.Since(eventTime) > FreshnessThreshold || time.Until(eventTime) > FreshnessThreshold { logger.Warnf("[Mesh Event] Dropping stale or future event from %s (timestamp: %d)", msg.ReceivedFrom, event.Timestamp) - RecordNodeEvent("security", "stale_event", msg.ReceivedFrom.String(), "stale or future event dropped") + RecordNodeEvent(EventCategorySecurity, EventTypeStaleEvent, msg.ReceivedFrom.String(), "stale or future event dropped") continue } switch event.Type { case api.MeshEvent_BANNED: - RecordNodeEvent("mesh_event", "banned", event.PeerId, "peer banned by hub") + RecordNodeEvent(EventCategoryMesh, EventTypeBanned, event.PeerId, "peer banned by hub") n.handleBannedEvent(&event) case api.MeshEvent_KEY_ROTATION: - RecordNodeEvent("mesh_event", "key_rotation", event.PeerId, "hub key rotation") + RecordNodeEvent(EventCategoryMesh, EventTypeKeyRotation, event.PeerId, "hub key rotation") n.handleKeyRotationEvent(&event) case api.MeshEvent_POLICY_UPDATE: - RecordNodeEvent("mesh_event", "policy_update", event.PeerId, "mesh policy update received") + RecordNodeEvent(EventCategoryMesh, EventTypePolicyUpdate, event.PeerId, "mesh policy update received") logger.Infof("[Mesh Event] Received POLICY_UPDATE event from %s, triggering sync", msg.ReceivedFrom) go func() { maxJitter := n.config.PolicySyncJitter From 5547d57f9952a15eaab6cb413707943b668d2ad8 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 13 Aug 2026 20:39:02 +0000 Subject: [PATCH 10/13] feat(node): get_control_plane_info MCP tool exposing /info routers and /keys --- internal/node/controlplane.go | 29 ++++++-- internal/node/controlplane_test.go | 103 +++++++++++++++++++++++++++++ internal/node/mcp.go | 6 ++ internal/node/mcp_handlers.go | 40 +++++++++++ 4 files changed, 174 insertions(+), 4 deletions(-) diff --git a/internal/node/controlplane.go b/internal/node/controlplane.go index 959fcb75..c7dce87e 100644 --- a/internal/node/controlplane.go +++ b/internal/node/controlplane.go @@ -28,14 +28,14 @@ import ( "google.golang.org/protobuf/proto" ) -// FetchControlPlaneInfo retrieves the latest configuration from the control plane's /info endpoint. -func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.ControlPlaneInfoResponse, error) { +// fetchControlPlaneProto GETs a control plane endpoint and returns the raw protobuf body. +func fetchControlPlaneProto(ctx context.Context, controlPlaneURL, path string) ([]byte, error) { if !strings.HasPrefix(controlPlaneURL, "http://") && !strings.HasPrefix(controlPlaneURL, "https://") { controlPlaneURL = "https://" + controlPlaneURL } controlPlaneURL = strings.TrimSuffix(controlPlaneURL, "/") - urlStr := controlPlaneURL + "/info" + urlStr := controlPlaneURL + path req, err := http.NewRequestWithContext(ctx, "GET", urlStr, nil) if err != nil { return nil, fmt.Errorf("failed to create HTTP request: %w", err) @@ -59,14 +59,35 @@ func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.Co return nil, fmt.Errorf("control plane returned status %s: %s", resp.Status, string(body)) } + return body, nil +} + +// FetchControlPlaneInfo retrieves the latest configuration from the control plane's /info endpoint. +func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.ControlPlaneInfoResponse, error) { + body, err := fetchControlPlaneProto(ctx, controlPlaneURL, "/info") + if err != nil { + return nil, err + } var info api.ControlPlaneInfoResponse if err := proto.Unmarshal(body, &info); err != nil { return nil, fmt.Errorf("failed to decode /info response: %w", err) } - return &info, nil } +// FetchControlPlaneKeys retrieves the currently valid signing keys from the control plane's /keys endpoint. +func FetchControlPlaneKeys(ctx context.Context, controlPlaneURL string) (*api.KeysResponse, error) { + body, err := fetchControlPlaneProto(ctx, controlPlaneURL, "/keys") + if err != nil { + return nil, err + } + var keys api.KeysResponse + if err := proto.Unmarshal(body, &keys); err != nil { + return nil, fmt.Errorf("failed to decode /keys response: %w", err) + } + return &keys, nil +} + // SyncMeshConfig loads the mesh configuration from the store, attempts to refresh it // via HTTP from the control plane, and updates the store if successful. // It returns the control plane public key and the latest multiaddresses. diff --git a/internal/node/controlplane_test.go b/internal/node/controlplane_test.go index 1ef342ad..43070296 100644 --- a/internal/node/controlplane_test.go +++ b/internal/node/controlplane_test.go @@ -16,6 +16,8 @@ package node import ( "context" + "encoding/base64" + "encoding/json" "net/http" "net/http/httptest" "reflect" @@ -23,6 +25,7 @@ import ( "testing" "github.com/google/sam/api" + "github.com/modelcontextprotocol/go-sdk/mcp" "google.golang.org/protobuf/proto" ) @@ -94,6 +97,106 @@ func TestFetchControlPlaneInfo_InvalidProto(t *testing.T) { } } +func TestFetchControlPlaneKeys(t *testing.T) { + expectedKeys := &api.KeysResponse{ + PublicKeys: [][]byte{[]byte("key-one"), []byte("key-two")}, + } + + body, err := proto.Marshal(expectedKeys) + if err != nil { + t.Fatalf("Failed to marshal keys: %v", err) + } + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/keys" { + t.Errorf("Expected path /keys, got %s", r.URL.Path) + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body) + })) + defer server.Close() + + keys, err := FetchControlPlaneKeys(context.Background(), server.URL) + if err != nil { + t.Fatalf("FetchControlPlaneKeys failed: %v", err) + } + + if !reflect.DeepEqual(keys.PublicKeys, expectedKeys.PublicKeys) { + t.Errorf("Expected PublicKeys %v, got %v", expectedKeys.PublicKeys, keys.PublicKeys) + } +} + +func TestHandleGetControlPlaneInfo(t *testing.T) { + infoBody, err := proto.Marshal(&api.ControlPlaneInfoResponse{ + RouterAddresses: []string{"/dns4/router-a/tcp/4001/p2p/12D3KooRouterA"}, + }) + if err != nil { + t.Fatalf("Failed to marshal info: %v", err) + } + keysBody, err := proto.Marshal(&api.KeysResponse{ + PublicKeys: [][]byte{[]byte("key-one")}, + }) + if err != nil { + t.Fatalf("Failed to marshal keys: %v", err) + } + + mux := http.NewServeMux() + mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(infoBody) }) + mux.HandleFunc("/keys", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(keysBody) }) + server := httptest.NewServer(mux) + defer server.Close() + + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatalf("Failed to create store: %v", err) + } + if err := store.SaveControlPlaneURL(server.URL); err != nil { + t.Fatalf("Failed to save control plane URL: %v", err) + } + + node := &SamNode{Store: store} + result, _, err := node.handleGetControlPlaneInfo(context.Background(), nil, GetControlPlaneInfoParams{}) + if err != nil { + t.Fatalf("handler error: %v", err) + } + + text := result.Content[0].(*mcp.TextContent).Text + var response struct { + ControlPlaneURL string `json:"control_plane_url"` + RouterAddresses []string `json:"router_addresses"` + SigningKeys []string `json:"signing_keys"` + } + if err := json.Unmarshal([]byte(text), &response); err != nil { + t.Fatalf("bad JSON: %v", err) + } + if response.ControlPlaneURL != server.URL { + t.Errorf("Expected control_plane_url %s, got %s", server.URL, response.ControlPlaneURL) + } + if !reflect.DeepEqual(response.RouterAddresses, []string{"/dns4/router-a/tcp/4001/p2p/12D3KooRouterA"}) { + t.Errorf("unexpected router_addresses: %v", response.RouterAddresses) + } + expectedKey := base64.StdEncoding.EncodeToString([]byte("key-one")) + if !reflect.DeepEqual(response.SigningKeys, []string{expectedKey}) { + t.Errorf("unexpected signing_keys: %v", response.SigningKeys) + } +} + +func TestHandleGetControlPlaneInfo_NoURL(t *testing.T) { + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatalf("Failed to create store: %v", err) + } + + node := &SamNode{Store: store} + _, _, err = node.handleGetControlPlaneInfo(context.Background(), nil, GetControlPlaneInfoParams{}) + if err == nil { + t.Fatal("Expected error for missing control plane URL, got nil") + } + if !strings.Contains(err.Error(), "no control plane URL stored") { + t.Errorf("Expected 'no control plane URL stored' error, got %v", err) + } +} + func TestSyncMeshConfig(t *testing.T) { expectedInfo := &api.ControlPlaneInfoResponse{ RouterAddresses: []string{"/ip4/127.0.0.1/tcp/4001"}, diff --git a/internal/node/mcp.go b/internal/node/mcp.go index 255a6427..fed33a53 100644 --- a/internal/node/mcp.go +++ b/internal/node/mcp.go @@ -159,6 +159,12 @@ func NewMCPServer(node *SamNode) *mcp.Server { Description: "Poll typed node events (mesh events, security events) with cursor semantics. Pass since_seq from the previous response; 0 returns everything buffered.", }, node.handlePollNodeEvents) + // Add the get_control_plane_info tool. + mcp.AddTool(mcpServer, &mcp.Tool{ + Name: "get_control_plane_info", + Description: "Fetch the control plane's public /info and /keys endpoints; returns active router addresses and valid signing keys as JSON.", + }, node.handleGetControlPlaneInfo) + return mcpServer } diff --git a/internal/node/mcp_handlers.go b/internal/node/mcp_handlers.go index d8dc0dc3..54442021 100644 --- a/internal/node/mcp_handlers.go +++ b/internal/node/mcp_handlers.go @@ -2,6 +2,7 @@ package node import ( "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -914,3 +915,42 @@ func (n *SamNode) handlePollNodeEvents(ctx context.Context, req *mcp.CallToolReq Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, }, nil, nil } + +// GetControlPlaneInfoParams defines parameters for the get_control_plane_info tool. +type GetControlPlaneInfoParams struct{} + +// handleGetControlPlaneInfo implements the get_control_plane_info tool. +func (n *SamNode) handleGetControlPlaneInfo(ctx context.Context, req *mcp.CallToolRequest, params GetControlPlaneInfoParams) (*mcp.CallToolResult, any, error) { + controlPlaneURL, err := n.Store.LoadControlPlaneURL() + if err != nil { + return nil, nil, fmt.Errorf("failed to load control plane URL: %w", err) + } + if controlPlaneURL == "" { + return nil, nil, fmt.Errorf("no control plane URL stored; is the node enrolled?") + } + + info, err := FetchControlPlaneInfo(ctx, controlPlaneURL) + if err != nil { + return nil, nil, err + } + keys, err := FetchControlPlaneKeys(ctx, controlPlaneURL) + if err != nil { + return nil, nil, err + } + + signingKeys := []string{} + for _, key := range keys.PublicKeys { + signingKeys = append(signingKeys, base64.StdEncoding.EncodeToString(key)) + } + data, err := json.Marshal(map[string]any{ + "control_plane_url": controlPlaneURL, + "router_addresses": info.RouterAddresses, + "signing_keys": signingKeys, + }) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, + }, nil, nil +} From a65a3c6ac412effec341495a7010754e4821aa8f Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Thu, 13 Aug 2026 20:39:02 +0000 Subject: [PATCH 11/13] feat(sam-cop): alert on router churn, signing-key changes, control-plane reachability --- development/examples/sam-cop/sam_cop.py | 77 ++++++++++++++++++- development/examples/sam-cop/test_sam_cop.py | 79 ++++++++++++++++++++ 2 files changed, 154 insertions(+), 2 deletions(-) diff --git a/development/examples/sam-cop/sam_cop.py b/development/examples/sam-cop/sam_cop.py index 2fdb1149..9833c29d 100644 --- a/development/examples/sam-cop/sam_cop.py +++ b/development/examples/sam-cop/sam_cop.py @@ -98,6 +98,10 @@ class SamCopState: miss_counts: dict = field(default_factory=dict) partitioned: bool = False first_snapshot: bool = True + routers: set = field(default_factory=set) + signing_keys: set = field(default_factory=set) + control_plane_first_snapshot: bool = True + control_plane_misses: int = 0 def detect_partition(state: SamCopState, connected_peer_count: int, min_peers: int) -> List[Alert]: @@ -178,12 +182,72 @@ def detect_churn(state: SamCopState, services_snapshot: set, miss_threshold: int return alerts +# Sentinel distinguishing "control plane not polled" (tests) from "poll failed" (None). +CONTROL_PLANE_UNPOLLED = object() + + +def router_peer_ids(router_addresses) -> set: + """One router leases several multiaddrs; collapse them to the trailing /p2p/ peer ID.""" + routers = set() + for address in router_addresses: + if "/p2p/" in address: + routers.add(address.rsplit("/p2p/", 1)[1]) + else: + routers.add(address) + return routers + + +def detect_control_plane(state: SamCopState, control_plane_info, miss_threshold: int) -> List[Alert]: + if control_plane_info is CONTROL_PLANE_UNPOLLED: + return [] + + alerts = [] + if control_plane_info is None: + state.control_plane_misses += 1 + if state.control_plane_misses == miss_threshold: + alerts.append(Alert("warning", "control plane unreachable", + f"/info and /keys fetch failed for {miss_threshold} consecutive polls")) + return alerts + if state.control_plane_misses >= miss_threshold: + alerts.append(Alert("info", "control plane reachable again", + "control plane info fetch succeeded after repeated failures")) + state.control_plane_misses = 0 + + routers = router_peer_ids(control_plane_info.get("router_addresses") or []) + signing_keys = set(control_plane_info.get("signing_keys") or []) + + if state.control_plane_first_snapshot: + state.routers = routers + state.signing_keys = signing_keys + state.control_plane_first_snapshot = False + return alerts + + # No miss-threshold debounce: the control plane's lease TTL already debounces router churn. + for router in sorted(routers - state.routers): + alerts.append(Alert("info", "router appeared", "new router leased on the control plane", peer_id=router)) + for router in sorted(state.routers - routers): + alerts.append(Alert("warning", "router disappeared", + "router lease expired or was dropped on the control plane", peer_id=router)) + state.routers = routers + + for signing_key in sorted(signing_keys - state.signing_keys): + alerts.append(Alert("critical", "new control-plane signing key", + f"unrecognized signing key published on /keys: {signing_key[:16]}...")) + for signing_key in sorted(state.signing_keys - signing_keys): + alerts.append(Alert("warning", "control-plane signing key retired", + f"signing key removed from /keys: {signing_key[:16]}...")) + state.signing_keys = signing_keys + return alerts + + def evaluate_cycle(state: SamCopState, connected_peer_count: int, node_events: List[dict], - latest_seq: int, services_snapshot: set, config: SamCopConfig): + latest_seq: int, services_snapshot: set, config: SamCopConfig, + control_plane_info=CONTROL_PLANE_UNPOLLED): alerts = [] alerts.extend(detect_partition(state, connected_peer_count, config.min_peers)) alerts.extend(detect_node_events(state, node_events, latest_seq)) alerts.extend(detect_churn(state, services_snapshot, config.miss_threshold)) + alerts.extend(detect_control_plane(state, control_plane_info, config.miss_threshold)) return state, alerts @@ -234,7 +298,16 @@ async def run_cycle(client, state: SamCopState, channels: List[Channel], config: services_snapshot = await fetch_services(client) - state, alerts = evaluate_cycle(state, connected_peer_count, node_events, latest_seq, services_snapshot, config) + # A control plane outage must not fail the cycle: node-event detection stays live + # and the failure must not feed the reconnect counter. + try: + control_plane_info = parse_tool_json(await client.call_tool("get_control_plane_info", {})) + except Exception as error: + control_plane_info = None + print(f"[-] control plane info fetch failed: {error}", flush=True) + + state, alerts = evaluate_cycle(state, connected_peer_count, node_events, latest_seq, + services_snapshot, config, control_plane_info) timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds") for alert in alerts: diff --git a/development/examples/sam-cop/test_sam_cop.py b/development/examples/sam-cop/test_sam_cop.py index 49c50121..b1f27201 100644 --- a/development/examples/sam-cop/test_sam_cop.py +++ b/development/examples/sam-cop/test_sam_cop.py @@ -414,3 +414,82 @@ async def fake_sleep(delay): assert old_client.closed is True assert len(connect_calls) == 2 assert new_client.calls >= 1 + + +from sam_cop import CONTROL_PLANE_UNPOLLED, detect_control_plane, router_peer_ids + +ROUTER_A_ADDRESSES = ["/dns4/router-a/tcp/4001/p2p/12D3KooRouterA", + "/dns4/router-a/udp/4001/quic-v1/p2p/12D3KooRouterA"] +ROUTER_B_ADDRESS = "/dns4/router-b/tcp/4001/p2p/12D3KooRouterB" +CONTROL_PLANE_INFO = {"router_addresses": ROUTER_A_ADDRESSES + [ROUTER_B_ADDRESS], + "signing_keys": ["a2V5LW9uZQ=="]} + + +def test_router_peer_ids_collapses_multiaddrs_per_router(): + assert router_peer_ids(ROUTER_A_ADDRESSES) == {"12D3KooRouterA"} + assert router_peer_ids(["/dns4/no-peer-component/tcp/4001"]) == {"/dns4/no-peer-component/tcp/4001"} + + +def test_control_plane_first_snapshot_sets_baseline_without_alerts(): + state = SamCopState() + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert alerts == [] + assert state.routers == {"12D3KooRouterA", "12D3KooRouterB"} + assert state.signing_keys == {"a2V5LW9uZQ=="} + + +def test_router_disappearance_alerts_warning_without_debounce(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": ROUTER_A_ADDRESSES, + "signing_keys": ["a2V5LW9uZQ=="]}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "router disappeared" in alerts[0].title + assert alerts[0].peer_id == "12D3KooRouterB" + + +def test_router_appearance_alerts_info(): + state = SamCopState() + detect_control_plane(state, {"router_addresses": ROUTER_A_ADDRESSES, "signing_keys": []}, CONFIG.miss_threshold) + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert any(a.severity == "info" and "router appeared" in a.title and a.peer_id == "12D3KooRouterB" + for a in alerts) + + +def test_new_signing_key_alerts_critical(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": CONTROL_PLANE_INFO["router_addresses"], + "signing_keys": ["a2V5LW9uZQ==", "a2V5LXR3bw=="]}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["critical"] + assert "new control-plane signing key" in alerts[0].title + + +def test_signing_key_retirement_alerts_warning(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": CONTROL_PLANE_INFO["router_addresses"], + "signing_keys": []}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "signing key retired" in alerts[0].title + + +def test_control_plane_unreachable_alerts_once_at_threshold_then_recovers(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + alerts = detect_control_plane(state, None, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "control plane unreachable" in alerts[0].title + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["info"] + assert "reachable again" in alerts[0].title + + +def test_unpolled_control_plane_is_noop(): + state = SamCopState() + assert detect_control_plane(state, CONTROL_PLANE_UNPOLLED, CONFIG.miss_threshold) == [] + assert state.control_plane_first_snapshot is True + assert state.control_plane_misses == 0 From 2920855d9c4edea6cad102927a25ee89a06401eb Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 25 Aug 2026 20:34:49 +0000 Subject: [PATCH 12/13] fix(sam-cop): graceful shutdown, park stale clients, SSE-safe HTTP timeouts Handle SIGINT/SIGTERM via a shutdown flag so session-loss CancelledError is distinguishable from a real shutdown. Park dead clients instead of closing them cross-task. Always keep the stdout channel on. Use create_mcp_http_client so SSE streams don't hit httpx's 5s read timeout. --- development/examples/sam-cop/channels.py | 4 +-- development/examples/sam-cop/sam_cop.py | 45 ++++++++++++++++++++---- sam-mcp-python/src/sam_mcp/client.py | 7 ++-- 3 files changed, 44 insertions(+), 12 deletions(-) diff --git a/development/examples/sam-cop/channels.py b/development/examples/sam-cop/channels.py index 859f2d29..78e576ff 100644 --- a/development/examples/sam-cop/channels.py +++ b/development/examples/sam-cop/channels.py @@ -60,8 +60,8 @@ async def send(self, message: str) -> None: def build_channels(env) -> List[Channel]: + # Stdout is always on so the log shows every alert, delivered or not. channels = [cls.from_env(env) for cls in Channel.registry if cls.required_env and all(env.get(var) for var in cls.required_env)] - if not channels: - channels.append(StdoutChannel()) + channels.append(StdoutChannel()) return channels diff --git a/development/examples/sam-cop/sam_cop.py b/development/examples/sam-cop/sam_cop.py index 9833c29d..994906ce 100644 --- a/development/examples/sam-cop/sam_cop.py +++ b/development/examples/sam-cop/sam_cop.py @@ -2,6 +2,7 @@ import json import os import re +import signal import sys from dataclasses import dataclass, field from datetime import datetime, timezone @@ -330,14 +331,28 @@ async def connect_with_retry(retries: int = 12, delay: float = 5.0) -> SamClient CONSECUTIVE_FAILURE_LIMIT = 3 +ABANDONED_CLIENT_LIMIT = 50 + +_shutdown = False + +# Dead clients are parked, never released: closing one raises from a foreign task, +# and letting the GC finalize it cancels a scope owned by ours. Reconnects are rare. +_abandoned: List[SamClient] = [] + + +def _request_shutdown() -> None: + global _shutdown + _shutdown = True async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" - try: - await client.close() - except Exception as error: - print(f"[-] error closing stale client: {error}", flush=True) + _abandoned.append(client) + # Each parked client keeps a file descriptor; exit before they pile up - + # the supervisor restart clears the list. + if len(_abandoned) >= ABANDONED_CLIENT_LIMIT: + print(f"[-] {len(_abandoned)} stale clients parked, exiting for a clean restart", flush=True) + sys.exit(1) client = await connect_with_retry() mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} node_peer_id = mesh_info.get("peer_id", "unknown") @@ -352,6 +367,10 @@ async def run_sam_cop(): print(f"[*] sam-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " f"min_peers={config.min_peers} channels=[{channel_names}]", flush=True) + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, _request_shutdown) + client = await connect_with_retry() try: mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} @@ -364,7 +383,11 @@ async def run_sam_cop(): try: state = await run_cycle(client, state, channels, config, node_peer_id) consecutive_failures = 0 - except Exception as error: + # A lost session surfaces as a bare CancelledError from anyio's cancel scope, + # indistinguishable by type from a real shutdown - hence the explicit flag. + except (Exception, asyncio.CancelledError) as error: + if _shutdown: + raise consecutive_failures += 1 print(f"[-] poll cycle failed ({consecutive_failures}/{CONSECUTIVE_FAILURE_LIMIT}): {error}", flush=True) if consecutive_failures >= CONSECUTIVE_FAILURE_LIMIT: @@ -372,9 +395,17 @@ async def run_sam_cop(): client, node_peer_id, state = await reconnect(client, state) consecutive_failures = 0 print(f"[+] reconnected to local node {node_peer_id}", flush=True) - await asyncio.sleep(config.poll_interval) + try: + await asyncio.sleep(config.poll_interval) + except (Exception, asyncio.CancelledError): + # A discarded session can still cancel us here, outside the cycle. + if _shutdown: + raise finally: - await client.close() + try: + await client.close() + except BaseException as error: + print(f"[-] error closing client on shutdown: {error}", flush=True) if __name__ == "__main__": diff --git a/sam-mcp-python/src/sam_mcp/client.py b/sam-mcp-python/src/sam_mcp/client.py index 7043d1bc..dfe0cb9b 100644 --- a/sam-mcp-python/src/sam_mcp/client.py +++ b/sam-mcp-python/src/sam_mcp/client.py @@ -3,6 +3,7 @@ from typing import Any, Dict, List, Optional from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client +from mcp.shared._httpx_utils import create_mcp_http_client class SamClient: """High-level developer interface for SAM MCP using official SDK.""" @@ -19,12 +20,12 @@ def __init__(self, server_url: Optional[str] = None, token: Optional[str] = None async def connect(self): """Connects to the SAM node via Streamable HTTP.""" - import httpx headers = {"Accept": "application/json, text/event-stream"} if self.token: headers["X-Sam-Authentication"] = f"Bearer {self.token}" - - self._http_client = httpx.AsyncClient(headers=headers) + + # Applies MCP's SSE-friendly timeouts; a plain httpx client reads for 5s and drops the stream. + self._http_client = create_mcp_http_client(headers=headers) try: self._sh_cm = streamable_http_client(self.server_url, http_client=self._http_client) res = await self._sh_cm.__aenter__() From 5efce991e135f123c14a9ca7a4042b23d751756f Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 25 Aug 2026 20:35:35 +0000 Subject: [PATCH 13/13] refactor(sam-cop): move to development/tools --- development/{examples => tools}/sam-cop/channels.py | 0 development/{examples => tools}/sam-cop/sam_cop.py | 0 development/{examples => tools}/sam-cop/test_sam_cop.py | 0 3 files changed, 0 insertions(+), 0 deletions(-) rename development/{examples => tools}/sam-cop/channels.py (100%) rename development/{examples => tools}/sam-cop/sam_cop.py (100%) rename development/{examples => tools}/sam-cop/test_sam_cop.py (100%) diff --git a/development/examples/sam-cop/channels.py b/development/tools/sam-cop/channels.py similarity index 100% rename from development/examples/sam-cop/channels.py rename to development/tools/sam-cop/channels.py diff --git a/development/examples/sam-cop/sam_cop.py b/development/tools/sam-cop/sam_cop.py similarity index 100% rename from development/examples/sam-cop/sam_cop.py rename to development/tools/sam-cop/sam_cop.py diff --git a/development/examples/sam-cop/test_sam_cop.py b/development/tools/sam-cop/test_sam_cop.py similarity index 100% rename from development/examples/sam-cop/test_sam_cop.py rename to development/tools/sam-cop/test_sam_cop.py