tiphys.channels

31class Peer(BaseModel):
32    """Represents a user or entity on a messaging platform."""
33
34    id: str
35    kind: PeerKind = PeerKind.USER
36    username: str | None = None
37    display_name: str | None = None
38    is_bot: bool = False
39    raw: dict[str, Any] = Field(default_factory=dict)

42class Attachment(BaseModel):
43    """File attachment in a message."""
44
45    url: str
46    mime_type: str
47    filename: str | None = None
48    size_bytes: int | None = None

51class InboundMessage(BaseModel):
52    """Normalized message from any channel."""
53
54    id: str
55    channel: str  # "telegram", "discord", "webchat"
56    sender: Peer
57    content: str  # Markdown text
58    attachments: list[Attachment] = Field(default_factory=list)
59    reply_to: str | None = None  # Parent message ID
60    thread_id: str | None = None  # Thread context
61    timestamp: datetime = Field(default_factory=datetime.utcnow)
62    raw: dict[str, Any] = Field(default_factory=dict)  # Original platform payload

65class OutboundMessage(BaseModel):
66    """Message to send to a channel."""
67
68    content: str  # Markdown text
69    attachments: list[Attachment] = Field(default_factory=list)
70    reply_to: str | None = None
71    thread_id: str | None = None
72    metadata: dict[str, Any] = Field(default_factory=dict)

75class ChannelCapabilities(BaseModel):
76    """What a channel can do."""
77
78    chat_types: list[str] = Field(default_factory=lambda: ["direct"])
79    supports_streaming: bool = False
80    supports_reactions: bool = False
81    supports_media: bool = True
82    supports_threads: bool = False
83    supports_editing: bool = False
84    text_chunk_limit: int = 4000  # Max characters per message

87class ChannelContext(BaseModel):
88    """Context for sending messages."""
89
90    channel_id: str
91    thread_id: str | None = None
92    reply_to: str | None = None
93    metadata: dict[str, Any] = Field(default_factory=dict)

 96class ChatContext(BaseModel):
 97    """Context for permission resolution."""
 98
 99    channel: str
100    peer_id: str
101    group_id: str | None = None
102    thread_id: str | None = None
103    is_dm: bool = True

