feat: message aggregator

[wangcham]

概述 / Overview

请在此部分填写你实现/解决/优化的内容:
Summary of what you implemented/solved/optimized:

添加防抖动机制

更改前后对比截图 / Screenshots

请在此部分粘贴更改前后对比截图（可以是界面截图、控制台输出、对话截图等）:
Please paste the screenshots of changes before and after here (can be interface screenshots, console output, conversation screenshots, etc.):

修改前 / Before:

修改后 / After:

检查清单 / Checklist

PR 作者完成 / For PR author

请在方括号间写x以打勾 / Please tick the box with x

• 阅读仓库贡献指引了吗？ / Have you read the contribution guide?
• 与项目所有者沟通过了吗？ / Have you communicated with the project maintainer?
• 我确定已自行测试所作的更改，确保功能符合预期。 / I have tested the changes and ensured they work as expected.

项目维护者完成 / For project maintainer

• 相关 issues 链接了吗？ / Have you linked the related issues?
• 配置项写好了吗？迁移写好了吗？生效了吗？ / Have you written the configuration items? Have you written the migration? Has it taken effect?
• 依赖加到 pyproject.toml 和 core/bootutils/deps.py 了吗 / Have you added the dependencies to pyproject.toml and core/bootutils/deps.py?
• 文档编写了吗？ / Have you written the documentation?

[Copilot]

Pull request overview

This PR introduces a message aggregation (debounce) mechanism in the trigger stage so that consecutive messages can be buffered briefly and merged before being sent into the pipeline for processing.

Changes:

• Add new message-aggregation trigger stage metadata and default config (enabled, delay).
• Route incoming friend/group messages through a new MessageAggregator instead of directly into QueryPool.
• Initialize and attach the aggregator to the application during boot.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/langbot/templates/metadata/pipeline/trigger.yaml	Adds pipeline trigger metadata/config schema for message aggregation.
src/langbot/templates/default-pipeline-config.json	Provides default values for the new trigger config.
src/langbot/pkg/platform/botmgr.py	Switches message ingestion from query_pool to msg_aggregator.
src/langbot/pkg/pipeline/aggregator.py	Implements buffering, delayed flush, and message merging logic.
src/langbot/pkg/core/stages/build_app.py	Instantiates and attaches MessageAggregator to the app.
src/langbot/pkg/core/app.py	Adds msg_aggregator to the application context.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

delay = ... float(delay) will raise ValueError if the pipeline config contains a non-numeric value (e.g. user-edited JSON/YAML, corrupted DB value), which would break message processing. Consider wrapping the cast in a try/except and falling back to default_delay before clamping.

Suggested change

	delay = aggregation_config.get('delay', default_delay)
	# Clamp delay to valid range
	delay = max(1.0, min(10.0, float(delay)))
	delay_raw = aggregation_config.get('delay', default_delay)
	try:
	delay = float(delay_raw)
	except (TypeError, ValueError):
	delay = default_delay
	# Clamp delay to valid range
	delay = max(1.0, min(10.0, delay))

[Copilot]

_merge_messages() mutates message_event.message_chain to the merged chain. Several adapters rely on message_source.message_chain.message_id when replying/quoting; replacing the chain risks losing the original message_id and other event metadata. Since QueryPool.add_query() already takes message_chain separately, keep message_event unmodified (or choose a specific event to keep for reply reference) and only pass the merged chain via the message_chain field.

Suggested change

	# Create merged message with updated chain
	# We need to update the message_event's message_chain as well
	merged_event = base_msg.message_event
	merged_event.message_chain = merged_chain
	return PendingMessage(
	bot_uuid=base_msg.bot_uuid,
	launcher_type=base_msg.launcher_type,
	launcher_id=base_msg.launcher_id,
	sender_id=base_msg.sender_id,
	message_event=merged_event,
	# Create merged message using the original event (do not mutate its chain)
	return PendingMessage(
	bot_uuid=base_msg.bot_uuid,
	launcher_type=base_msg.launcher_type,
	launcher_id=base_msg.launcher_id,
	sender_id=base_msg.sender_id,
	message_event=base_msg.message_event,

[Copilot]

The new aggregation/debounce behavior is non-trivial (buffering, keying, merge semantics). Add unit tests to cover: (1) messages within the delay window are merged, (2) different senders in the same group are not merged, and (3) reply-critical message_event/message_id stays intact when merging.

[Copilot]

_get_session_id() currently keys buffers only by bot_uuid, launcher_type, and launcher_id. For group messages this means messages from different senders in the same group can be merged together, and the merged query will also keep sender_id from the first buffered message (misattributing content). Include sender_id in the buffer/session key (at least for group chats) so aggregation matches the “same user” behavior described in the metadata and avoids cross-user merges.

Suggested change

	) -> str:
	"""Generate a unique session ID"""
	sender_id: typing.Union[int, str, None] = None,
	) -> str:
	"""Generate a unique session ID
	For group or multi-user contexts, callers SHOULD supply ``sender_id``
	so that aggregation is scoped per sender and avoids cross-user merges.
	If ``sender_id`` is omitted, the session ID falls back to the
	historical behavior and does not distinguish between senders.
	"""
	# Include sender_id in the session key when provided to avoid
	# aggregating messages from different users in the same launcher
	if sender_id is not None:
	return f'{bot_uuid}:{launcher_type.value}:{launcher_id}:{sender_id}'

[RockChinQ]