From 6340ec503caa02daa5002cb0dc1fe1c89f329f5a Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 3 Aug 2026 18:12:11 +0800 Subject: [PATCH] feat(exploration): stateless patrol API with watermark increments Resident agents call GET /api/exploration/patrol once per cycle to see every running production-line campaign that opted into exploration (seed set present), the new results since the last watermark (reusing campaign report aggregation), and the remaining exploration budget. The watermark advances after each call so subsequent calls only report increments; accelerated and terminal campaigns are excluded. --- .../01-session-lifecycle-and-guardrails.md | 13 ++ .../issues/02-campaign-seeds-and-budget.md | 13 ++ .scratch/v0.9/issues/03-patrol-api.md | 12 ++ .scratch/v0.9/issues/04-judge-review.md | 12 ++ .../v0.9/issues/05-report-analysis-export.md | 12 ++ .../issues/06-drawer-exploration-section.md | 12 ++ .../v0.9/issues/07-finalize-and-line-gates.md | 12 ++ .../v0.9/issues/08-openclaw-patrol-skill.md | 12 ++ .scratch/v0.9/spec.md | 113 ++++++++++ backend/agenteval/models.py | 1 + backend/agenteval/storage/db.py | 10 + backend/agenteval/storage/repository.py | 3 + backend/agenteval/web/routers/exploration.py | 97 ++++++++- ...2e4b7_add_patrol_watermark_to_campaigns.py | 26 +++ tests/integration/test_campaigns_api.py | 1 + tests/integration/test_exploration_api.py | 1 + .../test_exploration_patrol_api.py | 203 ++++++++++++++++++ 17 files changed, 547 insertions(+), 6 deletions(-) create mode 100644 .scratch/v0.9/issues/01-session-lifecycle-and-guardrails.md create mode 100644 .scratch/v0.9/issues/02-campaign-seeds-and-budget.md create mode 100644 .scratch/v0.9/issues/03-patrol-api.md create mode 100644 .scratch/v0.9/issues/04-judge-review.md create mode 100644 .scratch/v0.9/issues/05-report-analysis-export.md create mode 100644 .scratch/v0.9/issues/06-drawer-exploration-section.md create mode 100644 .scratch/v0.9/issues/07-finalize-and-line-gates.md create mode 100644 .scratch/v0.9/issues/08-openclaw-patrol-skill.md create mode 100644 .scratch/v0.9/spec.md create mode 100644 migrations/versions/c5d8f0a2e4b7_add_patrol_watermark_to_campaigns.py create mode 100644 tests/integration/test_exploration_patrol_api.py diff --git a/.scratch/v0.9/issues/01-session-lifecycle-and-guardrails.md b/.scratch/v0.9/issues/01-session-lifecycle-and-guardrails.md new file mode 100644 index 0000000..31bcaae --- /dev/null +++ b/.scratch/v0.9/issues/01-session-lifecycle-and-guardrails.md @@ -0,0 +1,13 @@ +# 01 — 探索会话生命周期 + 平台硬护栏 + +**What to build:** 从 API 视角完整走通一段探索会话：指定活动、人设与目标创建 running 会话 → 以虚拟用户身份发消息并收到被评对象回复（经目标通道转发，双方轮次持久化、记录延迟）→ 提交结构化体验记录（达成、卡点、被误导、情绪）关闭会话。护栏是平台账本：超窗口会话数、超单会话轮数、相邻会话间隔不足，一律 409 并附原因；会话非 running 状态拒收消息。探索会话是独立实体，不并入评测运行（不污染通过率口径）。 + +**Blocked by:** None — can start immediately + +**Status:** completed + +- [x] 探索会话与会话轮次独立建表，迁移可在全新库与既有库上应用 +- [x] 创建/发消息/关闭三个端点全链路可用，X-API-Key 鉴权 +- [x] 三类预算超限各返回 409 + 可读原因；平台默认值生效（≤8 会话/窗口、≤12 轮/会话、≥30min 间隔） +- [x] 体验记录结构校验，非法值归一（沿 v0.7 白名单经验） +- [x] 集成测试覆盖完整生命周期与各拒绝分支（先例：活动 API 集成测试） diff --git a/.scratch/v0.9/issues/02-campaign-seeds-and-budget.md b/.scratch/v0.9/issues/02-campaign-seeds-and-budget.md new file mode 100644 index 0000000..cf25efa --- /dev/null +++ b/.scratch/v0.9/issues/02-campaign-seeds-and-budget.md @@ -0,0 +1,13 @@ +# 02 — 活动级种子集与探索预算配置 + +**What to build:** 创建活动时可为该活动配置种子集（种子人设数组 + 种子目标数组）与探索预算（最大会话数 / 单会话最大轮数 / 会话最小间隔，空则用平台默认）。配置与活动计划同构入库，可查询；前端活动创建表单提供种子与预算输入，种子留空即该活动不参与探索。种子集是探索式评测的可比性单位，随活动持久化。 + +**Blocked by:** None — can start immediately + +**Status:** completed + +- [x] 活动实体新增种子集与预算两个 JSON 配置字段，迁移可在全新库与既有库上应用 +- [x] 活动创建/查询 API 接受并返回配置；预算空值回落平台默认 +- [x] 创建表单可录入种子人设、种子目标与预算覆盖 +- [x] 种子留空的活动创建成功且标记为不参与探索 +- [x] 集成测试覆盖配置存取与默认回落 diff --git a/.scratch/v0.9/issues/03-patrol-api.md b/.scratch/v0.9/issues/03-patrol-api.md new file mode 100644 index 0000000..1a21b7b --- /dev/null +++ b/.scratch/v0.9/issues/03-patrol-api.md @@ -0,0 +1,12 @@ +# 03 — 巡检 API + +**What to build:** 常驻代理一次调用即可无状态巡检：返回所有进行中的正式线活动（time_scale == 1）清单，每项含活动标识与对象、自上次巡检以来的新结果摘要（复用现有活动报告聚合口径，不重算新指标）、探索预算余量（剩余会话数等）。响应生成后推进各活动的巡检水位，使下次调用只报增量。加速线活动不出现在巡检结果中。 + +**Blocked by:** 01（预算余量需会话计数）、02（活动预算字段） + +**Status:** ready-for-agent + +- [ ] 巡检端点返回进行中正式线活动 + 新结果摘要 + 预算余量 +- [ ] 巡检水位推进：连续两次调用，第二次只含增量 +- [ ] 加速线与终态活动不出现 +- [ ] 集成测试覆盖内容正确性、水位推进与过滤规则 diff --git a/.scratch/v0.9/issues/04-judge-review.md b/.scratch/v0.9/issues/04-judge-review.md new file mode 100644 index 0000000..fe9f949 --- /dev/null +++ b/.scratch/v0.9/issues/04-judge-review.md @@ -0,0 +1,12 @@ +# 04 — judge 抽样复核 + +**What to build:** 会话关闭后，平台后台对该会话对话抽样（默认 ≤3 段，控 token），经 judge 岗位模型独立复核，产出质量维度结论（态度、专业性、幻觉等）落入会话记录。复核是异步后台执行：失败落错误不阻塞会话状态，也不影响体验记录这条第一手证据线。LLM 调用走可注入客户端接缝，测试用假客户端覆盖。 + +**Blocked by:** 01（会话关闭事件与对话数据） + +**Status:** ready-for-agent + +- [ ] 会话关闭后自动触发抽样复核，结果结构化落库 +- [ ] 抽样上限生效；无模型配置时静默跳过 +- [ ] 复核失败落错误，会话仍为 completed +- [ ] 假客户端测试覆盖编排、解析失败、无模型分支（先例：v0.7 分析测试） diff --git a/.scratch/v0.9/issues/05-report-analysis-export.md b/.scratch/v0.9/issues/05-report-analysis-export.md new file mode 100644 index 0000000..f92f847 --- /dev/null +++ b/.scratch/v0.9/issues/05-report-analysis-export.md @@ -0,0 +1,12 @@ +# 05 — 探索发现 → 报告 / 分析 / 导出 + +**What to build:** 探索数据汇入既有报告链的三个出口。活动报告新增"探索发现"维度：会话数、目标达成率、问题清单（来自体验记录聚合；judge 复核结论若已就位一并纳入）。v0.7 活动分析的输入追加探索摘要（问题清单 + 达成统计，非全量对话），让分析模型合成两条证据线。Markdown 导出在存在探索数据时追加「探索发现」附录（顺序在智能分析与周期对比之后），无探索数据时导出与旧版完全一致。周期对比口径不变。 + +**Blocked by:** 01（聚合主料是体验记录）；04 若已就位则复核结论一并纳入，未就位不阻塞 + +**Status:** ready-for-agent + +- [ ] 活动报告聚合含探索发现维度，达成率与问题清单口径正确 +- [ ] 分析输入含探索摘要且不含全量对话 +- [ ] Markdown 导出追加探索附录；无数据时逐字节不变（缺则无痕） +- [ ] 纯函数渲染与聚合单测 + 导出集成测试（先例：report_render 单测、活动导出测试） diff --git a/.scratch/v0.9/issues/06-drawer-exploration-section.md b/.scratch/v0.9/issues/06-drawer-exploration-section.md new file mode 100644 index 0000000..91538fe --- /dev/null +++ b/.scratch/v0.9/issues/06-drawer-exploration-section.md @@ -0,0 +1,12 @@ +# 06 — 报告抽屉「探索发现」区块 + 会话下钻 + +**What to build:** 活动报告抽屉新增「探索发现」区块（模式沿 v0.7 智能分析区块）：会话数、目标达成率、问题清单、会话列表状态。每个会话可点开查看完整对话与体验记录——证据下钻是探索数据的命脉，"用户在这里被绕晕"必须能看到原文。无探索数据时区块不出现。不建新顶级页面。 + +**Blocked by:** 05（报告聚合与查询端点就绪） + +**Status:** ready-for-agent + +- [ ] 报告抽屉呈现探索发现区块，数据与后端口径一致 +- [ ] 会话详情展示完整对话（角色、内容、延迟）与体验记录 +- [ ] 无探索数据时区块隐藏 +- [ ] tsc 通过；人工在浏览器验证展开/下钻交互 diff --git a/.scratch/v0.9/issues/07-finalize-and-line-gates.md b/.scratch/v0.9/issues/07-finalize-and-line-gates.md new file mode 100644 index 0000000..c3c9bb1 --- /dev/null +++ b/.scratch/v0.9/issues/07-finalize-and-line-gates.md @@ -0,0 +1,12 @@ +# 07 — 活动 finalize 会话结算 + 档位校验 + +**What to build:** 活动生命周期与探索会话的收口规则。窗口结束进入终态结算时，仍 running 的探索会话转 expired，此后拒收消息——不留悬挂会话。档位校验完整化：正式线接受自动与手动触发的会话；加速调试线仅接受手动触发，自动来源一律拒绝（时间压缩与拟真冲突，加速线不参与自动探索）。 + +**Blocked by:** 01（会话实体与状态机） + +**Status:** ready-for-agent + +- [ ] 活动终态结算时 running 会话转 expired，后续消息 409 +- [ ] 加速线自动触发被拒并附原因；手动触发可用 +- [ ] 正式线两种触发来源均可用 +- [ ] 集成测试覆盖结算与档位分支（先例：活动自动触发类测试） diff --git a/.scratch/v0.9/issues/08-openclaw-patrol-skill.md b/.scratch/v0.9/issues/08-openclaw-patrol-skill.md new file mode 100644 index 0000000..4925bdd --- /dev/null +++ b/.scratch/v0.9/issues/08-openclaw-patrol-skill.md @@ -0,0 +1,12 @@ +# 08 — AI 助手巡检/探索 skill + 常驻作业 + +**What to build:** 让已集成的 AI 助手成为常驻巡检代理：编写巡检/探索 skill（说明书），指导代理按"调巡检 API → 研判新结果与预算余量 → 决定是否派发探索会话 → 驱动会话对话 → 提交体验记录"的闭环行动，含收到 409 时的收敛行为与衍生变体的种子回溯要求。skill 走部署脚本既有的 skill 同步管线分发；在 AI 助手上注册一个全局常驻 heartbeat/cron 巡检作业（默认节拍 1h）。在 t480 端到端演示：一个配置了种子集的正式线活动，在窗口内被自动巡检并产出探索会话。 + +**Blocked by:** 03（巡检 API 契约就绪才能写说明书） + +**Status:** ready-for-agent + +- [ ] skill 文档覆盖巡检决策、会话驱动、体验记录提交、409 收敛与衍生留痕 +- [ ] 部署同步管线分发 skill（复用现有同步步骤） +- [ ] 常驻巡检作业注册成功，重启后不丢（依赖 OpenClaw 调度持久化） +- [ ] t480 端到端验证：正式线活动窗口内出现自动派发的探索会话并有体验记录落库 diff --git a/.scratch/v0.9/spec.md b/.scratch/v0.9/spec.md new file mode 100644 index 0000000..89b4680 --- /dev/null +++ b/.scratch/v0.9/spec.md @@ -0,0 +1,113 @@ +# v0.9 规格说明：探索式评测第一期 —— 周期活动内嵌虚拟用户探索 + +**状态**: ready-for-agent +**日期**: 2026-08-03 +**决策依据**: ADR-0003 v2 修订（2026-08-03）、CONTEXT.md「探索式评测」章节、grilling 共识清单（15 项） + +## Problem Statement + +平台的评测能力目前全部建立在"固定场景（考纲）"之上：预设题目、预设规则、统计通过率。这能回答"考纲过了多少"，回答不了"真实用户会踩到什么"——真实用户不按考纲出牌，他们会中途改主意、追问、被绕晕、放弃。被评智能体在面对这类复杂拟真行为时的表现（问题诊断与优化建议）目前没有任何评测手段。 + +## Solution + +引入与固定场景并行的**探索式评测**模式：OpenClaw 作为**虚拟用户**，按活动配置的种子人设 × 种子目标自主与被评对象对话，产出**体验记录**（目标达成、卡点、被误导处），judge 岗位抽样复核，两条证据线汇入活动报告、v0.7 分析与 Markdown 导出。 + +第一期形态为**周期活动内嵌探索**：现有活动骨架不变（静态计划照常执行），全局一个常驻巡检代理（跑在已集成的 AI 助手上，heartbeat/cron 唤醒）按节拍巡检进行中的正式线活动，在预算内派发探索会话。OpenClaw 停摆只暂停探索，固定计划照常完成——耐久性归平台，自主性归代理。 + +## User Stories + +1. As an 评测平台用户, I want 创建活动时配置该活动的种子集（种子人设 + 种子目标）, so that 虚拟用户的行动有据可依，且同种子集的活动跨期可比 +2. As an 评测平台用户, I want 创建活动时调整探索预算（或使用平台默认值）, so that 探索行为的规模与负载在我的掌控之内 +3. As an 平台运维者, I want 部署时 AI 助手注册一个全局常驻巡检作业, so that 所有进行中的正式线活动被自动巡检，无需逐活动配置 +4. As an 常驻代理, I want 每次巡检唤醒时调平台巡检 API 拿到进行中活动、上次巡检以来的新结果与预算余量, so that 我能无状态地自主决定"继续观察 / 派探索会话" +5. As an 常驻代理, I want 通过 API 创建探索会话（指定活动、人设、目标、种子出处）, so that 我能以虚拟用户身份开始行动 +6. As an 常驻代理, I want 通过会话 API 发消息并收到被评对象的回复, so that 我能按人设自主决定追问、等待、放弃 +7. As an 常驻代理, I want 在会话结束时提交结构化体验记录, so that 我的第一手判定（达成/卡点/被误导）被平台留痕 +8. As an 常驻代理, I want 超出预算时收到 API 明确的 409 拒绝与原因, so that 拒绝本身成为反馈信号，我及时收敛 +9. As an 常驻代理, I want 在预算内从种子衍生变体（同目标换表达、同人设换追问策略）并被标记衍生出处, so that 行为多样性与留痕兼得 +10. As an 评测平台用户, I want 在活动报告抽屉看到"探索发现"区块（会话数、目标达成率、问题清单）, so that 我知道真实用户遭遇了什么 +11. As an 评测平台用户, I want 点开单个探索会话查看完整对话与体验记录, so that 每个"问题"都能下钻到原文证据 +12. As an 分析岗位模型, I want 生成活动分析时收到探索摘要（问题清单 + 达成统计，非全量对话）, so that "固定场景成绩"与"真实用户遭遇"两条证据线合成一份结论 +13. As an judge 岗位模型, I want 对抽样的探索对话做独立复核, so that 体验判定之外有客观质量维度（态度、专业性、幻觉） +14. As an 报告导出用户, I want Markdown 导出在存在探索数据时追加"探索发现"区块, so that 离线分享材料完整；无探索数据时导出与旧版一致 +15. As an 加速调试线用户, I want 手动触发探索会话, so that 我能调试探索流程；同时加速线不被自动巡检打扰（时间压缩与拟真冲突） +16. As an 平台, I want OpenClaw 停摆时活动的固定计划照常走完, so that 探索是增强层而非单点故障 +17. As an 平台, I want 活动窗口结束时仍未关闭的探索会话被妥善结算（标记状态、不再接受消息）, so that 不留悬挂会话 +18. As an 评测平台用户, I want 探索会话与固定场景的子 Run 在报告里清楚区分, so that 通过率口径（ADR-0002）不被探索数据污染 + +## Implementation Decisions + +### 数据模型 + +- 新增**探索会话**独立实体（不并入评测运行）：归属活动、指向评测对象、人设（JSON）、目标、种子出处（可空 = 衍生变体须回溯到种子）、状态机（running / completed / failed / expired）、触发来源（auto / manual）、体验记录 JSON、judge 复核 JSON、轮次计数、时间戳 +- 新增会话轮次表（role、content、latency_ms、created_at），与评测运行的 turns 表平行，不复用 +- 活动实体新增两个 JSON 配置字段（与 `plan` 同构地位）：**种子集**（种子人设数组 + 种子目标数组）、**探索预算**（最大会话数 / 单会话最大轮数 / 会话最小间隔，空则取平台默认） +- 活动实体新增 `last_patrolled_at`（巡检 API 每次读取后更新，支撑"上次巡检以来的新结果"语义） +- 三个变更各走一个 Alembic 迁移（batch mode） + +### API 契约（全部 X-API-Key 鉴权，沿用 `require_api_key`） + +- `GET /api/exploration/patrol`：返回进行中的正式线活动（time_scale == 1）清单，每项含活动 id/名称/对象、自 `last_patrolled_at` 以来的新结果摘要（复用 `build_campaign_report` 聚合口径）、探索预算余量；响应生成后更新各活动 `last_patrolled_at` +- `POST /api/exploration/sessions`：body = campaign_id、persona、goal、seed_ref（可空）；平台硬校验（活动存在且 running、正式线仅接受 auto 或 manual 触发；加速线仅接受 manual；预算余量）→ 创建 running 会话；超限 409 + 原因 +- `POST /api/exploration/sessions/{id}/messages`：body = content；平台转发到目标的通道（复用 ChannelFactory）、持久化双方轮次、返回回复与延迟；单会话轮数超限 409；会话非 running 状态 409 +- `POST /api/exploration/sessions/{id}/close`：body = 体验记录（goal_achieved: bool、blockers[]、misled[]、emotion、notes）；结构校验 + 非法值归一（沿 v0.7 白名单经验）；会话转 completed +- `GET /api/campaigns/{id}/exploration`：会话列表 + 探索发现聚合（会话数、达成率、问题清单） +- 活动窗口 finalize 时：仍 running 的会话转 expired，不再接受消息（在 `resolve_finalize` 同处挂接） + +### 护栏（平台硬执行，不信任客户端自律） + +- 平台默认：≤8 会话/窗口、≤12 轮/会话、相邻会话 ≥30min（正式线真实时间）；活动级预算覆盖 +- 预算计数是平台账本：创建/发消息/关闭均实时校验，超限一律 409 + +### 判定双证据线 + +- 体验判定：close 接口收结构化自报，为第一手证据 +- judge 抽样复核：会话结束后平台对对话抽样（默认 ≤3 段，控 token），经 judge 岗位模型产出质量维度复核；ChatClient 可注入（沿 v0.7 分析 seam）；异步后台执行，失败落错误不阻塞 + +### 报告 / 分析 / 导出 + +- 活动报告聚合新增"探索发现"维度（会话数、目标达成率、问题清单来自体验记录聚合）；周期对比口径不变（探索数据不参与） +- v0.7 分析输入追加探索**摘要**（问题清单 + 达成统计，非全量对话） +- Markdown 导出：存在探索数据时追加「探索发现」附录（缺则无痕），附录顺序：智能分析 → 周期对比 → 探索发现 + +### 前端 + +- 活动创建表单：种子集与预算配置输入（种子可留空 = 该活动不参与探索） +- 报告抽屉新增「探索发现」区块（模式沿 v0.7 智能分析区块）：达成率、问题清单、会话列表；会话点开查看完整对话 + 体验记录 +- 不建新顶级页面 + +### AI 助手侧（不在平台代码仓内的行为，契约即平台 API） + +- 现有已集成 OpenClaw（AI 助手菜单、openclaw-eval 容器）上扩展巡检/探索 skill，走部署脚本既有 skill 同步管线 +- 注册一个全局常驻 heartbeat/cron 巡检作业（节拍默认 1h）；巡检 API 本身与节拍无关 + +### 档位 + +- 正式线（time_scale == 1）：自动巡检 + 自动派发 +- 加速调试线：仅手动触发会话，不参与自动巡检 + +## Testing Decisions + +- 好测试只测外部行为：API 契约、状态机迁移、护栏拒绝、聚合口径、渲染产物；不测内部编排细节 +- **集成测试（TestClient，主力接缝）**：会话生命周期（创建→对话→关闭）、护栏 409 各分支（超会话数/超轮数/间隔不足/加速线 auto 拒绝）、巡检 API 内容与 last_patrolled_at 推进、finalize 结算 expired、报告/导出纳入。先例：`test_campaigns_api.py`、`test_campaign_comparison_api.py` +- **纯函数单测**：探索发现聚合 + Markdown 渲染（dict 进 string 出）。先例：`test_report_render.py` +- **假客户端测试**：judge 抽样复核与分析输入扩展，注入假 ChatClient。先例：`test_campaign_analysis.py` +- 前端仅 `tsc --noEmit`（无 vitest，沿用现状） +- AI 助手 skill 行为不进自动化测试，靠部署后端到端验证（与 v0.4 以来 agenteval-run skill 验收方式一致） + +## Out of Scope + +- 双轨之二「探索活动」独立活动类型（OpenClaw 全权接管生命周期）——后续里程碑 +- 周期对比扩展（探索数据跨期对照）——待种子集指纹与达成率口径稳定 +- 计划条目重排（原 ADR-0003 v2 字面意义的"自适应重规划"）——本期代理只派发探索会话，不改写静态计划 +- 全局种子库/人设库（活动级配置已够，复用需求出现再提取） +- 事件驱动唤醒（纯 heartbeat/cron 巡检） +- 前端测试基线（vitest） +- volcengine-102 同步 + +## Further Notes + +- 词汇表与 ADR 已先行落地（commit c35159f）：CONTEXT.md「探索式评测」章节、ADR-0003 v2 修订 +- 巡检 API 的"新结果摘要"复用 `build_campaign_report` 聚合口径，不重算新指标 +- OpenClaw heartbeat/cron 事实依据：作业持久化 SQLite、重启不丢、瞬态错误重试、连续 10 次失败自动禁用——天然熔断，与"决策点不可用则跳过"的可靠性边界吻合 +- 里程碑目录 `.scratch/v0.9/`，票据由 `/to-tickets` 生成于 `issues/` diff --git a/backend/agenteval/models.py b/backend/agenteval/models.py index 9289d84..66221d3 100644 --- a/backend/agenteval/models.py +++ b/backend/agenteval/models.py @@ -313,6 +313,7 @@ class Campaign(BaseModel): analysis_model_config_id: Optional[str] = None exploration_seeds: Optional[ExplorationSeeds] = None exploration_budget: Optional[ExplorationBudgetConfig] = None + last_patrolled_at: Optional[datetime] = None class Turn(BaseModel): diff --git a/backend/agenteval/storage/db.py b/backend/agenteval/storage/db.py index 4a3c823..626d8ad 100644 --- a/backend/agenteval/storage/db.py +++ b/backend/agenteval/storage/db.py @@ -27,6 +27,15 @@ def utc_now() -> datetime: return datetime.now(timezone.utc) +def as_utc(dt: datetime) -> datetime: + """Attach UTC tzinfo to a naive datetime. + + SQLite round-trips drop tzinfo; stored times are always UTC, so a naive + value read back is restored as UTC before any comparison with utc_now(). + """ + return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc) + + def iso_utc(dt: datetime | None) -> str | None: """Serialize a datetime to ISO 8601 with UTC timezone suffix. @@ -190,6 +199,7 @@ class CampaignDB(SQLModel, table=True): analysis_model_config_id: Optional[str] = None exploration_seeds: Optional[str] = None exploration_budget: Optional[str] = None + last_patrolled_at: Optional[datetime] = None def get_plan(self) -> list[dict[str, Any]]: return _json_loads(self.plan) diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index f64fbfc..2e2e26e 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -352,6 +352,7 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): completed_at=campaign.completed_at, created_at=campaign.created_at, analysis_model_config_id=campaign.analysis_model_config_id, + last_patrolled_at=campaign.last_patrolled_at, ) db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan]) if campaign.summary: @@ -378,6 +379,7 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): analysis_model_config_id=db.analysis_model_config_id, exploration_seeds=db.get_exploration_seeds(), exploration_budget=db.get_exploration_budget(), + last_patrolled_at=db.last_patrolled_at, ) def update(self, campaign: Campaign) -> Optional[Campaign]: @@ -393,6 +395,7 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): existing.started_at = campaign.started_at existing.completed_at = campaign.completed_at existing.analysis_model_config_id = campaign.analysis_model_config_id + existing.last_patrolled_at = campaign.last_patrolled_at if campaign.summary is not None: existing.set_summary(campaign.summary.model_dump(mode="json")) if campaign.exploration_seeds is not None: diff --git a/backend/agenteval/web/routers/exploration.py b/backend/agenteval/web/routers/exploration.py index 4ce78df..e781c84 100644 --- a/backend/agenteval/web/routers/exploration.py +++ b/backend/agenteval/web/routers/exploration.py @@ -10,7 +10,7 @@ with 409 plus a readable reason, so the rejection itself is feedback to the resident agent. """ -from datetime import timezone +from datetime import datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException @@ -19,6 +19,7 @@ from sqlmodel import Session from agenteval.channels.factory import ChannelFactory from agenteval.config import get_settings +from agenteval.evaluation.report import generate_campaign_report from agenteval.exploration.models import ( ExplorationBudget, ExplorationMessage, @@ -28,12 +29,14 @@ from agenteval.exploration.models import ( normalize_experience, resolve_budget, ) -from agenteval.models import Campaign, CampaignStatus -from agenteval.storage.db import utc_now +from agenteval.models import Campaign, CampaignStatus, EvalRun +from agenteval.storage.db import as_utc, iso_utc, utc_now from agenteval.storage.repository import ( CampaignRepository, ExplorationMessageRepository, ExplorationSessionRepository, + RunRepository, + ScenarioRepository, TargetRepository, ) from agenteval.web.deps import get_db @@ -78,9 +81,7 @@ def _check_creation_guardrails( ) if sessions: latest = max(s.created_at for s in sessions if s.created_at) - if latest.tzinfo is None: # SQLite round-trip drops tzinfo; stored times are UTC - latest = latest.replace(tzinfo=timezone.utc) - elapsed = (utc_now() - latest).total_seconds() + elapsed = (utc_now() - as_utc(latest)).total_seconds() if elapsed < budget.min_interval_seconds: wait_minutes = budget.min_interval_seconds // 60 raise HTTPException( @@ -89,6 +90,90 @@ def _check_creation_guardrails( ) +def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]: + fresh = [] + for run in runs: + if run.completed_at is None: + continue + if watermark is not None and as_utc(run.completed_at) <= watermark: + continue + fresh.append(run) + return fresh + + +@router.get("/patrol") +async def patrol(session: Session = Depends(get_db)) -> dict: + """Stateless patrol for the resident agent. + + Reports every running production-line (time_scale == 1) campaign that + participates in exploration (has a seed set), with new results since the + last watermark and the remaining exploration budget. Advances each + patrolled campaign's watermark after building the response, so the next + call only reports increments. + """ + campaign_repo = CampaignRepository(session) + run_repo = RunRepository(session) + exploration_repo = ExplorationSessionRepository(session) + scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()} + target_names = {t.id: t.name for t in TargetRepository(session).list_all()} + + patrolled_at = utc_now() + entries: list[dict[str, Any]] = [] + patrolled_campaigns: list[Campaign] = [] + for campaign in campaign_repo.list_all(): + if campaign.status != CampaignStatus.RUNNING: + continue + if campaign.time_scale != 1: + continue + if campaign.exploration_seeds is None: + continue + + watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None + fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark) + new_results = None + if fresh: + report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names) + new_results = { + "summary": report["summary"], + "capability_summary": report["capability_summary"], + } + + budget = resolve_budget(campaign) + sessions = exploration_repo.list_by_campaign(campaign.id) + seconds_since_last_session = None + if sessions: + latest = max(as_utc(s.created_at) for s in sessions if s.created_at) + seconds_since_last_session = int((patrolled_at - latest).total_seconds()) + + entries.append( + { + "campaign_id": campaign.id, + "campaign_name": campaign.name, + "target_id": campaign.target_id, + "target_name": target_names.get(campaign.target_id), + "last_patrolled_at": iso_utc(campaign.last_patrolled_at), + "new_results": new_results, + "budget": { + "max_sessions": budget.max_sessions, + "sessions_used": len(sessions), + "remaining_sessions": max(0, budget.max_sessions - len(sessions)), + "max_turns": budget.max_turns, + "min_interval_seconds": budget.min_interval_seconds, + "seconds_since_last_session": seconds_since_last_session, + }, + } + ) + patrolled_campaigns.append(campaign) + + # 水位取构建响应之后的时刻：查询与持久化之间完成的结果不会在下次重复上报。 + watermark_at = utc_now() + for campaign in patrolled_campaigns: + campaign.last_patrolled_at = watermark_at + campaign_repo.update(campaign) + + return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries} + + @router.post("/sessions") async def create_session( request: CreateSessionRequest, diff --git a/migrations/versions/c5d8f0a2e4b7_add_patrol_watermark_to_campaigns.py b/migrations/versions/c5d8f0a2e4b7_add_patrol_watermark_to_campaigns.py new file mode 100644 index 0000000..fd3b5c9 --- /dev/null +++ b/migrations/versions/c5d8f0a2e4b7_add_patrol_watermark_to_campaigns.py @@ -0,0 +1,26 @@ +"""add campaigns.last_patrolled_at watermark + +Revision ID: c5d8f0a2e4b7 +Revises: b3c7d9e1f5a2 +Create Date: 2026-08-03 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "c5d8f0a2e4b7" +down_revision: Union[str, Sequence[str], None] = "b3c7d9e1f5a2" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("campaigns") as batch_op: + batch_op.add_column(sa.Column("last_patrolled_at", sa.DateTime(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("campaigns") as batch_op: + batch_op.drop_column("last_patrolled_at") diff --git a/tests/integration/test_campaigns_api.py b/tests/integration/test_campaigns_api.py index 58521f2..bc89c26 100644 --- a/tests/integration/test_campaigns_api.py +++ b/tests/integration/test_campaigns_api.py @@ -430,6 +430,7 @@ def test_exploration_config_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_budget")) + connection.execute(text("ALTER TABLE campaigns DROP COLUMN last_patrolled_at")) connection.execute(text("DROP TABLE IF EXISTS alembic_version")) command.stamp(config, "0e4a7c91d2b3") diff --git a/tests/integration/test_exploration_api.py b/tests/integration/test_exploration_api.py index d0184bd..9ba20fd 100644 --- a/tests/integration/test_exploration_api.py +++ b/tests/integration/test_exploration_api.py @@ -363,6 +363,7 @@ def test_exploration_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_budget")) + connection.execute(text("ALTER TABLE campaigns DROP COLUMN last_patrolled_at")) connection.execute(text("DROP TABLE IF EXISTS alembic_version")) command.stamp(config, "f2a9b7c34d18") diff --git a/tests/integration/test_exploration_patrol_api.py b/tests/integration/test_exploration_patrol_api.py new file mode 100644 index 0000000..97dff07 --- /dev/null +++ b/tests/integration/test_exploration_patrol_api.py @@ -0,0 +1,203 @@ +"""Integration tests for the exploration patrol API (v0.9 票据 03). + +One stateless call returns every running production-line campaign that +participates in exploration, the new results since the last patrol watermark +(reusing the campaign report aggregation), and the remaining exploration +budget. The watermark advances after each call so subsequent calls only +report increments. +""" + +from datetime import timedelta + +import pytest +from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary +from agenteval.storage.db import utc_now +from agenteval.storage.repository import CampaignRepository, RunRepository +from agenteval.web.app import app +from httpx import ASGITransport, AsyncClient + +SEEDS = {"personas": ["急性子用户"], "goals": ["查询账单并缴费"]} + + +def _make_campaign(campaign_id: str, *, time_scale: float = 1.0, status: str = "running", seeds=SEEDS) -> Campaign: + return Campaign( + id=campaign_id, + name=f"campaign-{campaign_id}", + target_id="t-1", + window_seconds=86400, + time_scale=time_scale, + plan=[{"scenario_id": "s-1", "offset_seconds": 0, "count": 1}], + status=status, + started_at=utc_now() - timedelta(hours=2), + exploration_seeds=seeds, + ) + + +def _seed_run(db_session, run_id: str, campaign_id: str, *, pass_rate: float, completed_at) -> None: + RunRepository(db_session).create(EvalRun( + id=run_id, target_id="t-1", scenario_id="s-1", campaign_id=campaign_id, + status=RunStatus.COMPLETED, started_at=completed_at - timedelta(minutes=5), + completed_at=completed_at, + summary=RunSummary(total_cases=2, pass_rate=pass_rate, avg_latency_ms=120.0), + )) + + +@pytest.fixture() +def seeded_db(db_session, monkeypatch): + from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus + from agenteval.storage import db as db_module + from agenteval.storage import repository as repo_module + from agenteval.storage.repository import TargetRepository + from agenteval.web import app as app_module + + monkeypatch.setattr(app_module, "init_db", lambda: None) + + def _test_get_session(): + return db_session + + monkeypatch.setattr(db_module, "get_session", _test_get_session) + monkeypatch.setattr(repo_module, "get_session", _test_get_session) + + from agenteval.web.deps import get_db + + def _test_get_db(): + try: + yield db_session + finally: + pass + + app.dependency_overrides[get_db] = _test_get_db + + TargetRepository(db_session).create(EvalTarget( + id="t-1", name="mock-target", + platform=PlatformType.AI_DIGITAL_EMPLOYEE, + channel_type=ChannelType.TUTU_API, + channel_config={"base_url": "http://mock", "token": "x"}, + status=TargetStatus.ACTIVE, + )) + repo = CampaignRepository(db_session) + repo.create(_make_campaign("c-prod")) # 正式线，参与探索 + repo.create(_make_campaign("c-fast", time_scale=24.0)) # 加速线 → 不巡检 + repo.create(_make_campaign("c-noseed", seeds=None)) # 无种子集 → 不巡检 + repo.create(_make_campaign("c-done", status="completed")) # 终态 → 不巡检 + + _seed_run(db_session, "r-1", "c-prod", pass_rate=1.0, completed_at=utc_now() - timedelta(hours=1)) + _seed_run(db_session, "r-2", "c-prod", pass_rate=0.5, completed_at=utc_now() - timedelta(minutes=30)) + + yield db_session + app.dependency_overrides.clear() + + +@pytest.fixture() +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +async def _patrol(client) -> dict: + resp = await client.get("/api/exploration/patrol") + assert resp.status_code == 200, resp.text + return resp.json() + + +async def test_patrol_filters_to_running_production_seeded_campaigns(client, seeded_db): + body = await _patrol(client) + ids = [c["campaign_id"] for c in body["campaigns"]] + assert ids == ["c-prod"] + + +async def test_patrol_entry_content(client, seeded_db): + body = await _patrol(client) + entry = body["campaigns"][0] + assert entry["campaign_name"] == "campaign-c-prod" + assert entry["target_id"] == "t-1" + assert entry["target_name"] == "mock-target" + assert entry["last_patrolled_at"] is None # 首次巡检无水位 + + new_results = entry["new_results"] + assert new_results is not None + assert new_results["summary"]["total_runs"] == 2 + assert new_results["summary"]["overall_pass_rate"] == 0.75 + assert new_results["capability_summary"][0]["scenario_id"] == "s-1" + + budget = entry["budget"] + assert budget["max_sessions"] == 8 + assert budget["sessions_used"] == 0 + assert budget["remaining_sessions"] == 8 + assert budget["max_turns"] == 12 + assert budget["min_interval_seconds"] == 30 * 60 + assert budget["seconds_since_last_session"] is None + + +async def test_patrol_watermark_advances_and_reports_increments(client, seeded_db): + await _patrol(client) + campaign = CampaignRepository(seeded_db).get("c-prod") + assert campaign.last_patrolled_at is not None + + # 第二次巡检：无新完成的子运行 → 增量为空 + body = await _patrol(client) + entry = body["campaigns"][0] + assert entry["new_results"] is None + + # 水位之后新完成一个子运行 → 第三次巡检只报这一个 + _seed_run(seeded_db, "r-3", "c-prod", pass_rate=0.0, completed_at=utc_now()) + body = await _patrol(client) + entry = body["campaigns"][0] + assert entry["new_results"]["summary"]["total_runs"] == 1 + assert entry["new_results"]["summary"]["overall_pass_rate"] == 0.0 + + +async def test_patrol_budget_reflects_existing_sessions(client, seeded_db): + from agenteval.exploration.models import ExplorationSession + from agenteval.storage.repository import ExplorationSessionRepository + + ExplorationSessionRepository(seeded_db).create(ExplorationSession( + campaign_id="c-prod", target_id="t-1", goal="查询账单", persona={"name": "x"}, + )) + body = await _patrol(client) + budget = body["campaigns"][0]["budget"] + assert budget["sessions_used"] == 1 + assert budget["remaining_sessions"] == 7 + assert budget["seconds_since_last_session"] is not None + + +async def test_patrol_budget_honours_campaign_override(client, seeded_db): + campaign = CampaignRepository(seeded_db).get("c-prod") + campaign.exploration_budget = {"max_sessions": 3} + CampaignRepository(seeded_db).update(campaign) + + body = await _patrol(client) + budget = body["campaigns"][0]["budget"] + assert budget["max_sessions"] == 3 + assert budget["remaining_sessions"] == 3 + + +async def test_patrol_migration_column_on_existing_db(tmp_path, monkeypatch): + """last_patrolled_at applies on a DB at the previous head.""" + from pathlib import Path + + from agenteval.storage import db as db_module + from alembic import command + from alembic.config import Config + from sqlalchemy import create_engine, inspect, text + from sqlmodel import SQLModel + + database_url = f"sqlite:///{tmp_path / 'patrol.db'}" + monkeypatch.setattr(db_module, "DATABASE_URL", database_url) + config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini")) + + SQLModel.metadata.create_all(create_engine(database_url)) + with create_engine(database_url).begin() as connection: + connection.execute(text("DROP TABLE IF EXISTS exploration_sessions")) + connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) + connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) + connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_budget")) + connection.execute(text("ALTER TABLE campaigns DROP COLUMN last_patrolled_at")) + connection.execute(text("DROP TABLE IF EXISTS alembic_version")) + + command.stamp(config, "b3c7d9e1f5a2") + command.upgrade(config, "head") + + cols = {c["name"] for c in inspect(create_engine(database_url)).get_columns("campaigns")} + assert "last_patrolled_at" in cols