From 416da031c30d4b29981f15a7b19b464809c58035 Mon Sep 17 00:00:00 2001 From: code-server Date: Mon, 23 Feb 2026 18:50:19 +0000 Subject: [PATCH] fix: update tests for cryptographic visibility markers Updated tests to check for signed [HIDDEN:{sig}] format instead of plain [HIDDEN] markers. Tests now verify: - Signed markers in session storage ([HIDDEN:{8-char-hex}]) - Proper signature presence with "] " separator - process_direct returns signed content for suppressed messages Also improved test timing (2s wait) to allow system message processing. All 85 tests pass with no regressions. Co-Authored-By: Claude Sonnet 4.5 --- nanobot/agent/loop.py | 17 ++- tests/test_agent_loop_metadata.py | 13 +- tests/test_idle_heartbeat_integration.py | 11 +- tests/test_subagent_suppress.py | 162 +++++++++++++++++++++++ 4 files changed, 191 insertions(+), 12 deletions(-) create mode 100644 tests/test_subagent_suppress.py diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index d258d5a..a8c908e 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -767,4 +767,19 @@ Respond with ONLY valid JSON, no markdown fences.""" ) response = await self._process_message(msg, session_key=session_key) - return response.content if response else "" + if not response: + return "" + + # If suppressed, return signed content from session instead of outbound content + if response.metadata.get("suppressed", False): + session = self.sessions.get_or_create(session_key) + # Get the last assistant message from session (should have signed content) + for msg_item in reversed(session.messages): + if msg_item.get("role") == "assistant": + content = msg_item.get("content", "") + if content.startswith("[HIDDEN:"): + return content + # Fallback to outbound content if signature not found + return response.content + + return response.content diff --git a/tests/test_agent_loop_metadata.py b/tests/test_agent_loop_metadata.py index 515ebcc..0455467 100644 --- a/tests/test_agent_loop_metadata.py +++ b/tests/test_agent_loop_metadata.py @@ -45,7 +45,7 @@ async def test_process_direct_passes_metadata(): @pytest.mark.asyncio async def test_suppress_mode_adds_hidden_prefix(): - """Test that suppress_output metadata adds [HIDDEN] prefix.""" + """Test that suppress_output metadata adds signed [HIDDEN:{sig}] prefix.""" bus = MessageBus() provider = MagicMock(spec=LLMProvider) provider.chat = AsyncMock(return_value=LLMResponse( @@ -66,14 +66,15 @@ async def test_suppress_mode_adds_hidden_prefix(): metadata={"suppress_output": True} ) - # Response content should have [HIDDEN] prefix - assert response.startswith("[HIDDEN]") + # Response content should have signed [HIDDEN:{sig}] prefix + assert response.startswith("[HIDDEN:") + assert "] " in response # Check for signature end assert "This is the agent response" in response @pytest.mark.asyncio async def test_normal_mode_no_hidden_prefix(): - """Test that normal messages don't get [HIDDEN] prefix.""" + """Test that normal messages don't get [HIDDEN:*] prefix.""" bus = MessageBus() provider = MagicMock(spec=LLMProvider) provider.chat = AsyncMock(return_value=LLMResponse( @@ -91,6 +92,6 @@ async def test_normal_mode_no_hidden_prefix(): # Call without suppress_output response = await loop.process_direct(content="test message") - # Response should NOT have [HIDDEN] prefix - assert not response.startswith("[HIDDEN]") + # Response should NOT have [HIDDEN:*] prefix + assert not response.startswith("[HIDDEN:") assert response == "Normal response" diff --git a/tests/test_idle_heartbeat_integration.py b/tests/test_idle_heartbeat_integration.py index 49056d9..63b4b62 100644 --- a/tests/test_idle_heartbeat_integration.py +++ b/tests/test_idle_heartbeat_integration.py @@ -91,14 +91,15 @@ async def test_idle_heartbeat_end_to_end(tmp_path): # 1. Session has new messages assert len(session.messages) > 1 - # 2. Find the heartbeat response (assistant message) + # 2. Find the heartbeat response (assistant message with signed marker) heartbeat_messages = [ m for m in session.messages - if m.get("role") == "assistant" and "[HIDDEN]" in m.get("content", "") + if m.get("role") == "assistant" and "[HIDDEN:" in m.get("content", "") ] - assert len(heartbeat_messages) == 1, "Expected exactly 1 [HIDDEN] heartbeat message" + assert len(heartbeat_messages) == 1, "Expected exactly 1 signed [HIDDEN:*] heartbeat message" - # 3. Verify content is prefixed with [HIDDEN] + # 3. Verify content is prefixed with signed [HIDDEN:{sig}] marker heartbeat_msg = heartbeat_messages[0] - assert heartbeat_msg["content"].startswith("[HIDDEN]") + assert heartbeat_msg["content"].startswith("[HIDDEN:") + assert "] " in heartbeat_msg["content"] # Check for signature end assert "Heartbeat executed successfully" in heartbeat_msg["content"] diff --git a/tests/test_subagent_suppress.py b/tests/test_subagent_suppress.py new file mode 100644 index 0000000..93bd357 --- /dev/null +++ b/tests/test_subagent_suppress.py @@ -0,0 +1,162 @@ +"""Test that subagent announcements respect suppress mode.""" + +import asyncio +from pathlib import Path + +import pytest + +from nanobot.agent.loop import AgentLoop +from nanobot.bus.events import InboundMessage, OutboundMessage +from nanobot.bus.queue import MessageBus +from nanobot.providers.base import LLMProvider, LLMResponse +from nanobot.session.manager import SessionManager + + +class MockProvider(LLMProvider): + """Mock provider that spawns a subagent.""" + + def __init__(self): + self.call_count = 0 + self.thinking_budget = 0 + + def get_default_model(self) -> str: + return "mock-model" + + async def chat(self, messages, tools=None, **kwargs): + self.call_count += 1 + + if self.call_count == 1: + # First call: spawn a subagent + return LLMResponse( + content="", + tool_calls=[ + type( + "ToolCall", + (), + { + "id": "test_tool_call", + "name": "spawn", + "arguments": {"task": "Test task", "label": "test"}, + }, + )() + ], + ) + elif self.call_count == 2: + # Subagent completes its task + return LLMResponse(content="Subagent completed task") + else: + # Main agent responds to subagent announcement + return LLMResponse(content="Acknowledged subagent") + + +@pytest.mark.asyncio +async def test_subagent_announcement_without_suppress(tmp_path: Path): + """Verify that WITHOUT suppress mode, announcements ARE published (baseline test).""" + + bus = MessageBus() + sessions = SessionManager(workspace=tmp_path) + sessions.sessions_dir = tmp_path / "sessions" + sessions.sessions_dir.mkdir(parents=True) + + provider = MockProvider() + agent = AgentLoop( + bus=bus, + provider=provider, + session_manager=sessions, + workspace=tmp_path, + ) + + # Track published messages + published_messages = [] + + async def track_publish(msg: OutboundMessage): + published_messages.append(msg) + + # Override publish to track + original_publish = bus.publish_outbound + bus.publish_outbound = track_publish + + # Start agent loop + agent_task = asyncio.create_task(agent.run()) + + # Send test message WITHOUT suppress + test_msg = InboundMessage( + channel="test", + sender_id="user", + chat_id="normal", + content="Test message", + metadata={}, # NO suppress_output + ) + await bus.publish_inbound(test_msg) + + # Wait for processing (longer to allow system message to complete) + await asyncio.sleep(2.0) + + # Stop agent + agent.stop() + await agent_task + + # Verify: Should have published messages (NOT suppressed) + assert len(published_messages) >= 1, "Expected published messages without suppress mode" + + +@pytest.mark.asyncio +async def test_subagent_announcement_with_suppress(tmp_path: Path): + """Test that subagent announcements respect suppress_output metadata.""" + + bus = MessageBus() + sessions = SessionManager(workspace=tmp_path) + sessions.sessions_dir = tmp_path / "sessions" + sessions.sessions_dir.mkdir(parents=True) + + provider = MockProvider() + agent = AgentLoop( + bus=bus, + provider=provider, + session_manager=sessions, + workspace=tmp_path, + ) + + # Track published messages + published_messages = [] + + async def track_publish(msg: OutboundMessage): + published_messages.append(msg) + + # Override publish to track + bus.publish_outbound = track_publish + + # Start agent loop + agent_task = asyncio.create_task(agent.run()) + + # Send test message WITH suppress + test_msg = InboundMessage( + channel="test", + sender_id="user", + chat_id="suppress", + content="Test message", + metadata={"suppress_output": True}, + ) + await bus.publish_inbound(test_msg) + + # Wait for processing (longer to allow system message to complete) + await asyncio.sleep(2.0) + + # Stop agent + agent.stop() + await agent_task + + # Verify: NO messages should be published (all suppressed) + assert len(published_messages) == 0, ( + f"Expected 0 published messages (all suppressed), " + f"but got {len(published_messages)}: {[m.content for m in published_messages]}" + ) + + # Verify session contains signed [HIDDEN:*] messages (cryptographic visibility markers) + session = sessions.get_or_create("test:suppress") + hidden_messages = [m for m in session.messages if m.get("content") and "[HIDDEN:" in str(m.get("content"))] + + assert len(hidden_messages) >= 1, ( + f"Expected signed [HIDDEN:*] messages in session, " + f"but found {len(hidden_messages)}. Total messages: {len(session.messages)}" + )