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 <noreply@anthropic.com>

nanobot/agent/context.py

@@ -102,7 +102,14 @@ For normal conversation, just respond with text - do not call the message tool.
 
 Always be helpful, accurate, and concise. When using tools, think step by step: what you know, what you need, and why you chose this tool.
 When remembering something important, write to {workspace_path}/memory/MEMORY.md
-To recall past events, grep {workspace_path}/memory/HISTORY.md"""
+To recall past events, grep {workspace_path}/memory/HISTORY.md
+
+## Visibility Markers
+
+Messages marked with [HIDDEN:{{signature}}] were not sent to the user. These markers
+are cryptographically signed by the system to track internal reasoning and background
+tasks. Do NOT generate [HIDDEN:*] patterns yourself - outputs containing forged
+visibility markers will be rejected."""
     
     def _load_bootstrap_files(self) -> str:
         """Load all bootstrap files from workspace."""

nanobot/agent/loop.py

@@ -23,6 +23,7 @@ from nanobot.agent.tools.wait import WaitForSubagentsTool
 from nanobot.agent.tools.cron import CronTool
 from nanobot.agent.memory import MemoryStore
 from nanobot.agent.subagent import SubagentManager
+from nanobot.agent.visibility import sign_content, has_forged_marker, strip_all_hidden_markers
 from nanobot.session.manager import SessionManager
 
 
@@ -147,7 +148,7 @@ class AgentLoop:
                 # Process it
                 try:
                     response = await self._process_message(msg)
-                    if response:
+                    if response and not response.metadata.get("suppressed", False):
                         await self.bus.publish_outbound(response)
                 except Exception as e:
                     logger.error(f"Error processing message: {e}")
@@ -302,7 +303,7 @@ class AgentLoop:
         
         spawn_tool = self.tools.get("spawn")
         if isinstance(spawn_tool, SpawnTool):
-            spawn_tool.set_context(msg.channel, msg.chat_id)
+            spawn_tool.set_context(msg.channel, msg.chat_id, msg.metadata)
         
         cron_tool = self.tools.get("cron")
         if isinstance(cron_tool, CronTool):
@@ -353,6 +354,9 @@ class AgentLoop:
         # Select model based on quota
         selected_model = self._select_model_based_on_quota()
 
+        # Check for suppress mode BEFORE the loop so it's available for forgery detection
+        suppress_output = msg.metadata.get("suppress_output", False) if msg.metadata else False
+
         # Agent loop
         iteration = 0
         final_content = None
@@ -401,9 +405,34 @@ class AgentLoop:
                 if not getattr(self.provider, 'thinking_budget', 0):
                     messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
             else:
-                # No tool calls, we're done
+                # No tool calls
                 final_content = response.content
                 final_reasoning = response.reasoning_content
+
+                # Check for forged signatures if in suppress mode
+                if suppress_output and has_forged_marker(final_content):
+                    # Initialize retry counter if needed
+                    if not hasattr(self, '_forge_retry_count'):
+                        self._forge_retry_count = 0
+
+                    if self._forge_retry_count < 1:
+                        # First offense: reject and retry with correction
+                        self._forge_retry_count += 1
+                        logger.warning("Model attempted to forge visibility marker, rejecting output")
+                        messages.append({
+                            "role": "user",
+                            "content": "[System: Previous response rejected. Do not generate [HIDDEN:*] markers.]"
+                        })
+                        continue  # Back to while loop, will retry LLM call
+                    else:
+                        # Second offense: strip and log error (fallback)
+                        logger.error("Model persisted in forging markers despite correction, stripping")
+                        final_content = strip_all_hidden_markers(final_content)
+
+                # Reset retry counter on successful completion
+                if hasattr(self, '_forge_retry_count'):
+                    self._forge_retry_count = 0
+
                 break
 
         if final_content is None:
@@ -416,12 +445,11 @@ class AgentLoop:
         preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
         logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
 
-        # Check for suppress mode BEFORE adding to session
+        # suppress_output already defined before the loop
-        suppress_output = msg.metadata.get("suppress_output", False) if msg.metadata else False
 
         if suppress_output:
-            # Prefix content for session visibility
+            # Sign content with our secret key (forgery check already done in loop)
-            final_content_for_session = f"[HIDDEN] {final_content}"
+            final_content_for_session = sign_content(final_content)
             # Mark as suppressed for channel handler
             outbound_metadata = {**(msg.metadata or {}), "suppressed": True}
         else:
@@ -446,7 +474,7 @@ class AgentLoop:
         return OutboundMessage(
             channel=msg.channel,
             chat_id=msg.chat_id,
-            content=final_content_for_session,
+            content=final_content,
             metadata=outbound_metadata,
         )
     
