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