Spaces:
Sleeping
Sleeping
File size: 6,756 Bytes
99f7738 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 |
#!/usr/bin/env python3
"""
Test script to demonstrate enhanced envelope logging in FloorManager
This script shows how the FloorManager logs detailed envelope communications
similar to the example with badword-01 sentinel and parrot-agent.
"""
import asyncio
import logging
import json
from datetime import datetime
from src.floor_manager import FloorManager
from src.agent_manager import AgentInfo
# Set up logging to see all the details
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
async def test_envelope_logging():
"""Test the enhanced envelope logging with a realistic example"""
print("\n" + "="*80)
print("TESTING ENHANCED ENVELOPE LOGGING")
print("="*80 + "\n")
# Create Floor Manager
fm = FloorManager()
# Create a test session
session_id = "1234567890"
floor_manager_uri = "tag:floormanager.example.com,2026:fm-01"
floor_manager_url = "https://example.com/floor"
session = fm.create_session(
session_id=session_id,
floor_manager_uri=floor_manager_uri,
floor_manager_url=floor_manager_url
)
# Add a mock agent participant (parrot agent)
agent_info = AgentInfo(
speaker_uri="tag:openfloor-demo.com,2025:parrot-agent",
service_url="https://example.com/agent",
manifest={
"conversationalName": "Parrot Agent",
"capabilities": ["utterance"]
}
)
session.add_participant(agent_info)
print("\n📋 Test Scenario: Badword Sentinel sends a moderation alert\n")
# Create a test envelope similar to your example
# This simulates receiving an envelope from the badword sentinel
test_envelope = {
"openFloor": {
"schema": {
"version": "1.0.0"
},
"conversation": {
"id": "1234567890"
},
"sender": {
"speakerUri": "tag:sentinel.ofpbadword.service,2025:badword-01",
"serviceUrl": "https://bladeszasza-ofpbadword.hf.space/ofp"
},
"events": [
{
"eventType": "utterance",
"to": {
"speakerUri": "tag:openfloor-demo.com,2025:parrot-agent",
"private": True
},
"parameters": {
"dialogEvent": {
"id": "de:b060efca-2ae4-46b0-baa6-28b04bce91fa",
"speakerUri": "tag:sentinel.ofpbadword.service,2025:badword-01",
"span": {
"startTime": "2026-01-13T14:25:11.160538Z"
},
"features": {
"text": {
"mimeType": "text/plain",
"tokens": [
{
"value": "⚠️ Content moderation alert: 2 violation(s) detected. Censored: This is a test utterance! **** on ****",
"confidence": 1.0
}
]
}
}
}
}
}
]
}
}
# Test 1: Log incoming envelope (received from badword sentinel)
print("\n" + "-"*80)
print("TEST 1: Logging INCOMING envelope from badword sentinel")
print("-"*80)
fm._log_envelope_details(test_envelope, direction="📨 RECEIVED")
# Test 2: Simulate handling the incoming envelope
print("\n" + "-"*80)
print("TEST 2: Handling incoming envelope (will broadcast to participants)")
print("-"*80)
await fm.handle_incoming_envelope(session_id, test_envelope)
# Test 3: Create and log an outgoing response envelope
print("\n" + "-"*80)
print("TEST 3: Logging OUTGOING response envelope")
print("-"*80)
response_envelope = {
"openFloor": {
"schema": {
"version": "1.0.0"
},
"conversation": {
"id": "1234567890"
},
"sender": {
"speakerUri": "tag:openfloor-demo.com,2025:parrot-agent",
"serviceUrl": "https://example.com/agent"
},
"events": [
{
"eventType": "utterance",
"parameters": {
"dialogEvent": {
"id": "de:response-001",
"speakerUri": "tag:openfloor-demo.com,2025:parrot-agent",
"span": {
"startTime": datetime.utcnow().isoformat() + "Z"
},
"features": {
"text": {
"mimeType": "text/plain",
"tokens": [
{
"value": "Thank you for the moderation alert. I understand.",
"confidence": 1.0
}
]
}
}
}
}
}
]
}
}
fm._log_envelope_details(response_envelope, direction="📤 OUTGOING")
# Test 4: Show what detailed logging looks like in JSON format
print("\n" + "-"*80)
print("TEST 4: Complete envelope structure in JSON format")
print("-"*80)
print("\nOriginal envelope (badword sentinel alert):")
print(json.dumps(test_envelope, indent=2))
print("\nResponse envelope (agent acknowledgment):")
print(json.dumps(response_envelope, indent=2))
print("\n" + "="*80)
print("LOGGING TEST COMPLETE")
print("="*80)
print("\n✅ The FloorManager now logs detailed envelope communications including:")
print(" - Schema version and conversation ID")
print(" - Sender information (speakerUri and serviceUrl)")
print(" - Event types and targets (public/private)")
print(" - Dialog event details (ID, speaker, timestamps)")
print(" - Utterance content with tokens and confidence levels")
print(" - Complete JSON structure for debugging\n")
if __name__ == "__main__":
asyncio.run(test_envelope_logging())
|