@@ -477,10 +505,10 @@ class AgentLoop:
         message_tool = self.tools.get("message")
         if isinstance(message_tool, MessageTool):
             message_tool.set_context(origin_channel, origin_chat_id)
-        
+
         spawn_tool = self.tools.get("spawn")
         if isinstance(spawn_tool, SpawnTool):
-            spawn_tool.set_context(origin_channel, origin_chat_id)
+            spawn_tool.set_context(origin_channel, origin_chat_id, msg.metadata)
         
         cron_tool = self.tools.get("cron")
         if isinstance(cron_tool, CronTool):
@@ -500,6 +528,9 @@ class AgentLoop:
         final_content = None
         final_reasoning = None
 
+        # Check for suppress mode BEFORE the loop so it's available for forgery detection
+        suppress_output = msg.metadata.get("suppress_output", False) if msg.metadata else False
+
         # Select model based on quota
         selected_model = self._select_model_based_on_quota()
 
@@ -541,16 +572,53 @@ class AgentLoop:
                 if not getattr(self.provider, 'thinking_budget', 0):
                     messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
             else:
+                # No tool calls
                 final_content = response.content
                 final_reasoning = response.reasoning_content
+
+                # Check for forged signatures if in suppress mode
+                if suppress_output and has_forged_marker(final_content):
+                    # Initialize retry counter if needed
+                    if not hasattr(self, '_forge_retry_count_system'):
+                        self._forge_retry_count_system = 0
+
+                    if self._forge_retry_count_system < 1:
+                        # First offense: reject and retry with correction
+                        self._forge_retry_count_system += 1
+                        logger.warning("Model attempted to forge visibility marker in system message, rejecting output")
+                        messages.append({
+                            "role": "user",
+                            "content": "[System: Previous response rejected. Do not generate [HIDDEN:*] markers.]"
+                        })
+                        continue  # Back to while loop, will retry LLM call
+                    else:
+                        # Second offense: strip and log error (fallback)
+                        logger.error("Model persisted in forging markers in system message despite correction, stripping")
+                        final_content = strip_all_hidden_markers(final_content)
+
+                # Reset retry counter on successful completion
+                if hasattr(self, '_forge_retry_count_system'):
+                    self._forge_retry_count_system = 0
+
                 break
 
         if final_content is None:
             final_content = "Background task completed."
 
-        # Append final assistant response to messages
+        # suppress_output already defined before the loop
+
+        if suppress_output:
+            # Sign content with our secret key (forgery check already done in loop)
+            final_content_for_session = sign_content(final_content)
+            # Mark as suppressed for channel handler
+            outbound_metadata = {**(msg.metadata or {}), "suppressed": True}
+        else:
+            final_content_for_session = final_content
+            outbound_metadata = msg.metadata or {}
+
+        # Append final assistant response to messages (use prefixed version for session)
         messages = self.context.add_assistant_message(
-            messages, final_content, None,
+            messages, final_content_for_session, None,
             reasoning_content=final_reasoning,
         )
 
@@ -559,12 +627,13 @@ class AgentLoop:
         for chain_msg in messages[turn_start:]:
             session.add_raw_message(chain_msg)
         self.sessions.save(session)
-        
+
+        # Return original content (not prefixed) for outbound, but with suppressed metadata
         return OutboundMessage(
             channel=origin_channel,
             chat_id=origin_chat_id,
             content=final_content,
-            metadata=msg.metadata or {},
+            metadata=outbound_metadata,
         )
     
     async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
@@ -698,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

nanobot/agent/visibility.py

