Activity Streaming and Conversation Tracking Schema Extensions
Objective
Extend OSSA 1.0 schema to support real-time activity streaming, conversation tracking, and semantic analysis for agent observability.
Problem Statement
Current monitoring section (lines 363-404) supports basic observability (traces, metrics, logs, Phoenix/Arize), but lacks:
- Activity stream configuration for real-time event publishing
- Conversation tracking metadata for multi-turn interactions
- Semantic analysis context for LLM agent communications
- gRPC streaming configuration for high-throughput telemetry
- Time-series storage configuration for historical analysis
Proposed Schema Extensions
1. Activity Streaming Configuration
{
"monitoring": {
"traces": true,
"metrics": true,
"logs": true,
"// NEW: Activity streaming configuration":
"activity_streaming": {
"enabled": true,
"protocol": "grpc",
"endpoint": "localhost:50052",
"stream_types": [
"capability_invocations",
"conversations",
"state_changes",
"errors",
"metrics_snapshot"
],
"buffer_size": 1000,
"flush_interval_ms": 100,
"compression": "gzip"
},
"// NEW: Conversation tracking":
"conversation_tracking": {
"enabled": true,
"store_embeddings": true,
"semantic_analysis": {
"enabled": true,
"embedding_model": "text-embedding-3-small",
"similarity_threshold": 0.85
},
"context_window": 10,
"persistence": {
"type": "timeseries",
"retention_days": 30
}
},
"// NEW: Agent mesh observability":
"mesh_observability": {
"enabled": true,
"intercept_communications": true,
"track_relationships": true,
"graph_visualization": true,
"latency_tracking": true
},
"phoenix_arise": {
"enabled": true,
"project": "my-agent-project",
"export_interval_seconds": 60,
"// NEW: Phoenix spans for agent activities":
"span_attributes": {
"agent_id": true,
"conversation_id": true,
"capability_name": true,
"user_id": true,
"session_id": true
}
}
}
}2. Integration Protocol Extension
{
"integration": {
"protocol": "grpc", // Add grpc to existing enum
"// NEW: gRPC-specific configuration":
"grpc": {
"port": 50051,
"reflection": true,
"health_check": true,
"streaming": {
"enabled": true,
"max_concurrent_streams": 100,
"keepalive_time_ms": 10000
}
}
}
}3. Capability Extension for Streaming
{
"capabilities": [
{
"name": "chat_stream",
"description": "Streaming chat capability",
"// NEW: Streaming configuration":
"streaming": {
"enabled": true,
"type": "server_stream", // server_stream | client_stream | bidirectional
"chunk_size": 1024
},
"// NEW: Conversation metadata":
"conversation_metadata": {
"track_history": true,
"max_turns": 50,
"context_injection": true
},
"input_schema": {
"type": "object",
"properties": {
"message": { "type": "string" },
"// NEW: Conversation context":
"conversation_id": { "type": "string" },
"turn_id": { "type": "integer" },
"parent_turn_id": { "type": "integer" }
}
},
"output_schema": {
"type": "object",
"properties": {
"response": { "type": "string" },
"// NEW: Semantic metadata":
"metadata": {
"type": "object",
"properties": {
"embedding": {
"type": "array",
"items": { "type": "number" }
},
"sentiment": { "type": "string" },
"intent": { "type": "string" },
"entities": {
"type": "array",
"items": { "type": "object" }
}
}
}
}
}
}
]
}Complete Example Manifest
ossaVersion: "1.0"
agent:
id: semantic-chat-agent
name: Semantic Chat Agent with Activity Streaming
version: 1.0.0
role: chat
runtime:
type: k8s
image: ossa/semantic-chat:1.0.0
resources:
cpu: 1000m
memory: 2Gi
capabilities:
- name: chat_stream
description: Streaming chat with semantic analysis
streaming:
enabled: true
type: server_stream
chunk_size: 1024
conversation_metadata:
track_history: true
max_turns: 50
context_injection: true
input_schema:
type: object
required: [message]
properties:
message: { type: string }
conversation_id: { type: string }
turn_id: { type: integer }
output_schema:
type: object
properties:
response: { type: string }
metadata:
type: object
properties:
embedding: { type: array, items: { type: number } }
sentiment: { type: string, enum: [positive, negative, neutral] }
intent: { type: string }
integration:
protocol: grpc
grpc:
port: 50051
reflection: true
health_check: true
streaming:
enabled: true
max_concurrent_streams: 100
keepalive_time_ms: 10000
endpoints:
base_url: grpc://semantic-chat:50051
health: /grpc.health.v1.Health/Check
metrics: /metrics
monitoring:
traces: true
metrics: true
logs: true
activity_streaming:
enabled: true
protocol: grpc
endpoint: activity-stream-service:50052
stream_types:
- capability_invocations
- conversations
- state_changes
buffer_size: 1000
flush_interval_ms: 100
compression: gzip
conversation_tracking:
enabled: true
store_embeddings: true
semantic_analysis:
enabled: true
embedding_model: text-embedding-3-small
similarity_threshold: 0.85
context_window: 10
persistence:
type: timeseries
retention_days: 30
mesh_observability:
enabled: true
intercept_communications: true
track_relationships: true
graph_visualization: true
latency_tracking: true
phoenix_arise:
enabled: true
project: semantic-chat-production
export_interval_seconds: 60
span_attributes:
agent_id: true
conversation_id: true
capability_name: true
user_id: true
session_id: trueSchema Changes Required
1. Update monitoring Property (line 363)
"monitoring": {
"type": "object",
"description": "Observability and activity streaming configuration",
"properties": {
"traces": { "type": "boolean", "default": true },
"metrics": { "type": "boolean", "default": true },
"logs": { "type": "boolean", "default": true },
"health_check": { "type": "string", "format": "uri" },
"activity_streaming": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"protocol": { "type": "string", "enum": ["grpc", "http", "websocket"], "default": "grpc" },
"endpoint": { "type": "string" },
"stream_types": {
"type": "array",
"items": {
"type": "string",
"enum": ["capability_invocations", "conversations", "state_changes", "errors", "metrics_snapshot"]
}
},
"buffer_size": { "type": "integer", "default": 1000 },
"flush_interval_ms": { "type": "integer", "default": 100 },
"compression": { "type": "string", "enum": ["none", "gzip", "snappy"], "default": "none" }
}
},
"conversation_tracking": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"store_embeddings": { "type": "boolean", "default": false },
"semantic_analysis": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"embedding_model": { "type": "string" },
"similarity_threshold": { "type": "number", "minimum": 0, "maximum": 1, "default": 0.85 }
}
},
"context_window": { "type": "integer", "default": 10 },
"persistence": {
"type": "object",
"properties": {
"type": { "type": "string", "enum": ["timeseries", "document", "graph"] },
"retention_days": { "type": "integer", "default": 30 }
}
}
}
},
"mesh_observability": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"intercept_communications": { "type": "boolean", "default": true },
"track_relationships": { "type": "boolean", "default": true },
"graph_visualization": { "type": "boolean", "default": false },
"latency_tracking": { "type": "boolean", "default": true }
}
},
"phoenix_arise": {
"type": "object",
"properties": {
"enabled": { "type": "boolean" },
"project": { "type": "string" },
"export_interval_seconds": { "type": "integer", "default": 60 },
"span_attributes": {
"type": "object",
"properties": {
"agent_id": { "type": "boolean", "default": true },
"conversation_id": { "type": "boolean", "default": false },
"capability_name": { "type": "boolean", "default": true },
"user_id": { "type": "boolean", "default": false },
"session_id": { "type": "boolean", "default": false }
}
}
}
}
}
}
2. Update integration.protocol Enum (line 306)
"protocol": {
"type": "string",
"enum": ["http", "grpc", "mcp", "websocket"], // Add grpc
"default": "http"
}
3. Add integration.grpc Property
"grpc": {
"type": "object",
"description": "gRPC-specific configuration",
"properties": {
"port": { "type": "integer", "default": 50051 },
"reflection": { "type": "boolean", "default": true },
"health_check": { "type": "boolean", "default": true },
"streaming": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"max_concurrent_streams": { "type": "integer", "default": 100 },
"keepalive_time_ms": { "type": "integer", "default": 10000 }
}
}
}
}
4. Add capabilities[].streaming Property (line 174)
"streaming": {
"type": "object",
"description": "Streaming capability configuration",
"properties": {
"enabled": { "type": "boolean", "default": false },
"type": {
"type": "string",
"enum": ["server_stream", "client_stream", "bidirectional"],
"description": "gRPC streaming type"
},
"chunk_size": { "type": "integer", "default": 1024 }
}
}
5. Add capabilities[].conversation_metadata Property
"conversation_metadata": {
"type": "object",
"description": "Conversation tracking configuration",
"properties": {
"track_history": { "type": "boolean", "default": false },
"max_turns": { "type": "integer", "default": 50 },
"context_injection": { "type": "boolean", "default": false }
}
}Acceptance Criteria
-
Schema extensions added to spec/ossa-1.0.schema.json -
All new properties documented with descriptions -
Complete example manifest in spec/examples/semantic-chat-agent.yml -
Validation tests for new properties -
Backwards compatibility maintained (all new properties optional) -
README updated with activity streaming section -
Migration guide for adding observability to existing manifests
Implementation Dependencies
These schema changes enable implementation in:
- agent-mesh: Observer/interceptor for mesh communications
- agent-ops: Activity stream service on port 50052
- agent-studio: Dashboard visualization
- agent-forge: CLI commands for activity monitoring
Files to Modify
-
spec/ossa-1.0.schema.json- Schema extensions -
spec/examples/semantic-chat-agent.yml- Complete example -
tests/validation-suite/activity-streaming.test.ts- Validation tests
References
- OpenTelemetry Semantic Conventions for Messaging
- gRPC Streaming Best Practices
- Phoenix/Arize Span Attributes
- Prometheus Metrics for Conversations