106@runtime_checkable
107class ChannelDock(Protocol):
108    """Abstract interface for messaging channel adapters."""
109
110    @property
111    def name(self) -> str:
112        """Channel identifier (e.g., 'telegram', 'discord')."""
113        ...
114
115    @property
116    def capabilities(self) -> ChannelCapabilities:
117        """What this channel supports."""
118        ...
119
120    async def start(self) -> None:
121        """Start listening for messages."""
122        ...
123
124    async def stop(self) -> None:
125        """Stop the channel adapter."""
126        ...
127
128    async def normalize_inbound(self, raw_event: Any) -> InboundMessage:
129        """Convert platform-specific event to standard format."""
130        ...
131
132    async def send_outbound(
133        self,
134        message: OutboundMessage,
135        context: ChannelContext,
136    ) -> None:
137        """Deliver message to the platform."""
138        ...
139
140    def build_session_key(self, message: InboundMessage) -> str:
141        """Build a session key for the message context."""
142        ...
143
144    def resolve_allow_from(self, context: ChatContext) -> list[str]:
145        """Determine allowed user IDs for this context."""
146        ...

 30class ChannelBridge:
 31    """
 32    Bridges a channel adapter to the core message pipeline.
 33
 34    Responsibilities:
 35    - Receives raw events from channel adapter
 36    - Normalizes to pipeline InboundMessage
 37    - Routes through InboundPipeline
 38    - Handles outbound delivery through OutboundPipeline
 39
 40    Each channel gets its own bridge instance, allowing channel-specific
 41    middleware configuration.
 42    """
 43
 44    def __init__(
 45        self,
 46        channel: ChannelDock,
 47        context: AppContext,
 48        inbound_pipeline: InboundPipeline | None = None,
 49        outbound_pipeline: OutboundPipeline | None = None,
 50    ) -> None:
 51        """
 52        Initialize the channel bridge.
 53
 54        Args:
 55            channel: The channel adapter to bridge.
 56            context: Application context.
 57            inbound_pipeline: Optional custom inbound pipeline.
 58            outbound_pipeline: Optional custom outbound pipeline.
 59        """
 60        self.channel = channel
 61        self.context = context
 62        self._channel_name = channel.name
 63
 64        # Use provided pipelines or create defaults
 65        self.inbound = inbound_pipeline or InboundPipeline()
 66        self.outbound = outbound_pipeline or OutboundPipeline()
 67
 68        # Set default handlers if not configured
 69        if self.inbound._handler is None:
 70            self.inbound.set_handler(self._default_inbound_handler)
 71        if self.outbound._handler is None:
 72            self.outbound.set_handler(self._default_outbound_handler)
 73
 74    @property
 75    def name(self) -> str:
 76        """Channel name."""
 77        return self._channel_name
 78
 79    @property
 80    def capabilities(self) -> ChannelCapabilities:
 81        """Channel capabilities."""
 82        return self.channel.capabilities
 83
 84    async def on_raw_event(self, raw_event: dict[str, Any]) -> None:
 85        """
 86        Handle a raw event from the channel adapter.
 87
 88        This is the entry point for all inbound messages from the channel.
 89        Normalizes the event and routes through the inbound pipeline.
 90
 91        Args:
 92            raw_event: Raw event data from the channel.
 93        """
 94        try:
 95            # Normalize using channel adapter
 96            channel_msg = await self.channel.normalize_inbound(raw_event)
 97
 98            # Convert to pipeline message format
 99            message = InboundMessage(
100                content=channel_msg.content,
101                peer_id=channel_msg.sender.id,
102                channel_name=self._channel_name,
103                message_id=channel_msg.id,
104                group_id=channel_msg.thread_id,
105                reply_to=channel_msg.reply_to,
106                attachments=[a.url for a in channel_msg.attachments],
107                raw=channel_msg.raw,
108            )
109
110            # Build session key using channel's logic
111            session_key = self.channel.build_session_key(channel_msg)
112
113            # Create pipeline context
114            ctx = PipelineContext(
115                app=self.context,
116                channel_name=self._channel_name,
117                peer_id=channel_msg.sender.id,
118                group_id=channel_msg.thread_id,
119                session_key=session_key,
120                metadata={
121                    "chunk_size": self.channel.capabilities.text_chunk_limit,
122                },
123            )
124
125            # Process through inbound pipeline
126            await self.inbound.process(message, ctx)
127
128        except Exception as e:
129            await logger.aerror(
130                "Failed to process inbound event",
131                channel=self._channel_name,
132                error=str(e),
133            )
134            raise
135
136    async def send(
137        self,
138        message: OutboundMessage,
139        context: PipelineContext | None = None,
140    ) -> None:
141        """
142        Send a message through the outbound pipeline.
143
144        The pipeline will handle chunking, rate limiting, etc.
145
146        Args:
147            message: The message to send.
148            context: Optional pipeline context. Creates one if not provided.
149        """
150        if context is None:
151            context = PipelineContext(
152                app=self.context,
153                channel_name=self._channel_name,
154                peer_id=message.peer_id,
155                metadata={
156                    "chunk_size": self.channel.capabilities.text_chunk_limit,
157                },
158            )
159
160        await self.outbound.process(message, context)
161
162    async def deliver(
163        self,
164        message: OutboundMessage,
165        context: PipelineContext,
166    ) -> None:
167        """
168        Deliver message directly to channel (bypassing pipeline).
169
170        Use send() instead for normal message sending.
171
172        Args:
173            message: The message to deliver.
174            context: Pipeline context.
175        """
176        await self._default_outbound_handler(message, context)
177
178    async def _default_inbound_handler(
179        self,
180        message: InboundMessage,
181        context: PipelineContext,
182    ) -> None:
183        """
184        Default inbound message handler.
185
186        Routes the message to the agent runner for processing.
187        """
188        from tiphys.agent.runner import AgentRunner
189
190        runner = AgentRunner(context=self.context)
191
192        # Stream response (blocks until complete)
193        async for _block in runner.stream(
194            message=message.content,
195            session_key=context.session_key or f"{self._channel_name}:default",
196        ):
197            pass  # Blocks are emitted via event_bus
198
199    async def _default_outbound_handler(
200        self,
201        message: OutboundMessage,
202        context: PipelineContext,
203    ) -> None:
204        """
205        Default outbound message handler.
206
207        Delivers the message to the channel adapter.
208        """
209        from tiphys.channels.base import ChannelContext
210        from tiphys.channels.base import OutboundMessage as ChannelOutboundMessage
211
212        # Convert pipeline message to channel format
213        channel_msg = ChannelOutboundMessage(
214            content=message.content,
215            reply_to=message.reply_to,
216            thread_id=message.thread_id,
217            metadata=message.metadata,
218        )
219
220        # Build channel context
221        channel_ctx = ChannelContext(
222            channel_id=message.peer_id or "",
223            thread_id=message.thread_id,
224            reply_to=message.reply_to,
225        )
226
227        await self.channel.send_outbound(channel_msg, channel_ctx)
228
229    async def start(self) -> None:
230        """Start the underlying channel adapter."""
231        await self.channel.start()
232
233    async def stop(self) -> None:
234        """Stop the underlying channel adapter."""
235        await self.channel.stop()

