tiphys.pipeline

103class InboundMiddleware(ABC):
104    """
105    Middleware for processing inbound messages.
106
107    Implement this to add cross-cutting concerns like validation,
108    rate limiting, or logging to the inbound message flow.
109    """
110
111    @abstractmethod
112    async def process(
113        self,
114        message: InboundMessage,
115        context: PipelineContext,
116        next_handler: InboundHandler,
117    ) -> None:
118        """
119        Process the message and call next handler.
120
121        Args:
122            message: The inbound message to process.
123            context: Pipeline context with app and metadata.
124            next_handler: Callback to invoke the next middleware or final handler.
125        """
126        ...

129class OutboundMiddleware(ABC):
130    """
131    Middleware for processing outbound messages.
132
133    Implement this to add cross-cutting concerns like chunking,
134    rate limiting, or logging to the outbound message flow.
135    """
136
137    @abstractmethod
138    async def process(
139        self,
140        message: OutboundMessage,
141        context: PipelineContext,
142        next_handler: OutboundHandler,
143    ) -> None:
144        """
145        Process the message and call next handler.
146
147        Args:
148            message: The outbound message to process.
149            context: Pipeline context with app and metadata.
150            next_handler: Callback to invoke the next middleware or final handler.
151        """
152        ...

155class InboundPipeline:
156    """
157    Pipeline for processing inbound messages through middleware chain.
158
159    Usage:
160        pipeline = InboundPipeline()
161        pipeline.use(ValidationMiddleware())
162        pipeline.use(RateLimitMiddleware())
163        pipeline.use(LoggingMiddleware())
164        pipeline.set_handler(process_message)
165
166        await pipeline.process(message, context)
167    """
168
169    def __init__(self) -> None:
170        """Initialize an empty pipeline."""
171        self._middleware: list[InboundMiddleware] = []
172        self._handler: InboundHandler | None = None
173
174    def use(self, middleware: InboundMiddleware) -> InboundPipeline:
175        """
176        Add middleware to the pipeline.
177
178        Args:
179            middleware: The middleware to add.
180
181        Returns:
182            Self for method chaining.
183        """
184        self._middleware.append(middleware)
185        return self
186
187    def set_handler(self, handler: InboundHandler) -> None:
188        """
189        Set the final handler.
190
191        Args:
192            handler: The function that processes the message after all middleware.
193        """
194        self._handler = handler
195
196    async def process(self, message: InboundMessage, context: PipelineContext) -> None:
197        """
198        Process a message through the pipeline.
199
200        Args:
201            message: The inbound message.
202            context: Pipeline context.
203
204        Raises:
205            RuntimeError: If no handler is set.
206        """
207        if self._handler is None:
208            raise RuntimeError("No handler set for inbound pipeline")
209
210        # Build the middleware chain recursively
211        async def run_chain(
212            idx: int,
213            msg: InboundMessage,
214            ctx: PipelineContext,
215        ) -> None:
216            if idx >= len(self._middleware):
217                # End of middleware chain, call final handler
218                await self._handler(msg, ctx)  # type: ignore[misc]
219            else:
220                # Call current middleware with next in chain
221                await self._middleware[idx].process(msg, ctx, lambda m, c: run_chain(idx + 1, m, c))
222
223        await run_chain(0, message, context)
224
225    @property
226    def middleware_count(self) -> int:
227        """Number of middleware in the pipeline."""
228        return len(self._middleware)

231class OutboundPipeline:
232    """
233    Pipeline for processing outbound messages through middleware chain.
234
235    Usage:
236        pipeline = OutboundPipeline()
237        pipeline.use(ChunkingMiddleware())
238        pipeline.use(RateLimitMiddleware())
239        pipeline.set_handler(deliver_message)
240
241        await pipeline.process(message, context)
242    """
243
244    def __init__(self) -> None:
245        """Initialize an empty pipeline."""
246        self._middleware: list[OutboundMiddleware] = []
247        self._handler: OutboundHandler | None = None
248
249    def use(self, middleware: OutboundMiddleware) -> OutboundPipeline:
250        """
251        Add middleware to the pipeline.
252
253        Args:
254            middleware: The middleware to add.
255
256        Returns:
257            Self for method chaining.
258        """
259        self._middleware.append(middleware)
260        return self
261
262    def set_handler(self, handler: OutboundHandler) -> None:
263        """
264        Set the final handler (delivery to channel).
265
266        Args:
267            handler: The function that delivers the message.
268        """
269        self._handler = handler
270
271    async def process(self, message: OutboundMessage, context: PipelineContext) -> None:
272        """
273        Process a message through the pipeline.
274
275        Args:
276            message: The outbound message.
277            context: Pipeline context.
278
279        Raises:
280            RuntimeError: If no handler is set.
281        """
282        if self._handler is None:
283            raise RuntimeError("No handler set for outbound pipeline")
284
285        # Build the middleware chain recursively
286        async def run_chain(
287            idx: int,
288            msg: OutboundMessage,
289            ctx: PipelineContext,
290        ) -> None:
291            if idx >= len(self._middleware):
292                # End of middleware chain, call final handler
293                await self._handler(msg, ctx)  # type: ignore[misc]
294            else:
295                # Call current middleware with next in chain
296                await self._middleware[idx].process(msg, ctx, lambda m, c: run_chain(idx + 1, m, c))
297
298        await run_chain(0, message, context)
299
300    @property
301    def middleware_count(self) -> int:
302        """Number of middleware in the pipeline."""
303        return len(self._middleware)