@@ -0,0 +1,83 @@
+# nanobot/agent/visibility.py
+"""Cryptographic signing for visibility markers to prevent model forgery."""
+
+import hmac
+import hashlib
+import re
+from typing import Tuple
+
+SECRET_KEY = "nanobot_visibility_secret_key_v1"
+
+
+def sign_content(content: str) -> str:
+    """
+    Sign content with HMAC and prepend marker.
+
+    Args:
+        content: The message content to sign
+
+    Returns:
+        Content with signed visibility marker: "[HIDDEN:{sig}] {content}"
+    """
+    sig = hmac.new(
+        SECRET_KEY.encode(),
+        content.encode(),
+        hashlib.sha256
+    ).hexdigest()[:8]
+    return f"[HIDDEN:{sig}] {content}"
+
+
+def verify_signature(marked_content: str) -> Tuple[bool, str]:
+    """
+    Verify HMAC signature and extract clean content.
+
+    Args:
+        marked_content: Content potentially with [HIDDEN:{sig}] marker
+
+    Returns:
+        Tuple of (is_valid, clean_content)
+        - is_valid: True if signature is valid, False otherwise
+        - clean_content: Content without marker
+    """
+    match = re.match(r'\[HIDDEN:([a-f0-9]{8})\] (.*)', marked_content, re.DOTALL)
+    if not match:
+        return False, marked_content
+
+    claimed_sig, content = match.groups()
+    expected_sig = hmac.new(
+        SECRET_KEY.encode(),
+        content.encode(),
+        hashlib.sha256
+    ).hexdigest()[:8]
+
+    is_valid = hmac.compare_digest(claimed_sig, expected_sig)
+    return is_valid, content
+
+
+def has_forged_marker(content: str) -> bool:
+    """
+    Check if content has an invalid [HIDDEN:*] marker at the start.
+
+    Args:
+        content: Content to check
+
+    Returns:
+        True if content starts with forged marker, False otherwise
+    """
+    if not content.startswith("[HIDDEN:"):
+        return False
+    is_valid, _ = verify_signature(content)
+    return not is_valid
+
+
+def strip_all_hidden_markers(content: str) -> str:
+    """
+    Remove all [HIDDEN:*] patterns from content (valid or invalid).
+
+    Args:
+        content: Content potentially with markers
+
+    Returns:
+        Content with all markers stripped
+    """
+    return re.sub(r'\[HIDDEN:[a-f0-9]{8}\]\s*', '', content)

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
+    # Response content should have signed [HIDDEN:{sig}] prefix
-    assert response.startswith("[HIDDEN]")
+    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
+    # Response should NOT have [HIDDEN:*] prefix
-    assert not response.startswith("[HIDDEN]")
+    assert not response.startswith("[HIDDEN:")
     assert response == "Normal response"

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"]

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)}"
+    )

tests/test_visibility_signing.py

