fix(maf): harden phase 5 approval safety

This commit is contained in:
董宏凯 committed 2026-08-02 14:32:49 +08:00
1 parent f82363d81f
commit a55992e31c
11 files changed
+489 -146

No files matched your search

@@ -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`。补偿动作必须独立审批,不能由通用框架自动猜测或自动重试。
@@ -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 ???
@@ -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)
+1 -1
View File
@@ -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,因此这里只登记两个只读工具。
@@ -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",
)
@@ -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.
@@ -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:
@@ -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)
@@ -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()
@@ -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)
@@ -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"])