22@dataclass
23class PipelineContext:
24    """
25    Context passed through the pipeline.
26
27    Carries application context, channel info, and request metadata
28    through the entire middleware chain.
29    """
30
31    app: AppContext
32    channel_name: str
33    peer_id: str | None = None
34    group_id: str | None = None
35    session_key: str | None = None
36    metadata: dict[str, Any] = field(default_factory=dict)
37
38    # Correlation ID for request tracing
39    correlation_id: str = field(default_factory=lambda: str(uuid4())[:8])
40
41    def with_metadata(self, **kwargs: Any) -> PipelineContext:
42        """
43        Return a new context with additional metadata.
44
45        Args:
46            **kwargs: Metadata fields to add.
47
48        Returns:
49            New PipelineContext with merged metadata.
50        """
51        new_meta = {**self.metadata, **kwargs}
52        return PipelineContext(
53            app=self.app,
54            channel_name=self.channel_name,
55            peer_id=self.peer_id,
56            group_id=self.group_id,
57            session_key=self.session_key,
58            metadata=new_meta,
59            correlation_id=self.correlation_id,
60        )

63class InboundMessage(BaseModel):
64    """
65    Normalized inbound message.
66
67    Platform-agnostic representation of a message from any channel.
68    """
69
70    content: str
71    peer_id: str
72    channel_name: str
73    message_id: str = Field(default_factory=lambda: str(uuid4()))
74    group_id: str | None = None
75    reply_to: str | None = None
76    thread_id: str | None = None
77    attachments: list[str] = Field(default_factory=list)
78    raw: dict[str, Any] = Field(default_factory=dict)

81class OutboundMessage(BaseModel):
82    """
83    Normalized outbound message.
84
85    Platform-agnostic representation of a message to send.
86    """
87
88    content: str
89    channel_name: str
90    peer_id: str | None = None  # None = broadcast
91    group_id: str | None = None
92    reply_to: str | None = None
93    thread_id: str | None = None
94    priority: int = 0  # Higher = more urgent
95    metadata: dict[str, Any] = Field(default_factory=dict)

 89class ValidationMiddleware(InboundMiddleware):
 90    """
 91    Middleware that validates inbound messages.
 92
 93    Checks message length constraints and optionally sanitizes content.
 94    Drops messages that don't meet validation criteria.
 95
 96    Uses per-channel configuration when available (max_message_length).
 97    """
 98
 99    def __init__(
100        self,
101        max_length: int = 10000,
102        min_length: int = 1,
103        strip_whitespace: bool = True,
104        use_channel_config: bool = True,
105    ) -> None:
106        """
107        Initialize validation middleware.
108
109        Args:
110            max_length: Default maximum allowed message length.
111            min_length: Minimum required message length.
112            strip_whitespace: Whether to strip leading/trailing whitespace.
113            use_channel_config: Whether to read max_length from channel config.
114        """
115        self.max_length = max_length
116        self.min_length = min_length
117        self.strip_whitespace = strip_whitespace
118        self.use_channel_config = use_channel_config
119
120    def _get_max_length(self, context: PipelineContext) -> int:
121        """Get max message length from channel config or default."""
122        if self.use_channel_config and context.app:
123            try:
124                channel_config = context.app.config.get_channel_config(context.channel_name)
125                return channel_config.max_message_length
126            except (AttributeError, KeyError):
127                pass
128        return self.max_length
129
130    async def process(
131        self,
132        message: InboundMessage,
133        context: PipelineContext,
134        next_handler: InboundHandler,
135    ) -> None:
136        """Validate and sanitize the message."""
137        content = message.content
138        if self.strip_whitespace:
139            content = content.strip()
140
141        # Check minimum length
142        if len(content) < self.min_length:
143            await logger.awarning(
144                "Message too short, dropping",
145                correlation_id=context.correlation_id,
146                length=len(content),
147                min_length=self.min_length,
148            )
149            return
150
151        # Get max length from channel config or use default
152        max_length = self._get_max_length(context)
153
154        # Check maximum length
155        if len(content) > max_length:
156            await logger.awarning(
157                "Message too long, truncating",
158                correlation_id=context.correlation_id,
159                original_length=len(content),
160                max_length=max_length,
161            )
162            content = content[:max_length]
163
164        # Create validated message
165        validated = message.model_copy(update={"content": content})
166        await next_handler(validated, context)

