diff --git a/.env.example b/.env.example index a0018f9..1361aea 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,10 @@ AGENTMESH_FEISHU_CHAT_ID= AGENTMESH_FEISHU_TASK_BASE_URL= # Default cards contain only status and Task ID; opt in to objective/result excerpts. 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 +AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1 # Comma-separated trusted in-process extensions. Empty disables every installed extension. AGENTMESH_RUNTIME_EXTENSIONS=agentmesh.music-studio # Immutable operator allowlist used before importing third-party extension code. diff --git a/compose.yaml b/compose.yaml index cb75663..499bb24 100644 --- a/compose.yaml +++ b/compose.yaml @@ -31,6 +31,7 @@ x-agentmesh-environment: &agentmesh-environment # coordinated DAG across published employees. Operators can override the full # feature-gate set with an explicit value. AGENTMESH_FEATURE_GATES: ${AGENTMESH_FEATURE_GATES:-coordinated_execution=true,agent_registry_management=true} + AGENTMESH_FEISHU_SYNC_COLLABORATION: ${AGENTMESH_FEISHU_SYNC_COLLABORATION:-false} AGENTMESH_RUNTIME_EXTENSIONS: ${AGENTMESH_RUNTIME_EXTENSIONS:-agentmesh.music-studio} AGENTMESH_EXTENSION_LOCK_PATH: ${AGENTMESH_EXTENSION_LOCK_PATH:-/app/extensions.lock} AGENTMESH_IDENTITY_PRINCIPALS_JSON: ${AGENTMESH_IDENTITY_PRINCIPALS_JSON:-[]} @@ -171,6 +172,7 @@ services: AGENTMESH_FEISHU_CHAT_ID: ${AGENTMESH_FEISHU_CHAT_ID:-} 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} depends_on: migrate: condition: service_completed_successfully diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 1806c14..25cd39a 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -1,6 +1,6 @@ # Implementation status -Last updated: 2026-09-30 · Status: Alpha, supported single-team v1 baseline +Last updated: 2026-10-07 · Status: Alpha, supported single-team v1 baseline This page describes the current repository, not the full target architecture. The [v1 scope](v1-completion-scope.md) is implementation-complete and release-qualified for evaluation @@ -14,7 +14,7 @@ and non-critical single-team use. It is **not** a production HA or multi-tenant | Deliverable acceptance | Optional pinned Coordinated primary checks, typed/unit-aware rates, audited human decisions, and accepted-result export | Checks cover only configured fields/facts, not arbitrary prose truth or already executed tool actions; see [guide](deliverable-acceptance.md) | | Agents and models | Published immutable Agent Versions, role-bound execution, deterministic keyless demo, encrypted OpenAI/DeepSeek model connections in the authenticated Console | A private DeepSeek Direct and three-employee Coordinated run passed; OpenAI has fixture qualification, not a live-provider acceptance claim | | Collaboration and tools | Governed MCP registry/read tools/idempotent writes; trusted A2A peer registration, delegation, polling, cancellation, and unknown-outcome reconciliation | External providers/peers require configuration; irreversible writes and A2A streaming/push are not supported | -| External notifications | Optional Feishu bot cards for Task completion, failure, pending human review, and governed approval requests, backed by durable PostgreSQL delivery jobs | Gate is off by default; app credentials and a test group are required; inbound commands/approvals are not yet supported | +| External notifications | Optional Feishu bot cards for Task completion, failure, pending human review and governed approvals; separately opt-in live coordinated employee starts, received deliveries and provisional business summaries; durable PostgreSQL delivery jobs | Gate and collaboration sync are off by default; content excerpts require another opt-in; app credentials and a controlled group are required; inbound commands/approvals are not supported | | Governance | Optional Identity/RBAC, Policy approvals, one-time Permits, budgets, quotas, audit, usage, and Langfuse metadata export | Identity is not enabled by default; do not expose credential entry over anonymous public HTTP | | Company | Company/Position/Appointment, goals, Operations, Business Objects, reviewed PostgreSQL memory with optional subject-scoped review hints, internal finance controls, and installable Packs | No automatic semantic contradiction verdict; external memory ranking, accounting, payments, and most live connectors are not part of the default product | | Console | Guided task/team creation, model and employee setup, English/Chinese UI, work cards, pinned coordinated context transfers, and durable Mission Map replay | Transfer records start with new Runs; a dependency line alone is not evidence of an Agent message. The game-style Office is retired | diff --git a/docs/integrations/feishu-notifications.md b/docs/integrations/feishu-notifications.md index 3ff40b2..9655a69 100644 --- a/docs/integrations/feishu-notifications.md +++ b/docs/integrations/feishu-notifications.md @@ -42,8 +42,34 @@ The gate is off in every profile, including `full`. outbound-only release. If `policy_approval` is also enabled, pending governed actions produce separate approval cards. +## Live employee collaboration + +After enabling the notification gate, explicitly opt in to coordinated employee updates: + +```dotenv +AGENTMESH_FEISHU_SYNC_COLLABORATION=true +# Only if every group member may see business excerpts: +AGENTMESH_FEISHU_INCLUDE_CONTENT=true +AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1 +``` + +Restart API, Worker and notifier. This extra setting defaults off. Only coordinated execution +workers with pinned work-item evidence emit start, received-delivery, provisional-result and +failure cards; technical supervisors are not presented as employee discussion. Existing Tasks +are not automatically backfilled. Multiple incoming deliveries are grouped into one start card. + +These are actual work-item transfers and bounded public business summaries, not chain-of-thought, +full prompts, raw memory or free-form chat. Metadata-only mode hides summaries. Content opt-in +uses credential-pattern filtering, **not a guarantee of semantic confidentiality**: normal +business conclusions may use permitted knowledge. Use a controlled group and reviewed data +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. + ## Delivery and operations +The [live qualification record](../qualification/feishu-collaboration.md) documents a real +four-employee model run and nine acknowledged Feishu messages, with test scope and limitations. + Task transitions and notification jobs commit atomically in PostgreSQL. The notifier independently claims jobs, sends with the stable job UUID as Feishu's deduplication UUID, and retries transient errors with bounded backoff. After eight failed attempts a job becomes `DEAD`; inspect the diff --git a/docs/integrations/feishu-notifications.zh-CN.md b/docs/integrations/feishu-notifications.zh-CN.md index bccbd59..371e62b 100644 --- a/docs/integrations/feishu-notifications.zh-CN.md +++ b/docs/integrations/feishu-notifications.zh-CN.md @@ -38,8 +38,35 @@ 如果还开启了 `policy_approval`，待处理的治理审批也会产生单独卡片。 机器人需要在群内并有发言权限。单向通知不需要配置飞书回调地址。 +## 运行中的员工协作 + +在原有通知 Gate 开启后，可额外配置： + +```dotenv +AGENTMESH_FEISHU_SYNC_COLLABORATION=true +# 只有群成员有权看到业务内容时才开启摘要： +AGENTMESH_FEISHU_INCLUDE_CONTENT=true +AGENTMESH_FEISHU_SEND_INTERVAL_SECONDS=1 +``` + +重启 API、Worker 和通知进程。该选项默认关闭，适用于协作模式下具有已固定工作输入的执行员工； +不会自动回放旧任务，也不会将技术汇总节点冒充真实员工发言。 + +群内会看到「员工开始工作」「收到其他员工的交付」「阶段成果」「执行失败」卡片。 +接收关系来自真正固定的交接快照，成果来自该次执行的公开业务摘要，不是模型内部思考、 +完整提示词、原始记忆记录或自由聊天。每位员工的多个输入交接合并进一条开始卡片， +避免每条依赖单独刷屏。低负载时通常数秒内发送，积压、限流或重试时会延后。 + +不开启 `INCLUDE_CONTENT` 时仅发送身份、关系和状态。开启后会发送有长度限制、 +凭据模式遮盖的业务摘要；它不是任意敏感信息的语义脱敏保证，员工可能在正常业务结论中 +使用被允许读取的知识。因此只接入受控群，发送前确认数据权限，不把密钥填写到业务内容中。 +这不是飞书双向指挥功能：群中回复不会直接改变任务、员工记忆或批准作品。 + ## 可靠性与边界 +[真实验收记录](../qualification/feishu-collaboration.md)记录了四员工真实模型协作及九条飞书消息 +成功接收的结果，并明确验收范围和未覆盖的边界。 + 任务状态和通知待办在同一个 PostgreSQL 事务提交。通知进程独立领取待办，使用固定 UUID 去重，失败后有界退避重试；连续失败八次进入 `DEAD`，可先检查 `feishu_notifications` 表和通知日志，再由管理员手动恢复。状态已变化或过期的旧审批通知会跳过。飞书故障不会让任务失败。 diff --git a/docs/qualification/feishu-collaboration.md b/docs/qualification/feishu-collaboration.md new file mode 100644 index 0000000..258b724 --- /dev/null +++ b/docs/qualification/feishu-collaboration.md @@ -0,0 +1,48 @@ +# Live Feishu collaboration qualification + +[Setup and limits](../integrations/feishu-notifications.md) · [简体中文配置](../integrations/feishu-notifications.zh-CN.md) + +Validated on 2026-10-07 in an authenticated, private single-team deployment using the owner's +existing bot and controlled test group. Credentials and deployment addresses are not published. + +## Scenario and observed result + +A coordinated original-song planning task used four separately published employees backed by +the live DeepSeek model connection: creative director, researcher, lyricist, and composer. +The director supplied creative constraints; the researcher used the supplied evidence; lyrics +and arrangement work consumed pinned predecessor results. No live chart research, audio +generation or audio listening is claimed by this text-only qualification. + +The task completed, and the private scenario entered its waiting-for-audio stage. PostgreSQL +delivery records showed: + +| Event | Delivered | Attempts per event | +| --- | ---: | ---: | +| Employee started work, including pinned incoming deliveries | 4 | 1 | +| Employee provisional result | 4 | 1 | +| Task completed | 1 | 1 | + +All nine deliveries received successful acknowledgements from the real Feishu send API. +This confirms API acceptance, not a claim that a human read the messages. Transfers are actual +pinned collaboration evidence, not model internal reasoning or invented chat. + +## Regression evidence + +- 50 focused configuration, projection, filtering and notifier tests passed. +- 23 tests passed against an isolated PostgreSQL database, including atomic rollback, + duplicate-save protection, gate defaults, tenant/task/run binding and delivery after completion. +- The full non-PostgreSQL regression suite passed. +- GitHub CI passed quality, unit coverage, PostgreSQL integration, Compose E2E, + dependency review and CodeQL checks. + +## Limits + +The shared public demo keeps live collaboration sync disabled. New deployments default to +metadata-only notifications with both the parent feature gate and collaboration opt-in off. +Private business excerpts require explicit content opt-in. Credential-pattern filtering is not +semantic confidentiality; do not place secrets in Task objectives or employee summaries. +No prompts, raw memory, raw provider errors or full creative artifacts are selected for cards. +This is outbound notification, not two-way group chat or approval from Feishu. + +The live run did not exercise provider outage, exhausted retries, scale, rate limits, audio +review or duplicate delivery after Feishu's deduplication window; those are not qualified here. diff --git a/src/agentmesh/bootstrap.py b/src/agentmesh/bootstrap.py index c910720..a2cdc9d 100644 --- a/src/agentmesh/bootstrap.py +++ b/src/agentmesh/bootstrap.py @@ -365,6 +365,9 @@ def _database_components(settings: Settings): return engine, session_factory, SqlAlchemyUnitOfWorkFactory( session_factory, feishu_notifications_enabled=gates.is_enabled(Feature.FEISHU_NOTIFICATIONS), + feishu_collaboration_enabled=( + gates.is_enabled(Feature.FEISHU_NOTIFICATIONS) and settings.feishu_sync_collaboration + ), ) diff --git a/src/agentmesh/config.py b/src/agentmesh/config.py index fdd64c1..57bfb28 100644 --- a/src/agentmesh/config.py +++ b/src/agentmesh/config.py @@ -105,6 +105,8 @@ class Settings(BaseSettings): feishu_chat_id: str | None = None feishu_task_base_url: str | None = None feishu_include_content: bool = False + feishu_sync_collaboration: bool = False + 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) runtime_extensions: str = "agentmesh.music-studio" diff --git a/src/agentmesh/entrypoints/feishu_notifier.py b/src/agentmesh/entrypoints/feishu_notifier.py index cac7ed2..9292b57 100644 --- a/src/agentmesh/entrypoints/feishu_notifier.py +++ b/src/agentmesh/entrypoints/feishu_notifier.py @@ -63,6 +63,8 @@ def main() -> None: client=client, task_base_url=settings.feishu_task_base_url, include_content=settings.feishu_include_content, + sync_collaboration=settings.feishu_sync_collaboration, + send_interval_seconds=settings.feishu_send_interval_seconds, ) try: while True: diff --git a/src/agentmesh/infrastructure/postgres/repositories.py b/src/agentmesh/infrastructure/postgres/repositories.py index 3dc6fa3..961d335 100644 --- a/src/agentmesh/infrastructure/postgres/repositories.py +++ b/src/agentmesh/infrastructure/postgres/repositories.py @@ -1,5 +1,5 @@ from datetime import datetime, timezone -from uuid import UUID, uuid4 +from uuid import NAMESPACE_URL, UUID, uuid4, uuid5 from sqlalchemy import Select, delete, select from sqlalchemy import func as sa_func @@ -264,11 +264,13 @@ def _to_domain(record: TaskResolutionRecord) -> TaskResolution: class SqlAlchemyTaskRunRepository: - 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, run: TaskRun) -> None: self._session.add(self._to_record(run)) + self._queue_collaboration(run, previous_status=None, previously_pinned=False) def get(self, run_id: UUID, *, for_update: bool = False) -> TaskRun | None: record = self._session.get(TaskRunRecord, run_id, with_for_update=for_update) @@ -278,6 +280,8 @@ def save(self, run: TaskRun) -> None: record = self._session.get(TaskRunRecord, run.id) if record is None: raise LookupError(f"Task run record {run.id} was not found") + previous_status = record.status + previously_pinned = record.work_item_snapshot is not None if ( record.runtime_authority != run.runtime_authority or record.comparison_mode != run.comparison_mode @@ -311,6 +315,76 @@ def save(self, run: TaskRun) -> None: record.paused_from_status = ( run.paused_from_status.value if run.paused_from_status is not None else None ) + if record.task_id == run.task_id: + self._queue_collaboration( + run, previous_status=previous_status, previously_pinned=previously_pinned + ) + + def _queue_collaboration( + self, run: TaskRun, *, previous_status: str | None, previously_pinned: bool + ) -> None: + if ( + not self._feishu_collaboration_enabled + or run.role is not RunRole.EXECUTOR + or run.subtask_id is None + or run.work_item_pinned_at is None + ): + return + snapshot = run.work_item_snapshot + if not isinstance(snapshot, dict): + return + item = snapshot.get("work_item") + if ( + type(snapshot.get("schema_version")) is not int + or snapshot["schema_version"] != 1 + or not isinstance(item, dict) + or not isinstance(item.get("objective"), str) + or not item["objective"].strip() + or not isinstance(item.get("input"), dict) + or not isinstance(snapshot.get("transfers"), list) + ): + return + # Notification tenancy comes only from the persisted Task. Queue entries + # carry references, never prompt inputs, transfer payloads or model output. + with self._session.no_autoflush: + task = self._session.get(TaskRecord, run.task_id) + subtask = self._session.get(SubtaskRecord, run.subtask_id) + if ( + task is None + or task.execution_mode != TaskExecutionMode.COORDINATED.value + or subtask is None + or subtask.task_id != task.id + ): + return + events: list[tuple[str, datetime]] = [] + if not previously_pinned: + events.append(("COLLAB_STARTED", run.work_item_pinned_at)) + if previous_status != run.status.value: + if run.status is RunStatus.SUCCEEDED: + events.append(("COLLAB_RESULT", run.completed_at or run.work_item_pinned_at)) + elif run.status in {RunStatus.FAILED, RunStatus.CANCELED}: + events.append(("COLLAB_FAILED", run.completed_at or run.work_item_pinned_at)) + for event_kind, occurred_at in events: + self._session.execute( + insert(FeishuNotificationRecord) + .values( + id=uuid5(NAMESPACE_URL, f"agentmesh:feishu:run:{run.id}:{event_kind}"), + tenant_id=task.tenant_id, + subject_type="TASK_RUN", + subject_id=run.id, + subject_revision=1, + event_kind=event_kind, + status="PENDING", + created_at=occurred_at, + available_at=occurred_at, + claimed_by=None, + claimed_until=None, + attempt_count=0, + delivered_at=None, + last_error=None, + ) + .on_conflict_do_nothing() + ) def list_for_task(self, task_id: UUID, *, for_update: bool = False) -> list[TaskRun]: statement = ( diff --git a/src/agentmesh/infrastructure/postgres/uow.py b/src/agentmesh/infrastructure/postgres/uow.py index 36813e9..efabd6e 100644 --- a/src/agentmesh/infrastructure/postgres/uow.py +++ b/src/agentmesh/infrastructure/postgres/uow.py @@ -92,9 +92,13 @@ def __init__( session_factory: sessionmaker[Session], *, feishu_notifications_enabled: bool = False, + feishu_collaboration_enabled: bool = False, ) -> None: self._session_factory = session_factory self._feishu_notifications_enabled = feishu_notifications_enabled + self._feishu_collaboration_enabled = ( + feishu_notifications_enabled and feishu_collaboration_enabled + ) def __enter__(self) -> SqlAlchemyUnitOfWork: self._session = self._session_factory() @@ -118,7 +122,9 @@ def __enter__(self) -> SqlAlchemyUnitOfWork: self._session ) self.handoffs = SqlAlchemyHandoffRepository(self._session) - self.runs = SqlAlchemyTaskRunRepository(self._session) + self.runs = SqlAlchemyTaskRunRepository( + self._session, feishu_collaboration_enabled=self._feishu_collaboration_enabled + ) self.runtimes = SqlAlchemyRuntimeRepository(self._session) self.runtime_comparisons = SqlAlchemyRuntimeComparisonRepository(self._session) self.attempts = SqlAlchemyTaskAttemptRepository(self._session) @@ -177,12 +183,17 @@ def __init__( session_factory: sessionmaker[Session], *, feishu_notifications_enabled: bool = False, + feishu_collaboration_enabled: bool = False, ) -> None: self._session_factory = session_factory self._feishu_notifications_enabled = feishu_notifications_enabled + self._feishu_collaboration_enabled = ( + feishu_notifications_enabled and feishu_collaboration_enabled + ) def __call__(self) -> SqlAlchemyUnitOfWork: return SqlAlchemyUnitOfWork( self._session_factory, feishu_notifications_enabled=self._feishu_notifications_enabled, + feishu_collaboration_enabled=self._feishu_collaboration_enabled, ) diff --git a/src/agentmesh/integrations/feishu_notifications.py b/src/agentmesh/integrations/feishu_notifications.py index f183167..9454b11 100644 --- a/src/agentmesh/integrations/feishu_notifications.py +++ b/src/agentmesh/integrations/feishu_notifications.py @@ -4,9 +4,16 @@ import json import logging +import math +import re import time +import unicodedata +from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timedelta, timezone +from hashlib import sha256 +from typing import Any +from urllib.parse import parse_qsl, unquote, urlsplit from uuid import UUID import httpx @@ -16,8 +23,11 @@ from agentmesh.infrastructure.postgres.models import ( FeishuNotificationRecord, GovernedActionRecord, + SubtaskRecord, TaskRecord, + TaskRunRecord, ) +from agentmesh.runtime_sdk.canonical import canonical_json_bytes logger = logging.getLogger(__name__) _API_ROOT = "https://open.feishu.cn/open-apis" @@ -26,7 +36,213 @@ "FAILED": "AgentMesh · 任务失败", "WAITING_APPROVAL": "AgentMesh · 等待人工处理", "PENDING_APPROVAL": "AgentMesh · 等待治理审批", + "COLLAB_STARTED": "AgentMesh · 员工开始工作", + "COLLAB_RESULT": "AgentMesh · 阶段成果", + "COLLAB_FAILED": "AgentMesh · 执行失败", } +_COLLAB_EVENTS = frozenset({"COLLAB_STARTED", "COLLAB_RESULT", "COLLAB_FAILED"}) +_MAX_TRANSFERS = 20 +_SECRET = re.compile( + r"(?i)(?:\b(?:sk-|mpg-|AQ\.)[\w.~-]{4,}|\bbearer\s+\S+|" + r"\b(?:api[ _-]?key|access[ _-]?token|refresh[ _-]?token|authorization|" + r"password|client[ _-]?secret|app[ _-]?secret)\s*[:=]\s*\S+)" +) +_URL = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE) +_TOKEN_QUERY = re.compile( + r"(?i)(?:token|secret|password|credential|signature|api.?key|authorization|" + r"access.?key|^auth$|^key$|^sig$|^code$)" +) + + +def safe_content(value: object, *, limit: int = 500) -> str | None: + """Allow bounded prose, never truncate before inspecting its entire secret surface. + + This prevents known credential formats, not semantic disclosure of private + facts in otherwise innocuous prose. Raw inputs and memory are never selected. + """ + if not isinstance(value, str) or len(value) > limit: + return None + text = unicodedata.normalize("NFKC", value) + text = "".join(char for char in text if unicodedata.category(char) not in {"Cc", "Cf"}) + if any(0xD800 <= ord(char) <= 0xDFFF for char in text) or not text.strip(): + return None + inspected = unquote(unquote(text)) + if _SECRET.search(inspected): + return None + for match in _URL.finditer(inspected): + try: + url = urlsplit(match.group()) + if url.username is not None or url.password is not None: + return None + if any(_TOKEN_QUERY.search(key) for key, _ in parse_qsl(url.query)): + return None + except ValueError: + return None + return text.strip() if len(text) <= limit else None + + +@dataclass(frozen=True) +class CollaborationTransfer: + kind: str + source_key: str + source_agent_id: str | None + target_key: str + target_agent_id: str + payload_sha256: str + summary: str | None + + +@dataclass(frozen=True) +class CollaborationSubject: + task_id: UUID + run_id: UUID + agent_id: str + subtask_key: str + role_label: str + occurred_at: datetime + status: str + transfers: tuple[CollaborationTransfer, ...] + output_summary: str | None + + +def project_collaboration_subject( + notification: ClaimedNotification, + *, + task: TaskRecord, + run: TaskRunRecord, + subtask: SubtaskRecord, + get_record: Callable[[Any, UUID], Any], +) -> CollaborationSubject | None: + """Project an exact Run's pinned coordinator evidence, never latest Subtask output.""" + if ( + notification.subject_type != "TASK_RUN" + or notification.event_kind not in _COLLAB_EVENTS + or run.id != notification.subject_id + or task.tenant_id != notification.tenant_id + or task.id != run.task_id + or task.execution_mode != "COORDINATED" + or run.role != "EXECUTOR" + or subtask.id != run.subtask_id + or subtask.task_id != task.id + ): + return None + snapshot = run.work_item_snapshot + pinned_at = run.work_item_pinned_at + if ( + not isinstance(snapshot, dict) + or type(snapshot.get("schema_version")) is not int + or snapshot["schema_version"] != 1 + or not isinstance(snapshot.get("work_item"), dict) + or not isinstance(snapshot["work_item"].get("objective"), str) + or not snapshot["work_item"]["objective"].strip() + or not isinstance(snapshot["work_item"].get("input"), dict) + or not isinstance(snapshot.get("transfers"), list) + or not isinstance(pinned_at, datetime) + or pinned_at.tzinfo is None + ): + return None + try: + if len(canonical_json_bytes(snapshot)) > 262_144: + return None + except (TypeError, ValueError, RecursionError): + return None + if notification.event_kind == "COLLAB_STARTED": + occurred_at, status = pinned_at, "STARTED" + elif notification.event_kind == "COLLAB_RESULT" and run.status == "SUCCEEDED": + occurred_at, status = run.completed_at, "SUCCEEDED" + elif notification.event_kind == "COLLAB_FAILED" and run.status in {"FAILED", "CANCELED"}: + occurred_at, status = run.completed_at, run.status + else: + return None + if not isinstance(occurred_at, datetime) or occurred_at.tzinfo is None: + return None + agent_id = safe_content(run.agent_id, limit=128) + subtask_key = safe_content(subtask.key, limit=128) + if agent_id is None or subtask_key is None: + return None + input_value = snapshot["work_item"]["input"] + dependency_outputs = input_value.get("dependency_outputs", {}) + accepted_handoffs = input_value.get("accepted_handoffs", []) + 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 + 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", + }: + return None + try: + source_id = UUID(raw["source_subtask_id"]) + source_run_id = UUID(raw["source_run_id"]) if raw.get("source_run_id") else None + 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"] + 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") + )) + ): + return None + payload = raw["payload"] + digest = sha256(canonical_json_bytes(payload)).hexdigest() + if digest != raw["payload_sha256"]: + return None + if raw["kind"] == "DEPENDENCY_RESULT" and ( + raw["source_key"] not in dependency_outputs + or dependency_outputs[raw["source_key"]] != payload + ): + return None + if raw["kind"] == "ACCEPTED_HANDOFF": + UUID(raw["handoff_id"]) + 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", + ) + ): + 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): + 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 = ( + 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 + 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, + ) @dataclass(frozen=True) @@ -82,8 +298,25 @@ def claim(self, *, worker_id: str, limit: int = 5) -> list[ClaimedNotification]: def subject( self, notification: ClaimedNotification - ) -> TaskRecord | GovernedActionRecord | None: + ) -> TaskRecord | GovernedActionRecord | CollaborationSubject | None: + if notification.tenant_id != self._tenant_id: + return None with self._session_factory() as session: + 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) + ) + if run is None or run.subtask_id is None: + return None + task = session.get(TaskRecord, run.task_id) + subtask = session.get(SubtaskRecord, run.subtask_id) + 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, + ) model = ( TaskRecord if notification.subject_type == "TASK" else GovernedActionRecord if notification.subject_type == "GOVERNED_ACTION" else None @@ -211,29 +444,62 @@ def send(self, *, notification_id: UUID, card: dict[str, object]) -> None: def build_card( notification: ClaimedNotification, - subject: TaskRecord | GovernedActionRecord, + subject: TaskRecord | GovernedActionRecord | CollaborationSubject, *, task_base_url: str | None, include_content: bool, ) -> dict[str, object]: title = _EVENT_TITLES[notification.event_kind] + if isinstance(subject, CollaborationSubject): + elements: list[dict[str, object]] = [] + + 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}") + 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 () + 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}") + 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 未记录可安全同步的业务摘要。" + )) + _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" - elements: list[dict[str, object]] = [ + elements = [ {"tag": "div", "text": {"tag": "plain_text", "content": f"{subject_label}：{subject.id}"}} ] if include_content and isinstance(subject, TaskRecord): - elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": subject.objective[:300]}} - ) + objective = safe_content(subject.objective, limit=300) + if objective: + elements.append( + {"tag": "div", "text": {"tag": "plain_text", "content": objective}} + ) if ( notification.event_kind == "COMPLETED" and isinstance(subject.output, dict) and "agentmesh_deliverable_acceptance" not in (subject.input or {}) ): - summary = subject.output.get("summary") - if isinstance(summary, str) and summary.strip(): + summary = safe_content(subject.output.get("summary")) + if summary: elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": summary[:500]}} + {"tag": "div", "text": {"tag": "plain_text", "content": summary}} ) elif ( notification.event_kind == "COMPLETED" @@ -243,7 +509,8 @@ def build_card( "执行已完成；交付物已启用独立验收，请在 AgentMesh 查看验收状态。"}}) elif notification.event_kind == "FAILED" and subject.error: elements.append( - {"tag": "div", "text": {"tag": "plain_text", "content": subject.error[:300]}} + {"tag": "div", "text": {"tag": "plain_text", "content": + "执行失败；详细诊断请在 AgentMesh 内查看。"}} ) elif include_content and isinstance(subject, GovernedActionRecord): elements.append( @@ -251,12 +518,27 @@ def build_card( "tag": "div", "text": { "tag": "plain_text", - "content": f"{subject.action_type} · {subject.resource_type}", + "content": safe_content( + f"{subject.action_type} · {subject.resource_type}", limit=160 + ) or "治理审批请求", }, } ) - if task_base_url: - query = "task" if notification.subject_type == "TASK" else "approval" + query = "task" if notification.subject_type == "TASK" else "approval" + _add_link(elements, task_base_url, query=query, subject_id=subject.id) + return _card(title, elements) + + +def _add_link( + 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: + url = urlsplit(task_base_url) + if url.scheme != "https" or not url.hostname or url.query or url.fragment: + return + except ValueError: + return elements.append( { "tag": "action", @@ -265,11 +547,14 @@ def build_card( "tag": "button", "text": {"tag": "plain_text", "content": "在 AgentMesh 查看"}, "type": "primary", - "url": f"{task_base_url.rstrip('/')}/?{query}={subject.id}", + "url": f"{task_base_url.rstrip('/')}/?{query}={subject_id}", } ], } ) + + +def _card(title: str, elements: list[dict[str, object]]) -> dict[str, object]: return { "config": {"wide_screen_mode": True}, "header": {"title": {"tag": "plain_text", "content": title}, "template": "blue"}, @@ -286,17 +571,29 @@ def __init__( client: FeishuClient, task_base_url: str | None, include_content: bool, + sync_collaboration: bool = False, + send_interval_seconds: float = 1.0, + sleep: Callable[[float], None] = time.sleep, ) -> None: self._worker_id = worker_id self._store = store self._client = client self._task_base_url = task_base_url self._include_content = include_content + 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._send_interval_seconds = send_interval_seconds + self._sleep = sleep + self._last_send_at: float | None = None 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: + self._store.finish(notification, worker_id=self._worker_id, status="SKIPPED") + continue subject = self._store.subject(notification) stale_task = ( notification.subject_type == "TASK" @@ -310,17 +607,26 @@ def run_once(self) -> int: or subject.expires_at <= datetime.now(timezone.utc) ) ) - if stale_task or stale_approval or subject is None: + invalid_collaboration = ( + notification.subject_type == "TASK_RUN" + and not isinstance(subject, CollaborationSubject) + ) + if stale_task or stale_approval or invalid_collaboration 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, + include_content=self._include_content, + ) + if self._last_send_at is not None: + elapsed = time.monotonic() - self._last_send_at + remaining = self._send_interval_seconds - elapsed + if remaining > 0: + self._sleep(remaining) + self._last_send_at = time.monotonic() self._client.send( notification_id=notification.id, - card=build_card( - notification, - subject, - task_base_url=self._task_base_url, - include_content=self._include_content, - ), + card=card, ) self._store.finish(notification, worker_id=self._worker_id, status="DELIVERED") except Exception as exc: @@ -331,6 +637,8 @@ def run_once(self) -> int: self._store.fail( notification, worker_id=self._worker_id, - error=f"{type(exc).__name__}: {str(exc)[:180]}", + error=f"{type(exc).__name__}: " + ( + safe_content(str(exc), limit=180) or "delivery error details withheld" + ), ) return len(claimed) diff --git a/tests/integration/test_feishu_notifications_postgres.py b/tests/integration/test_feishu_notifications_postgres.py index bb16dc2..7eee36a 100644 --- a/tests/integration/test_feishu_notifications_postgres.py +++ b/tests/integration/test_feishu_notifications_postgres.py @@ -2,16 +2,17 @@ import os from datetime import datetime, timedelta, timezone -from uuid import uuid4 +from uuid import NAMESPACE_URL, uuid4, uuid5 import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import Session, sessionmaker from agentmesh.config import get_settings +from agentmesh.domain.coordination import Subtask from agentmesh.domain.policy import GovernedAction, GovernedActionType, PolicyResult -from agentmesh.domain.tasks import Task, TaskStatus -from agentmesh.infrastructure.postgres.models import FeishuNotificationRecord +from agentmesh.domain.tasks import RunRole, RunStatus, Task, TaskExecutionMode, TaskRun, TaskStatus +from agentmesh.infrastructure.postgres.models import FeishuNotificationRecord, TaskRecord from agentmesh.infrastructure.postgres.uow import SqlAlchemyUnitOfWorkFactory from agentmesh.integrations.feishu_notifications import ( FeishuNotificationStore, @@ -144,3 +145,308 @@ def test_governed_approval_creates_a_separate_notification_job() -> None: assert job.event_kind == "PENDING_APPROVAL" finally: engine.dispose() + + +@pytest.fixture +def collaboration_sessions(): + engine = create_engine(get_settings().database_url) + sessions = sessionmaker(bind=engine, expire_on_commit=False, class_=Session) + try: + yield sessions + finally: + engine.dispose() + + +def _run_fixture(sessions, *, role=RunRole.EXECUTOR, mode=TaskExecutionMode.COORDINATED): + task = Task.create( + tenant_id=f"feishu-collaboration-{uuid4().hex}", + objective="Collaborate without exposing private work input", + execution_mode=mode, + plan_version=1 if mode is TaskExecutionMode.COORDINATED else None, + plan_digest=f"sha256:{'a' * 64}" if mode is TaskExecutionMode.COORDINATED else None, + ) + subtask = Subtask.create( + subtask_id=uuid4(), + task_id=task.id, + key="analysis", + objective="Analyze supplied input", + input={}, + required_capabilities=(), + preferred_agent_id=None, + initially_ready=True, + ) + factory = SqlAlchemyUnitOfWorkFactory(sessions) + with factory() as uow: + uow.tasks.add(task) + uow.flush() + uow.subtasks.add(subtask) + uow.commit() + run = TaskRun.request(task.id, "collaboration-test-agent", role=role, subtask_id=subtask.id) + return task, subtask, run + + +def _snapshot(): + return { + "schema_version": 1, + "work_item": {"objective": "Bound input", "input": {"private": "do not queue this"}}, + "transfers": [], + } + + +def _jobs(sessions, task): + with sessions() as session: + return list( + session.scalars( + select(FeishuNotificationRecord).where( + FeishuNotificationRecord.tenant_id == task.tenant_id + ) + ) + ) + + +def _enabled(sessions): + return SqlAlchemyUnitOfWorkFactory( + sessions, feishu_notifications_enabled=True, feishu_collaboration_enabled=True + ) + + +def test_real_store_delivers_historical_start_and_succeeded_result(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.start() + run.pin_work_item_snapshot(_snapshot()) + run.succeed({"summary": "Verified public work summary", "private": "not for chat"}) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + with sessions() as session, session.begin(): + session.get(TaskRecord, task.id).status = "COMPLETED" + + class Client: + sent = [] + + def send(self, **kwargs): + self.sent.append(kwargs) + + client = Client() + worker = FeishuNotificationWorker( + worker_id="test", store=FeishuNotificationStore(sessions, tenant_id=task.tenant_id), + client=client, task_base_url=None, include_content=True, sync_collaboration=True, + send_interval_seconds=0, + ) + assert worker.run_once() == 2 + assert worker.run_once() == 0 + assert len(client.sent) == 2 + rendered = str(client.sent) + assert "Verified public work summary" in rendered + assert "not for chat" not in rendered and "do not queue this" not in rendered + assert {value.status for value in _jobs(sessions, task)} == {"DELIVERED"} + + +def test_collaboration_pin_and_terminal_notification_are_atomic_and_deduplicated( + collaboration_sessions, +): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + factory = _enabled(sessions) + with factory() as uow: + uow.runs.add(run) + uow.commit() + run.start() + run.pin_work_item_snapshot(_snapshot()) + with factory() as uow: + uow.runs.save(run) + uow.rollback() + assert _jobs(sessions, task) == [] + with factory() as uow: + persisted = uow.runs.get(run.id) + assert persisted.work_item_snapshot is None + assert persisted.status is RunStatus.QUEUED + uow.runs.save(run) + uow.commit() + run.succeed({"summary": "Done", "private": "do not queue output"}) + with factory() as uow: + uow.runs.save(run) + uow.runs.save(run) + uow.commit() + with factory() as uow: + uow.runs.save(run) + uow.commit() + jobs = _jobs(sessions, task) + assert {job.event_kind for job in jobs} == {"COLLAB_STARTED", "COLLAB_RESULT"} + assert len(jobs) == 2 + for job in jobs: + assert job.subject_type == "TASK_RUN" + assert job.subject_id == run.id + assert job.subject_revision == 1 + assert job.id == uuid5(NAMESPACE_URL, f"agentmesh:feishu:run:{run.id}:{job.event_kind}") + assert job.status == "PENDING" + + +def test_adding_pinned_terminal_run_creates_both_notices(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.start() + run.pin_work_item_snapshot(_snapshot()) + run.succeed({"summary": "Done"}) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert {job.event_kind for job in _jobs(sessions, task)} == {"COLLAB_STARTED", "COLLAB_RESULT"} + + +@pytest.mark.parametrize("terminal", [RunStatus.FAILED, RunStatus.CANCELED]) +def test_pinned_failure_or_cancellation_creates_one_failure_notice( + collaboration_sessions, terminal +): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.start() + run.pin_work_item_snapshot(_snapshot()) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + if terminal is RunStatus.FAILED: + run.fail("Private failure details stay in the execution record") + else: + run.cancel() + with _enabled(sessions)() as uow: + uow.runs.save(run) + uow.runs.save(run) + uow.commit() + assert {job.event_kind for job in _jobs(sessions, task)} == {"COLLAB_STARTED", "COLLAB_FAILED"} + + +@pytest.mark.parametrize("parent,collaboration", [(False, False), (False, True), (True, False)]) +def test_collaboration_requires_both_parent_and_child_opt_in( + collaboration_sessions, + parent, + collaboration, +): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.pin_work_item_snapshot(_snapshot()) + factory = SqlAlchemyUnitOfWorkFactory( + sessions, feishu_notifications_enabled=parent, feishu_collaboration_enabled=collaboration + ) + with factory() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + + +@pytest.mark.parametrize("role", [RunRole.SUPERVISOR, RunRole.REVIEWER]) +def test_only_executor_work_items_create_collaboration_notices(collaboration_sessions, role): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions, role=role) + run.pin_work_item_snapshot(_snapshot()) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + + +@pytest.mark.parametrize("terminal", [RunStatus.SUCCEEDED, RunStatus.CANCELED]) +def test_unpinned_terminal_run_does_not_claim_actual_collaboration( + collaboration_sessions, terminal +): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.status = terminal + run.completed_at = datetime.now(timezone.utc) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + + +@pytest.mark.parametrize( + "bad", + [ + "boolean_version", + "missing_transfers", + "invalid_work_item", + "empty_objective", + "invalid_input", + ], +) +def test_malformed_snapshot_never_creates_collaboration_notices(collaboration_sessions, bad): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + snapshot = _snapshot() + if bad == "boolean_version": + snapshot["schema_version"] = True + elif bad == "missing_transfers": + del snapshot["transfers"] + elif bad == "invalid_work_item": + snapshot["work_item"] = [] + elif bad == "empty_objective": + snapshot["work_item"]["objective"] = " " + else: + snapshot["work_item"]["input"] = [] + run.pin_work_item_snapshot(snapshot) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + + +def test_cross_task_subtask_reference_does_not_leak_to_either_tenant(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + other_task, other_subtask, _ = _run_fixture(sessions) + run.subtask_id = other_subtask.id + run.pin_work_item_snapshot(_snapshot()) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + assert _jobs(sessions, other_task) == [] + + +def test_direct_task_run_never_creates_collaboration_notices(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions, mode=TaskExecutionMode.DIRECT) + run.pin_work_item_snapshot(_snapshot()) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + assert _jobs(sessions, task) == [] + + +def test_terminal_result_and_notice_rollback_together(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + run.start() + run.pin_work_item_snapshot(_snapshot()) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + run.succeed({"summary": "Do not announce a rolled-back result"}) + with _enabled(sessions)() as uow: + uow.runs.save(run) + uow.rollback() + assert {job.event_kind for job in _jobs(sessions, task)} == {"COLLAB_STARTED"} + with _enabled(sessions)() as uow: + persisted = uow.runs.get(run.id) + assert persisted.status is RunStatus.RUNNING + assert persisted.output is None + uow.runs.save(run) + uow.commit() + assert {job.event_kind for job in _jobs(sessions, task)} == {"COLLAB_STARTED", "COLLAB_RESULT"} + + +def test_spoofed_domain_task_identity_cannot_redirect_collaboration_notice(collaboration_sessions): + sessions = collaboration_sessions + task, _, run = _run_fixture(sessions) + other, _, _ = _run_fixture(sessions) + with _enabled(sessions)() as uow: + uow.runs.add(run) + uow.commit() + run.pin_work_item_snapshot(_snapshot()) + run.task_id = other.id + with _enabled(sessions)() as uow: + uow.runs.save(run) + uow.commit() + assert _jobs(sessions, task) == [] + assert _jobs(sessions, other) == [] diff --git a/tests/test_feishu_collaboration.py b/tests/test_feishu_collaboration.py new file mode 100644 index 0000000..8ada575 --- /dev/null +++ b/tests/test_feishu_collaboration.py @@ -0,0 +1,215 @@ +"""Project only real pinned work exchanges, not private context or imagined chat.""" + +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from hashlib import sha256 +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from agentmesh.infrastructure.postgres.models import SubtaskRecord, TaskRecord, TaskRunRecord +from agentmesh.integrations.feishu_notifications import ( + ClaimedNotification, + FeishuNotificationWorker, + build_card, + project_collaboration_subject, + safe_content, +) +from agentmesh.runtime_sdk.canonical import canonical_json_bytes + + +@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"}) + 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"}) + 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, + "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], + } + 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, + 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) + + +def test_real_delivery_card_only_selects_public_business_summary(exchange): + notice, subject = exchange.project() + assert subject is not None + rendered = str(build_card(notice, subject, task_base_url=None, include_content=True)) + assert "lyricist / lyrics → composer / composition" in rendered + assert "歌词围绕夜归与灯光" in rendered + assert "内容指纹" in rendered + assert "never-send" not in rendered + assert "private-full-lyrics" not in rendered + metadata = str(build_card(notice, subject, task_base_url=None, include_content=False)) + assert "歌词围绕夜归与灯光" not in metadata + + +def test_result_uses_succeeded_run_not_task_completed_and_is_provisional(exchange): + notice, subject = exchange.project("COLLAB_RESULT") + assert subject is not None and subject.status == "SUCCEEDED" + card = str(build_card(notice, subject, task_base_url=None, include_content=True)) + assert "编曲使用已交接的歌词" in card + assert "尚不代表最终交付或人工批准" in card + exchange.target.output = {"summary": "wrong-latest-retry-output"} + _, subject = exchange.project("COLLAB_RESULT") + assert subject.output_summary == exchange.run.output["summary"] + + +@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, + )) + + +@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 + return + if mutation == "task": + exchange.run.task_id = uuid4() + elif mutation == "subtask": + exchange.target.task_id = uuid4() + elif mutation == "supervisor": + exchange.run.role = "SUPERVISOR" + elif mutation == "schema": + exchange.run.work_item_snapshot["schema_version"] = True + elif mutation == "digest": + exchange.transfer["payload_sha256"] = "0" * 64 + elif mutation == "source_task": + exchange.source.task_id = uuid4() + elif mutation == "source_run": + exchange.upstream.agent_id = "wrong-agent" + elif mutation == "target_agent": + exchange.transfer["target_agent_id"] = "wrong-agent" + elif mutation == "unpinned": + exchange.run.work_item_pinned_at = None + elif mutation == "naive_timestamp": + exchange.run.work_item_pinned_at = datetime.now() + elif mutation == "missing_payload": + del exchange.transfer["payload"] + 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", +]) +def test_credential_surfaces_and_oversized_text_are_withheld(value): + assert safe_content(value) is None + + +def test_control_characters_removed_and_plain_text_escapes_chat_mentions(): + assert safe_content("夜\u200b归\x00的灯") == "夜归的灯" + + +def test_worker_delivers_historic_start_and_result_after_task_completed(exchange): + items = [exchange.project(kind) for kind in ("COLLAB_STARTED", "COLLAB_RESULT")] + finished, sent, slept = [], [], [] + subjects = {notice.id: subject for notice, subject in items} + store = SimpleNamespace( + claim=lambda **kwargs: [notice for notice, _ in items], + subject=lambda notice: subjects[notice.id], + finish=lambda notice, **kwargs: finished.append(kwargs["status"]), + 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) + assert worker.run_once() == 2 + assert finished == ["DELIVERED", "DELIVERED"] + assert len(sent) == 2 and len(slept) == 1 + assert [value["notification_id"] for value in sent] == [notice.id for notice, _ in items] + + +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, + finish=lambda value, **kwargs: finished.append(kwargs["status"]), + ) + worker = FeishuNotificationWorker( + worker_id="test", store=store, + client=SimpleNamespace(send=lambda **kwargs: pytest.fail("no egress")), + task_base_url=None, include_content=True, + ) + assert worker.run_once() == 1 and finished == ["SKIPPED"] + + +def test_entire_transfer_list_checked_not_only_displayed_prefix(exchange): + snapshot = exchange.run.work_item_snapshot + snapshot["transfers"] = [deepcopy(exchange.transfer) for _ in range(21)] + snapshot["transfers"][-1]["payload_sha256"] = "0" * 64 + assert exchange.project()[1] is None + + +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()) + inputs = exchange.run.work_item_snapshot["work_item"]["input"] + inputs["dependency_outputs"] = {} + inputs["accepted_handoffs"] = [payload] + notice, subject = exchange.project() + assert subject is not None + rendered = str(build_card(notice, subject, task_base_url=None, include_content=True)) + assert "已接受的交接" in rendered and "请沿用这份歌词" in rendered + assert "never-send-constraints" not in rendered diff --git a/tests/test_feishu_configuration.py b/tests/test_feishu_configuration.py new file mode 100644 index 0000000..3e1ff07 --- /dev/null +++ b/tests/test_feishu_configuration.py @@ -0,0 +1,45 @@ +"""Collaboration egress requires both explicit notification and detail opt-ins.""" + +import pytest +from pydantic import ValidationError + +from agentmesh import bootstrap +from agentmesh.config import Settings + + +@pytest.mark.parametrize( + ("gates", "requested", "enabled"), + [ + ("", False, False), + ("", True, False), + ("feishu_notifications=true", False, False), + ("feishu_notifications=true", True, True), + ], +) +def test_collaboration_queue_requires_parent_gate(monkeypatch, gates, requested, enabled): + captured = {} + monkeypatch.setattr(bootstrap, "create_engine", lambda *args, **kwargs: object()) + monkeypatch.setattr(bootstrap, "sessionmaker", lambda **kwargs: object()) + + def factory(*args, **kwargs): + captured.update(kwargs) + return object() + + monkeypatch.setattr(bootstrap, "SqlAlchemyUnitOfWorkFactory", factory) + bootstrap._database_components( + Settings(_env_file=None, feature_gates=gates, feishu_sync_collaboration=requested) + ) + assert captured["feishu_collaboration_enabled"] is enabled + + +def test_collaboration_settings_default_to_no_content_egress(): + settings = Settings(_env_file=None) + assert settings.feishu_sync_collaboration is False + assert settings.feishu_include_content is False + assert settings.feishu_send_interval_seconds == 1.0 + + +@pytest.mark.parametrize("value", [0, -1, 11, float("nan"), float("inf")]) +def test_notification_pacing_is_bounded(value): + with pytest.raises(ValidationError): + Settings(_env_file=None, feishu_send_interval_seconds=value)