238class BridgeManager:
239    """
240    Manages all channel bridges.
241
242    Provides a central registry for channel bridges and coordinates
243    lifecycle and message routing across channels.
244    """
245
246    def __init__(self, context: AppContext) -> None:
247        """
248        Initialize the bridge manager.
249
250        Args:
251            context: Application context.
252        """
253        self.context = context
254        self._bridges: dict[str, ChannelBridge] = {}
255
256    def register(self, bridge: ChannelBridge) -> None:
257        """
258        Register a channel bridge.
259
260        Args:
261            bridge: The bridge to register.
262
263        Raises:
264            ValueError: If a bridge with the same name is already registered.
265        """
266        if bridge.name in self._bridges:
267            raise ValueError(f"Bridge '{bridge.name}' is already registered")
268        self._bridges[bridge.name] = bridge
269
270    def unregister(self, channel_name: str) -> bool:
271        """
272        Unregister a channel bridge.
273
274        Args:
275            channel_name: Name of the channel to unregister.
276
277        Returns:
278            True if unregistered, False if not found.
279        """
280        if channel_name in self._bridges:
281            del self._bridges[channel_name]
282            return True
283        return False
284
285    def get(self, channel_name: str) -> ChannelBridge | None:
286        """
287        Get a bridge by channel name.
288
289        Args:
290            channel_name: The channel name.
291
292        Returns:
293            The bridge, or None if not found.
294        """
295        return self._bridges.get(channel_name)
296
297    def list_channels(self) -> list[str]:
298        """List all registered channel names."""
299        return list(self._bridges.keys())
300
301    def get_all(self) -> dict[str, ChannelBridge]:
302        """Get all registered bridges."""
303        return self._bridges.copy()
304
305    async def broadcast(self, message: OutboundMessage) -> dict[str, bool]:
306        """
307        Send a message to all channels.
308
309        Args:
310            message: The message to broadcast.
311
312        Returns:
313            Dict mapping channel names to success status.
314        """
315        results: dict[str, bool] = {}
316
317        for name, bridge in self._bridges.items():
318            try:
319                await bridge.send(message)
320                results[name] = True
321            except Exception as e:
322                await logger.aerror(
323                    "Broadcast failed",
324                    channel=name,
325                    error=str(e),
326                )
327                results[name] = False
328
329        return results
330
331    async def start_all(self) -> None:
332        """Start all registered channel bridges."""
333        for name, bridge in self._bridges.items():
334            try:
335                await bridge.start()
336                await logger.ainfo("Bridge started", channel=name)
337            except Exception as e:
338                await logger.aerror(
339                    "Failed to start bridge",
340                    channel=name,
341                    error=str(e),
342                )
343
344    async def stop_all(self) -> None:
345        """Stop all registered channel bridges."""
346        for name, bridge in self._bridges.items():
347            try:
348                await bridge.stop()
349                await logger.ainfo("Bridge stopped", channel=name)
350            except Exception as e:
351                await logger.aerror(
352                    "Failed to stop bridge",
353                    channel=name,
354                    error=str(e),
355                )
356
357    def clear(self) -> int:
358        """
359        Clear all bridges.
360
361        Returns:
362            Number of bridges cleared.
363        """
364        count = len(self._bridges)
365        self._bridges.clear()
366        return count