169class RateLimitMiddleware(InboundMiddleware):
170    """
171    Token bucket rate limiter per peer.
172
173    Limits the rate of inbound messages to prevent abuse.
174    Messages exceeding the rate limit are dropped with a warning.
175
176    Uses per-channel configuration when available (rate_limit_requests_per_minute,
177    rate_limit_burst).
178    """
179
180    def __init__(
181        self,
182        requests_per_minute: int = 20,
183        burst_size: int = 5,
184        use_channel_config: bool = True,
185    ) -> None:
186        """
187        Initialize rate limiter.
188
189        Args:
190            requests_per_minute: Default sustained request rate.
191            burst_size: Default maximum burst size (token bucket capacity).
192            use_channel_config: Whether to read rate limits from channel config.
193        """
194        self.default_rate = requests_per_minute / 60.0  # Tokens per second
195        self.default_burst = burst_size
196        self.use_channel_config = use_channel_config
197        # bucket key -> (tokens, last_update_time)
198        self._buckets: dict[str, tuple[float, float]] = defaultdict(
199            lambda: (float(burst_size), time.monotonic())
200        )
201
202    def _get_rate_config(self, context: PipelineContext) -> tuple[float, int]:
203        """Get rate limit config from channel or defaults."""
204        if self.use_channel_config and context.app:
205            try:
206                channel_config = context.app.config.get_channel_config(context.channel_name)
207                rate = channel_config.rate_limit_requests_per_minute / 60.0
208                burst = channel_config.rate_limit_burst
209                return rate, burst
210            except (AttributeError, KeyError):
211                pass
212        return self.default_rate, self.default_burst
213
214    async def process(
215        self,
216        message: InboundMessage,
217        context: PipelineContext,
218        next_handler: InboundHandler,
219    ) -> None:
220        """Check rate limit and process if allowed."""
221        # Get rate config from channel or defaults
222        rate, burst = self._get_rate_config(context)
223
224        # Build unique key for rate limiting
225        key = f"{context.channel_name}:{context.peer_id or 'anonymous'}"
226
227        tokens, last_update = self._buckets[key]
228
229        # Refill tokens based on elapsed time
230        now = time.monotonic()
231        elapsed = now - last_update
232        tokens = min(burst, tokens + elapsed * rate)
233
234        if tokens < 1:
235            await logger.awarning(
236                "Rate limit exceeded",
237                correlation_id=context.correlation_id,
238                peer_id=context.peer_id,
239                channel=context.channel_name,
240            )
241            # Drop the message
242            return
243
244        # Consume one token
245        self._buckets[key] = (tokens - 1, now)
246        await next_handler(message, context)
247
248    def reset(self, key: str | None = None) -> None:
249        """
250        Reset rate limit state.
251
252        Args:
253            key: Specific key to reset, or None to reset all.
254        """
255        if key is None:
256            self._buckets.clear()
257        else:
258            self._buckets.pop(key, None)

32class LoggingMiddleware(InboundMiddleware, OutboundMiddleware):
33    """
34    Middleware that logs messages with context.
35
36    Logs message processing start/end with timing information.
37    Can be used for both inbound and outbound pipelines.
38    """
39
40    def __init__(self, log_content: bool = False) -> None:
41        """
42        Initialize logging middleware.
43
44        Args:
45            log_content: Whether to log message content (may be verbose).
46        """
47        self.log_content = log_content
48
49    async def process(
50        self,
51        message: InboundMessage | OutboundMessage,
52        context: PipelineContext,
53        next_handler: InboundHandler | OutboundHandler,
54    ) -> None:
55        """Process message with logging."""
56        start = time.monotonic()
57
58        # Determine direction
59        direction = "inbound" if isinstance(message, InboundMessage) else "outbound"
60
61        # Build log context
62        log_data: dict[str, Any] = {
63            "correlation_id": context.correlation_id,
64            "channel": context.channel_name,
65            "direction": direction,
66            "content_length": len(message.content),
67        }
68
69        if context.peer_id:
70            log_data["peer_id"] = context.peer_id
71
72        if self.log_content:
73            # Truncate for logging
74            log_data["content_preview"] = message.content[:100]
75
76        await logger.ainfo(f"Processing {direction} message", **log_data)
77
78        try:
79            await next_handler(message, context)  # type: ignore[arg-type]
80        finally:
81            elapsed_ms = (time.monotonic() - start) * 1000
82            await logger.ainfo(
83                f"Completed {direction} message",
84                correlation_id=context.correlation_id,
85                elapsed_ms=round(elapsed_ms, 2),
86            )

