Feat/safty control

flocks/bus/bus.py

@@ -95,15 +95,6 @@ async def publish(
         # Create payload
         payload = EventPayload(definition.type, properties)
 
-        # Run hook pipeline (event stage)
-        try:
-            from flocks.hooks.pipeline import HookPipeline
-            ctx = await HookPipeline.run_event(payload.to_dict())
-            if ctx and isinstance(ctx.input, dict) and "type" in ctx.input and "properties" in ctx.input:
-                payload = EventPayload(ctx.input["type"], ctx.input["properties"])
-        except Exception as exc:
-            log.error("bus.hook.error", {"type": definition.type, "error": str(exc)})
-
         log.info("bus.publishing", {"type": definition.type})
 
         # Collect callbacks for this event type and wildcard ("*")

flocks/channel/base.py

@@ -243,6 +243,20 @@ async def handle_webhook(
             f"Channel '{self.meta().id}' does not support webhook mode"
         )
 
+    async def webhook_authentication_evidence(
+        self,
+        body: bytes,
+        headers: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Return opaque verification evidence before a public webhook effect.
+
+        The default reports no verification.  Flocks does not decide whether
+        this evidence is sufficient; extensions may apply their own policy at
+        the generic Channel webhook lifecycle boundary.
+        """
+        del body, headers
+        return {"plugin_authenticated": False}
+
     async def stop(self) -> None:
         """Stop listening and release resources."""
 

flocks/channel/builtin/feishu/channel.py

@@ -220,6 +220,38 @@ async def _ensure_webhook_dedup_ready(self, account_id: str, dedup) -> None:
         if task is None or task.done():
             self._webhook_dedup_flush_tasks[account_id] = await dedup.start_background_flush()
 
+    async def webhook_authentication_evidence(
+        self,
+        body: bytes,
+        headers: dict,
+    ) -> dict:
+        """Expose Feishu's existing verification result as neutral evidence."""
+        from flocks.channel.builtin.feishu.config import (
+            resolve_webhook_account_config,
+            verify_webhook_timestamp,
+        )
+
+        try:
+            data = json.loads(body)
+        except (TypeError, ValueError):
+            return {"plugin_authenticated": False, "provider": "feishu"}
+        if not isinstance(data, dict) or not verify_webhook_timestamp(headers):
+            return {"plugin_authenticated": False, "provider": "feishu"}
+        resolved = resolve_webhook_account_config(
+            self._config,
+            body=body,
+            headers=headers,
+            data=data,
+        )
+        if not resolved:
+            return {"plugin_authenticated": False, "provider": "feishu"}
+        return {
+            "plugin_authenticated": True,
+            "provider": "feishu",
+            # The concrete handler owns and applies event-id/nonce dedup.
+            "replay_protection": "plugin_dedup",
+        }
+
     async def handle_webhook(self, body: bytes, headers: dict) -> Optional[dict]:
         from flocks.channel.builtin.feishu.config import (
             build_webhook_replay_key,

flocks/channel/gateway/manager.py

@@ -14,6 +14,7 @@
 
 from flocks.channel.base import ChannelPlugin, ChannelStatus, NonRetryableChannelError
 from flocks.channel.inbound.dispatcher import InboundDispatcher
+from flocks.identity import mint_channel_ingress_provenance
 from flocks.channel.registry import ChannelRegistry, default_registry
 from flocks.utils.log import Log
 
@@ -344,10 +345,21 @@ def _make_on_message(
         plugin: ChannelPlugin,
         dispatch: Callable,
     ) -> Callable:
-        """Wrap *dispatch* so that each inbound message records a timestamp."""
+        """Mint neutral provenance before forwarding a gateway message."""
         async def _on_message(msg) -> None:
             plugin.record_message()
-            await dispatch(msg)
+            await dispatch(
+                msg,
+                provenance=mint_channel_ingress_provenance(
+                    channel_id=msg.channel_id,
+                    account_id=msg.account_id,
+                    message_id=msg.message_id,
+                    sender_id=msg.sender_id,
+                    chat_type=msg.chat_type.value,
+                    message=msg,
+                    evidence=msg.raw,
+                ),
+            )
         return _on_message
 
     def record_message(self, channel_id: str) -> None:

flocks/channel/inbound/dispatcher.py

@@ -22,6 +22,9 @@
     is_channel_media_placeholder,
 )
 from flocks.config.config import ChannelConfig
+from flocks.hooks.execution import execute_with_hooks
+from flocks.hooks.pipeline import HookPipeline
+from flocks.identity import ChannelIngressProvenance
 from flocks.utils.log import Log
 
 log = Log.create(service="channel.dispatcher")
@@ -274,6 +277,18 @@ def is_stale(self) -> bool:
 
 _channel_config_cache: dict[str, _CachedConfig] = {}
 
+
+def invalidate_channel_config_cache(channel_id: str | None = None) -> None:
+    """Invalidate cached channel config entries.
+
+    Args:
+        channel_id: When provided, drop only this channel. Otherwise clear all.
+    """
+    if channel_id is None:
+        _channel_config_cache.clear()
+        return
+    _channel_config_cache.pop(str(channel_id), None)
+
 _SESSION_LOCK_MAX = 5000
 
 
@@ -293,7 +308,39 @@ def __init__(self) -> None:
         self._session_locks: OrderedDict[str, asyncio.Lock] = OrderedDict()
         self._group_context: OrderedDict[str, deque[_GroupContextEntry]] = OrderedDict()
 
-    async def dispatch(self, msg: InboundMessage) -> None:
+    async def dispatch(
+        self,
+        msg: InboundMessage,
+        *,
+        provenance: ChannelIngressProvenance | None = None,
+    ) -> None:
+        ingress_payload = {
+            "operation": "channel.inbound.dispatch",
+            "transport": "channel",
+            "entry": "channel_webhook",
+            "channel_id": msg.channel_id,
+            "account_id": msg.account_id,
+            "chat_id": msg.chat_id,
+            "sender_id": msg.sender_id,
+            "chat_type": msg.chat_type.value,
+            "message_id": msg.message_id,
+            "authentication": {
+                "plugin_authenticated": bool(provenance is not None)
+            },
+            "provenance": provenance,
+            "message": msg,
+            "evidence": msg.raw,
+            "raw": msg.raw if isinstance(msg.raw, dict) else {},
+        }
+        return await execute_with_hooks(
+            ingress_payload,
+            lambda: self._dispatch(msg),
+            before=HookPipeline.run_ingress_before,
+            after=HookPipeline.run_ingress_after,
+        )
+
+    async def _dispatch(self, msg: InboundMessage) -> None:
+
         # 1. dedup
         if self.dedup.is_duplicate(msg.message_id):
             log.debug("dispatcher.dedup", {"message_id": msg.message_id})
@@ -314,8 +361,7 @@ async def dispatch(self, msg: InboundMessage) -> None:
 
         # 4. channel.inbound hook — allows plugins to inspect/block/modify
         try:
-            from flocks.hooks.pipeline import HookPipeline
-            hook_ctx = await HookPipeline.run_channel_inbound({
+            hook_ctx = await HookPipeline.run_channel_inbound_before({
                 "channel_id": msg.channel_id,
                 "sender_id": msg.sender_id,
                 "chat_id": msg.chat_id,
@@ -342,6 +388,7 @@ async def dispatch(self, msg: InboundMessage) -> None:
         # ``rex`` only via ``Agent.get(name) or Agent.get("rex")``, hiding the
         # real default and making behaviour diverge between WebUI and channel.
         default_agent = channel_config.default_agent
+        visible_agents = _resolve_visible_agents(channel_config)
         scope_override = None
         if msg.channel_id == "feishu" and msg.chat_type == ChatType.GROUP:
             scope_override, feishu_agent = _resolve_feishu_group_overrides(
@@ -358,10 +405,13 @@ async def dispatch(self, msg: InboundMessage) -> None:
                     "error": str(exc),
                 })
                 default_agent = "rex"
+        if visible_agents and default_agent not in visible_agents:
+            default_agent = visible_agents[0]
 
         binding = await self.binding_service.resolve_or_create(
             msg,
             default_agent=default_agent,
+            visible_agents=visible_agents,
             scope_override=scope_override,
             directory=channel_config.workspace_dir,
         )
@@ -898,6 +948,23 @@ async def _handle_session_command(
             **Session.inherited_model_kwargs(session),
             **owner_kwargs,
         )
+        try:
+            from flocks.session.execution_profile import (
+                upsert_session_execution_profile,
+            )
+
+            await upsert_session_execution_profile(
+                new_session.id,
+                patch={
+                    "entry": "channel",
+                    "channel_id": msg.channel_id,
+                    "account_id": msg.account_id,
+                    "default_agent": str(new_session.agent or "").strip(),
+                },
+                source="channel.command.new_session",
+            )
+        except Exception:
+            pass
         new_binding = await self.binding_service.rebind(
             msg,
             new_session.id,
@@ -1029,9 +1096,13 @@ async def _trigger_command_hook(action: str, session_id: str, context: dict[str,
             })
 
     @staticmethod
-    async def _get_channel_config(channel_id: str) -> ChannelConfig:
+    async def _get_channel_config(
+        channel_id: str,
+        *,
+        force_refresh: bool = False,
+    ) -> ChannelConfig:
         cached = _channel_config_cache.get(channel_id)
-        if cached and not cached.is_stale():
+        if cached and not cached.is_stale() and not force_refresh:
             return cached.config
         try:
             from flocks.config.config import Config
@@ -1040,6 +1111,8 @@ async def _get_channel_config(channel_id: str) -> ChannelConfig:
             _channel_config_cache[channel_id] = _CachedConfig(ch_cfg)
             return ch_cfg
         except Exception:
+            if cached is not None:
+                return cached.config
             return ChannelConfig()
 
     @staticmethod
@@ -1454,6 +1527,13 @@ def _resolve_feishu_group_overrides(
     return scope, agent
 
 
+def _resolve_visible_agents(channel_config: ChannelConfig) -> list[str]:
+    raw = channel_config.visible_agents
+    if not isinstance(raw, list):
+        return []
+    return [str(item).strip() for item in raw if str(item).strip()]
+
+
 async def _expand_merge_forward(
     message_id: str,
     channel_config: "ChannelConfig",