diff --git a/.env.example b/.env.example
index 1361aea..ad87368 100644
--- a/.env.example
+++ b/.env.example
@@ -81,6 +81,11 @@ AGENTMESH_FEISHU_INCLUDE_CONTENT=false
 # Opt-in coordinated employee start/result cards. Requires feishu_notifications.
 # To include short business summaries, also opt in to INCLUDE_CONTENT above.
 AGENTMESH_FEISHU_SYNC_COLLABORATION=false
+# Optional independent employee bot identities; mount the protected JSON in notifier only.
+AGENTMESH_FEISHU_EMPLOYEE_BOTS_ENABLED=false
+AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE=
+# Unmapped employees use the summary bot; mapped bot failures NEVER silently fall back.
+AGENTMESH_FEISHU_EMPLOYEE_BOT_FALLBACK=true
 AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1
 # Comma-separated trusted in-process extensions. Empty disables every installed extension.
 AGENTMESH_RUNTIME_EXTENSIONS=agentmesh.music-studio
diff --git a/.gitignore b/.gitignore
index d6a1051..58501b3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,6 +6,7 @@
 compose.public-demo.yaml
 *.pem
 *.key
+feishu-employee-bots.json
 
 # IDE and OS
 .idea/
diff --git a/compose.yaml b/compose.yaml
index 499bb24..d26f077 100644
--- a/compose.yaml
+++ b/compose.yaml
@@ -173,6 +173,9 @@ services:
       AGENTMESH_FEISHU_TASK_BASE_URL: ${AGENTMESH_FEISHU_TASK_BASE_URL:-}
       AGENTMESH_FEISHU_INCLUDE_CONTENT: ${AGENTMESH_FEISHU_INCLUDE_CONTENT:-false}
       AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS: ${AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS:-1}
+      AGENTMESH_FEISHU_EMPLOYEE_BOTS_ENABLED: ${AGENTMESH_FEISHU_EMPLOYEE_BOTS_ENABLED:-false}
+      AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE: ${AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE:-}
+      AGENTMESH_FEISHU_EMPLOYEE_BOT_FALLBACK: ${AGENTMESH_FEISHU_EMPLOYEE_BOT_FALLBACK:-true}
     depends_on:
       migrate:
         condition: service_completed_successfully
diff --git a/docs/integrations/feishu-notifications.md b/docs/integrations/feishu-notifications.md
index 9655a69..77fd084 100644
--- a/docs/integrations/feishu-notifications.md
+++ b/docs/integrations/feishu-notifications.md
@@ -65,6 +65,54 @@ business conclusions may use permitted knowledge. Use a controlled group and rev
 policy. Low-volume notifications normally arrive within seconds; queues, rate limits and retries
 can delay delivery. Replying in Feishu does not execute a command, teach memory or approve work.
 