261class ChunkingMiddleware(OutboundMiddleware):
262    """
263    Middleware that chunks long messages based on channel limits.
264
265    Splits messages at natural boundaries (paragraphs, sentences)
266    to respect platform message size limits.
267
268    Uses per-channel configuration when available (chunk_size).
269    """
270
271    def __init__(
272        self,
273        default_chunk_size: int = 2000,
274        use_channel_config: bool = True,
275    ) -> None:
276        """
277        Initialize chunking middleware.
278
279        Args:
280            default_chunk_size: Default maximum characters per chunk.
281            use_channel_config: Whether to read chunk_size from channel config.
282        """
283        self.default_chunk_size = default_chunk_size
284        self.use_channel_config = use_channel_config
285
286    def _get_chunk_size(self, context: PipelineContext) -> int:
287        """Get chunk size from metadata, channel config, or default."""
288        # First check context metadata (highest priority - set by bridge)
289        if "chunk_size" in context.metadata:
290            return int(context.metadata["chunk_size"])
291
292        # Then check channel config
293        if self.use_channel_config and context.app:
294            try:
295                channel_config = context.app.config.get_channel_config(context.channel_name)
296                return channel_config.chunk_size
297            except (AttributeError, KeyError):
298                pass
299
300        return self.default_chunk_size
301
302    async def process(
303        self,
304        message: OutboundMessage,
305        context: PipelineContext,
306        next_handler: OutboundHandler,
307    ) -> None:
308        """Chunk message if necessary and process each chunk."""
309        # Get channel-specific chunk size
310        chunk_size = self._get_chunk_size(context)
311        content = message.content
312
313        # If content fits, pass through
314        if len(content) <= chunk_size:
315            await next_handler(message, context)
316            return
317
318        # Split into chunks
319        chunks = self._split_content(content, chunk_size)
320
321        # Send each chunk
322        for i, chunk in enumerate(chunks):
323            chunk_msg = message.model_copy(update={"content": chunk})
324            chunk_ctx = context.with_metadata(
325                chunk_index=i,
326                total_chunks=len(chunks),
327                is_chunked=True,
328            )
329            await next_handler(chunk_msg, chunk_ctx)
330
331    def _split_content(self, content: str, max_size: int) -> list[str]:
332        """
333        Split content into chunks at natural boundaries.
334
335        Tries to split at paragraph breaks, then line breaks, then spaces.
336        """
337        chunks: list[str] = []
338
339        while content:
340            if len(content) <= max_size:
341                chunks.append(content)
342                break
343
344            # Find best split point
345            split_point = self._find_split_point(content, max_size)
346            chunks.append(content[:split_point].strip())
347            content = content[split_point:].strip()
348
349        return chunks
350
351    def _find_split_point(self, content: str, max_size: int) -> int:
352        """Find the best point to split content."""
353        # Try paragraph boundary first
354        split_point = content.rfind("\n\n", 0, max_size)
355        if split_point > 0:
356            return split_point
357
358        # Try line break
359        split_point = content.rfind("\n", 0, max_size)
360        if split_point > 0:
361            return split_point
362
363        # Try space
364        split_point = content.rfind(" ", 0, max_size)
365        if split_point > 0:
366            return split_point
367
368        # Force split at max_size
369        return max_size

372class ContextLoggingMiddleware(InboundMiddleware, OutboundMiddleware):
373    """
374    Middleware that sets up structured logging context for the request.
375
376    Should be placed early in the pipeline to ensure all subsequent
377    logging has proper context (correlation_id, channel, peer_id, etc.).
378
379    Example:
380        pipeline = InboundPipeline()
381        pipeline.use(ContextLoggingMiddleware())  # Add early in chain
382        pipeline.use(ValidationMiddleware())
383        pipeline.use(RateLimitMiddleware())
384    """
385
386    async def process(
387        self,
388        message: InboundMessage | OutboundMessage,
389        context: PipelineContext,
390        next_handler: InboundHandler | OutboundHandler,
391    ) -> None:
392        """Set up logging context and process."""
393        from tiphys.logging import LogContext, clear_log_context, set_log_context
394
395        # Extract message_id if available (inbound messages have it)
396        message_id = getattr(message, "message_id", None)
397
398        log_ctx = LogContext(
399            correlation_id=context.correlation_id,
400            channel_name=context.channel_name,
401            peer_id=context.peer_id,
402            session_key=context.session_key,
403            agent_id=context.app.config.agent_id if context.app else None,
404            message_id=message_id,
405        )
406        set_log_context(log_ctx)
407
408        try:
409            await next_handler(message, context)  # type: ignore[arg-type]
410        finally:
411            # Clear context after processing to avoid leaks between requests
412            clear_log_context()