369def create_bridge(
370    channel: ChannelDock,
371    context: AppContext,
372    *,
373    use_default_middleware: bool = True,
374) -> ChannelBridge:
375    """
376    Factory function to create a channel bridge with default configuration.
377
378    Args:
379        channel: The channel adapter.
380        context: Application context.
381        use_default_middleware: Whether to add default middleware.
382
383    Returns:
384        Configured ChannelBridge.
385    """
386    from tiphys.pipeline import (
387        ChunkingMiddleware,
388        LoggingMiddleware,
389        RateLimitMiddleware,
390        TransformMiddleware,
391        ValidationMiddleware,
392    )
393
394    # Create pipelines
395    inbound = InboundPipeline()
396    outbound = OutboundPipeline()
397
398    if use_default_middleware:
399        # Add default inbound middleware
400        inbound.use(LoggingMiddleware())
401        inbound.use(TransformMiddleware())
402        inbound.use(ValidationMiddleware())
403        inbound.use(RateLimitMiddleware())
404
405        # Add default outbound middleware
406        outbound.use(LoggingMiddleware())
407        outbound.use(ChunkingMiddleware(default_chunk_size=channel.capabilities.text_chunk_limit))
408
409    return ChannelBridge(
410        channel=channel,
411        context=context,
412        inbound_pipeline=inbound,
413        outbound_pipeline=outbound,
414    )

 15class ChannelRegistry:
 16    """
 17    Registry for channel adapters.
 18
 19    Manages registration, discovery, and lifecycle of channel implementations.
 20    """
 21
 22    def __init__(self) -> None:
 23        """Initialize the channel registry."""
 24        self._channels: dict[str, ChannelDock] = {}
 25        self._started: set[str] = set()
 26
 27    def register(self, channel: ChannelDock) -> None:
 28        """
 29        Register a channel adapter.
 30
 31        Args:
 32            channel: The channel to register.
 33
 34        Raises:
 35            ValueError: If a channel with the same name is already registered.
 36        """
 37        name = channel.name
 38        if name in self._channels:
 39            raise ValueError(f"Channel '{name}' is already registered")
 40
 41        self._channels[name] = channel
 42
 43    def unregister(self, name: str) -> bool:
 44        """
 45        Unregister a channel adapter.
 46
 47        Args:
 48            name: Name of the channel to unregister.
 49
 50        Returns:
 51            True if the channel was unregistered, False if not found.
 52        """
 53        if name in self._channels:
 54            del self._channels[name]
 55            self._started.discard(name)
 56            return True
 57        return False
 58
 59    def get(self, name: str) -> ChannelDock | None:
 60        """
 61        Get a channel by name.
 62
 63        Args:
 64            name: Name of the channel.
 65
 66        Returns:
 67            The channel adapter or None if not found.
 68        """
 69        return self._channels.get(name)
 70
 71    def list_names(self) -> list[str]:
 72        """List all registered channel names."""
 73        return list(self._channels.keys())
 74
 75    def get_all(self) -> dict[str, ChannelDock]:
 76        """Get all registered channels."""
 77        return self._channels.copy()
 78
 79    async def start_all(self) -> None:
 80        """Start all registered channels."""
 81        for name, channel in self._channels.items():
 82            if name not in self._started:
 83                try:
 84                    await channel.start()
 85                    self._started.add(name)
 86                    await logger.ainfo("Channel started", channel=name)
 87                except Exception as e:
 88                    await logger.aerror(
 89                        "Failed to start channel",
 90                        channel=name,
 91                        error=str(e),
 92                    )
 93
 94    async def stop_all(self) -> None:
 95        """Stop all running channels."""
 96        for name in list(self._started):
 97            channel = self._channels.get(name)
 98            if channel:
 99                try:
100                    await channel.stop()
101                    self._started.discard(name)
102                    await logger.ainfo("Channel stopped", channel=name)
103                except Exception as e:
104                    await logger.aerror(
105                        "Failed to stop channel",
106                        channel=name,
107                        error=str(e),
108                    )
109
110    async def start(self, name: str) -> bool:
111        """
112        Start a specific channel.
113
114        Args:
115            name: Name of the channel to start.
116
117        Returns:
118            True if started, False if channel not found.
119        """
120        channel = self._channels.get(name)
121        if not channel:
122            return False
123
124        if name not in self._started:
125            await channel.start()
126            self._started.add(name)
127
128        return True
129
130    async def stop(self, name: str) -> bool:
131        """
132        Stop a specific channel.
133
134        Args:
135            name: Name of the channel to stop.
136
137        Returns:
138            True if stopped, False if channel not found.
139        """
140        channel = self._channels.get(name)
141        if not channel:
142            return False
143
144        if name in self._started:
145            await channel.stop()
146            self._started.discard(name)
147
148        return True
149
150    def is_running(self, name: str) -> bool:
151        """Check if a channel is running."""
152        return name in self._started
153
154    def clear(self) -> int:
155        """
156        Clear all channels.
157
158        Returns:
159            Number of channels cleared.
160        """
161        count = len(self._channels)
162        self._channels.clear()
163        self._started.clear()
164        return count