+## Independent employee bot identities
+
+This optional mode uses a separate Feishu application bot per employee. An outgoing message
+cannot change the existing application's sender identity by setting a card title or avatar.
+Task/approval cards keep the original summary bot; collaboration cards use the exact pinned
+Run's `agent_id`, never its role label or model-provided output.
+
+1. Create distinct bot applications and authorize only the permissions you need. Feishu's
+   [official SDK app registration flow](https://github.com/larksuite/node-sdk#app-registration)
+   supports pre-filled names and explicit user confirmation. Use a minimal template with
+   `im:message:send_as_bot` for this outbound-only mode. App credentials must not appear in logs.
+2. Add each bot to the same controlled test group and check that it can speak. Registration
+   alone does not establish group membership. Adding bots via API needs separate group-member
+   management permission; the notification feature does not request that or read group history.
+3. Copy the [credential-file example](../../examples/feishu/employee-bots.example.json) to a
+   protected `feishu-employee-bots.json` outside Git and replace all placeholders. Bind the
+   deployment tenant and group exactly. `agent_id` is the execution identity recorded on Runs
+   (usually the employee's registered name), not an Agent Definition UUID or a work-item label.
+   No employee or App ID may be duplicated; the summary App ID must not be reused.
+4. On Linux, make the file readable only by its owner (`chmod 600`). On Windows, apply a
+   private file ACL. Mount it read-only into **only** the notifier, using a Compose override:
+
+   ```yaml
+   services:
+     feishu-notifier:
+       environment:
+         AGENTMESH_FEISHU_EMPLOYEE_BOTS_ENABLED: "true"
+         AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE: /run/agentmesh/feishu-employee-bots.json
+         AGENTMESH_FEISHU_EMPLOYEE_BOT_FALLBACK: "true"
+       volumes:
+         - ./secrets/feishu-employee-bots.json:/run/agentmesh/feishu-employee-bots.json:ro
+   ```
+
+5. Keep the parent notification gate and collaboration sync enabled, then restart the notifier
+   with this override. The identity option defaults off; malformed files fail startup with
+   sanitized errors. Files are limited to 64 KiB and 256 bindings. Each bot has an independent
+   token cache; delivery retains the existing notification UUID and shared send pacing.
+
+`EMPLOYEE_BOT_FALLBACK=true` allows unmapped employees to use the summary bot during gradual
+setup. Set it to `false` to withhold and retry those deliveries instead. A **mapped** bot's error
+never falls back to a different sender: missing group membership or revoked credentials stays
+a delivery failure, without affecting the Task. The existing bounded retry/dead-letter rules apply.
+
+Bindings are loaded once per notifier startup. Before remapping a bot, stop delivery and resolve
+its pending/retrying jobs, especially unknown send outcomes: Feishu deduplication can be scoped
+to the application, so changing App IDs during retries can cause duplicates. This slice does not
+persist sender-binding revisions, automate administrator approval or enable two-way bot chat.
+
 ## Delivery and operations
 
 The [live qualification record](../qualification/feishu-collaboration.md) documents a real
diff --git a/docs/integrations/feishu-notifications.zh-CN.md b/docs/integrations/feishu-notifications.zh-CN.md
index 371e62b..3057bd3 100644
--- a/docs/integrations/feishu-notifications.zh-CN.md
+++ b/docs/integrations/feishu-notifications.zh-CN.md
@@ -62,6 +62,47 @@ AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1
 使用被允许读取的知识。因此只接入受控群，发送前确认数据权限，不把密钥填写到业务内容中。
 这不是飞书双向指挥功能：群中回复不会直接改变任务、员工记忆或批准作品。
 
+## 每位员工使用独立机器人
+
+该模式可为每位员工绑定一个不同的飞书应用机器人。改卡片标题或头像并不能改变实际发送者；
+任务汇总和审批仍由原有机器人发送，员工协作消息按已固定执行记录的 `agent_id` 选择身份，
+不按工作项角色名称或模型输出选择。
+
+1. 创建不同的应用机器人。可使用[飞书官方 SDK 一键创建流程](https://github.com/larksuite/node-sdk#app-registration)
+   预填名称，再由用户确认授权。仅发送通知时，使用最小模板，只申请 `im:message:send_as_bot`，
+   不需要获取通讯录或读取群消息。凭证不得输出到日志。
+2. 把每个机器人加入同一个受控测试群，并确认允许发言。创建成功不代表已经入群；API 拉群需要
+   额外的群成员管理权限，本通知功能不会自动申请这些权限。
+3. 复制[配置样例](../../examples/feishu/employee-bots.example.json)，在 Git 之外保存为受保护的
+   `feishu-employee-bots.json`，替换占位值。租户和目标群必须与通知进程完全一致。
+   `agent_id` 是执行记录中的身份（通常是员工注册名称），不是 Agent Definition 的 UUID 或工作项
+   标签。员工绑定和 App ID 不得重复，也不能复用总机器人的 App ID。
+4. Linux 文件设为 `600` 权限；Windows 使用仅允许本人/运行账户访问的 ACL。通过 Compose 覆盖
+   配置把文件只读挂载到**通知进程**，不传给 API、Worker 或模型：
+
+   ```yaml
+   services:
+     feishu-notifier:
+       environment:
+         AGENTMESH_FEISHU_EMPLOYEE_BOTS_ENABLED: "true"
+         AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE: /run/agentmesh/feishu-employee-bots.json
+         AGENTMESH_FEISHU_EMPLOYEE_BOT_FALLBACK: "true"
+       volumes:
+         - ./secrets/feishu-employee-bots.json:/run/agentmesh/feishu-employee-bots.json:ro
+   ```
+
+5. 保持原有通知 Gate 和协作同步开启，加载上述覆盖配置后重启通知进程。独立身份开关默认关闭。
+   配置不合法会阻止启动，并且不打印密钥。文件最大 64 KiB，最多 256 项绑定；各机器人独立缓存
+   访问令牌，继续使用原通知 UUID 去重及统一发送节流。
+
+`EMPLOYEE_BOT_FALLBACK=true` 时，尚未绑定的员工继续用总机器人发送，便于逐步配置；改为
+`false` 会暂缓发送并进入重试。**已绑定**机器人的发送失败绝不偷偷改用另一身份，未入群、
+凭证撤销等只影响通知投递，不会让业务任务失败。重试和 `DEAD` 状态沿用已有机制。
+
+绑定在进程启动时读取。更换 App ID 前，应停止投递并处理该员工尚未结束/结果未知的通知；
+飞书去重可能限定在应用内，重试期间更换发送者可能重复发消息。当前未持久化发送者绑定版本，
+不会自动绕过管理员审批，也不是机器人之间的双向群聊。
+
 ## 可靠性与边界
 
 [真实验收记录](../qualification/feishu-collaboration.md)记录了四员工真实模型协作及九条飞书消息
diff --git a/examples/feishu/employee-bots.example.json b/examples/feishu/employee-bots.example.json
new file mode 100644
index 0000000..3533873
--- /dev/null
+++ b/examples/feishu/employee-bots.example.json
@@ -0,0 +1,17 @@
+{
+  "schema_version": 1,
+  "tenant_id": "default",
+  "chat_id": "oc_REPLACE_WITH_GROUP_ID",
+  "bots": [
+    {
+      "agent_id": "researcher",
+      "app_id": "cli_REPLACE_WITH_RESEARCH_APP_ID",
+      "app_secret": "REPLACE_WITH_RESEARCH_SECRET"
+    },
+    {
+      "agent_id": "writer",
+      "app_id": "cli_REPLACE_WITH_WRITER_APP_ID",
+      "app_secret": "REPLACE_WITH_WRITER_SECRET"
+    }
+  ]
+}
diff --git a/src/agentmesh/config.py b/src/agentmesh/config.py
index 57bfb28..f6b59d2 100644
--- a/src/agentmesh/config.py
+++ b/src/agentmesh/config.py
@@ -106,6 +106,9 @@ class Settings(BaseSettings):
     feishu_task_base_url: str | None = None
     feishu_include_content: bool = False
     feishu_sync_collaboration: bool = False
+    feishu_employee_bots_enabled: bool = False
+    feishu_employee_bots_file: str | None = None
+    feishu_employee_bot_fallback: bool = True
     feishu_send_interval_seconds: float = Field(default=1.0, ge=0.1, le=10)
     feishu_timeout_seconds: int = Field(default=5, ge=1, le=30)
     feishu_scan_seconds: int = Field(default=3, ge=1, le=60)
diff --git a/src/agentmesh/entrypoints/feishu_notifier.py b/src/agentmesh/entrypoints/feishu_notifier.py
index 9292b57..cda3f67 100644
--- a/src/agentmesh/entrypoints/feishu_notifier.py
+++ b/src/agentmesh/entrypoints/feishu_notifier.py
@@ -10,6 +10,7 @@
 
 from agentmesh.config import Settings, get_settings
 from agentmesh.features import Feature, FeatureGateSet
+from agentmesh.integrations.feishu_identities import load_employee_bot_credentials
 from agentmesh.integrations.feishu_notifications import (
     FeishuClient,
     FeishuNotificationStore,
@@ -41,12 +42,37 @@ def validate_configuration(settings: Settings) -> None:
             or url.fragment
         ):
             raise ValueError("AGENTMESH_FEISHU_TASK_BASE_URL must be a public HTTPS URL")
+    if settings.feishu_employee_bots_enabled:
+        if not settings.feishu_sync_collaboration:
+            raise ValueError("Employee bots require AGENTMESH_FEISHU_SYNC_COLLABORATION")
+        if not settings.feishu_employee_bots_file:
+            raise ValueError("AGENTMESH_FEISHU_EMPLOYEE_BOTS_FILE is required")
+
+
+def build_employee_clients(settings: Settings) -> dict[str, FeishuClient]:
+    if not settings.feishu_employee_bots_enabled:
+        return {}
+    validate_configuration(settings)
+    credentials = load_employee_bot_credentials(
+        settings.feishu_employee_bots_file or "",
+        tenant_id=settings.tenant_id,
+        chat_id=settings.feishu_chat_id or "",
+    )
+    if any(bot.app_id == settings.feishu_app_id for bot in credentials.values()):
+        raise ValueError("Employee bots must be distinct from the task-summary bot")
+    return {
+        agent_id: FeishuClient(
+            app_id=bot.app_id, app_secret=bot.app_secret.get_secret_value(),
+            chat_id=settings.feishu_chat_id or "", timeout_seconds=settings.feishu_timeout_seconds,
+        ) for agent_id, bot in credentials.items()
+    }
 
 
 def main() -> None:
     logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
     settings = get_settings()
     validate_configuration(settings)
+    employee_clients = build_employee_clients(settings)
     engine = create_engine(settings.database_url, pool_pre_ping=True)
     sessions = sessionmaker(bind=engine, expire_on_commit=False, class_=Session)
     client = FeishuClient(
@@ -64,6 +90,8 @@ def main() -> None:
         task_base_url=settings.feishu_task_base_url,
         include_content=settings.feishu_include_content,
         sync_collaboration=settings.feishu_sync_collaboration,
+        employee_clients=employee_clients if settings.feishu_employee_bots_enabled else None,
+        employee_bot_fallback=settings.feishu_employee_bot_fallback,
         send_interval_seconds=settings.feishu_send_interval_seconds,
     )
     try:
@@ -78,6 +106,8 @@ def main() -> None:
     except KeyboardInterrupt:
         pass
     finally:
+        for employee_client in employee_clients.values():
+            employee_client.close()
         client.close()
         engine.dispose()
 
diff --git a/src/agentmesh/integrations/feishu_identities.py b/src/agentmesh/integrations/feishu_identities.py
new file mode 100644
index 0000000..f4168cc
--- /dev/null
+++ b/src/agentmesh/integrations/feishu_identities.py
@@ -0,0 +1,95 @@
+"""Operator-owned, tenant/group-bound employee bot credentials. No registration or chat ingress."""
+
+import json
+import os
+import stat
+import unicodedata
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError, field_validator
+
+_MAX_FILE_BYTES = 65_536
+
+
+class EmployeeBotCredential(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    agent_id: str = Field(min_length=1, max_length=128)
+    app_id: str = Field(pattern=r"^cli_[A-Za-z0-9_]+$", max_length=128)
+    app_secret: SecretStr = Field(repr=False)
+
+    @field_validator("agent_id")
+    @classmethod
+    def exact_identifier(cls, value: str) -> str:
+        if value != value.strip() or any(unicodedata.category(c).startswith("C") for c in value):
+            raise ValueError("Invalid employee identifier")
+        return value
+
+    @field_validator("app_secret")
+    @classmethod
+    def bounded_secret(cls, value: SecretStr) -> SecretStr:
+        raw = value.get_secret_value()
+        if not raw.strip() or len(raw) > 1024 or any(c.isspace() for c in raw):
+            raise ValueError("Invalid bot credential")
+        return value
+
+
+class EmployeeBotFile(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal[1]
+    tenant_id: str = Field(min_length=1, max_length=128)
+    chat_id: str = Field(pattern=r"^oc_[A-Za-z0-9_]+$", max_length=128)
+    bots: list[EmployeeBotCredential] = Field(min_length=1, max_length=256)
+
+    @field_validator("schema_version", mode="before")
+    @classmethod
+    def exact_version(cls, value: object) -> object:
+        if type(value) is not int:
+            raise ValueError("Invalid schema version")
+        return value
+
+
+def _unique_json_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("Duplicate configuration field")
+        result[key] = value
+    return result
+
+
+def load_employee_bot_credentials(
+    path: str, *, tenant_id: str, chat_id: str
+) -> dict[str, EmployeeBotCredential]:
+    """Bound reads and reject ambiguity, wrong destinations and broadly readable POSIX files.
+
+    Validation errors deliberately omit the original inputs, which contain App Secrets.
+    Windows operators must use an ACL-protected file; Windows permission bits are not ACLs.
+    """
+    try:
+        with Path(path).open("rb") as stream:
+            info = os.fstat(stream.fileno())
+            if not stat.S_ISREG(info.st_mode):
+                raise ValueError("Not a regular configuration file")
+            if os.name != "nt" and info.st_mode & 0o077:
+                raise ValueError("Insecure configuration permissions")
+            raw = stream.read(_MAX_FILE_BYTES + 1)
+        if len(raw) > _MAX_FILE_BYTES:
+            raise ValueError("Configuration too large")
+        parsed = json.loads(raw, object_pairs_hook=_unique_json_keys)
+        configuration = EmployeeBotFile.model_validate(parsed)
+        if configuration.tenant_id != tenant_id or configuration.chat_id != chat_id:
+            raise ValueError("Wrong tenant or group")
+        credentials = {bot.agent_id: bot for bot in configuration.bots}
+        if len(credentials) != len(configuration.bots):
+            raise ValueError("Duplicate employee binding")
+        if len({bot.app_id for bot in configuration.bots}) != len(configuration.bots):
+            raise ValueError("A bot must not impersonate multiple employees")
+        return credentials
+    except (OSError, ValueError, ValidationError, RecursionError):
+        raise ValueError(
+            "Invalid Feishu employee bot file: check schema, tenant/group, unique bindings "
+            "and private file permissions; credential details withheld"
+        ) from None
diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py
index 9454b11..229cc2e 100644
--- a/src/agentmesh/integrations/feishu_notifications.py
+++ b/src/agentmesh/integrations/feishu_notifications.py
@@ -8,7 +8,7 @@
 import re
 import time
 import unicodedata
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from hashlib import sha256
@@ -572,6 +572,8 @@ def __init__(
         task_base_url: str | None,
         include_content: bool,
         sync_collaboration: bool = False,
+        employee_clients: Mapping[str, FeishuClient] | None = None,
+        employee_bot_fallback: bool = True,
         send_interval_seconds: float = 1.0,
         sleep: Callable[[float], None] = time.sleep,
     ) -> None:
@@ -583,6 +585,12 @@ def __init__(
         if not math.isfinite(send_interval_seconds) or not 0 <= send_interval_seconds <= 10:
             raise ValueError("Feishu send interval must be between 0 and 10 seconds")
         self._sync_collaboration = sync_collaboration
+        if employee_clients is not None and (not sync_collaboration or not employee_clients):
+            raise ValueError(
+                "Employee bot routing requires collaboration sync and nonempty bindings"
+            )
+        self._employee_clients = dict(employee_clients or {})
+        self._employee_bot_fallback = employee_bot_fallback
         self._send_interval_seconds = send_interval_seconds
         self._sleep = sleep
         self._last_send_at: float | None = None
@@ -624,7 +632,14 @@ def run_once(self) -> int:
                     if remaining > 0:
                         self._sleep(remaining)
                 self._last_send_at = time.monotonic()
-                self._client.send(
+                client = self._client
+                if isinstance(subject, CollaborationSubject) and self._employee_clients:
+                    employee_client = self._employee_clients.get(subject.agent_id)
+                    if employee_client is not None:
+                        client = employee_client
+                    elif not self._employee_bot_fallback:
+                        raise RuntimeError("Employee bot binding missing; delivery withheld")
+                client.send(
                     notification_id=notification.id,
                     card=card,
                 )
diff --git a/tests/test_feishu_identities.py b/tests/test_feishu_identities.py
new file mode 100644
index 0000000..fcf0e73
--- /dev/null
+++ b/tests/test_feishu_identities.py
@@ -0,0 +1,220 @@
+"""Independent sender identity must use the pinned Run employee, not card text or role labels."""
+
+import json
+import os
+import traceback
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+
+from agentmesh.config import Settings
+from agentmesh.entrypoints.feishu_notifier import build_employee_clients, validate_configuration
+from agentmesh.integrations.feishu_identities import load_employee_bot_credentials
+from agentmesh.integrations.feishu_notifications import (
+    ClaimedNotification,
+    CollaborationSubject,
+    FeishuNotificationWorker,
+)
+
+
+def config():
+    return {
+        "schema_version": 1, "tenant_id": "tenant", "chat_id": "oc_group",
+        "bots": [{"agent_id": "employee-one", "app_id": "cli_one", "app_secret": "test-secret"}],
+    }
+
+
+def write_config(tmp_path, value):
+    path = tmp_path / "bots.json"
+    path.write_text(json.dumps(value), encoding="utf-8")
+    path.chmod(0o600)
+    return str(path)
+
+
+def test_read_valid_bounded_private_file_and_redacted_credentials(tmp_path):
+    bots = load_employee_bot_credentials(
+        write_config(tmp_path, config()), tenant_id="tenant", chat_id="oc_group"
+    )
+    assert bots["employee-one"].app_secret.get_secret_value() == "test-secret"
+    assert "test-secret" not in repr(bots)
+
+
+@pytest.mark.parametrize("change", [
+    lambda v: v.update(schema_version=True),
+    lambda v: v.update(schema_version=2),
+    lambda v: v.update(tenant_id="other"),
+    lambda v: v.update(chat_id="oc_other"),
+    lambda v: v.update(bots=[]),
+    lambda v: v.update(endpoint="https://attacker.invalid"),
+    lambda v: v["bots"][0].update(agent_id=" employee-one"),
+    lambda v: v["bots"][0].update(agent_id="employee\u200bone"),
+    lambda v: v["bots"][0].update(app_secret=""),
+    lambda v: v["bots"][0].update(app_secret="credential with whitespace"),
+    lambda v: v["bots"][0].update(app_id="not-an-app"),
+    lambda v: v["bots"].append(dict(v["bots"][0])),
+    lambda v: v["bots"].append(dict(v["bots"][0], agent_id="employee-two")),
+])
+def test_invalid_files_fail_without_echoing_secrets(tmp_path, change):
+    value = config()
+    change(value)
+    with pytest.raises(ValueError, match="credential details withheld") as caught:
+        load_employee_bot_credentials(
+            write_config(tmp_path, value), tenant_id="tenant", chat_id="oc_group"
+        )
+    rendered = "".join(traceback.format_exception(caught.type, caught.value, caught.tb))
+    assert "test-secret" not in rendered
+
+
+@pytest.mark.parametrize(
+    "data", ['{"schema_version":1,"schema_version":1}', "[", "x" * 65537],
+    ids=["duplicate-keys", "invalid-json", "oversize"],
+)
+def test_duplicate_fields_malformed_and_oversize_files_rejected(tmp_path, data):
+    path = tmp_path / "bots.json"
+    path.write_text(data)
+    path.chmod(0o600)
+    with pytest.raises(ValueError, match="credential details withheld"):
+        load_employee_bot_credentials(str(path), tenant_id="tenant", chat_id="oc_group")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Windows requires operator-managed ACLs")
+def test_broad_posix_permissions_rejected(tmp_path):
+    path = write_config(tmp_path, config())
+    os.chmod(path, 0o644)
+    with pytest.raises(ValueError):
+        load_employee_bot_credentials(path, tenant_id="tenant", chat_id="oc_group")
+
+
+def settings(**changes):
+    return Settings(_env_file=None, **dict({
+        "feature_gates": "feishu_notifications=true", "tenant_id": "tenant",
+        "feishu_app_id": "cli_summary", "feishu_app_secret": "summary-secret",
+        "feishu_chat_id": "oc_group", "feishu_sync_collaboration": True,
+    }, **changes))
+
+
+def test_identity_opt_in_off_does_not_read_file(tmp_path):
+    assert not Settings(_env_file=None).feishu_employee_bots_enabled
+    assert build_employee_clients(settings(feishu_employee_bots_file="missing")) == {}
+
+
+def test_enabled_routing_requires_sync_and_file():
+    with pytest.raises(ValueError, match="SYNC_COLLABORATION"):
+        validate_configuration(settings(
+            feishu_employee_bots_enabled=True, feishu_sync_collaboration=False
+        ))
+    with pytest.raises(ValueError, match="BOTS_FILE"):
+        validate_configuration(settings(feishu_employee_bots_enabled=True))
+
+
+def test_clients_share_configured_group_but_not_tokens(tmp_path):
+    value = config()
+    value["bots"].append({
+        "agent_id": "employee-two", "app_id": "cli_two", "app_secret": "second-secret",
+    })
+    clients = build_employee_clients(settings(
+        feishu_employee_bots_enabled=True, feishu_employee_bots_file=write_config(tmp_path, value)
+    ))
+    try:
+        assert set(clients) == {"employee-one", "employee-two"}
+        assert clients["employee-one"] is not clients["employee-two"]
+        assert all(client._chat_id == "oc_group" for client in clients.values())
+    finally:
+        for client in clients.values():
+            client.close()
+
+
+def test_employee_must_not_reuse_summary_app(tmp_path):
+    value = config()
+    value["bots"][0]["app_id"] = "cli_summary"
+    with pytest.raises(ValueError, match="distinct"):
+        build_employee_clients(settings(
+            feishu_employee_bots_enabled=True,
+            feishu_employee_bots_file=write_config(tmp_path, value)
+        ))
+
+
+class Client:
+    def __init__(self, *, fail=False):
+        self.sent = []
+        self.failed = fail
+
+    def send(self, **kwargs):
+        self.sent.append(kwargs)
+        if self.failed:
+            raise RuntimeError("Feishu API code 230002")
+
+
+class Store:
+    def __init__(self, kind="COLLAB_RESULT", agent="employee-one"):
+        self.notification = ClaimedNotification(
+            uuid4(), "tenant", "TASK_RUN" if kind.startswith("COLLAB") else "TASK",
+            uuid4(), kind, 1,
+        )
+        self.value = CollaborationSubject(
+            uuid4(), self.notification.subject_id, agent, "role-is-not-an-identity", "Label",
+            datetime.now(timezone.utc), "SUCCEEDED", (), "Public summary",
+        ) if kind.startswith("COLLAB") else SimpleNamespace(
+            status="COMPLETED", id=self.notification.subject_id,
+            input={}, output={}, objective="test"
+        )
+        self.finished = []
+        self.failed = []
+
+    def claim(self, **kwargs):
+        return [self.notification]
+
+    def subject(self, notification):
+        return self.value
+
+    def finish(self, notification, **kwargs):
+        self.finished.append(kwargs["status"])
+
+    def fail(self, notification, **kwargs):
+        self.failed.append(kwargs["error"])
+
+
+def worker(store, summary, employees=None, fallback=True):
+    return FeishuNotificationWorker(
+        worker_id="test", store=store, client=summary, task_base_url=None, include_content=False,
+        sync_collaboration=True, employee_clients=employees, employee_bot_fallback=fallback,
+    )
+
+
+@pytest.mark.parametrize("kind", ["COLLAB_STARTED", "COLLAB_RESULT", "COLLAB_FAILED"])
+def test_routes_exact_employee_with_same_delivery_uuid(kind):
+    store, summary, employee = Store(kind), Client(), Client()
+    assert worker(store, summary, {"employee-one": employee}).run_once() == 1
+    assert not summary.sent
+    assert employee.sent[0]["notification_id"] == store.notification.id
+    assert store.finished == ["DELIVERED"]
+
+
+def test_mapped_failure_retries_without_impersonating_summary_bot():
+    store, summary, employee = Store(), Client(), Client(fail=True)
+    worker(store, summary, {"employee-one": employee}).run_once()
+    assert employee.sent and not summary.sent
+    assert store.failed == ["RuntimeError: Feishu API code 230002"]
+    assert not store.finished
+
+
+@pytest.mark.parametrize("fallback", [True, False])
+def test_unmapped_identity_obeys_explicit_fallback_not_role_label(fallback):
+    store, summary, employee = Store(agent="unknown"), Client(), Client()
+    worker(store, summary, {"role-is-not-an-identity": employee}, fallback).run_once()
+    assert not employee.sent
+    assert bool(summary.sent) is fallback
+    assert bool(store.failed) is not fallback
+
+
+def test_task_summary_keeps_original_sender():
+    store, summary, employee = Store("COMPLETED"), Client(), Client()
+    worker(store, summary, {"employee-one": employee}, False).run_once()
+    assert len(summary.sent) == 1 and not employee.sent
+
+
+def test_empty_mapping_cannot_silently_bypass_strict_mode():
+    with pytest.raises(ValueError, match="nonempty bindings"):
+        worker(Store(), Client(), {}, False)