From 91afecf47cd7709cad8efd922e97b4e30e150a28 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Thu, 8 Oct 2026 13:50:48 +0800 Subject: [PATCH 1/8] feat: persist extension employee activity with business revisions --- .../feishu-notifications.zh-CN.md | 8 + .../application/business_object_services.py | 65 ++-- src/agentmesh/domain/business_activity.py | 27 ++ .../infrastructure/postgres/repositories.py | 20 +- src/agentmesh/infrastructure/postgres/uow.py | 4 +- .../integrations/feishu_notifications.py | 332 +++++++++++++----- tests/test_business_activity_notifications.py | 188 ++++++++++ 7 files changed, 526 insertions(+), 118 deletions(-) create mode 100644 src/agentmesh/domain/business_activity.py create mode 100644 tests/test_business_activity_notifications.py diff --git a/docs/integrations/feishu-notifications.zh-CN.md b/docs/integrations/feishu-notifications.zh-CN.md index 3057bd36..65995f0d 100644 --- a/docs/integrations/feishu-notifications.zh-CN.md +++ b/docs/integrations/feishu-notifications.zh-CN.md @@ -105,6 +105,14 @@ AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1 ## 可靠性与边界 +可信扩展也可通过 `BusinessObjectService.apply_action(activity=BusinessActivityNotice(...))` +报告员工的 `STARTED`、`RESULT`、`FAILED` 工作状态。这个参数不是公开 HTTP API 的输入字段。 +通知只允许员工标识、角色名和最长 500 字的明确业务摘要，不接收原始记忆、提示词或推理。 +活动事件和业务版本在同一事务提交；飞书投递仍要求原通知 Gate 与协作同步同时开启。 +通知进程核对租户、公司、对象、确切版本和内容指纹，再按已有员工机器人绑定发送。 +默认不发送摘要内容；开启 `FEISHU_INCLUDE_CONTENT` 后仍应用凭据检查。此机制不依赖任何业务场景， +也不新增公网回调、群消息读取、自动批准或外部发布能力。 + [真实验收记录](../qualification/feishu-collaboration.md)记录了四员工真实模型协作及九条飞书消息 成功接收的结果，并明确验收范围和未覆盖的边界。 diff --git a/src/agentmesh/application/business_object_services.py b/src/agentmesh/application/business_object_services.py index 0d693c43..47007bc8 100644 --- a/src/agentmesh/application/business_object_services.py +++ b/src/agentmesh/application/business_object_services.py @@ -5,6 +5,7 @@ from uuid import UUID from agentmesh.application.ports import UnitOfWorkFactory +from agentmesh.domain.business_activity import ACTIVITY_SCHEMA, BusinessActivityNotice from agentmesh.domain.business_objects import ( BusinessObject, BusinessObjectRevision, @@ -70,9 +71,7 @@ def create_type(self, company_id: UUID, **values: Any) -> BusinessObjectType: uow.commit() return object_type - def transition_type( - self, company_id: UUID, type_id: UUID, action: str - ) -> BusinessObjectType: + def transition_type(self, company_id: UUID, type_id: UUID, action: str) -> BusinessObjectType: self._require_enabled() with self._uow_factory() as uow: self._active_company(uow, company_id) @@ -92,9 +91,7 @@ def transition_type( elif action == "deprecate": object_type.deprecate() else: - raise InvalidBusinessObject( - f"Unknown Business Object Type action '{action}'" - ) + raise InvalidBusinessObject(f"Unknown Business Object Type action '{action}'") uow.business_objects.save_type(object_type) self._emit( uow, @@ -141,13 +138,8 @@ def create_object( "Only a published Business Object Type can create Objects" ) validate_data(object_type.json_schema, data, label="Business Object data") - if ( - object_type.ownership_rules.get("position_required") - and owner_position_id is None - ): - raise InvalidBusinessObject( - "Business Object Type requires an owner Position" - ) + if object_type.ownership_rules.get("position_required") and owner_position_id is None: + raise InvalidBusinessObject("Business Object Type requires an owner Position") self._validate_owner(uow, company_id, owner_position_id) if external_ref and uow.business_objects.get_object_by_external_ref( type_id, external_ref.strip() @@ -199,6 +191,7 @@ def apply_action( evidence_refs: list[str] | None = None, actor_position_key: str | None = None, actor_capabilities: list[str] | None = None, + activity: BusinessActivityNotice | None = None, ) -> BusinessObjectSnapshot: self._require_enabled() with self._uow_factory() as uow: @@ -234,22 +227,16 @@ def apply_action( ) evidence = sorted(set(evidence_refs or [])) if action["required_evidence"] and not evidence: - raise BusinessObjectConflict( - f"Action '{action_key}' requires evidence references" - ) + raise BusinessObjectConflict(f"Action '{action_key}' requires evidence references") required_positions = set(action["required_position_keys"]) if required_positions and actor_position_key not in required_positions: - raise BusinessObjectConflict( - f"Action '{action_key}' requires an allowed Position" - ) + raise BusinessObjectConflict(f"Action '{action_key}' requires an allowed Position") required_capabilities = set(action["required_capabilities"]) if not required_capabilities <= set(actor_capabilities or []): raise BusinessObjectConflict( f"Action '{action_key}' requires declared capabilities" ) - previous = uow.business_objects.get_revision( - value.id, value.current_revision - ) + previous = uow.business_objects.get_revision(value.id, value.current_revision) if previous is None: raise BusinessObjectNotFound("Current Business Object revision was not found") next_data = {**previous.data, **dict(input)} @@ -258,9 +245,7 @@ def apply_action( next_data, label="Resulting Business Object data", ) - value.apply( - expected_revision=expected_revision, target_state=action["to"] - ) + value.apply(expected_revision=expected_revision, target_state=action["to"]) revision = BusinessObjectRevision.create( object_id=value.id, revision=value.current_revision, @@ -285,13 +270,29 @@ def apply_action( "side_effect_class": action["side_effect_class"], }, ) + if activity is not None: + # The notice and exact object revision commit together. API requests + # cannot supply this parameter: it is a trusted extension SDK hook. + notice = BusinessActivityNotice.model_validate(activity) + uow.outbox.add( + MessageEnvelope.domain_event( + schema_name=ACTIVITY_SCHEMA, + tenant_id=self._tenant_id, + aggregate_id=value.id, + payload={ + "company_id": str(company_id), + "object_id": str(value.id), + "revision": revision.revision, + "data_digest": revision.data_digest, + "activity": notice.model_dump(), + }, + ) + ) uow.commit() revisions = uow.business_objects.list_revisions(value.id) return self._snapshot(value, object_type, revisions) - def get_object( - self, company_id: UUID, object_id: UUID - ) -> BusinessObjectSnapshot: + def get_object(self, company_id: UUID, object_id: UUID) -> BusinessObjectSnapshot: self._require_enabled() with self._uow_factory() as uow: self._company(uow, company_id) @@ -332,9 +333,7 @@ def _validate_owner(uow: Any, company_id: UUID, owner_position_id: UUID | None) or position.company_id != company_id or position.status is not ResourceStatus.ACTIVE ): - raise InvalidBusinessObject( - "Business Object owner must be an active Company Position" - ) + raise InvalidBusinessObject("Business Object owner must be an active Company Position") @staticmethod def _event_payload( @@ -369,9 +368,7 @@ def _snapshot( ) for revision in revisions ] - return BusinessObjectSnapshot( - object=value, type=object_type, revisions=redacted - ) + return BusinessObjectSnapshot(object=value, type=object_type, revisions=redacted) def _company(self, uow: Any, company_id: UUID): company = uow.company_model.get_company(company_id) diff --git a/src/agentmesh/domain/business_activity.py b/src/agentmesh/domain/business_activity.py new file mode 100644 index 00000000..cb1ab2b9 --- /dev/null +++ b/src/agentmesh/domain/business_activity.py @@ -0,0 +1,27 @@ +"""Explicit, bounded employee activity notices supplied by trusted extensions. + +Only the summary is intended for external collaboration channels; never pass raw +prompts, memory, model reasoning, credentials, or provider responses here. +""" + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +ACTIVITY_SCHEMA = "agentmesh.company.object.activity" + + +class BusinessActivityNotice(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + agent_id: str = Field(min_length=1, max_length=128) + role_label: str = Field(min_length=1, max_length=80) + phase: Literal["STARTED", "RESULT", "FAILED"] + summary: str | None = Field(default=None, min_length=1, max_length=500) + + @field_validator("agent_id", "role_label", "summary") + @classmethod + def normalized_text(cls, value: str | None) -> str | None: + if value is not None and (value != value.strip() or any(ord(c) < 32 for c in value)): + raise ValueError("Activity text must be normalized and contain no control characters") + return value diff --git a/src/agentmesh/infrastructure/postgres/repositories.py b/src/agentmesh/infrastructure/postgres/repositories.py index 961d3354..955cec6a 100644 --- a/src/agentmesh/infrastructure/postgres/repositories.py +++ b/src/agentmesh/infrastructure/postgres/repositories.py @@ -7,6 +7,7 @@ from sqlalchemy.orm import Session from agentmesh.domain.budgets import BudgetSettlementSource, TaskBudget +from agentmesh.domain.business_activity import ACTIVITY_SCHEMA, BusinessActivityNotice from agentmesh.domain.coordination import ( CoordinationRuntimeDrain, CoordinationRuntimeDrainStatus, @@ -1065,8 +1066,9 @@ def _to_domain(record: UsageRecordModel) -> UsageRecord: class SqlAlchemyOutboxRepository: - def __init__(self, session: Session) -> None: + def __init__(self, session: Session, *, feishu_collaboration_enabled: bool = False) -> None: self._session = session + self._feishu_collaboration_enabled = feishu_collaboration_enabled def add(self, envelope: MessageEnvelope) -> None: self._session.add( @@ -1086,6 +1088,22 @@ def add(self, envelope: MessageEnvelope) -> None: last_error=None, ) ) + if self._feishu_collaboration_enabled and envelope.schema_name == ACTIVITY_SCHEMA: + notice = BusinessActivityNotice.model_validate(envelope.payload["activity"]) + self._session.add( + FeishuNotificationRecord( + id=uuid5(NAMESPACE_URL, f"agentmesh:feishu:activity:{envelope.message_id}"), + tenant_id=envelope.tenant_id, + subject_type="BUSINESS_ACTIVITY", + subject_id=envelope.message_id, + subject_revision=envelope.payload["revision"], + event_kind=f"ACTIVITY_{notice.phase}", + status="PENDING", + created_at=envelope.occurred_at, + available_at=envelope.occurred_at, + attempt_count=0, + ) + ) def add_if_absent(self, envelope: MessageEnvelope) -> bool: if self._session.get(OutboxEventRecord, envelope.message_id) is not None: diff --git a/src/agentmesh/infrastructure/postgres/uow.py b/src/agentmesh/infrastructure/postgres/uow.py index efabd6e3..85a9d288 100644 --- a/src/agentmesh/infrastructure/postgres/uow.py +++ b/src/agentmesh/infrastructure/postgres/uow.py @@ -129,7 +129,9 @@ def __enter__(self) -> SqlAlchemyUnitOfWork: self.runtime_comparisons = SqlAlchemyRuntimeComparisonRepository(self._session) self.attempts = SqlAlchemyTaskAttemptRepository(self._session) self.quotas = SqlAlchemyQuotaRepository(self._session) - self.outbox = SqlAlchemyOutboxRepository(self._session) + self.outbox = SqlAlchemyOutboxRepository( + self._session, feishu_collaboration_enabled=self._feishu_collaboration_enabled + ) self.inbox = SqlAlchemyInboxRepository(self._session) self.idempotency = SqlAlchemyIdempotencyRepository(self._session) self.agent_definitions = SqlAlchemyAgentDefinitionRepository(self._session) diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index 229cc2e4..932be24b 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -17,12 +17,19 @@ from uuid import UUID import httpx +from pydantic import ValidationError from sqlalchemy import or_, select, update from sqlalchemy.orm import Session, sessionmaker +from agentmesh.domain.business_activity import ACTIVITY_SCHEMA, BusinessActivityNotice +from agentmesh.domain.messaging import MessageEnvelope from agentmesh.infrastructure.postgres.models import ( + BusinessObjectRecord, + BusinessObjectRevisionRecord, + CompanyRecord, FeishuNotificationRecord, GovernedActionRecord, + OutboxEventRecord, SubtaskRecord, TaskRecord, TaskRunRecord, @@ -39,6 +46,9 @@ "COLLAB_STARTED": "AgentMesh · 员工开始工作", "COLLAB_RESULT": "AgentMesh · 阶段成果", "COLLAB_FAILED": "AgentMesh · 执行失败", + "ACTIVITY_STARTED": "AgentMesh · 员工开始工作", + "ACTIVITY_RESULT": "AgentMesh · 阶段成果", + "ACTIVITY_FAILED": "AgentMesh · 阶段未完成", } _COLLAB_EVENTS = frozenset({"COLLAB_STARTED", "COLLAB_RESULT", "COLLAB_FAILED"}) _MAX_TRANSFERS = 20 @@ -105,6 +115,64 @@ class CollaborationSubject: output_summary: str | None +@dataclass(frozen=True) +class BusinessActivitySubject: + object_id: UUID + revision: int + agent_id: str + role_label: str + phase: str + occurred_at: datetime + summary: str | None + + +def project_business_activity(notification, event, obj, revision, company): + """Bind a notice to its immutable, same-tenant committed object revision.""" + try: + envelope = MessageEnvelope.from_dict(event.envelope) + payload = envelope.payload + if ( + notification.subject_type != "BUSINESS_ACTIVITY" + or event.id != notification.subject_id + or event.tenant_id != notification.tenant_id + or envelope.tenant_id != notification.tenant_id + or envelope.message_id != event.id + or event.topic != ACTIVITY_SCHEMA + or envelope.schema_name != ACTIVITY_SCHEMA + or envelope.schema_version != 1 + or set(payload) != {"company_id", "object_id", "revision", "data_digest", "activity"} + or type(payload["revision"]) is not int + or payload["revision"] < 1 + or company.tenant_id != notification.tenant_id + or company.id != obj.company_id + or str(company.id) != payload["company_id"] + or str(obj.id) != payload["object_id"] + or envelope.correlation_id != obj.id + or revision.object_id != obj.id + or revision.revision != payload["revision"] + or revision.data_digest != payload["data_digest"] + ): + return None + notice = BusinessActivityNotice.model_validate(payload["activity"]) + if notification.event_kind != f"ACTIVITY_{notice.phase}": + return None + agent_id = safe_content(notice.agent_id, limit=128) + role_label = safe_content(notice.role_label, limit=80) + if not agent_id or not role_label: + return None + return BusinessActivitySubject( + obj.id, + revision.revision, + agent_id, + role_label, + notice.phase, + envelope.occurred_at, + safe_content(notice.summary) if notice.summary else None, + ) + except (ValueError, TypeError, KeyError, AttributeError, ValidationError): + return None + + def project_collaboration_subject( notification: ClaimedNotification, *, @@ -166,16 +234,17 @@ def project_collaboration_subject( if not isinstance(dependency_outputs, dict) or not isinstance(accepted_handoffs, list): return None role_input = input_value.get("subtask_input") - role_label = safe_content(role_input.get("role"), limit=80) if isinstance( - role_input, dict - ) else None + role_label = ( + safe_content(role_input.get("role"), limit=80) if isinstance(role_input, dict) else None + ) transfers: list[CollaborationTransfer] = [] # Validate the complete evidence list even when only a bounded prefix is shown. if len(snapshot["transfers"]) > 1000: return None for raw in snapshot["transfers"]: if not isinstance(raw, dict) or raw.get("kind") not in { - "DEPENDENCY_RESULT", "ACCEPTED_HANDOFF", + "DEPENDENCY_RESULT", + "ACCEPTED_HANDOFF", }: return None try: @@ -184,16 +253,22 @@ def project_collaboration_subject( source = get_record(SubtaskRecord, source_id) source_run = get_record(TaskRunRecord, source_run_id) if source_run_id else None if ( - source is None or source.task_id != task.id or source.key != raw["source_key"] + source is None + or source.task_id != task.id + or source.key != raw["source_key"] or raw["target_subtask_id"] != str(subtask.id) or raw["target_key"] != subtask.key or raw["target_run_id"] != str(run.id) or raw["target_agent_id"] != run.agent_id - or (source_run_id is not None and ( - source_run is None or source_run.task_id != task.id - or source_run.subtask_id != source_id - or source_run.agent_id != raw.get("source_agent_id") - )) + or ( + source_run_id is not None + and ( + source_run is None + or source_run.task_id != task.id + or source_run.subtask_id != source_id + or source_run.agent_id != raw.get("source_agent_id") + ) + ) ): return None payload = raw["payload"] @@ -210,38 +285,57 @@ def project_collaboration_subject( if source_run_id is None or payload not in accepted_handoffs: return None if not isinstance(payload, dict) or any( - payload.get(key) != raw.get(key) for key in ( - "handoff_id", "source_subtask_id", "source_run_id", - "source_agent_id", "target_agent_id", + payload.get(key) != raw.get(key) + for key in ( + "handoff_id", + "source_subtask_id", + "source_run_id", + "source_agent_id", + "target_agent_id", ) ): return None source_key = safe_content(raw["source_key"], limit=128) source_agent = safe_content(raw.get("source_agent_id"), limit=128) - if source_key is None or (raw.get("source_agent_id") is not None - and source_agent is None): + if source_key is None or ( + raw.get("source_agent_id") is not None and source_agent is None + ): return None except (TypeError, ValueError, KeyError, RecursionError): return None if len(transfers) < _MAX_TRANSFERS: - summary_field = "summary" if raw["kind"] == "DEPENDENCY_RESULT" else ( - "completed_work_summary" + summary_field = ( + "summary" if raw["kind"] == "DEPENDENCY_RESULT" else ("completed_work_summary") ) summary = ( safe_content(payload.get(summary_field)) if isinstance(payload, dict) else None ) - transfers.append(CollaborationTransfer( - kind=raw["kind"], source_key=source_key, source_agent_id=source_agent, - target_key=subtask_key, target_agent_id=agent_id, - payload_sha256=digest, summary=summary, - )) - output_summary = safe_content(run.output.get("summary")) if ( - notification.event_kind == "COLLAB_RESULT" and isinstance(run.output, dict) - ) else None + transfers.append( + CollaborationTransfer( + kind=raw["kind"], + source_key=source_key, + source_agent_id=source_agent, + target_key=subtask_key, + target_agent_id=agent_id, + payload_sha256=digest, + summary=summary, + ) + ) + output_summary = ( + safe_content(run.output.get("summary")) + if (notification.event_kind == "COLLAB_RESULT" and isinstance(run.output, dict)) + else None + ) return CollaborationSubject( - task_id=task.id, run_id=run.id, agent_id=agent_id, subtask_key=subtask_key, - role_label=role_label or subtask_key, occurred_at=occurred_at.astimezone(timezone.utc), - status=status, transfers=tuple(transfers), output_summary=output_summary, + task_id=task.id, + run_id=run.id, + agent_id=agent_id, + subtask_key=subtask_key, + role_label=role_label or subtask_key, + occurred_at=occurred_at.astimezone(timezone.utc), + status=status, + transfers=tuple(transfers), + output_summary=output_summary, ) @@ -298,15 +392,34 @@ def claim(self, *, worker_id: str, limit: int = 5) -> list[ClaimedNotification]: def subject( self, notification: ClaimedNotification - ) -> TaskRecord | GovernedActionRecord | CollaborationSubject | None: + ) -> TaskRecord | GovernedActionRecord | CollaborationSubject | BusinessActivitySubject | None: if notification.tenant_id != self._tenant_id: return None with self._session_factory() as session: + if notification.subject_type == "BUSINESS_ACTIVITY": + event = session.get(OutboxEventRecord, notification.subject_id) + if event is None or event.tenant_id != self._tenant_id: + return None + try: + payload = event.envelope["payload"] + obj = session.get(BusinessObjectRecord, UUID(payload["object_id"])) + if obj is None: + return None + company = session.get(CompanyRecord, obj.company_id) + revision = session.get( + BusinessObjectRevisionRecord, (obj.id, payload["revision"]) + ) + return project_business_activity(notification, event, obj, revision, company) + except (ValueError, TypeError, KeyError): + return None if notification.subject_type == "TASK_RUN": run = session.scalar( - select(TaskRunRecord).join(TaskRecord, TaskRecord.id == TaskRunRecord.task_id) - .where(TaskRunRecord.id == notification.subject_id, - TaskRecord.tenant_id == self._tenant_id) + select(TaskRunRecord) + .join(TaskRecord, TaskRecord.id == TaskRunRecord.task_id) + .where( + TaskRunRecord.id == notification.subject_id, + TaskRecord.tenant_id == self._tenant_id, + ) ) if run is None or run.subtask_id is None: return None @@ -315,11 +428,18 @@ def subject( if task is None or subtask is None: return None return project_collaboration_subject( - notification, task=task, run=run, subtask=subtask, get_record=session.get, + notification, + task=task, + run=run, + subtask=subtask, + get_record=session.get, ) model = ( - TaskRecord if notification.subject_type == "TASK" else GovernedActionRecord - if notification.subject_type == "GOVERNED_ACTION" else None + TaskRecord + if notification.subject_type == "TASK" + else GovernedActionRecord + if notification.subject_type == "GOVERNED_ACTION" + else None ) if model is None: return None @@ -444,12 +564,25 @@ def send(self, *, notification_id: UUID, card: dict[str, object]) -> None: def build_card( notification: ClaimedNotification, - subject: TaskRecord | GovernedActionRecord | CollaborationSubject, + subject: TaskRecord | GovernedActionRecord | CollaborationSubject | BusinessActivitySubject, *, task_base_url: str | None, include_content: bool, ) -> dict[str, object]: title = _EVENT_TITLES[notification.event_kind] + if isinstance(subject, BusinessActivitySubject): + content = ( + f"业务对象 ID：{subject.object_id} · 版本 {subject.revision}" + f"\n时间：{subject.occurred_at.isoformat()}" + f"\n员工：{subject.agent_id} · {subject.role_label}" + "\n这是明确记录的工作状态或业务摘要，不是模型内部推理或自由群聊。" + ) + elements = [{"tag": "div", "text": {"tag": "plain_text", "content": content}}] + if include_content and subject.summary: + elements.append( + {"tag": "div", "text": {"tag": "plain_text", "content": subject.summary}} + ) + return _card(title, elements) if isinstance(subject, CollaborationSubject): elements: list[dict[str, object]] = [] @@ -457,28 +590,36 @@ def text(content: str) -> None: elements.append({"tag": "div", "text": {"tag": "plain_text", "content": content}}) text(f"任务 ID：{subject.task_id}\nRun ID：{subject.run_id}") - text(f"时间：{subject.occurred_at.isoformat()}\n员工：{subject.agent_id}" - f"\n工作项：{subject.subtask_key} · {subject.role_label}") + text( + f"时间：{subject.occurred_at.isoformat()}\n员工：{subject.agent_id}" + f"\n工作项：{subject.subtask_key} · {subject.role_label}" + ) if notification.event_kind == "COLLAB_STARTED": text("以下为协调器已固定的输入交接，不代表自由聊天或模型内部推理。") elif notification.event_kind == "COLLAB_RESULT": text("本次 Run 的阶段成果，尚不代表最终交付或人工批准。") else: text("本次 Run 执行失败；详细诊断请在 AgentMesh 内查看。") - incoming = subject.transfers[:_MAX_TRANSFERS] if ( - notification.event_kind == "COLLAB_STARTED" - ) else () + incoming = ( + subject.transfers[:_MAX_TRANSFERS] + if (notification.event_kind == "COLLAB_STARTED") + else () + ) for transfer in incoming: label = "依赖结果" if transfer.kind == "DEPENDENCY_RESULT" else "已接受的交接" - text(f"{label}：{transfer.source_agent_id or '未记录员工'} / {transfer.source_key}" - f" → {transfer.target_agent_id} / {transfer.target_key}" - f"\n内容指纹：{transfer.payload_sha256}") + text( + f"{label}：{transfer.source_agent_id or '未记录员工'} / {transfer.source_key}" + f" → {transfer.target_agent_id} / {transfer.target_key}" + f"\n内容指纹：{transfer.payload_sha256}" + ) if include_content and transfer.summary: text(f"已固定的业务摘要：{transfer.summary}") if include_content and notification.event_kind == "COLLAB_RESULT": - text(f"阶段业务摘要：{subject.output_summary}" if subject.output_summary else ( - "本次 Run 未记录可安全同步的业务摘要。" - )) + text( + f"阶段业务摘要：{subject.output_summary}" + if subject.output_summary + else ("本次 Run 未记录可安全同步的业务摘要。") + ) _add_link(elements, task_base_url, query="task", subject_id=subject.task_id) return _card(title, elements) subject_label = "任务 ID" if notification.subject_type == "TASK" else "审批请求 ID" @@ -488,9 +629,7 @@ def text(content: str) -> None: if include_content and isinstance(subject, TaskRecord): objective = safe_content(subject.objective, limit=300) if objective: - elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": objective}} - ) + elements.append({"tag": "div", "text": {"tag": "plain_text", "content": objective}}) if ( notification.event_kind == "COMPLETED" and isinstance(subject.output, dict) @@ -498,19 +637,31 @@ def text(content: str) -> None: ): summary = safe_content(subject.output.get("summary")) if summary: - elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": summary}} - ) - elif ( - notification.event_kind == "COMPLETED" - and "agentmesh_deliverable_acceptance" in (subject.input or {}) + elements.append({"tag": "div", "text": {"tag": "plain_text", "content": summary}}) + elif notification.event_kind == "COMPLETED" and "agentmesh_deliverable_acceptance" in ( + subject.input or {} ): - elements.append({"tag": "div", "text": {"tag": "plain_text", "content": - "执行已完成；交付物已启用独立验收，请在 AgentMesh 查看验收状态。"}}) + elements.append( + { + "tag": "div", + "text": { + "tag": "plain_text", + "content": ( + "执行已完成；交付物已启用独立验收，" + "请在 AgentMesh 查看验收状态。" + ), + }, + } + ) elif notification.event_kind == "FAILED" and subject.error: elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": - "执行失败；详细诊断请在 AgentMesh 内查看。"}} + { + "tag": "div", + "text": { + "tag": "plain_text", + "content": "执行失败；详细诊断请在 AgentMesh 内查看。", + }, + } ) elif include_content and isinstance(subject, GovernedActionRecord): elements.append( @@ -520,7 +671,8 @@ def text(content: str) -> None: "tag": "plain_text", "content": safe_content( f"{subject.action_type} · {subject.resource_type}", limit=160 - ) or "治理审批请求", + ) + or "治理审批请求", }, } ) @@ -530,7 +682,11 @@ def text(content: str) -> None: def _add_link( - elements: list[dict[str, object]], task_base_url: str | None, *, query: str, subject_id: UUID, + elements: list[dict[str, object]], + task_base_url: str | None, + *, + query: str, + subject_id: UUID, ) -> None: if task_base_url and safe_content(task_base_url, limit=2048): try: @@ -599,31 +755,41 @@ def run_once(self) -> int: claimed = self._store.claim(worker_id=self._worker_id) for notification in claimed: try: - if notification.subject_type == "TASK_RUN" and not self._sync_collaboration: + if ( + notification.subject_type in {"TASK_RUN", "BUSINESS_ACTIVITY"} + and not self._sync_collaboration + ): self._store.finish(notification, worker_id=self._worker_id, status="SKIPPED") continue subject = self._store.subject(notification) - stale_task = ( - notification.subject_type == "TASK" - and (subject is None or subject.status != notification.event_kind) + stale_task = notification.subject_type == "TASK" and ( + subject is None or subject.status != notification.event_kind ) - stale_approval = ( - notification.subject_type == "GOVERNED_ACTION" - and ( - subject is None - or subject.approval_status != "PENDING" - or subject.expires_at <= datetime.now(timezone.utc) - ) + stale_approval = notification.subject_type == "GOVERNED_ACTION" and ( + subject is None + or subject.approval_status != "PENDING" + or subject.expires_at <= datetime.now(timezone.utc) + ) + invalid_collaboration = notification.subject_type == "TASK_RUN" and not isinstance( + subject, CollaborationSubject ) - invalid_collaboration = ( - notification.subject_type == "TASK_RUN" - and not isinstance(subject, CollaborationSubject) + invalid_activity = ( + notification.subject_type == "BUSINESS_ACTIVITY" + and not isinstance(subject, BusinessActivitySubject) ) - if stale_task or stale_approval or invalid_collaboration or subject is None: + if ( + stale_task + or stale_approval + or invalid_collaboration + or invalid_activity + or subject is None + ): self._store.finish(notification, worker_id=self._worker_id, status="SKIPPED") continue card = build_card( - notification, subject, task_base_url=self._task_base_url, + notification, + subject, + task_base_url=self._task_base_url, include_content=self._include_content, ) if self._last_send_at is not None: @@ -633,7 +799,10 @@ def run_once(self) -> int: self._sleep(remaining) self._last_send_at = time.monotonic() client = self._client - if isinstance(subject, CollaborationSubject) and self._employee_clients: + if ( + isinstance(subject, (CollaborationSubject, BusinessActivitySubject)) + and self._employee_clients + ): employee_client = self._employee_clients.get(subject.agent_id) if employee_client is not None: client = employee_client @@ -652,8 +821,7 @@ def run_once(self) -> int: self._store.fail( notification, worker_id=self._worker_id, - error=f"{type(exc).__name__}: " + ( - safe_content(str(exc), limit=180) or "delivery error details withheld" - ), + error=f"{type(exc).__name__}: " + + (safe_content(str(exc), limit=180) or "delivery error details withheld"), ) return len(claimed) diff --git a/tests/test_business_activity_notifications.py b/tests/test_business_activity_notifications.py new file mode 100644 index 00000000..3fcacb62 --- /dev/null +++ b/tests/test_business_activity_notifications.py @@ -0,0 +1,188 @@ +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from pydantic import ValidationError + +from agentmesh.domain.business_activity import ACTIVITY_SCHEMA, BusinessActivityNotice +from agentmesh.domain.messaging import MessageEnvelope +from agentmesh.infrastructure.postgres.repositories import SqlAlchemyOutboxRepository +from agentmesh.integrations.feishu_notifications import ( + ClaimedNotification, + FeishuNotificationWorker, + build_card, + project_business_activity, +) + + +def records(summary="Actual audio reviewed; revision recommended"): + obj = SimpleNamespace(id=uuid4(), company_id=uuid4()) + company = SimpleNamespace(id=obj.company_id, tenant_id="tenant") + revision = SimpleNamespace(object_id=obj.id, revision=3, data_digest="pinned-digest") + notice = BusinessActivityNotice( + agent_id="critic", role_label="Audio critic", phase="RESULT", summary=summary + ) + envelope = MessageEnvelope.domain_event( + schema_name=ACTIVITY_SCHEMA, + tenant_id="tenant", + aggregate_id=obj.id, + payload={ + "company_id": str(company.id), + "object_id": str(obj.id), + "revision": 3, + "data_digest": revision.data_digest, + "activity": notice.model_dump(), + }, + ) + event = SimpleNamespace( + id=envelope.message_id, + tenant_id="tenant", + topic=ACTIVITY_SCHEMA, + envelope=envelope.to_dict(), + ) + notification = ClaimedNotification( + uuid4(), "tenant", "BUSINESS_ACTIVITY", event.id, "ACTIVITY_RESULT", 1 + ) + return notification, event, obj, revision, company + + +def test_projection_binds_exact_revision_and_hides_content_by_default(): + notification, event, obj, revision, company = records() + subject = project_business_activity(notification, event, obj, revision, company) + assert subject and subject.revision == 3 + assert "Actual audio" not in str( + build_card(notification, subject, task_base_url=None, include_content=False) + ) + assert "Actual audio" in str( + build_card(notification, subject, task_base_url=None, include_content=True) + ) + + +@pytest.mark.parametrize("change", ["tenant", "digest", "revision", "company", "event", "extra"]) +def test_invalid_provenance_is_not_sent(change): + notification, event, obj, revision, company = records() + if change == "tenant": + company.tenant_id = "another" + elif change == "digest": + revision.data_digest = "changed" + elif change == "revision": + revision.revision = 4 + elif change == "company": + company.id = uuid4() + elif change == "event": + event.envelope["payload"]["activity"]["phase"] = "FAILED" + else: + event.envelope["payload"]["raw_memory"] = "must not leak" + assert project_business_activity(notification, event, obj, revision, company) is None + + +def test_credential_summary_is_withheld_without_losing_status(): + args = records("api_key=very-sensitive-secret-value") + subject = project_business_activity(*args) + assert subject and subject.summary is None + assert "sensitive" not in str( + build_card(args[0], subject, task_base_url=None, include_content=True) + ) + + +def test_notice_rejects_raw_unbounded_payload(): + with pytest.raises(ValidationError): + BusinessActivityNotice( + agent_id="critic", role_label="Critic", phase="RESULT", summary="x" * 501 + ) + with pytest.raises(ValidationError): + BusinessActivityNotice( + agent_id="critic", role_label="Critic", phase="RESULT", raw_memory="secret" + ) + + +@pytest.mark.parametrize("enabled,expected", [(False, 1), (True, 2)]) +def test_repository_queues_only_with_explicit_collaboration_gate(enabled, expected): + args = records() + session = SimpleNamespace(rows=[]) + session.add = session.rows.append + repository = SqlAlchemyOutboxRepository(session, feishu_collaboration_enabled=enabled) + repository.add(MessageEnvelope.from_dict(args[1].envelope)) + assert len(session.rows) == expected + if enabled: + job = session.rows[-1] + assert job.subject_type == "BUSINESS_ACTIVITY" and job.event_kind == "ACTIVITY_RESULT" + assert job.subject_revision == 3 and job.subject_id == args[1].id + + +@pytest.mark.parametrize("enabled,expected", [(False, "SKIPPED"), (True, "DELIVERED")]) +def test_worker_routes_to_employee_or_skips_when_disabled(enabled, expected): + args = records() + subject = project_business_activity(*args) + store = SimpleNamespace(status=None) + store.claim = lambda **kwargs: [args[0]] + store.subject = lambda notification: subject + store.finish = lambda notification, **kwargs: setattr(store, "status", kwargs["status"]) + generic = SimpleNamespace(send=lambda **kwargs: pytest.fail("Must use employee bot")) + employee = SimpleNamespace(sent=[]) + employee.send = lambda **kwargs: employee.sent.append(kwargs) + worker = FeishuNotificationWorker( + worker_id="test", + store=store, + client=generic, + task_base_url=None, + include_content=True, + sync_collaboration=enabled, + employee_clients={"critic": employee} if enabled else None, + send_interval_seconds=0, + ) + assert worker.run_once() == 1 + assert store.status == expected + assert len(employee.sent) == int(enabled) + + +def test_business_revision_and_notice_share_transaction( + company_service, business_object_service, uow_factory +): + from tests.test_business_objects import _company, _lead_type + + company, position = _company(company_service) + object_type = _lead_type(business_object_service, company.id) + created = business_object_service.create_object( + company.id, + type_id=object_type.id, + data={"name": "Ada", "email": "private@example.test"}, + owner_position_id=position.id, + actor="owner", + ) + notice = BusinessActivityNotice( + agent_id="employee", + role_label="Analyst", + phase="RESULT", + summary="Evidence-backed qualification complete", + ) + changed = business_object_service.apply_action( + company.id, + created.object.id, + action_key="qualify", + expected_revision=1, + input={"score": 95}, + actor="owner", + evidence_refs=["artifact:test"], + actor_position_key="sales-analyst", + actor_capabilities=["crm.qualify"], + activity=notice, + ) + events = [item for item in uow_factory.store.outbox if item.schema_name == ACTIVITY_SCHEMA] + assert len(events) == 1 + assert events[0].payload["revision"] == changed.object.current_revision == 2 + assert events[0].payload["data_digest"] == changed.revisions[-1].data_digest + assert "private@example.test" not in str(events[0].payload) + with pytest.raises(Exception, match="Stale"): + business_object_service.apply_action( + company.id, + created.object.id, + action_key="qualify", + expected_revision=1, + input={"score": 95}, + actor="owner", + activity=notice, + ) + assert ( + len([item for item in uow_factory.store.outbox if item.schema_name == ACTIVITY_SCHEMA]) == 1 + ) From 0e433617a4b678e69024bddaf242453cdf85686d Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Thu, 8 Oct 2026 17:04:51 +0800 Subject: [PATCH 2/8] feat: render grounded employee discussions as conversational Feishu cards --- .../integrations/feishu_notifications.py | 33 +++++++++++++++++-- tests/test_feishu_collaboration.py | 23 +++++++++++++ 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index 932be24b..bac685dc 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -113,6 +113,8 @@ class CollaborationSubject: status: str transfers: tuple[CollaborationTransfer, ...] output_summary: str | None + discussion_topic: str | None = None + discussion_turn: int | None = None @dataclass(frozen=True) @@ -336,6 +338,17 @@ def project_collaboration_subject( status=status, transfers=tuple(transfers), output_summary=output_summary, + discussion_topic=( + safe_content(role_input.get("discussion_topic"), limit=400) + if isinstance(role_input, dict) + and role_input.get("collaboration_mode") == "discussion" else None + ), + discussion_turn=( + role_input["discussion_turn"] + if isinstance(role_input, dict) + and type(role_input.get("discussion_turn")) is int + and 1 <= role_input["discussion_turn"] <= 20 else None + ), ) @@ -577,13 +590,29 @@ def build_card( f"\n员工：{subject.agent_id} · {subject.role_label}" "\n这是明确记录的工作状态或业务摘要，不是模型内部推理或自由群聊。" ) - elements = [{"tag": "div", "text": {"tag": "plain_text", "content": content}}] + elements = [] if include_content and subject.summary: elements.append( {"tag": "div", "text": {"tag": "plain_text", "content": subject.summary}} ) - return _card(title, elements) + elements.append({"tag": "note", "elements": [{"tag": "plain_text", "content": content}]}) + return _card(f"{subject.role_label} · {title.removeprefix('AgentMesh · ')}", elements) if isinstance(subject, CollaborationSubject): + if subject.discussion_topic and subject.discussion_turn: + if notification.event_kind == "COLLAB_RESULT": + reply = subject.output_summary if include_content else None + content = reply or "这条创作意见未开启外部内容同步，请在私有工作台查看。" + elif notification.event_kind == "COLLAB_STARTED": + content = "正在阅读前面的意见，准备回应。" + else: + content = "这次回应未完成；没有编造发言，请在工作台检查。" + elements = [{"tag": "div", "text": {"tag": "plain_text", "content": content}}] + elements.append({"tag": "note", "elements": [{"tag": "plain_text", "content": + f"讨论 {subject.discussion_turn} · {subject.discussion_topic}\n" + f"Task {subject.task_id} · Run {subject.run_id}\n" + "公开创作意见，不是内部推理或主人批准。"}]}) + _add_link(elements, task_base_url, query="task", subject_id=subject.task_id) + return _card(f"{subject.role_label} · 创作讨论", elements) elements: list[dict[str, object]] = [] def text(content: str) -> None: diff --git a/tests/test_feishu_collaboration.py b/tests/test_feishu_collaboration.py index 8ada575d..2a5ac310 100644 --- a/tests/test_feishu_collaboration.py +++ b/tests/test_feishu_collaboration.py @@ -89,6 +89,29 @@ def test_result_uses_succeeded_run_not_task_completed_and_is_provisional(exchang assert subject.output_summary == exchange.run.output["summary"] +def test_discussion_card_leads_with_real_reply_not_audit_metadata(exchange): + role = exchange.run.work_item_snapshot["work_item"]["input"]["subtask_input"] + role.update(collaboration_mode="discussion", discussion_topic="前奏与可唱性", discussion_turn=3) + exchange.run.output["summary"] = "词作人，我不同意整段删除；可以保留热汤意象，压缩其他句。" + notice, subject = exchange.project("COLLAB_RESULT") + card = build_card(notice, subject, task_base_url=None, include_content=True) + rendered = str(card) + assert "我不同意整段删除" in rendered + assert "创作讨论" in rendered and "不是内部推理或主人批准" in rendered + assert "never-send" not in rendered + assert "阶段业务摘要" not in rendered + hidden = str(build_card(notice, subject, task_base_url=None, include_content=False)) + assert "我不同意整段删除" not in hidden + + +def test_discussion_mode_does_not_export_secret_topic(exchange): + role = exchange.run.work_item_snapshot["work_item"]["input"]["subtask_input"] + role.update(collaboration_mode="discussion", + discussion_topic="Bearer private-credential", discussion_turn=True) + _, subject = exchange.project("COLLAB_RESULT") + assert subject.discussion_topic is None and subject.discussion_turn is None + + @pytest.mark.parametrize("status", ["FAILED", "CANCELED"]) def test_failed_run_does_not_export_error_body(exchange, status): exchange.run.status = status From cfec88a7df797c484f4729aaaea89d1b99a49b7f Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Fri, 9 Oct 2026 11:52:53 +0800 Subject: [PATCH 3/8] feat: collapse Feishu audit metadata and foreground employee discussion --- .../integrations/feishu_notifications.py | 118 +++++++++++++----- tests/test_feishu_collaboration.py | 7 +- tests/test_feishu_compact_cards.py | 55 ++++++++ 3 files changed, 151 insertions(+), 29 deletions(-) create mode 100644 tests/test_feishu_compact_cards.py diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index bac685dc..6700d448 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -115,6 +115,7 @@ class CollaborationSubject: output_summary: str | None discussion_topic: str | None = None discussion_turn: int | None = None + evidence_ids: tuple[str, ...] = () @dataclass(frozen=True) @@ -328,6 +329,31 @@ def project_collaboration_subject( if (notification.event_kind == "COLLAB_RESULT" and isinstance(run.output, dict)) else None ) + evidence_ids = () + if isinstance(role_input, dict) and role_input.get("collaboration_mode") == "discussion": + evidence = role_input.get("audio_evidence", []) + known = ( + { + e["id"] + for e in evidence + if isinstance(e, dict) + and isinstance(e.get("id"), str) + and re.fullmatch(r"[A-Za-z0-9:_-]{1,128}", e["id"]) + } + if isinstance(evidence, list) + else set() + ) + explicit = run.output.get("evidence_ids") if isinstance(run.output, dict) else None + if explicit is not None: + if ( + not isinstance(explicit, list) + or not 1 <= len(explicit) <= 14 + or any(not isinstance(ref, str) or ref not in known for ref in explicit) + ): + return None + evidence_ids = tuple(dict.fromkeys(explicit)) + elif output_summary: + evidence_ids = tuple(sorted(ref for ref in known if ref in output_summary)) return CollaborationSubject( task_id=task.id, run_id=run.id, @@ -338,16 +364,18 @@ def project_collaboration_subject( status=status, transfers=tuple(transfers), output_summary=output_summary, + evidence_ids=evidence_ids, discussion_topic=( safe_content(role_input.get("discussion_topic"), limit=400) - if isinstance(role_input, dict) - and role_input.get("collaboration_mode") == "discussion" else None + if isinstance(role_input, dict) and role_input.get("collaboration_mode") == "discussion" + else None ), discussion_turn=( role_input["discussion_turn"] if isinstance(role_input, dict) and type(role_input.get("discussion_turn")) is int - and 1 <= role_input["discussion_turn"] <= 20 else None + and 1 <= role_input["discussion_turn"] <= 20 + else None ), ) @@ -595,38 +623,57 @@ def build_card( elements.append( {"tag": "div", "text": {"tag": "plain_text", "content": subject.summary}} ) - elements.append({"tag": "note", "elements": [{"tag": "plain_text", "content": content}]}) - return _card(f"{subject.role_label} · {title.removeprefix('AgentMesh · ')}", elements) + elements.append(_details([_prose(content)])) + return _card(subject.role_label, elements) if isinstance(subject, CollaborationSubject): if subject.discussion_topic and subject.discussion_turn: + refs = ", ".join(subject.evidence_ids) if include_content else "内容同步关闭" if notification.event_kind == "COLLAB_RESULT": reply = subject.output_summary if include_content else None + if reply: + for ref in sorted(subject.evidence_ids, key=len, reverse=True): + reply = re.sub( + rf"(? None: elements.append({"tag": "div", "text": {"tag": "plain_text", "content": content}}) - text(f"任务 ID：{subject.task_id}\nRun ID：{subject.run_id}") - text( - f"时间：{subject.occurred_at.isoformat()}\n员工：{subject.agent_id}" - f"\n工作项：{subject.subtask_key} · {subject.role_label}" + details.append(_prose(f"任务 ID：{subject.task_id}\nRun ID：{subject.run_id}")) + details.append( + _prose( + f"时间：{subject.occurred_at.isoformat()}\n员工：{subject.agent_id}" + f"\n工作项：{subject.subtask_key} · {subject.role_label}" + ) ) if notification.event_kind == "COLLAB_STARTED": - text("以下为协调器已固定的输入交接，不代表自由聊天或模型内部推理。") + text("我接着处理这部分，先看看前面的材料。") + details.append(_prose("协调器已固定的输入交接，不是模型内部推理。")) elif notification.event_kind == "COLLAB_RESULT": - text("本次 Run 的阶段成果，尚不代表最终交付或人工批准。") + details.append(_prose("本次执行的阶段成果，尚不代表最终交付或人工批准。")) else: text("本次 Run 执行失败；详细诊断请在 AgentMesh 内查看。") incoming = ( @@ -636,25 +683,26 @@ def text(content: str) -> None: ) for transfer in incoming: label = "依赖结果" if transfer.kind == "DEPENDENCY_RESULT" else "已接受的交接" - text( - f"{label}：{transfer.source_agent_id or '未记录员工'} / {transfer.source_key}" - f" → {transfer.target_agent_id} / {transfer.target_key}" - f"\n内容指纹：{transfer.payload_sha256}" + details.append( + _prose( + f"{label}：{transfer.source_agent_id or '未记录员工'} / {transfer.source_key}" + f" → {transfer.target_agent_id} / {transfer.target_key}" + f"\n内容指纹：{transfer.payload_sha256}" + ) ) if include_content and transfer.summary: - text(f"已固定的业务摘要：{transfer.summary}") + text(transfer.summary) if include_content and notification.event_kind == "COLLAB_RESULT": text( - f"阶段业务摘要：{subject.output_summary}" + subject.output_summary if subject.output_summary else ("本次 Run 未记录可安全同步的业务摘要。") ) + elements.append(_details(details)) _add_link(elements, task_base_url, query="task", subject_id=subject.task_id) - return _card(title, elements) + return _card(subject.role_label, elements) subject_label = "任务 ID" if notification.subject_type == "TASK" else "审批请求 ID" - elements = [ - {"tag": "div", "text": {"tag": "plain_text", "content": f"{subject_label}：{subject.id}"}} - ] + elements = [] if include_content and isinstance(subject, TaskRecord): objective = safe_content(subject.objective, limit=300) if objective: @@ -676,8 +724,7 @@ def text(content: str) -> None: "text": { "tag": "plain_text", "content": ( - "执行已完成；交付物已启用独立验收，" - "请在 AgentMesh 查看验收状态。" + "执行已完成；交付物已启用独立验收，请在 AgentMesh 查看验收状态。" ), }, } @@ -706,10 +753,25 @@ def text(content: str) -> None: } ) query = "task" if notification.subject_type == "TASK" else "approval" + elements.append(_details([_prose(f"{subject_label}：{subject.id}")])) _add_link(elements, task_base_url, query=query, subject_id=subject.id) return _card(title, elements) +def _prose(content: str) -> dict[str, object]: + return {"tag": "div", "text": {"tag": "plain_text", "content": content}} + + +def _details(elements: list[dict[str, object]]) -> dict[str, object]: + """Feishu JSON 1.0 native panel; client-side expansion needs no callback.""" + return { + "tag": "collapsible_panel", + "expanded": False, + "header": {"title": {"tag": "plain_text", "content": "技术详情"}}, + "elements": elements, + } + + def _add_link( elements: list[dict[str, object]], task_base_url: str | None, diff --git a/tests/test_feishu_collaboration.py b/tests/test_feishu_collaboration.py index 2a5ac310..04491623 100644 --- a/tests/test_feishu_collaboration.py +++ b/tests/test_feishu_collaboration.py @@ -97,7 +97,12 @@ def test_discussion_card_leads_with_real_reply_not_audit_metadata(exchange): card = build_card(notice, subject, task_base_url=None, include_content=True) rendered = str(card) assert "我不同意整段删除" in rendered - assert "创作讨论" in rendered and "不是内部推理或主人批准" in rendered + assert "不是内部推理或主人批准" in rendered + assert card["header"]["title"]["content"] == subject.role_label + panel = card["elements"][1] + assert panel["tag"] == "collapsible_panel" and panel["expanded"] is False + assert str(subject.task_id) not in str(card["elements"][0]) + assert str(subject.run_id) in str(panel) assert "never-send" not in rendered assert "阶段业务摘要" not in rendered hidden = str(build_card(notice, subject, task_base_url=None, include_content=False)) diff --git a/tests/test_feishu_compact_cards.py b/tests/test_feishu_compact_cards.py new file mode 100644 index 00000000..88194c69 --- /dev/null +++ b/tests/test_feishu_compact_cards.py @@ -0,0 +1,55 @@ +from datetime import datetime, timezone +from uuid import uuid4 + +from agentmesh.integrations.feishu_notifications import ( + BusinessActivitySubject, + ClaimedNotification, + CollaborationSubject, + build_card, +) + + +def visible(card): + return str([e for e in card["elements"] if e["tag"] != "collapsible_panel"]) + + +def test_business_ids_and_disclaimers_are_collapsed_not_deleted(): + subject = BusinessActivitySubject( + uuid4(), + 3, + "music-agent", + "词作人", + "RESULT", + datetime.now(timezone.utc), + "这句留住灯光意象就好。", + ) + notice = ClaimedNotification(uuid4(), "t", "BUSINESS_ACTIVITY", uuid4(), "ACTIVITY_RESULT", 1) + card = build_card(notice, subject, task_base_url=None, include_content=True) + assert subject.summary in visible(card) and str(subject.object_id) not in visible(card) + panel = card["elements"][-1] + assert panel["expanded"] is False and str(subject.object_id) in str(panel) + assert "不是模型内部推理" in str(panel) + + +def test_discussion_keeps_citations_out_of_prose_and_technical_ids_in_panel(): + subject = CollaborationSubject( + uuid4(), + uuid4(), + "composer", + "response", + "编曲人", + datetime.now(timezone.utc), + "SUCCEEDED", + (), + "词作人，这句留着，尾奏再短点（audio-review）。", + "尾奏", + 3, + ("audio-review",), + ) + notice = ClaimedNotification(uuid4(), "t", "TASK_RUN", subject.run_id, "COLLAB_RESULT", 1) + card = build_card(notice, subject, task_base_url=None, include_content=True) + assert "这句留着" in visible(card) and "audio-review" not in visible(card) + assert str(subject.task_id) not in visible(card) + assert "audio-review" in str(card["elements"][-1]) + hidden = build_card(notice, subject, task_base_url=None, include_content=False) + assert "这句留着" not in str(hidden) From c686cebe12a7480127bd194b1e40028633dade1d Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Fri, 9 Oct 2026 11:55:31 +0800 Subject: [PATCH 4/8] test: bind compact discussion citations to pinned evidence --- tests/test_feishu_collaboration.py | 230 +++++++++++++++++++++-------- 1 file changed, 171 insertions(+), 59 deletions(-) diff --git a/tests/test_feishu_collaboration.py b/tests/test_feishu_collaboration.py index 04491623..94f46df7 100644 --- a/tests/test_feishu_collaboration.py +++ b/tests/test_feishu_collaboration.py @@ -22,47 +22,83 @@ @pytest.fixture def exchange(): now = datetime.now(timezone.utc) - task = TaskRecord(id=uuid4(), tenant_id="private", execution_mode="COORDINATED", - status="COMPLETED", input={"secret": "never-send-input"}) + task = TaskRecord( + id=uuid4(), + tenant_id="private", + execution_mode="COORDINATED", + status="COMPLETED", + input={"secret": "never-send-input"}, + ) source = SubtaskRecord(id=uuid4(), task_id=task.id, key="lyrics") target = SubtaskRecord(id=uuid4(), task_id=task.id, key="composition") - upstream = TaskRunRecord(id=uuid4(), task_id=task.id, subtask_id=source.id, - agent_id="lyricist", status="SUCCEEDED") - payload = {"summary": "歌词围绕夜归与灯光。", "lyrics": "private-full-lyrics", - "memory_candidates": [{"content": "never-send-memory"}]} - run = TaskRunRecord(id=uuid4(), task_id=task.id, subtask_id=target.id, - agent_id="composer", role="EXECUTOR", status="SUCCEEDED", - work_item_pinned_at=now, completed_at=now + timedelta(seconds=1), - output={"summary": "编曲使用已交接的歌词。", - "execution": "never-send-trace"}) + upstream = TaskRunRecord( + id=uuid4(), task_id=task.id, subtask_id=source.id, agent_id="lyricist", status="SUCCEEDED" + ) + payload = { + "summary": "歌词围绕夜归与灯光。", + "lyrics": "private-full-lyrics", + "memory_candidates": [{"content": "never-send-memory"}], + } + run = TaskRunRecord( + id=uuid4(), + task_id=task.id, + subtask_id=target.id, + agent_id="composer", + role="EXECUTOR", + status="SUCCEEDED", + work_item_pinned_at=now, + completed_at=now + timedelta(seconds=1), + output={"summary": "编曲使用已交接的歌词。", "execution": "never-send-trace"}, + ) transfer = { - "kind": "DEPENDENCY_RESULT", "source_subtask_id": str(source.id), - "source_key": source.key, "source_run_id": str(upstream.id), - "source_agent_id": upstream.agent_id, "target_subtask_id": str(target.id), - "target_key": target.key, "target_run_id": str(run.id), - "target_agent_id": run.agent_id, "payload": payload, + "kind": "DEPENDENCY_RESULT", + "source_subtask_id": str(source.id), + "source_key": source.key, + "source_run_id": str(upstream.id), + "source_agent_id": upstream.agent_id, + "target_subtask_id": str(target.id), + "target_key": target.key, + "target_run_id": str(run.id), + "target_agent_id": run.agent_id, + "payload": payload, "payload_sha256": sha256(canonical_json_bytes(payload)).hexdigest(), } run.work_item_snapshot = { "schema_version": 1, - "work_item": {"objective": "never-send-full-prompt", "input": { - "subtask_input": {"role": "编曲", "private": "never-send-context"}, - "dependency_outputs": {source.key: payload}, "accepted_handoffs": [], - "organizational_memory": {"content": "never-send-private-memory"}, - }}, "transfers": [transfer], + "work_item": { + "objective": "never-send-full-prompt", + "input": { + "subtask_input": {"role": "编曲", "private": "never-send-context"}, + "dependency_outputs": {source.key: payload}, + "accepted_handoffs": [], + "organizational_memory": {"content": "never-send-private-memory"}, + }, + }, + "transfers": [transfer], } records = {(SubtaskRecord, source.id): source, (TaskRunRecord, upstream.id): upstream} def project(kind="COLLAB_STARTED"): notice = ClaimedNotification(uuid4(), task.tenant_id, "TASK_RUN", run.id, kind, 1) return notice, project_collaboration_subject( - notice, task=task, run=run, subtask=target, + notice, + task=task, + run=run, + subtask=target, get_record=lambda model, key: records.get((model, key)), ) - return SimpleNamespace(task=task, source=source, target=target, upstream=upstream, - run=run, transfer=transfer, payload=payload, records=records, - project=project) + return SimpleNamespace( + task=task, + source=source, + target=target, + upstream=upstream, + run=run, + transfer=transfer, + payload=payload, + records=records, + project=project, + ) def test_real_delivery_card_only_selects_public_business_summary(exchange): @@ -111,36 +147,81 @@ def test_discussion_card_leads_with_real_reply_not_audit_metadata(exchange): def test_discussion_mode_does_not_export_secret_topic(exchange): role = exchange.run.work_item_snapshot["work_item"]["input"]["subtask_input"] - role.update(collaboration_mode="discussion", - discussion_topic="Bearer private-credential", discussion_turn=True) + role.update( + collaboration_mode="discussion", + discussion_topic="Bearer private-credential", + discussion_turn=True, + ) _, subject = exchange.project("COLLAB_RESULT") assert subject.discussion_topic is None and subject.discussion_turn is None +def test_separate_discussion_citations_are_bound_to_pinned_evidence(exchange): + role = exchange.run.work_item_snapshot["work_item"]["input"]["subtask_input"] + role.update( + collaboration_mode="discussion", + discussion_topic="歌词", + discussion_turn=2, + audio_evidence=[{"id": "finding-1", "text": "public finding"}], + ) + exchange.run.output.update( + summary="这句很好，但我建议把第二句缩短。", evidence_ids=["finding-1"] + ) + _, subject = exchange.project("COLLAB_RESULT") + assert subject.evidence_ids == ("finding-1",) + exchange.run.output["evidence_ids"] = ["invented"] + assert exchange.project("COLLAB_RESULT")[1] is None + + @pytest.mark.parametrize("status", ["FAILED", "CANCELED"]) def test_failed_run_does_not_export_error_body(exchange, status): exchange.run.status = status exchange.run.error = "private provider response and token" notice, subject = exchange.project("COLLAB_FAILED") assert subject is not None and subject.status == status - assert "private provider" not in str(build_card( - notice, subject, task_base_url=None, include_content=True, - )) + assert "private provider" not in str( + build_card( + notice, + subject, + task_base_url=None, + include_content=True, + ) + ) -@pytest.mark.parametrize("mutation", [ - "tenant", "task", "subtask", "supervisor", "schema", "digest", "source_task", - "source_run", "target_agent", "unpinned", "naive_timestamp", "missing_payload", -]) +@pytest.mark.parametrize( + "mutation", + [ + "tenant", + "task", + "subtask", + "supervisor", + "schema", + "digest", + "source_task", + "source_run", + "target_agent", + "unpinned", + "naive_timestamp", + "missing_payload", + ], +) def test_projection_fails_closed_on_untrusted_binding_or_evidence(exchange, mutation): if mutation == "tenant": exchange.task.tenant_id = "other" - notice = ClaimedNotification(uuid4(), "private", "TASK_RUN", exchange.run.id, - "COLLAB_STARTED", 1) - assert project_collaboration_subject( - notice, task=exchange.task, run=exchange.run, subtask=exchange.target, - get_record=lambda model, key: exchange.records.get((model, key)), - ) is None + notice = ClaimedNotification( + uuid4(), "private", "TASK_RUN", exchange.run.id, "COLLAB_STARTED", 1 + ) + assert ( + project_collaboration_subject( + notice, + task=exchange.task, + run=exchange.run, + subtask=exchange.target, + get_record=lambda model, key: exchange.records.get((model, key)), + ) + is None + ) return if mutation == "task": exchange.run.task_id = uuid4() @@ -167,12 +248,21 @@ def test_projection_fails_closed_on_untrusted_binding_or_evidence(exchange, muta assert exchange.project()[1] is None -@pytest.mark.parametrize("value", [ - "sk-test-secret", "mpg-test-secret", "AQ.example_secret", "Bearer secret", - "api_key=secret", "https://example.com/?access_token=secret", - "https://user:secret@example.com", "https://example.com/?%61pi_key=secret", - "safe text " + "x" * 501, "\ud800", -]) +@pytest.mark.parametrize( + "value", + [ + "sk-test-secret", + "mpg-test-secret", + "AQ.example_secret", + "Bearer secret", + "api_key=secret", + "https://example.com/?access_token=secret", + "https://user:secret@example.com", + "https://example.com/?%61pi_key=secret", + "safe text " + "x" * 501, + "\ud800", + ], +) def test_credential_surfaces_and_oversized_text_are_withheld(value): assert safe_content(value) is None @@ -192,9 +282,15 @@ def test_worker_delivers_historic_start_and_result_after_task_completed(exchange fail=lambda *args, **kwargs: pytest.fail("delivery should succeed"), ) client = SimpleNamespace(send=lambda **kwargs: sent.append(kwargs)) - worker = FeishuNotificationWorker(worker_id="test", store=store, client=client, - task_base_url=None, include_content=True, - sync_collaboration=True, sleep=slept.append) + worker = FeishuNotificationWorker( + worker_id="test", + store=store, + client=client, + task_base_url=None, + include_content=True, + sync_collaboration=True, + sleep=slept.append, + ) assert worker.run_once() == 2 assert finished == ["DELIVERED", "DELIVERED"] assert len(sent) == 2 and len(slept) == 1 @@ -205,13 +301,16 @@ def test_worker_skips_collaboration_when_extra_opt_in_is_off(exchange): notice, subject = exchange.project() finished = [] store = SimpleNamespace( - claim=lambda **kwargs: [notice], subject=lambda value: subject, + claim=lambda **kwargs: [notice], + subject=lambda value: subject, finish=lambda value, **kwargs: finished.append(kwargs["status"]), ) worker = FeishuNotificationWorker( - worker_id="test", store=store, + worker_id="test", + store=store, client=SimpleNamespace(send=lambda **kwargs: pytest.fail("no egress")), - task_base_url=None, include_content=True, + task_base_url=None, + include_content=True, ) assert worker.run_once() == 1 and finished == ["SKIPPED"] @@ -226,13 +325,26 @@ def test_entire_transfer_list_checked_not_only_displayed_prefix(exchange): def test_accepted_handoff_uses_pinned_summary_not_raw_arguments(exchange): transfer = exchange.transfer handoff_id = str(uuid4()) - payload = {key: transfer[key] for key in ( - "source_subtask_id", "source_run_id", "source_agent_id", "target_agent_id", - )} - payload.update(handoff_id=handoff_id, completed_work_summary="请沿用这份歌词。", - constraints={"private": "never-send-constraints"}) - transfer.update(kind="ACCEPTED_HANDOFF", handoff_id=handoff_id, payload=payload, - payload_sha256=sha256(canonical_json_bytes(payload)).hexdigest()) + payload = { + key: transfer[key] + for key in ( + "source_subtask_id", + "source_run_id", + "source_agent_id", + "target_agent_id", + ) + } + payload.update( + handoff_id=handoff_id, + completed_work_summary="请沿用这份歌词。", + constraints={"private": "never-send-constraints"}, + ) + transfer.update( + kind="ACCEPTED_HANDOFF", + handoff_id=handoff_id, + payload=payload, + payload_sha256=sha256(canonical_json_bytes(payload)).hexdigest(), + ) inputs = exchange.run.work_item_snapshot["work_item"]["input"] inputs["dependency_outputs"] = {} inputs["accepted_handoffs"] = [payload] From c6c25d9231a5d42e24d1d1493d74e879f8eb5574 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Fri, 9 Oct 2026 12:09:52 +0800 Subject: [PATCH 5/8] feat: suppress redundant typing notices in employee discussions --- .../integrations/feishu_notifications.py | 12 ++++++ tests/test_feishu_compact_cards.py | 38 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index 6700d448..8a6aa25e 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -819,6 +819,7 @@ def __init__( task_base_url: str | None, include_content: bool, sync_collaboration: bool = False, + discussion_results_only: bool = True, employee_clients: Mapping[str, FeishuClient] | None = None, employee_bot_fallback: bool = True, send_interval_seconds: float = 1.0, @@ -832,6 +833,7 @@ 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 + self._discussion_results_only = discussion_results_only 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" @@ -877,6 +879,16 @@ def run_once(self) -> int: ): self._store.finish(notification, worker_id=self._worker_id, status="SKIPPED") continue + if ( + self._discussion_results_only + and isinstance(subject, CollaborationSubject) + and subject.discussion_topic + and subject.discussion_turn + and notification.event_kind == "COLLAB_STARTED" + ): + # Keep the durable Run/event; do not flood a discussion with typing notices. + self._store.finish(notification, worker_id=self._worker_id, status="SKIPPED") + continue card = build_card( notification, subject, diff --git a/tests/test_feishu_compact_cards.py b/tests/test_feishu_compact_cards.py index 88194c69..57232342 100644 --- a/tests/test_feishu_compact_cards.py +++ b/tests/test_feishu_compact_cards.py @@ -1,10 +1,12 @@ from datetime import datetime, timezone +from types import SimpleNamespace from uuid import uuid4 from agentmesh.integrations.feishu_notifications import ( BusinessActivitySubject, ClaimedNotification, CollaborationSubject, + FeishuNotificationWorker, build_card, ) @@ -53,3 +55,39 @@ def test_discussion_keeps_citations_out_of_prose_and_technical_ids_in_panel(): assert "audio-review" in str(card["elements"][-1]) hidden = build_card(notice, subject, task_base_url=None, include_content=False) assert "这句留着" not in str(hidden) + + +def test_discussion_reading_notice_is_suppressed_but_actual_reply_delivered(): + subject = CollaborationSubject( + uuid4(), + uuid4(), + "composer", + "response", + "编曲人", + datetime.now(timezone.utc), + "SUCCEEDED", + (), + "保留这句。", + "歌词", + 2, + ) + notices = [ + ClaimedNotification(uuid4(), "t", "TASK_RUN", subject.run_id, kind, 1) + for kind in ("COLLAB_STARTED", "COLLAB_RESULT") + ] + finishes, sends = [], [] + store = SimpleNamespace( + claim=lambda **kwargs: notices, + subject=lambda n: subject, + finish=lambda n, **kwargs: finishes.append(kwargs["status"]), + ) + worker = FeishuNotificationWorker( + worker_id="test", + store=store, + client=SimpleNamespace(send=lambda **kw: sends.append(kw)), + task_base_url=None, + include_content=True, + sync_collaboration=True, + ) + assert worker.run_once() == 2 + assert finishes == ["SKIPPED", "DELIVERED"] and len(sends) == 1 From 33150dfc830d458f85eff07ce9bb7fa92ee8b404 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Mon, 12 Oct 2026 21:16:53 +0800 Subject: [PATCH 6/8] feat: bind trusted adapters to live principals and suppress private notifications --- .../application/identity_services.py | 24 +++++++- .../integrations/feishu_notifications.py | 15 +++++ tests/test_feishu_notifications.py | 59 ++++++++++++++++--- tests/test_persistent_identity.py | 16 +++++ 4 files changed, 104 insertions(+), 10 deletions(-) diff --git a/src/agentmesh/application/identity_services.py b/src/agentmesh/application/identity_services.py index 167daf7c..e54aded2 100644 --- a/src/agentmesh/application/identity_services.py +++ b/src/agentmesh/application/identity_services.py @@ -3,7 +3,7 @@ import hashlib import hmac import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import Any, Protocol from uuid import UUID @@ -119,6 +119,28 @@ def authorize(self, principal: PrincipalContext, permission: Permission) -> None f"Principal '{principal.principal_id}' lacks permission '{permission.value}'" ) + def resolve_configured_delegate(self, principal_id: str) -> PrincipalContext: + """Trusted in-process adapters only; not an authentication endpoint. + + The adapter must authenticate its external transport and explicitly bind + the external human to a deployment-configured principal. No role supplied + by an event is trusted, and persistent roles are read on every dispatch. + This avoids copying an operator's bearer credential into an adapter. + """ + matched = next((p for p in self._principals if p.principal_id == principal_id), None) + if not self.enabled or matched is None: + raise AuthenticationFailed("Configured delegate is unavailable") + self._validate_configured(matched) + if self.persistent: + try: + identity = UUID(principal_id) + except ValueError as exc: + raise AuthenticationFailed("Configured delegate ID must be a UUID") from exc + return self._persistent_context(identity, "trusted_adapter_binding") + return replace( + self._configured_context(matched), authentication_method="trusted_adapter_binding" + ) + def local_context(self) -> PrincipalContext: return PrincipalContext( principal_id="local-development", diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index 8a6aa25e..39a4ee3c 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -187,6 +187,10 @@ def project_collaboration_subject( """Project an exact Run's pinned coordinator evidence, never latest Subtask output.""" if ( notification.subject_type != "TASK_RUN" + or ( + isinstance(task.input, dict) + and task.input.get("agentmesh_notification_visibility") == "PRIVATE" + ) or notification.event_kind not in _COLLAB_EVENTS or run.id != notification.subject_id or task.tenant_id != notification.tenant_id @@ -487,6 +491,12 @@ def subject( subject = session.get(model, notification.subject_id) if subject is None or subject.tenant_id != notification.tenant_id: return None + if ( + isinstance(subject, TaskRecord) + and isinstance(subject.input, dict) + and (subject.input.get("agentmesh_notification_visibility") == "PRIVATE") + ): + return None session.expunge(subject) return subject @@ -872,6 +882,11 @@ def run_once(self) -> int: ) if ( stale_task + or ( + isinstance(subject, TaskRecord) + and isinstance(subject.input, dict) + and subject.input.get("agentmesh_notification_visibility") == "PRIVATE" + ) or stale_approval or invalid_collaboration or invalid_activity diff --git a/tests/test_feishu_notifications.py b/tests/test_feishu_notifications.py index fe8d5ede..7ef1a9f6 100644 --- a/tests/test_feishu_notifications.py +++ b/tests/test_feishu_notifications.py @@ -24,9 +24,9 @@ def _notification(kind: str = "COMPLETED") -> ClaimedNotification: def test_gate_is_explicit_opt_in_even_in_full_profile() -> None: assert not FeatureGateSet.from_config("full").is_enabled(Feature.FEISHU_NOTIFICATIONS) - assert FeatureGateSet.from_config( - "minimal", "feishu_notifications=true" - ).is_enabled(Feature.FEISHU_NOTIFICATIONS) + assert FeatureGateSet.from_config("minimal", "feishu_notifications=true").is_enabled( + Feature.FEISHU_NOTIFICATIONS + ) def test_notifier_rejects_missing_configuration_and_insecure_link() -> None: @@ -60,9 +60,7 @@ def test_card_hides_task_content_by_default_and_links_to_exact_task() -> None: rendered = str(card) assert "confidential" not in rendered assert f"https://mesh.example/?task={notification.subject_id}" in rendered - with_content = build_card( - notification, task, task_base_url=None, include_content=True - ) + with_content = build_card(notification, task, task_base_url=None, include_content=True) assert "confidential result" in str(with_content) @@ -136,7 +134,10 @@ def send(self, *, notification_id, card): store = Store() worker = FeishuNotificationWorker( - worker_id="test", store=store, client=Client(), task_base_url=None, + worker_id="test", + store=store, + client=Client(), + task_base_url=None, include_content=False, ) assert worker.run_once() == 1 @@ -166,13 +167,50 @@ def send(self, **kwargs): store = Store() worker = FeishuNotificationWorker( - worker_id="test", store=store, client=Client(), task_base_url=None, + worker_id="test", + store=store, + client=Client(), + task_base_url=None, include_content=False, ) assert worker.run_once() == 1 assert store.status == "SKIPPED" +def test_private_task_never_goes_to_group_even_when_content_sync_enabled(): + notification = _notification() + task = TaskRecord( + id=notification.subject_id, + tenant_id="tenant", + status="COMPLETED", + input={"agentmesh_notification_visibility": "PRIVATE"}, + output={"summary": "private expert conversation"}, + ) + + class Store: + status = None + + def claim(self, **kwargs): + return [notification] + + def subject(self, claimed): + return task + + def finish(self, claimed, **kwargs): + self.status = kwargs["status"] + + class Client: + def send(self, **kwargs): + pytest.fail("Private task must not be sent") + + store = Store() + worker = FeishuNotificationWorker( + worker_id="test", store=store, client=Client(), task_base_url=None, include_content=True + ) + assert worker.run_once() == 1 + assert store.status == "SKIPPED" + + def test_governed_approval_card_links_to_approval_view_without_arguments() -> None: notification = ClaimedNotification( uuid4(), "tenant", "GOVERNED_ACTION", uuid4(), "PENDING_APPROVAL", 1 @@ -218,7 +256,10 @@ def send(self, **kwargs): store = Store() worker = FeishuNotificationWorker( - worker_id="test", store=store, client=Client(), task_base_url=None, + worker_id="test", + store=store, + client=Client(), + task_base_url=None, include_content=False, ) assert worker.run_once() == 1 diff --git a/tests/test_persistent_identity.py b/tests/test_persistent_identity.py index ab5c7a83..37f21848 100644 --- a/tests/test_persistent_identity.py +++ b/tests/test_persistent_identity.py @@ -64,6 +64,22 @@ def _persistent_services(uow_factory: InMemoryUnitOfWorkFactory): return identity, administration +def test_trusted_delegate_has_live_roles_and_no_disabled_authentication_bypass(): + identity, administration = _persistent_services(InMemoryUnitOfWorkFactory()) + delegated = identity.resolve_configured_delegate(str(ADMIN_ID)) + assert delegated.authenticated + assert delegated.roles == frozenset({Role.TENANT_ADMIN}) + assert delegated.authentication_method == "trusted_adapter_binding" + with pytest.raises(AuthenticationFailed): + identity.resolve_configured_delegate("arbitrary-event-actor") + administration.change_status(ADMIN_ID, status=PrincipalStatus.SUSPENDED, actor=str(ADMIN_ID)) + with pytest.raises(AuthenticationFailed): + identity.resolve_configured_delegate(str(ADMIN_ID)) + disabled = IdentityService(enabled=False, tenant_id="test-tenant") + with pytest.raises(AuthenticationFailed): + disabled.resolve_configured_delegate(str(ADMIN_ID)) + + def test_persistent_roles_are_resolved_and_revoked_for_every_request() -> None: uow_factory = InMemoryUnitOfWorkFactory() identity, administration = _persistent_services(uow_factory) From 1ba1043a455e1d8e3d3e681697177003e15603af Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Sun, 18 Oct 2026 15:36:55 +0800 Subject: [PATCH 7/8] feat: support explicit knowledge snapshots without automatic memory --- .../application/memory_runtime_services.py | 5 ++++ tests/test_employee_memory_runtime_binding.py | 26 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/agentmesh/application/memory_runtime_services.py b/src/agentmesh/application/memory_runtime_services.py index 281659ba..b3b4051e 100644 --- a/src/agentmesh/application/memory_runtime_services.py +++ b/src/agentmesh/application/memory_runtime_services.py @@ -403,6 +403,11 @@ def _runtime_scope_in_unit_of_work( context = task.input.get("company_context") if not isinstance(context, dict): return None + # A caller may supply a governed, frozen knowledge snapshot explicitly. + # Keep company/workforce binding for scheduling, without hidden retrieval + # or automatic capture. This grants no permission to read any knowledge. + if context.get("memory_mode") == "EXPLICIT": + return None try: company_id = UUID(str(context["company_id"])) except (KeyError, ValueError): diff --git a/tests/test_employee_memory_runtime_binding.py b/tests/test_employee_memory_runtime_binding.py index 20a9c7ea..e7b78567 100644 --- a/tests/test_employee_memory_runtime_binding.py +++ b/tests/test_employee_memory_runtime_binding.py @@ -103,6 +103,32 @@ def _scopes(value): return result[2] +def test_explicit_memory_mode_preserves_work_item_without_search(employee_memory, monkeypatch): + from agentmesh.application.ports import WorkflowWorkItem + + value = employee_memory + value.task.input["company_context"]["memory_mode"] = "EXPLICIT" + monkeypatch.setattr(value.memory, "search", lambda *a, **k: pytest.fail("Hidden retrieval")) + item = WorkflowWorkItem(objective="Use the selected snapshot", input={"selected": []}) + assembled = value.runtime.assemble(value.task, value.run, item) + assert assembled.work_item is item and assembled.search is None + assert value.task.input["company_context"]["company_id"] == str(value.company.id) + + +def test_explicit_memory_mode_also_disables_capture_scope(employee_memory): + value = employee_memory + value.task.input["company_context"]["memory_mode"] = "EXPLICIT" + with value.uow_factory() as uow: + assert value.runtime._runtime_scope_in_unit_of_work(uow, value.task, value.run) is None + + +@pytest.mark.parametrize("mode", [None, "AUTOMATIC", False]) +def test_existing_memory_mode_keeps_automatic_scope(employee_memory, mode): + value = employee_memory + value.task.input["company_context"]["memory_mode"] = mode + assert _scopes(value) + + def test_employee_and_position_are_bound_without_caller_workforce(employee_memory): value = employee_memory scopes = _scopes(value) From 2885b86e05669fb9603097722bb582be4d14abfa Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Sun, 18 Oct 2026 15:52:25 +0800 Subject: [PATCH 8/8] docs: explain explicit runtime memory mode and authority boundary --- docs/model-and-memory-setup.md | 6 ++++++ docs/model-and-memory-setup.zh-CN.md | 5 +++++ 2 files changed, 11 insertions(+) diff --git a/docs/model-and-memory-setup.md b/docs/model-and-memory-setup.md index c339b239..bf754d1c 100644 --- a/docs/model-and-memory-setup.md +++ b/docs/model-and-memory-setup.md @@ -158,6 +158,12 @@ AgentMesh does not silently increase limits or accept a cut-off answer as comple ## What to verify +For trusted extensions that assemble their own policy-checked knowledge snapshots, setting +`company_context.memory_mode` to `EXPLICIT` disables automatic runtime retrieval **and capture** +for that Task, while retaining company/workforce scheduling bindings. It does not authorize any +memory read or validate caller-supplied knowledge: the extension must enforce ownership, policy, +review status, expiry and input integrity. Omitted modes retain existing automatic behavior. + - Executing employee versions and models match the configuration you selected. - Each step waits for its prerequisites, and intermediate output remains inspectable. - The real output meets your criteria; workflow success alone does not prove content quality. diff --git a/docs/model-and-memory-setup.zh-CN.md b/docs/model-and-memory-setup.zh-CN.md index afcd0b72..fe6722af 100644 --- a/docs/model-and-memory-setup.zh-CN.md +++ b/docs/model-and-memory-setup.zh-CN.md @@ -133,6 +133,11 @@ API 创建协作任务时可设置 `output_policy`： ## 验收时看什么 +可信扩展自行组装经过策略检查的固定知识时，可设置 `company_context.memory_mode=EXPLICIT`， +关闭该任务的自动运行时检索与记忆提取，但保留公司、员工调度绑定。这个字段不授予读取权限， +也不验证调用方传入的知识；扩展仍须检查归属、策略、审核状态、有效期与输入完整性。 +不设置时保留原有自动行为。 + - 员工实际执行的版本、模型和任务状态是否与选择一致。 - 每一步是否等待其前置工作完成，以及中间结果能否查看。 - 真实模型输出是否满足你的成功条件；流程成功不等于内容必然正确。