From a55992e31c461e4c75f185aa10d8da1f18ae9665 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Sun, 2 Aug 2026 14:32:49 +0800 Subject: [PATCH] fix(maf): harden phase 5 approval safety --- ...t-agent-framework-多智能体会商-实施计划.md | 10 +- .../backend/app/gateway/app.py | 215 ++++++++++-------- .../app/maf_roundtable/action_coordinator.py | 6 +- offline-backend-20260512/backend/config.yaml | 2 +- .../deerflow/config/maf_roundtable_config.py | 4 +- .../persistence/maf_roundtable/actions.py | 37 ++- .../scripts/migrate_maf_roundtable_phase5.py | 128 ++++++++--- .../backend/tests/test_maf_action_leases.py | 82 ++++++- .../tests/test_maf_migration_phase5.py | 110 +++++++++ .../tests/test_maf_phase5_entry_e2e.py | 39 +++- .../tests/test_maf_real_write_http_e2e.py | 2 + 11 files changed, 489 insertions(+), 146 deletions(-) diff --git a/frontend-web/docs/microsoft-agent-framework-多智能体会商-实施计划.md b/frontend-web/docs/microsoft-agent-framework-多智能体会商-实施计划.md index 40e616c7..07bda50d 100644 --- a/frontend-web/docs/microsoft-agent-framework-多智能体会商-实施计划.md +++ b/frontend-web/docs/microsoft-agent-framework-多智能体会商-实施计划.md @@ -25,10 +25,10 @@ 本次重新运行的证据： -- 后端安全闭环重点用例：`293 passed in 140.10s`（包含真实入口 ToolNode/outbox、能力快照、WriteGuard、审批过期与目标租约、旧 DeerFlow 回归）。 +- 后端安全闭环重点用例：`299 passed in 151.09s`（包含真实入口 ToolNode/outbox、能力快照、WriteGuard、审批过期与目标租约、旧 DeerFlow 回归）。 - 后端编排重点用例：`40 passed in 5.46s`（真实 SDK structured/group_chat 路径、Agent Adapter、阶段 3 主持人流程）。 - 前端 MAF 逻辑用例：`19 passed`；`pnpm build` 通过。 -- 全量 `test_maf_*.py` 已在无短时命令限制的终端完成，最终结果为 `293 passed in 140.10s`；PowerShell 下使用文档中的文件枚举方式展开测试文件。 +- 全量 `test_maf_*.py` 已在无短时命令限制的终端完成，最终结果为 `299 passed in 151.09s`；PowerShell 下使用文档中的文件枚举方式展开测试文件。 阶段 1 完成并不表示旧会商被替换。当前配置已打开新的 MAF 功能开关；前端入口由健康检查和权限控制。旧 DeerFlow、旧 roundtable、AI 写作不携带 `maf_run_id`，因此不进入 MAF WriteGuard 分支。 @@ -63,11 +63,11 @@ uv run ruff check app/maf_roundtable packages/harness/deerflow 本轮已完成前阶段 5 的代码与自动化验收项，但按约定仍不将阶段 5 标记为最终验收完成：真实 UAT 写入系统、测试租户、最小权限凭据、核验 API/补偿协议以及生产环境密钥注入仍待业务方和运维提供。 - `MAF-05-03A`：Action 服务层使用公开白名单 DTO；`target_key` 与 `idempotency_key` 仅留在持久层和内部仲裁路径，不能通过两个 actions GET 接口、事件或前端类型返回。目标键只读取服务端登记的稳定标量字段，并使用 `MAF_ACTION_SECRET_KEY` 做域隔离 HMAC-SHA-256；缺少密钥时真实写入失败关闭。 -- `MAF-05-03B`：`scripts/migrate_maf_roundtable_phase5.py` 按数据库方言生成 DDL：PostgreSQL 使用 `TIMESTAMP WITH TIME ZONE`，MySQL/MariaDB 使用 `DATETIME(6)`，SQLite 使用 `DATETIME`。迁移可重复执行，表、缺失索引和可安全补齐的列会自动补齐；已覆盖 SQLite 实际迁移以及三种方言 DDL 测试。 +- `MAF-05-03B`：`scripts/migrate_maf_roundtable_phase5.py` 按数据库方言生成 DDL：PostgreSQL 使用 `TIMESTAMP WITH TIME ZONE`，MySQL/MariaDB 使用 `DATETIME(6)`，SQLite 使用 `DATETIME`。迁移会校验非空列、主键、外键级联、唯一约束和两个索引；仅缺失索引时自动补回，空的残缺表删除后按当前方言完整重建，有数据的残缺表明确失败并要求人工备份核验；已覆盖 SQLite 实际迁移以及三种方言 DDL 测试。 - `MAF-05-04A`：Gateway lifespan 启动和关闭 MAF 专用审批过期扫描任务，间隔由 `maf_roundtable.approval_scan_interval_seconds` 配置。扫描使用已有条件式过期 CAS，然后只对受影响 run 调用 `dispatcher.resume()`；普通审批 GET 保持只读，决定/恢复仍在操作前执行失败关闭过期检查。 - `MAF-05-08A`：入口级测试从 HTTP 创建 session/run 开始，确定性假实现只位于模型边界；真实 Agent Runtime、WriteGuard、LangGraph ToolNode、加密参数和 outbox 均参与执行。覆盖 HTTP 审批后恰好一次写入，以及同目标并发时第二个 run 的 `waiting_target_lease` 与显式 resume。 -本轮证据：全部 `test_maf_*.py` 共 `293 passed`；阶段 5 定向回归 `15 passed`；`git diff --check` 通过。仓库现有的全目录 Ruff 基线仍有与本轮无关的历史问题，提交前需在 CI 中单独清理；本轮新增/修改文件已做定向检查。 +本轮证据：全部 `test_maf_*.py` 共 `299 passed`；阶段 5 定向回归 `21 passed`；包含 `app/gateway/app.py` 在内的本轮修改范围 Ruff 检查通过；`git diff --check` 通过。仓库现有的全目录 Ruff 基线仍有与本轮无关的历史问题，提交前需在 CI 中单独清理。 ### 0.4 下一批开工顺序与责任边界 @@ -861,7 +861,7 @@ frontend-web/src/maf-roundtable/ - [x] `MAF-05-02`（D3）已实现 `auto`、`confirm_high_risk`、`confirm_all_writes` 策略解析，并在 WriteGuard 之前执行分类。 - [x] `MAF-05-03`（D5）服务端目标租约与一次性执行核心已实现，并已完成 03A/03B 的公开边界和跨数据库迁移收口；阶段 5 仍受真实 UAT 工具与部署输入约束，不能据此标记为最终验收完成。 - [x] `MAF-05-03A`（D4，分配同事 A）已封闭目标租约键的公开边界：`target_key` 与内部 `idempotency_key` 仅留在持久层和内部仲裁路径，两个 actions GET、事件、日志和前端类型均不返回；目标键只接受服务端登记的稳定标量业务字段，并使用域隔离 HMAC-SHA-256 派生。HTTP 回归同时证明存储层仍能使用目标键完成租约仲裁。 -- [x] `MAF-05-03B`（D4，分配同事 A）已修复 `migrate_maf_roundtable_phase5.py` 的跨数据库 DDL：PostgreSQL 使用 `TIMESTAMP WITH TIME ZONE`、MySQL/MariaDB 使用 `DATETIME(6)`、SQLite 使用 `DATETIME`；表、缺失索引和可安全补齐列可重复自动补齐。已覆盖 SQLite 实际迁移及 PostgreSQL/MySQL DDL 测试。 +- [x] `MAF-05-03B`（D4，分配同事 A）已修复 `migrate_maf_roundtable_phase5.py` 的跨数据库 DDL：PostgreSQL 使用 `TIMESTAMP WITH TIME ZONE`、MySQL/MariaDB 使用 `DATETIME(6)`、SQLite 使用 `DATETIME`；迁移会校验非空列、主键、外键级联、唯一约束和两个索引，仅对缺失索引自动补回，空的残缺表完整重建，有数据的残缺表明确失败。已覆盖 SQLite 实际迁移及 PostgreSQL/MySQL DDL 测试。 - [x] `MAF-05-04`（D5）审批创建时写入部署级超时，到期后在同一事务内标记 Approval=`expired`、Action=`rejected` 和审计事件。已覆盖真实 HTTP：`POST /approvals/{id}/decision` → `dispatcher.resume()` → `WorkflowRunner.resume()` → `ToolExecutor` → LangGraph `ToolNode` → 测试 outbox；重复决策/恢复仍只执行一次。 - [x] `MAF-05-04A`（D4，分配同事 A）已将审批过期收口为 Gateway lifespan 中可取消的 MAF 周期扫描任务；间隔可配置，扫描使用条件式 CAS 后恢复受影响 run，普通 GET 保持只读。已覆盖并发扫描、重复/重启补扫、决策并发和过期后 Action 拒绝且工具执行次数为零，测试不使用 `sleep`。 - [ ] `MAF-05-05`（D5）为指定的第一个真实写工具实现结果未知核验：外部请求必须携带业务幂等 operation id；网络超时/断连后先调用目标系统查询接口核验，再决定 `succeeded`、`failed` 或保留 `unknown`。补偿动作必须独立审批，不能由通用框架自动猜测或自动重试。 diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 8a863873..3eb666df 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -1,4 +1,7 @@ -import asyncio +# ruff: noqa: E402 +# load_dotenv() must run before application imports because they read environment +# configuration during import-time initialisation. +import asyncio import logging import os import queue @@ -21,83 +24,83 @@ from fastapi.middleware.cors import CORSMiddleware # origins, and "Failed to fetch" in the browser. load_dotenv() -from app.gateway.auth.disabled_mode import is_auth_disabled # noqa: E402 -from app.gateway.auth_middleware import AuthMiddleware -from app.gateway.config import get_gateway_config -from app.gateway.csrf_middleware import CSRFMiddleware -from app.gateway.deps import langgraph_runtime -from app.gateway.routers import ( - ai_writing, - admin_active_runs, - admin_users, - agents, - artifacts, - assistants_compat, - auth, - browser_context, - business_mapping, - channels, - checkpoint_migration, - feedback, - dashboard_sessions, - html_page_favorites, - intent, - knowledge, - knowledge_ingest, - light_apps, - llm_metrics, - maf_roundtable, - mcp, - memory, - menu_overrides, - models, - multi_agent, - notifications, - open_3qfx, - open_chat, - parallel_agent_tasks, - parallel_agents, - position_roundtable, - positions, - position_roles, - public_config, - public_embed, - public_leaderboard, - public_scheduled_tasks, - recommend, - flow_extract, - recommended_questions, - roundtable_chains, - roundtable_diagnostics, - roundtable_draft_shares, - roundtable_drafts, - roundtable_jobs, - roundtable_task_drafts, - runs, - scheduled_tasks, - sensitive_words, - skill_addresses, - skills, - suggestions, - system_settings, - tags, - task_buttons, - taskcop_tasks, - task_reports, - thread_runs, - tool_metrics, - thread_shares, - threads, - uploads, - user_preferences, - user_prompts, - weknora_sources, - writing, -) -from deerflow.config import app_config as deerflow_app_config -from deerflow.config.app_config import apply_logging_level -from deerflow.runtime.scheduler import ScheduledTaskService, set_scheduled_task_service - +from app.gateway.auth.disabled_mode import is_auth_disabled # noqa: E402 +from app.gateway.auth_middleware import AuthMiddleware +from app.gateway.config import get_gateway_config +from app.gateway.csrf_middleware import CSRFMiddleware +from app.gateway.deps import langgraph_runtime +from app.gateway.routers import ( + admin_active_runs, + admin_users, + agents, + ai_writing, + artifacts, + assistants_compat, + auth, + browser_context, + business_mapping, + channels, + checkpoint_migration, + dashboard_sessions, + feedback, + flow_extract, + html_page_favorites, + intent, + knowledge, + knowledge_ingest, + light_apps, + llm_metrics, + maf_roundtable, + mcp, + memory, + menu_overrides, + models, + multi_agent, + notifications, + open_3qfx, + open_chat, + parallel_agent_tasks, + parallel_agents, + position_roles, + position_roundtable, + positions, + public_config, + public_embed, + public_leaderboard, + public_scheduled_tasks, + recommend, + recommended_questions, + roundtable_chains, + roundtable_diagnostics, + roundtable_draft_shares, + roundtable_drafts, + roundtable_jobs, + roundtable_task_drafts, + runs, + scheduled_tasks, + sensitive_words, + skill_addresses, + skills, + suggestions, + system_settings, + tags, + task_buttons, + task_reports, + taskcop_tasks, + thread_runs, + thread_shares, + threads, + tool_metrics, + uploads, + user_preferences, + user_prompts, + weknora_sources, + writing, +) +from deerflow.config import app_config as deerflow_app_config +from deerflow.config.app_config import apply_logging_level +from deerflow.runtime.scheduler import ScheduledTaskService, set_scheduled_task_service + AppConfig = deerflow_app_config.AppConfig get_app_config = deerflow_app_config.get_app_config @@ -333,10 +336,10 @@ async def _run_db_migrations() -> None: deadline = asyncio.get_running_loop().time() + timeout_s while result_queue.empty(): if asyncio.get_running_loop().time() >= deadline: - raise asyncio.TimeoutError + raise TimeoutError await asyncio.sleep(0.2) result = result_queue.get_nowait() - except asyncio.TimeoutError: + except TimeoutError: logger.error( "Database migration did not finish within %.0fs; continuing startup on create_all schema. " "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 " @@ -1003,7 +1006,7 @@ async def _maf_approval_scanner_loop(app: Any) -> None: service = getattr(app.state, "maf_roundtable_service", None) config = getattr(service, "config", None) - interval = max(1, int(getattr(config, "approval_scan_interval_seconds", 10) or 10)) + interval = max(5, int(getattr(config, "approval_scan_interval_seconds", 30) or 30)) while True: try: await _scan_maf_approval_expiry(app) @@ -1014,6 +1017,36 @@ async def _maf_approval_scanner_loop(app: Any) -> None: await asyncio.sleep(interval) +def _start_maf_approval_scanner(app: Any) -> asyncio.Task[None] | None: + """Start the MAF-only approval scanner when the feature is enabled.""" + + service = getattr(app.state, "maf_roundtable_service", None) + if service is None or not getattr(service, "enabled", False): + return None + task = asyncio.create_task(_maf_approval_scanner_loop(app)) + app.state.maf_approval_scanner_task = task + logger.info( + "MAF approval expiry scanner started (interval=%ss)", + getattr(service.config, "approval_scan_interval_seconds", 30), + ) + return task + + +async def _stop_maf_approval_scanner(app: Any) -> None: + """Cancel and await the MAF approval scanner during Gateway shutdown.""" + + task = getattr(app.state, "maf_approval_scanner_task", None) + if task is None: + return + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("MAF approval expiry scanner shutdown error") + + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """Application lifespan handler.""" @@ -1187,15 +1220,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # MAF-only approval expiry scanner. Legacy DeerFlow, old roundtable # and AI-writing paths never enter this loop. try: - maf_service = getattr(app.state, "maf_roundtable_service", None) - if maf_service is not None and getattr(maf_service, "enabled", False): - app.state.maf_approval_scanner_task = asyncio.create_task( - _maf_approval_scanner_loop(app) - ) - logger.info( - "MAF approval expiry scanner started (interval=%ss)", - getattr(maf_service.config, "approval_scan_interval_seconds", 10), - ) + _start_maf_approval_scanner(app) except Exception: logger.exception("MAF approval expiry scanner failed to start (non-fatal)") @@ -1267,15 +1292,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Concurrency sampler shutdown error") - maf_approval_scanner_task = getattr(app.state, "maf_approval_scanner_task", None) - if maf_approval_scanner_task is not None: - maf_approval_scanner_task.cancel() - try: - await maf_approval_scanner_task - except asyncio.CancelledError: - pass - except Exception: - logger.exception("MAF approval expiry scanner shutdown error") + await _stop_maf_approval_scanner(app) # ?? 7 ????AI ?? graph ???? checkpointer / in-flight task? # thread ??? LangGraph Server ? checkpointer ???lifespan ??? diff --git a/offline-backend-20260512/backend/app/maf_roundtable/action_coordinator.py b/offline-backend-20260512/backend/app/maf_roundtable/action_coordinator.py index e92bf294..7cbd6d85 100644 --- a/offline-backend-20260512/backend/app/maf_roundtable/action_coordinator.py +++ b/offline-backend-20260512/backend/app/maf_roundtable/action_coordinator.py @@ -223,7 +223,11 @@ def derive_target_key( def _is_stable_scalar(value: Any) -> bool: """Whether a configured target field can safely identify one target.""" - if value is None or isinstance(value, (str, bool, int)): + if isinstance(value, str): + return bool(value.strip()) + if value is None: + return False + if isinstance(value, (bool, int)): return True return isinstance(value, float) and isfinite(value) diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 352c5b2f..cb4a3e81 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -564,7 +564,7 @@ maf_roundtable: default_max_rounds: 8 # 群聊模式轮次硬上限 admin_diagnostic_access: true # 管理员是否可跨用户读取所有 MAF 会商用于诊断（仅读，不含写） seat_timeout_seconds: 600 # 结构化流程单席位超时（秒） - approval_scan_interval_seconds: 10 # 审批主动过期扫描间隔（秒） + approval_scan_interval_seconds: 30 # 审批主动过期扫描间隔（秒） schema_repair_enabled: true # 席位 JSON 校验失败时允许一次定向修复 # 必须使用 MCP 实际发现到的运行时 tool name；不要改成 server 名或前缀猜测。 # 当前部署仅启用检索 MCP，因此这里只登记两个只读工具。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/maf_roundtable_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/maf_roundtable_config.py index 574e4e59..9bdfd551 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/maf_roundtable_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/maf_roundtable_config.py @@ -147,8 +147,8 @@ class MafRoundtableConfig(BaseModel): description="外部写入审批有效期（秒）；到期后自动拒绝，绝不执行。", ) approval_scan_interval_seconds: int = Field( - default=10, - ge=1, + default=30, + ge=5, le=3600, description="MAF approval expiry scan interval in seconds", ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/actions.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/actions.py index 19ea594a..411c22c0 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/actions.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/actions.py @@ -26,7 +26,7 @@ def _now() -> datetime: def action_to_dict(row: MafRoundtableActionRow) -> dict[str, Any]: - """Serialise an action ORM row to a plain dict (public — reused by the store facade).""" + """Serialise an action ORM row to the internal persistence representation.""" return { "id": row.id, "session_id": row.session_id, @@ -52,16 +52,35 @@ def action_to_dict(row: MafRoundtableActionRow) -> dict[str, Any]: def action_to_public_dict(action: Mapping[str, Any]) -> dict[str, Any]: - """Return the allow-listed action representation exposed by MAF HTTP APIs. + """Return only the stable, audit-safe Action DTO fields.""" - Target lease keys and idempotency keys are persistence/coordinator - details. They must never cross the public boundary. - """ + return { + field: action[field] + for field in _PUBLIC_ACTION_FIELDS + if field in action + } - public = dict(action) - public.pop("target_key", None) - public.pop("idempotency_key", None) - return public + +_PUBLIC_ACTION_FIELDS: tuple[str, ...] = ( + "id", + "session_id", + "run_id", + "agent_id", + "tool_name", + "logical_action_id", + "turn_index", + "risk_level", + "status", + "arguments", + "result", + "external_object_id", + "approval_id", + "attempt_count", + "started_at", + "finished_at", + "created_at", + "updated_at", +) # Back-compat alias for internal callers that used the old private name. diff --git a/offline-backend-20260512/backend/scripts/migrate_maf_roundtable_phase5.py b/offline-backend-20260512/backend/scripts/migrate_maf_roundtable_phase5.py index e52bb3f9..d9fcf7e5 100644 --- a/offline-backend-20260512/backend/scripts/migrate_maf_roundtable_phase5.py +++ b/offline-backend-20260512/backend/scripts/migrate_maf_roundtable_phase5.py @@ -58,27 +58,70 @@ def _create_lease_table_sql(conn_or_dialect: Any) -> str: """ -def _add_missing_columns(conn: Any, *, inspector: Any, execute: Any) -> None: - """Complete a partially created table without rewriting existing data. +def validate_lease_table_schema(inspector: Any) -> list[str]: + """Return every lease-table contract violation found by SQLAlchemy.""" - The migration is additive. A missing primary key or action foreign-key - column cannot be safely added to a populated table, so fail loudly rather - than creating a lease table that the arbiter could misinterpret. - """ + if TABLE not in inspector.get_table_names(): + return [f"missing table: {TABLE}"] - columns = {column["name"] for column in inspector.get_columns(TABLE)} - required = {"target_key", "action_id", "lease_expires_at", "created_at", "updated_at"} - missing = required - columns - if not missing: - return - if {"target_key", "action_id"} & missing: - raise RuntimeError( - f"{TABLE} is incomplete ({sorted(missing)}); recreate it with the deployment migration" + required_columns = ("target_key", "action_id", "lease_expires_at", "created_at", "updated_at") + columns = {column["name"]: column for column in inspector.get_columns(TABLE)} + issues: list[str] = [] + for name in required_columns: + column = columns.get(name) + if column is None: + issues.append(f"missing column: {name}") + elif column.get("nullable") is not False: + issues.append(f"nullable column: {name}") + + primary_key = inspector.get_pk_constraint(TABLE).get("constrained_columns") or [] + if list(primary_key) != ["target_key"]: + issues.append(f"primary key must be (target_key), found {list(primary_key)}") + + foreign_key_ok = False + for foreign_key in inspector.get_foreign_keys(TABLE): + options = foreign_key.get("options") or {} + if ( + list(foreign_key.get("constrained_columns") or []) == ["action_id"] + and foreign_key.get("referred_table") == "maf_roundtable_actions" + and list(foreign_key.get("referred_columns") or []) == ["id"] + and str(options.get("ondelete") or "").upper() == "CASCADE" + ): + foreign_key_ok = True + break + if not foreign_key_ok: + issues.append( + "missing foreign key: action_id -> maf_roundtable_actions.id ON DELETE CASCADE" ) - timestamp_type = _timestamp_sql(conn) - for column in ("lease_expires_at", "created_at", "updated_at"): - if column in missing: - execute(f"ALTER TABLE {TABLE} ADD COLUMN {column} {timestamp_type}") + + unique_action_id = any( + list(constraint.get("column_names") or []) == ["action_id"] + for constraint in inspector.get_unique_constraints(TABLE) + ) + # Some dialects expose a named UNIQUE constraint only as a unique index. + unique_action_id = unique_action_id or any( + bool(index.get("unique")) and list(index.get("column_names") or []) == ["action_id"] + for index in inspector.get_indexes(TABLE) + ) + if not unique_action_id: + issues.append("missing unique constraint: action_id") + + indexes = {index.get("name"): list(index.get("column_names") or []) for index in inspector.get_indexes(TABLE)} + for name, column in ((ACTION_INDEX, "action_id"), (EXPIRY_INDEX, "lease_expires_at")): + if name not in indexes: + issues.append(f"missing index: {name}") + elif indexes[name] != [column]: + issues.append(f"invalid index: {name} must cover ({column})") + return issues + + +def _table_has_rows(conn: Any, text: Any) -> bool: + return conn.execute(text(f"SELECT 1 FROM {TABLE} LIMIT 1")).first() is not None + + +def _create_indexes(execute: Any) -> None: + execute(f"CREATE INDEX {ACTION_INDEX} ON {TABLE} (action_id)") + execute(f"CREATE INDEX {EXPIRY_INDEX} ON {TABLE} (lease_expires_at)") async def _migrate(dry_run: bool) -> int: @@ -111,22 +154,53 @@ def _migrate_sync(conn: Any, *, dry_run: bool) -> None: if TABLE not in inspector.get_table_names(): logger.info("Creating %s", TABLE) execute(_create_lease_table_sql(conn)) + _create_indexes(execute) if dry_run: - execute(f"CREATE INDEX {ACTION_INDEX} ON {TABLE} (action_id)") - execute(f"CREATE INDEX {EXPIRY_INDEX} ON {TABLE} (lease_expires_at)") return inspector = inspect(conn) else: logger.info("Table %s already exists", TABLE) - _add_missing_columns(conn, inspector=inspector, execute=execute) - if not dry_run: + issues = validate_lease_table_schema(inspector) + constraint_issues = [ + issue for issue in issues + if not issue.startswith("missing index:") + ] + if constraint_issues: + has_rows = _table_has_rows(conn, text) + detail = "; ".join(constraint_issues) + if has_rows: + raise RuntimeError( + f"{TABLE} has data but its lease safety schema is incomplete: {detail}. " + "Back up and manually rebuild the table after operational review." + ) + logger.warning( + "%s is empty but incomplete; rebuilding it with the current full schema: %s", + TABLE, + detail, + ) + execute(f"DROP TABLE {TABLE}") + execute(_create_lease_table_sql(conn)) + _create_indexes(execute) + if dry_run: + return + inspector = inspect(conn) + else: + for index_name, column in ( + (ACTION_INDEX, "action_id"), + (EXPIRY_INDEX, "lease_expires_at"), + ): + if f"missing index: {index_name}" in issues: + execute(f"CREATE INDEX {index_name} ON {TABLE} ({column})") + if dry_run: + return inspector = inspect(conn) - indexes = {index.get("name") for index in inspector.get_indexes(TABLE)} - if ACTION_INDEX not in indexes: - execute(f"CREATE INDEX {ACTION_INDEX} ON {TABLE} (action_id)") - if EXPIRY_INDEX not in indexes: - execute(f"CREATE INDEX {EXPIRY_INDEX} ON {TABLE} (lease_expires_at)") + final_issues = validate_lease_table_schema(inspector) + if final_issues: + raise RuntimeError( + f"{TABLE} migration did not produce the required lease schema: " + + "; ".join(final_issues) + ) def main() -> int: diff --git a/offline-backend-20260512/backend/tests/test_maf_action_leases.py b/offline-backend-20260512/backend/tests/test_maf_action_leases.py index 25693260..85bd9434 100644 --- a/offline-backend-20260512/backend/tests/test_maf_action_leases.py +++ b/offline-backend-20260512/backend/tests/test_maf_action_leases.py @@ -12,9 +12,14 @@ from maf_test_helpers import make_maf_service from sqlalchemy import update from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine -from app.gateway.app import _scan_maf_approval_expiry +from app.gateway.app import ( + _scan_maf_approval_expiry, + _start_maf_approval_scanner, + _stop_maf_approval_scanner, +) from app.maf_roundtable.action_coordinator import ( ActionCoordinator, + MafActionSecretKeyError, derive_target_key, encrypt_params, ) @@ -24,6 +29,7 @@ from app.maf_roundtable.workflow_executors import RoundtableState from app.maf_roundtable.workflow_runner import WorkflowRunner from deerflow.config.maf_roundtable_config import MafRoundtableConfig from deerflow.persistence.maf_roundtable import make_maf_roundtable_store +from deerflow.persistence.maf_roundtable.actions import action_to_public_dict from deerflow.persistence.maf_roundtable.model import ( Base, MafRoundtableActionLeaseRow, @@ -96,6 +102,7 @@ async def _approved_action( def test_target_key_is_server_derived_and_opaque(monkeypatch) -> None: + monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "explicit-hmac-test-secret") snapshot = _snapshot() original = { "tenant_id": "tenant-a", @@ -130,10 +137,60 @@ def test_target_key_is_server_derived_and_opaque(monkeypatch) -> None: ) assert nested == nested_other + conservative_inputs = [ + {"tenant_id": "tenant-a"}, + {"tenant_id": "tenant-a", "object_id": None}, + {"tenant_id": "tenant-a", "object_id": ""}, + {"tenant_id": "tenant-a", "object_id": {"id": "task-7"}}, + {"tenant_id": "tenant-a", "object_id": ["task-7"]}, + ] + conservative_keys = { + derive_target_key(TEST_TOOL, args, snapshot, "seat-1") + for args in conservative_inputs + } + assert len(conservative_keys) == 1 + + monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False) + with pytest.raises(MafActionSecretKeyError): + derive_target_key(TEST_TOOL, original, snapshot, "seat-1") + monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "another-test-domain-secret") assert target != derive_target_key(TEST_TOOL, original, snapshot, "seat-1") +def test_action_public_dto_is_an_explicit_allowlist() -> None: + internal = { + "id": "action-1", + "session_id": "session-1", + "run_id": "run-1", + "agent_id": "seat-1", + "tool_name": TEST_TOOL, + "logical_action_id": "tool-call-1", + "turn_index": 1, + "risk_level": "R3", + "status": "waiting_approval", + "arguments": {"safe": "redacted"}, + "result": None, + "external_object_id": None, + "approval_id": "approval-1", + "attempt_count": 0, + "started_at": None, + "finished_at": None, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": "2026-01-01T00:00:00+00:00", + "target_key": "private-target", + "idempotency_key": "private-idempotency", + "internal_canary": "must-not-leak", + } + public = action_to_public_dict(internal) + assert set(public) == { + "id", "session_id", "run_id", "agent_id", "tool_name", + "logical_action_id", "turn_index", "risk_level", "status", + "arguments", "result", "external_object_id", "approval_id", + "attempt_count", "started_at", "finished_at", "created_at", "updated_at", + } + + @pytest.mark.asyncio async def test_two_agents_same_target_are_serialized(store, svc): """The DB lease, not a process lock, blocks the second approved write.""" @@ -394,3 +451,26 @@ async def test_approval_expiry_scan_is_repeatable_and_get_is_observational(monke assert fake.calls == [(run["id"], "u1")] assert (await store.approvals.get_approval(action["approval_id"]))["status"] == "expired" assert (await store.actions.get_action(action["id"]))["status"] == "rejected" + + +@pytest.mark.asyncio +async def test_maf_approval_scanner_lifecycle_is_feature_gated(): + config = MafRoundtableConfig(enabled=True, approval_scan_interval_seconds=5) + enabled_app = SimpleNamespace( + state=SimpleNamespace( + maf_roundtable_service=SimpleNamespace(enabled=True, config=config), + ) + ) + task = _start_maf_approval_scanner(enabled_app) + assert task is not None + assert enabled_app.state.maf_approval_scanner_task is task + await _stop_maf_approval_scanner(enabled_app) + assert task.done() + + disabled_app = SimpleNamespace( + state=SimpleNamespace( + maf_roundtable_service=SimpleNamespace(enabled=False, config=config), + ) + ) + assert _start_maf_approval_scanner(disabled_app) is None + await _stop_maf_approval_scanner(disabled_app) diff --git a/offline-backend-20260512/backend/tests/test_maf_migration_phase5.py b/offline-backend-20260512/backend/tests/test_maf_migration_phase5.py index 65db626f..5ede916d 100644 --- a/offline-backend-20260512/backend/tests/test_maf_migration_phase5.py +++ b/offline-backend-20260512/backend/tests/test_maf_migration_phase5.py @@ -21,6 +21,25 @@ def _migration_module(): return module +def _assert_complete_lease_schema(sync_conn, module) -> None: + inspector = inspect(sync_conn) + assert module.validate_lease_table_schema(inspector) == [] + columns = {column["name"]: column for column in inspector.get_columns(module.TABLE)} + assert all(columns[name]["nullable"] is False for name in columns) + assert inspector.get_pk_constraint(module.TABLE)["constrained_columns"] == ["target_key"] + assert any( + constraint["column_names"] == ["action_id"] + for constraint in inspector.get_unique_constraints(module.TABLE) + ) + assert any( + foreign_key["constrained_columns"] == ["action_id"] + and foreign_key["referred_table"] == "maf_roundtable_actions" + and foreign_key["referred_columns"] == ["id"] + and str((foreign_key.get("options") or {}).get("ondelete")).upper() == "CASCADE" + for foreign_key in inspector.get_foreign_keys(module.TABLE) + ) + + @pytest.mark.asyncio async def test_phase5_lease_migration_is_additive_and_idempotent(): module = _migration_module() @@ -51,6 +70,97 @@ async def test_phase5_lease_migration_is_additive_and_idempotent(): await engine.dispose() +@pytest.mark.asyncio +async def test_empty_incomplete_lease_table_is_rebuilt_with_full_constraints(): + module = _migration_module() + engine = create_async_engine("sqlite+aiosqlite://", echo=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + await conn.execute(text(f"DROP TABLE {module.TABLE}")) + await conn.execute(text(f""" + CREATE TABLE {module.TABLE} ( + target_key VARCHAR(255), + action_id VARCHAR(64), + lease_expires_at DATETIME, + created_at DATETIME, + updated_at DATETIME + ) + """)) + await conn.run_sync(module._migrate_sync, dry_run=False) + await conn.run_sync(_assert_complete_lease_schema, module) + await engine.dispose() + + +@pytest.mark.asyncio +async def test_nonempty_incomplete_lease_table_fails_without_rewriting_data(): + module = _migration_module() + engine = create_async_engine("sqlite+aiosqlite://", echo=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + await conn.execute(text(f"DROP TABLE {module.TABLE}")) + await conn.execute(text(f""" + CREATE TABLE {module.TABLE} ( + target_key VARCHAR(255), + action_id VARCHAR(64), + lease_expires_at DATETIME, + created_at DATETIME, + updated_at DATETIME + ) + """)) + await conn.execute(text( + f"INSERT INTO {module.TABLE} " + "(target_key, action_id, lease_expires_at, created_at, updated_at) " + "VALUES ('opaque-1', 'action-1', '2026-01-01', '2026-01-01', '2026-01-01')" + )) + + with pytest.raises(RuntimeError, match="has data.*incomplete.*manual"): + async with engine.begin() as conn: + await conn.run_sync(module._migrate_sync, dry_run=False) + + async with engine.begin() as conn: + row = ( + await conn.execute(text(f"SELECT target_key, action_id FROM {module.TABLE}")) + ).one() + assert tuple(row) == ("opaque-1", "action-1") + await engine.dispose() + + +@pytest.mark.asyncio +async def test_nonempty_complete_columns_missing_fk_and_unique_constraint_fails(): + module = _migration_module() + engine = create_async_engine("sqlite+aiosqlite://", echo=False) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + await conn.execute(text(f"DROP TABLE {module.TABLE}")) + await conn.execute(text(f""" + CREATE TABLE {module.TABLE} ( + target_key VARCHAR(255) NOT NULL PRIMARY KEY, + action_id VARCHAR(64) NOT NULL, + lease_expires_at DATETIME NOT NULL, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL + ) + """)) + await conn.execute(text(f"CREATE INDEX {module.ACTION_INDEX} ON {module.TABLE} (action_id)")) + await conn.execute(text(f"CREATE INDEX {module.EXPIRY_INDEX} ON {module.TABLE} (lease_expires_at)")) + await conn.execute(text( + f"INSERT INTO {module.TABLE} " + "(target_key, action_id, lease_expires_at, created_at, updated_at) " + "VALUES ('opaque-2', 'action-2', '2026-01-01', '2026-01-01', '2026-01-01')" + )) + + with pytest.raises(RuntimeError) as exc_info: + async with engine.begin() as conn: + await conn.run_sync(module._migrate_sync, dry_run=False) + message = str(exc_info.value) + assert "foreign key" in message + assert "unique constraint" in message + + async with engine.begin() as conn: + assert (await conn.execute(text(f"SELECT COUNT(*) FROM {module.TABLE}"))).scalar_one() == 1 + await engine.dispose() + + @pytest.mark.asyncio async def test_phase5_migration_dry_run_is_safe_when_table_is_absent(): module = _migration_module() diff --git a/offline-backend-20260512/backend/tests/test_maf_phase5_entry_e2e.py b/offline-backend-20260512/backend/tests/test_maf_phase5_entry_e2e.py index c677fdfe..4c5b7398 100644 --- a/offline-backend-20260512/backend/tests/test_maf_phase5_entry_e2e.py +++ b/offline-backend-20260512/backend/tests/test_maf_phase5_entry_e2e.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import threading +from datetime import UTC, datetime, timedelta from types import SimpleNamespace import pytest @@ -13,8 +14,10 @@ from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration, ChatResult from langgraph.checkpoint.memory import InMemorySaver +from sqlalchemy import update from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from app.gateway.app import _scan_maf_approval_expiry from app.gateway.routers.maf_roundtable import router as maf_router from app.maf_roundtable.dispatcher import get_dispatcher from app.maf_roundtable.write_guard_bridge import register_write_guard_service @@ -24,6 +27,7 @@ from deerflow.config.tool_config import ToolConfig from deerflow.config.write_guard_config import WriteGuardConfig, WriteGuardProviderConfig from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store +from deerflow.persistence.maf_roundtable.model import MafRoundtableApprovalRow from deerflow.runtime import RunManager from deerflow.runtime.events.store.memory import MemoryRunEventStore from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge @@ -115,7 +119,7 @@ def _app_config(): enabled=True, default_write_policy="confirm_all_writes", approval_timeout_seconds=300, - approval_scan_interval_seconds=1, + approval_scan_interval_seconds=5, tool_metadata={ TEST_TOOL: MafToolMetadata( side_effect="write", @@ -266,6 +270,39 @@ async def test_entry_http_approval_reaches_real_toolnode_once(tmp_path, determin await engine.dispose() +@pytest.mark.asyncio +async def test_entry_approval_scanner_expires_and_finishes_without_tool(tmp_path, deterministic_agent): + app, service, engine = await _make_app(tmp_path, deterministic_agent) + dispatcher = get_dispatcher(app) + try: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + _session_id, run = await _create_waiting_run( + client, dispatcher, task_text="scanner expiry", service=service + ) + approval = (await client.get( + f"/api/maf-roundtable/runs/{run['id']}/approvals" + )).json()["approvals"][0] + async with service.store.session_factory() as db: + await db.execute( + update(MafRoundtableApprovalRow) + .where(MafRoundtableApprovalRow.id == approval["id"]) + .values(expires_at=datetime.now(UTC) - timedelta(seconds=1)) + ) + await db.commit() + + assert await _scan_maf_approval_expiry(app) == [run["id"]] + final_run = (await client.get(f"/api/maf-roundtable/runs/{run['id']}")).json() + assert final_run["status"] == "completed" + action = (await client.get( + f"/api/maf-roundtable/runs/{run['id']}/actions" + )).json()["actions"][0] + assert action["status"] == "rejected" + assert not (tmp_path / "phase5-target.jsonl").exists() + finally: + await engine.dispose() + + @pytest.mark.asyncio async def test_entry_http_same_target_waits_for_explicit_resume(tmp_path, deterministic_agent): app, service, engine = await _make_app(tmp_path, deterministic_agent) diff --git a/offline-backend-20260512/backend/tests/test_maf_real_write_http_e2e.py b/offline-backend-20260512/backend/tests/test_maf_real_write_http_e2e.py index 02ed3cb0..aa0a8e73 100644 --- a/offline-backend-20260512/backend/tests/test_maf_real_write_http_e2e.py +++ b/offline-backend-20260512/backend/tests/test_maf_real_write_http_e2e.py @@ -182,12 +182,14 @@ async def test_http_approval_dispatches_real_tool_once(monkeypatch, tmp_path): listed_action = listed.json()["actions"][0] assert "target_key" not in listed_action assert "idempotency_key" not in listed_action + assert "internal_canary" not in listed_action fetched = await client.get(f"/api/maf-roundtable/actions/{action['id']}") assert fetched.status_code == 200 fetched_action = fetched.json() assert "target_key" not in fetched_action assert "idempotency_key" not in fetched_action + assert "internal_canary" not in fetched_action # The lease arbiter still needs the private key in persistence. stored_before_decision = await store.actions.get_action(action["id"])