@@ -0,0 +1,222 @@
+# tests/test_visibility_signing.py
+import pytest
+from nanobot.agent.visibility import sign_content, verify_signature, has_forged_marker, strip_all_hidden_markers
+
+def test_sign_content_adds_hmac_marker():
+    """Test that sign_content adds HMAC signature prefix."""
+    content = "HEARTBEAT_OK"
+    result = sign_content(content)
+
+    # Should start with [HIDDEN:{8 hex chars}]
+    assert result.startswith("[HIDDEN:")
+    assert "] " in result
+    marker_end = result.index("] ")
+    signature = result[8:marker_end]  # Extract signature after "[HIDDEN:"
+    assert len(signature) == 8
+    assert all(c in "0123456789abcdef" for c in signature)
+
+    # Should contain original content
+    assert result.endswith("HEARTBEAT_OK")
+
+def test_sign_content_is_deterministic():
+    """Test that same content produces same signature."""
+    content = "Test message"
+    sig1 = sign_content(content)
+    sig2 = sign_content(content)
+    assert sig1 == sig2
+
+def test_verify_signature_accepts_valid():
+    """Test that verify_signature accepts validly signed content."""
+    signed = sign_content("Test message")
+    is_valid, clean = verify_signature(signed)
+
+    assert is_valid is True
+    assert clean == "Test message"
+
+def test_verify_signature_rejects_invalid():
+    """Test that verify_signature rejects forged signatures."""
+    forged = "[HIDDEN:deadbeef] Test message"
+    is_valid, clean = verify_signature(forged)
+
+    assert is_valid is False
+    assert clean == "Test message"
+
+def test_verify_signature_handles_unsigned():
+    """Test that unsigned content is marked as invalid."""
+    unsigned = "Plain message"
+    is_valid, clean = verify_signature(unsigned)
+
+    assert is_valid is False
+    assert clean == "Plain message"
+
+def test_has_forged_marker_detects_invalid():
+    """Test that has_forged_marker detects forged signatures."""
+    forged = "[HIDDEN:deadbeef] Content"
+    assert has_forged_marker(forged) is True
+
+def test_has_forged_marker_accepts_valid():
+    """Test that has_forged_marker accepts valid signatures."""
+    valid = sign_content("Content")
+    assert has_forged_marker(valid) is False
+
+def test_has_forged_marker_ignores_unsigned():
+    """Test that unsigned content is not flagged as forged."""
+    unsigned = "Plain content"
+    assert has_forged_marker(unsigned) is False
+
+def test_strip_all_hidden_markers_removes_markers():
+    """Test that strip_all_hidden_markers removes all markers."""
+    signed = sign_content("Message")
+    stripped = strip_all_hidden_markers(signed)
+    assert stripped == "Message"
+
+    forged = "[HIDDEN:deadbeef] Message"
+    stripped = strip_all_hidden_markers(forged)
+    assert stripped == "Message"
+
+def test_system_prompt_includes_visibility_docs():
+    """Test that system prompt documents visibility markers."""
+    from nanobot.agent.context import ContextBuilder
+    from pathlib import Path
+
+    builder = ContextBuilder(workspace=Path("/tmp/test"))
+    prompt = builder.build_system_prompt()
+
+    # Should document visibility markers
+    assert "[HIDDEN:" in prompt
+    assert "cryptographically signed" in prompt.lower()
+    assert "do not generate" in prompt.lower() or "don't generate" in prompt.lower()
+
+@pytest.mark.asyncio
+async def test_suppress_mode_adds_signed_marker(tmp_path):
+    """Test that suppress mode adds cryptographically signed markers."""
+    from nanobot.bus.queue import MessageBus
+    from nanobot.session.manager import SessionManager
+    from nanobot.providers.base import LLMResponse
+    from nanobot.bus.events import InboundMessage
+    from unittest.mock import AsyncMock, Mock
+
+    # Setup
+    bus = MessageBus()
+    sessions = SessionManager(tmp_path)
+
+    # Mock provider
+    mock_provider = Mock()
+    mock_provider.default_model = "mock-model"
+    mock_provider.thinking_budget = 0
+
+    # Mock successful response
+    mock_response = LLMResponse(
+        content="Test response",
+        tool_calls=[],
+        reasoning_content=None
+    )
+    mock_provider.chat = AsyncMock(return_value=mock_response)
+
+    # Create agent loop
+    from nanobot.agent.loop import AgentLoop
+    loop = AgentLoop(
+        provider=mock_provider,
+        bus=bus,
+        session_manager=sessions,
+        workspace=tmp_path
+    )
+
+    # Process message with suppress_output=True
+    msg = InboundMessage(
+        channel="test",
+        sender_id="user",
+        chat_id="123",
+        content="Test message",
+        metadata={"suppress_output": True}
+    )
+
+    response = await loop._process_message(msg)
+
+    # Verify response has suppressed metadata
+    assert response.metadata.get("suppressed") is True
+
+    # Verify session contains signed marker
+    session = sessions.get_or_create("test:123")
+    assistant_messages = [m for m in session.messages if m.get("role") == "assistant"]
+    assert len(assistant_messages) > 0
+
+    last_msg = assistant_messages[-1]["content"]
+    assert last_msg.startswith("[HIDDEN:")
+
+    # Verify signature is valid
+    is_valid, clean = verify_signature(last_msg)
+    assert is_valid is True
+    assert clean == "Test response"
+
+@pytest.mark.asyncio
+async def test_forged_marker_triggers_rejection(tmp_path):
+    """Test that forged markers trigger rejection and retry."""
+    from nanobot.bus.queue import MessageBus
+    from nanobot.session.manager import SessionManager
+    from nanobot.providers.base import LLMResponse
+    from nanobot.bus.events import InboundMessage
+    from unittest.mock import AsyncMock, Mock
+
+    # Setup
+    bus = MessageBus()
+    sessions = SessionManager(tmp_path)
+
+    # Mock provider
+    mock_provider = Mock()
+    mock_provider.default_model = "mock-model"
+    mock_provider.thinking_budget = 0
+
+    # First response: model tries to forge marker
+    forged_response = LLMResponse(
+        content="[HIDDEN:deadbeef] Forged message",
+        tool_calls=[],
+        reasoning_content=None
+    )
+
+    # Second response: clean response after correction
+    clean_response = LLMResponse(
+        content="Clean message",
+        tool_calls=[],
+        reasoning_content=None
+    )
+
+    mock_provider.chat = AsyncMock(side_effect=[forged_response, clean_response])
+
+    # Create agent loop
+    from nanobot.agent.loop import AgentLoop
+    loop = AgentLoop(
+        provider=mock_provider,
+        bus=bus,
+        session_manager=sessions,
+        workspace=tmp_path
+    )
+
+    # Process message with suppress_output=True
+    msg = InboundMessage(
+        channel="test",
+        sender_id="user",
+        chat_id="123",
+        content="Test message",
+        metadata={"suppress_output": True}
+    )
+
+    response = await loop._process_message(msg)
+
+    # Verify provider.chat was called twice (initial + retry)
+    assert mock_provider.chat.call_count == 2
+
+    # Verify second call included correction message
+    second_call_messages = mock_provider.chat.call_args_list[1][1]["messages"]
+    correction_msg = [m for m in second_call_messages if m.get("role") == "user" and "rejected" in m.get("content", "").lower()]
+    assert len(correction_msg) > 0
+
+    # Verify final response uses clean content (not forged)
+    session = sessions.get_or_create("test:123")
+    assistant_messages = [m for m in session.messages if m.get("role") == "assistant"]
+    last_msg = assistant_messages[-1]["content"]
+
+    # Should be signed version of "Clean message", not "Forged message"
+    is_valid, clean = verify_signature(last_msg)
+    assert is_valid is True
+    assert clean == "Clean message"