From d957ea81c137b71d63bfd83c9308493db783024a Mon Sep 17 00:00:00 2001 From: Fiber Date: Sat, 17 Oct 2026 16:05:50 +0800 Subject: [PATCH 1/2] fix: isolate consolidation step budgets and cap pair backlog --- README.md | 27 +- src/iris/consolidation.py | 151 +++++++- .../034_consolidation_step_budgets.sql | 11 + src/iris/person_summary.py | 8 +- src/iris/settings_api.py | 5 +- tests/test_consolidation_step_budgets.py | 362 ++++++++++++++++++ tests/test_entry_profile_consolidation.py | 8 +- tests/test_maintenance_cognition.py | 20 +- tests/test_merge_scoring_v2.py | 2 +- tests/test_person_summary_admin.py | 2 +- tests/test_person_summary_consolidation.py | 17 +- 11 files changed, 557 insertions(+), 56 deletions(-) create mode 100644 src/iris/migrations/034_consolidation_step_budgets.sql create mode 100644 tests/test_consolidation_step_budgets.py diff --git a/README.md b/README.md index 18336834..b1fd05a4 100644 --- a/README.md +++ b/README.md @@ -706,7 +706,7 @@ prepare 和 search 为每条最终返回的记忆追加 `subject_annotations`， persona 在记忆整理后调用既有到期判断：自上一版至少五条自我记忆新增／变化，或满七天且有变化；“待更新”的依据计入变化。只在生成候选、逐句检查和版本／依据校验全部通过后发布，变化大时默认待管理员确认，仍遵循 persona 的 `small_medium_auto / all_auto / all_manual` 设置。未到期、缺模型、用途暂停、每日 token 限额或预算不足都有报告状态，原版保持可用。目标复核随后逐目标检查内部依据的当前状态，添加／结束依据标注，保留目标正文和状态；过期沿用目标模块按截止时间的只读标识，不自动完成或放弃。 -全部模型阶段共享每次最多 50 次调用的持久预算，包括 HTTP 重试和 JSON 修正。记忆任务按重要度及变化时间先处理，未完成留到下次。persona 开始前至少剩两次额度，生成的重试／修正还须为检查保留一次；预算中途不足或检查失败不会发布部分候选。目标复核不调用模型，预算耗尽或模型暂停仍能执行。无 gateway 的离线调用跳过模型阶段，确定性维护和目标复核照常。中断后已完成项目不重做；已完成 persona 尝试可补回报告，不重新生成，未完成尝试标为中断后留待下一次整理。 +入口画像、记忆（合并／矛盾／依赖）、人物摘要和 persona 分别使用自己的每轮调用上限，互不占用；HTTP 重试和 JSON 修正均逐次计入所属步骤。每日 token 限额、用途暂停和停止信号优先于各步上限。某一步耗尽额度只停止该步，后续步骤仍可运行，但仍须通过共同的停止条件。记忆任务按重要度降序、变化时间降序、ID 升序处理，未完成留到下次。persona 和人物摘要开始生成前须在自己的额度内留出生成、检查各一次；生成重试／修正仍须为检查保留一次，预算中途不足或检查失败不会发布部分候选。目标复核不调用模型，预算耗尽或模型暂停仍能执行。无 gateway 的离线调用跳过模型阶段，确定性维护和目标复核照常。中断后已完成项目不重做；已完成 persona 尝试可补回报告，不重新生成，未完成尝试标为中断后留待下一次整理。 管理员接口沿用会话、CSRF 和 Host 校验，前端接入点： @@ -715,16 +715,27 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 | `GET /admin/api/settings` 的 `consolidation` | 返回以下设置字段 | | `PATCH /admin/api/settings/consolidation` | 部分更新，未提供字段保持原值；写 `consolidation_settings_saved` 操作记录 | | `maintenance_time` | 默认 `03:00`，严格 `HH:MM`；与 `lifecycle.maintenance_time` 共用存储，两条设置接口保持同步 | -| `max_calls` | 默认 50，整数 0—50；0 停止模型调用，目标复核照常 | +| `max_calls` | 记忆步骤每轮上限，默认 300，整数 0—1000；0 只停止记忆步骤的模型调用 | +| `entry_profile_max_calls` | 入口画像每轮上限，默认 18，整数 0—1000 | +| `person_summary_max_calls` | 人物摘要每轮上限，默认 40，整数 0—1000 | +| `persona_max_calls` | persona 每轮上限，默认 6，整数 0—1000 | | `merge_enabled / conflict_enabled / dependency_enabled` | 合并／矛盾建议／模型依赖复核开关，默认全部 true；不关闭 M2 确定性维护 | | `persona_enabled / goal_review_enabled` | 定期 persona／目标依据复核开关，默认全部 true | | `person_summary_enabled` | 人物摘要整理开关，默认 true；只影响后续整理运行，不关闭已发布摘要的宿主读取 | | `POST /admin/api/maintenance` | `{}`，202 返回 `run_id`；已有未完成运行时返回同一个 ID | | `GET /admin/api/maintenance`、`/maintenance/{run_id}` | 分页列表及完整报告 | -一次整理在接受请求时固定设置、时区与对象上界；改时间、预算或开关只影响之后接受的运行，续跑使用原快照。persona 的发布方式仍从既有 `/settings/persona` 修改，并在该次 persona 尝试被接受时固定。 +一次整理在接受请求时固定设置、时区与对象上界；四项调用上限均写入 `consolidation_runs.settings_json`，改时间、预算或开关只影响之后接受的运行，续跑使用原快照，并按持久调用账本统计各步已用次数（包括中断时已准入的调用），不重置、不重复扣计。旧设置／快照缺少新字段时按上述默认值解释，旧快照原文不改。persona 的发布方式仍从既有 `/settings/persona` 修改，并在该次 persona 尝试被接受时固定。 -报告的 `items` 只列实际变化与失败：合并与继承来源、带摘录的矛盾／依赖模型建议、persona 发布／待确认／拒绝、目标依据变更及失败原因。`summary.skipped.reasons` 汇总跳过，`summary.model_calls` 汇总调用数、已报告输入／输出 token、耗时及未知用量调用数；`consolidation` 给出本次设置快照、延期项数及停止原因。`persona` 给出 `status`（`published / pending / rejected / not_due / skipped / failed / conflict`，运行中为 `running`）、`reason`、`due`、尝试与版本 ID，候选结果附正文、检查结果和变化程度；`goal_review` 给出是否开启、检查数或关闭原因。未到期和无变化成功检查不逐条留日志；来源报告中的建议不等于已确认事实。 +新数据库的记忆上限为 300；升级时，仅当已保存的 `max_calls` 为整数 50，且操作记录中从未出现 `consolidation_settings_saved`，迁移才将其改为 300，其余字段不变。管理员保存过整理设置（包括只改开关、显式保存 50）则保留原值。首次设置不写整理设置，数据库导入／备份恢复保留设置及操作记录；直接 SQL 或内部 `Store.set_setting` 写入且未留管理员记录的 50 无法与历史默认值区分，也会升级。 + +其他三项默认值按已有公开评测／确定性测试中单轮 HTTP 调用数的 P95 × 2 向上取整（线性分位数，包含重试／修正）：入口画像用 [EF-15](evals/entry_profile_eval/EF-15-report.md) 六个入口在同一时间点的合计，14 轮共 62 次（含一次 JSON 修正），P95=9，取 18；人物摘要用 [EF-02](evals/person_summary_eval/EF-02-20261013-report.md) 两次公开集完整运行的 16、16 次及 [PS1](evals/person_summary_eval/PS1-report.md) 已有十人定期步骤测试的 20 次，P95=19.6，取 40；persona 用 [维护集成测试](tests/test_maintenance_cognition.py) 三个正常生成／检查用例和四个 HTTP 重试／JSON 修正用例的 2、2、2、1、3、3、1 次，P95=3，取 6。这是小样本参考值；公开入口按时间点汇总，人物摘要包括完整评测批次，不能视为上线流量的 P95。管理员可按负载调整；0 停止对应步骤的模型调用，其他步骤及目标复核照常。 + +每轮规划仍至多扫描 128 个新种子。待处理 pair（`pending / classified / decided`）超过内部上限 2000 时，按既有重要度降序、变化时间降序、ID 升序保留前 2000 项，其余记为终态 `backlog_trimmed`，不生成 `consolidation_receipts`，不改变记忆。相同指纹不反复入队；任一侧记忆以后发生语义变化、扫描指纹改变时可重新规划。依赖任务不参与修剪。维护报告 `items` 中同名 outcome／reason 的 `details.count` 记录该轮修剪数量，`details.limit` 为 2000。 + +步骤准入会写入 `items` 的 `outcome=model_step_budget`：`details.step` 为 `entry_profile / memory / person_summary / persona`，`details.max_calls` 为该步快照上限，`details.calls_used` 为持久账本已用数；停止时 `reason` 记录 `call_budget / usage_limit / paused / interrupted` 等原因。未到期、关闭或没有任务的步骤可无预算项。旧报告的 `summary.model_calls` 仍为四步合计，`consolidation.skip_reason` 仍表示记忆整理阶段原因；判断其他步骤请读取各自条目。 + +报告的 `items` 列实际变化、失败和步骤预算：合并与继承来源、带摘录的矛盾／依赖模型建议、persona 发布／待确认／拒绝、目标依据变更及失败原因。`summary.skipped.reasons` 汇总跳过，`summary.model_calls` 汇总调用数、已报告输入／输出 token、耗时及未知用量调用数；`consolidation` 给出本次设置快照、延期项数及停止原因。`persona` 给出 `status`（`published / pending / rejected / not_due / skipped / failed / conflict`，运行中为 `running`）、`reason`、`due`、尝试与版本 ID，候选结果附正文、检查结果和变化程度；`goal_review` 给出是否开启、检查数或关闭原因。未到期和无变化成功检查不逐条留日志；来源报告中的建议不等于已确认事实。 每日维护还分批清理运行数据，每批最多处理 128 行并单独提交：`model_calls` 保留 90 天；`recalls` 及其 `recall_items`、`host_recalls` 保留 30 天；`batch_attempts` 和 `consolidation_calls` 保留记录，只清除超过 90 天的原始／修正正文字段。数据目录中的迁移前 `iris.db.<时间>.bak` 保留最近 3 份。未满保留期及边界时刻的数据不清理。支持中断后续跑；有资格自动重试的批次目标消息受清理保护。停机超过 24 小时后的补跑若在当天预定时间后开始，占用当天维护时段，避免同日再次定时衰减。 @@ -739,7 +750,7 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 - **代码累计证据，决定是否展示**：依据作为 sources 累积，独立性按确定性规则计算（转引、复述、接龙、刷屏、同一件事的跨日后续都不算新讨论）。达到门槛才送检查：话题至少两次独立讨论且跨两个本地日期；说法至少两次独立使用（保守地合并自然变体）；群规需要一次面向全群的明确公告（个人请求、一次性安排、外部规定都不算）。代码只硬拦能确定判断的情形：具体实体（专名、作品名、地点、型号、数字、日期）必须出现在依据中；明显的个人请求、一次性安排、外部规定不能成为群规；回复或引用链、同一发起人对同一对象的跟进合为同一讨论。其余语义判断交给逐条检查。 - **逐条检查**：候选刚达到门槛、或需要修正时，只带这一条和它自己的依据检查一次（提示词 entry_profile_admit_v3）：核对依据、性别代词、引文隐私、含义依据、局部做法不能写成群体习惯、不写容易过时的描述。检查只能删减或缩短依据，不改写正文；失败只影响这一条。 - **稳定性靠反复出现**：候选 importance 30、准入后提为 50（仍走普通衰减）；话题和说法准入时，保留强度补足到 importance 50 的初始值并重置衰减（#117；更高的保留强度不降低，同一已准入正文的复查和重放不重复补足）；不再被确认的主题按生命周期淡出，被确认会增加保留强度。同一讨论只加一次。 -- **预算与恢复**：观察和检查各自 180 秒（含 HTTP 重试和一次 JSON 修正），在事务外调用；观察、检查分别持久保存，重启可续跑，不重复加分。与其他整理共用每轮调用上限与每日 token 限额。 +- **预算与恢复**：观察和检查各自 180 秒（含 HTTP 重试和一次 JSON 修正），在事务外调用；观察、检查分别持久保存，重启可续跑，不重复加分。使用独立的 `entry_profile_max_calls`（默认 18，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。 - **宿主**：`prepare.entry_profile` 的 `{text, version, generated_at, possibly_stale}` 不变，兼容新增 `memory_ids`（与 text 中展示的条目一一对应）。version 是每个入口的单调整数，只在可展示的画像集合变化时加一，读取不改变它。依据失效的条目直接不返回。 - **管理**：入口画像面板按条显示画像记忆（子类、正文、依据原话、作者、准入状态、独立天数／次数、保留强度、最近确认时间、修订历史）和宿主投影预览。逐条编辑、删除、恢复、人工新建走通用记忆接口（修订号校验）。旧的整版编辑、回滚、确认、拒绝接口返回 410，旧版本保留为只读历史。全自动，没有“待确认候选”。 - **迁移**（032、033）：管理员手写的群规逐条转为人工创建的画像记忆（无法分类的显示为“管理员说明”）；旧的模型条目不直接导入，从原消息按新方式重建。 @@ -760,7 +771,7 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 `entry_profile.py` 确定性选择本入口最近 14 天的实际发言（message／self_output，不含行动结果、事件、未来消息），按角色时区分天，各天平分 6000 token 预算；每一天依次取最早、最晚和区间中点，摘录最多约 300 token，连同元数据计入预算。有效、在本入口可见且有本入口直接来源的记忆按重要度降序、ID 升序选取，合计 2000 token；每条最多附最早一条及最近三条本入口原话。材料使用 S／M 编号，入库换为消息 ID 与指纹，或记忆 ID、修订号、来源指纹。 -梦境整理的记忆模型阶段增加入口画像步骤，与其他整理共用 `max_calls`（默认 50，含 HTTP 重试）及每日 token 限额。首版要求窗口内至少 50 条消息；更新要求自上一版以来至少 50 条新消息，或自生成时间经过至少 168 小时且有新消息。依据被清理、修改、删除、遗忘或可见范围变化时只读计算 `possibly_stale` 和 `stale_basis`，下一次整理重新生成；全部生成路径每入口每个角色本地日期最多一次。手动重新生成可跳过更新数量／七天条件，首版门槛、每日上限、开关和用途暂停仍有效。停用不删除历史。 +梦境整理的入口画像步骤使用独立的 `entry_profile_max_calls`（默认 18，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。首版要求窗口内至少 50 条消息；更新要求自上一版以来至少 50 条新消息，或自生成时间经过至少 168 小时且有新消息。依据被清理、修改、删除、遗忘或可见范围变化时只读计算 `possibly_stale` 和 `stale_basis`，下一次整理重新生成；全部生成路径每入口每个角色本地日期最多一次。手动重新生成可跳过更新数量／七天条件，首版门槛、每日上限、开关和用途暂停仍有效。停用不删除历史。 生成和逐条检查各共享 120 秒总预算（含调用内重试），在数据库事务外执行，沿用原用途和整理调用上限。代码将空白连续段规整为一个空格，除此之外不改大小写、标点或字词；引语必须是材料中本入口消息正文或其 `quote.content` 的连续子串，也可来自入选记忆附带的本入口原始消息。材料摘录不包含人工添加的截断提示。只引用记忆编号而无消息原话不能通过。明确的账号号段、账号标签、邮箱、平台提及语法以及已知不透明账号 ID 作确定性过滤，其余禁止内容由检查调用判断。 @@ -829,7 +840,7 @@ EP2 本回合只做假模型机制验证，未运行真实模型或质量双判 默认开启、检查通过即发布，可按人物关闭或改全部人工确认。代码先核对引用，再由检查调用逐条判断依据、泛化、违规和时间范围，删去超过一半的模型条目即整版作废，保留上一版。模型调用均在事务外，写回核对版本、主体／设置修订、依据修订／生命周期和可见性；管理员手写内容保留，未改条目继续携带原依据。历史版本可读，编辑／回滚会新建版本；操作写入 admin_operations。 -整理接入要求至少 8 条相关全局 active 记忆：首版，或至少 5 条相关记忆新增／变化，或满 14 天且有变化；已过时版本可提前满足重生成条件，处理顺序仍按最近活跃排序。依据已删除／遗忘／彻底清除或近期事项到期的条目在返回前移除，剩余为空不返回。近期事项未给明确截止日时，默认从依据的最新事件时间（无则创建时间）起 30 天到期。被拒或确定性失败（无效输出、删条过多、长度超限等）后，自动整理等相关记忆出现新变化才再尝试；超时、限流、服务端错误等模型故障和 interrupted 可在后续整理重试，仍受模型健康与预算限制。管理员重生成绕过这一退避和日常更新阈值；普通确认不算内容变化。默认每次至多 10 人，最近活跃者优先，与整理共享 max_calls 和每日 token 上限。主体合并后，被并入方版本作废，存续方下次重生成。手动生成也受初始数量、开关和模型预算约束。 +整理接入要求至少 8 条相关全局 active 记忆：首版，或至少 5 条相关记忆新增／变化，或满 14 天且有变化；已过时版本可提前满足重生成条件，处理顺序仍按最近活跃排序。依据已删除／遗忘／彻底清除或近期事项到期的条目在返回前移除，剩余为空不返回。近期事项未给明确截止日时，默认从依据的最新事件时间（无则创建时间）起 30 天到期。被拒或确定性失败（无效输出、删条过多、长度超限等）后，自动整理等相关记忆出现新变化才再尝试；超时、限流、服务端错误等模型故障和 interrupted 可在后续整理重试，仍受模型健康与预算限制。管理员重生成绕过这一退避和日常更新阈值；普通确认不算内容变化。默认每次至多 10 人，最近活跃者优先，使用独立的 `person_summary_max_calls`（默认 40）和共同的每日 token 上限。主体合并后，被并入方版本作废，存续方下次重生成。手动生成也受初始数量、开关和模型预算约束。 PS2 接入 prepare 的 `people` 分区、管理接口和整理流程；界面由 UX 后续实现。`people_limit` 为人数上限，默认 3、最大 3、0 关闭；按显式参与者顺序优先，否则最近发言顺序。`participant_names` 按主体 ID 或显式 participants 中的名字／别名提供本次显示名，主体 ID 键优先，仅影响 people.name。每人正文 250 token，整个数组独立限 1000 个估算 token：给 3 人的正文预留 750，剩余约 250 用于通常长度的名字、主体 ID、时间和字段；长显示名时按实际序列化结果整人跳过，不截断限定语。详情见 [宿主 v1 兼容新增说明](docs/host-api.md)。新增分区在召回写事务结束后使用独立只读快照，先筛有当前已发布版的参与者，再复核依据；已有分区、排序、学习材料和召回判断输入保持不变。一次人物分区读取共用按需构建的名字表；名字形式参与者复用召回解析的主体 ID，再按当前公开别名来源复核。人物分区内部异常时返回 `people: []`，hints 附固定 `people_unavailable` 提示“人物摘要暂不可用。”；宿主运行记录复用该请求的单条 `prepare` 事件，使用固定 `result=error` 分类，不写异常正文或额外事件。试用回复会把 prepare 的 people 与 entry_profile 一起放入模型输入；冻结的 trial_reply_v1.md 未改。 @@ -850,7 +861,7 @@ PS2 接入 prepare 的 `people` 分区、管理接口和整理流程；界面由 所有写操作核对当前已发布版本号 `expected_version`（尚无发布版为 0）并记操作记录；版本、依据或设置冲突返回 409，已有在途管理生成任务返回 409。服务关闭先等待后台任务完成，再关闭网关和数据库；重启将中断任务记为失败。手写或未改条目保留规则沿用 PS1，清空版不可回滚复活。 -整理在记忆合并、矛盾处理和依赖复核全部结束后调用 `update_summaries_for_run`，仅使用剩余调用预算，保留其他步骤的相对顺序；关闭上述记忆步骤时仍可运行人物摘要。人物发现每轮共用一次名字表，并预计算每个主体独有的 label。每运行的设置快照含 `person_summary_enabled`；主体上限沿用 `runtime_settings.person_summary.max_subjects_per_run`（默认 10，范围 1—100）。生成／检查及网关内部重试计入共享 max_calls；每次生成准入预留一个检查名额，共享 chat 每日 token 上限。单个人物错误写入记录后继续其他人物。管理员正在重生成时，为本轮该人物记录一次 skipped／generation_in_progress，继续其他人物并计入人数上限；不会让维护停在整理阶段或反复写运行记录。步骤异常用固定错误码写入报告，再继续其他维护步骤。维护报告 `items` 中 outcome=person_summary，details 含 subject_id、status、version、attempt_id；步骤异常不含主体，status=failed。不重复调用已完成尝试，停止信号可中断后续工作。 +整理在记忆合并、矛盾处理和依赖复核全部结束后调用 `update_summaries_for_run`，使用独立的 `person_summary_max_calls`，保留其他步骤的相对顺序；关闭上述记忆步骤时仍可运行人物摘要。人物发现每轮共用一次名字表，并预计算每个主体独有的 label。每运行的设置快照含 `person_summary_enabled`；主体上限沿用 `runtime_settings.person_summary.max_subjects_per_run`（默认 10，范围 1—100）。生成／检查、JSON 修正及网关内部重试计入人物摘要的独立上限；每次生成准入预留一个检查名额，共享 chat 每日 token 上限。单个人物错误写入记录后继续其他人物。管理员正在重生成时，为本轮该人物记录一次 skipped／generation_in_progress，继续其他人物并计入人数上限；不会让维护停在整理阶段或反复写运行记录。步骤异常用固定错误码写入报告，再继续其他维护步骤。维护报告 `items` 中 outcome=person_summary，details 含 subject_id、status、version、attempt_id；步骤异常不含主体，status=failed。不重复调用已完成尝试，停止信号可中断后续工作。 给 UX：人物页增加当前摘要、分类／时间范围、逐条记忆依据、检查／删条记录、过时与清空原因、候选确认／拒绝、历史／对比、编辑／回滚、重生成任务轮询和每人开关／发布方式。编辑不设 150 字下限，按接口 edit_limits 提示上限；409 要刷新版本再提交，清空旧版禁用回滚。设置页增加 person_summary_enabled；整理报告按 outcome=person_summary 显示每人结果并链接到对应版本。generation_in_progress 显示“管理员生成中，本轮已跳过”；模型暂时失败允许重试，确定性拒绝等待新依据，但管理员仍可主动重新生成。interrupted 显示为可重试，不要求先改记忆；重生成仍受最少 8 条、开关、版本冲突及模型预算限制。前端本 PR 不修改。 diff --git a/src/iris/consolidation.py b/src/iris/consolidation.py index 673aa5dd..07271fb1 100644 --- a/src/iris/consolidation.py +++ b/src/iris/consolidation.py @@ -20,12 +20,17 @@ # Candidate order and parameters are frozen before the first real-model probe. METHODS = {'balanced': 0.40, 'strict': 0.65, 'broad': 0.20} RESOLUTIONS = ('report_only_v1',) -DEFAULTS = {'enabled': True, 'max_calls': 50, 'method': 'broad', 'resolution': 'report_only_v1', +DEFAULTS = {'enabled': True, 'max_calls': 300, + 'entry_profile_max_calls': 18, 'person_summary_max_calls': 40, 'persona_max_calls': 6, + 'method': 'broad', 'resolution': 'report_only_v1', 'merge_enabled': True, 'conflict_enabled': True, 'dependency_enabled': True, 'persona_enabled': True, 'goal_review_enabled': True, 'entry_profile_enabled': True, 'person_summary_enabled': True} PROMPT_VERSION = 'consolidation_report_only_v2' MERGE_PROMPT_VERSION = 'consolidation_pair_v3' MAX_ANCESTORS = 64 +MAX_PENDING_PAIRS = 2000 +BUDGET_FIELDS = {'memory': 'max_calls', 'entry_profile': 'entry_profile_max_calls', + 'person_summary': 'person_summary_max_calls', 'persona': 'persona_max_calls'} class InvalidOutput(ValueError): @@ -45,8 +50,8 @@ def consolidation_settings(conn): """One snapshot for new runs; evaluation-only method knobs remain internal.""" row = conn.execute("SELECT value_json FROM runtime_settings WHERE key='consolidation'").fetchone() value = {**DEFAULTS, **(json.loads(row[0]) if row else {}), 'resolution': 'report_only_v1'} - if (value['method'] not in METHODS or type(value['max_calls']) is not int - or not 0 <= value['max_calls'] <= 50 + if (value['method'] not in METHODS + or any(type(value[k]) is not int or not 0 <= value[k] <= 1000 for k in BUDGET_FIELDS.values()) or any(type(value[k]) is not bool for k in DEFAULTS if k.endswith('enabled'))): raise ValueError('invalid consolidation settings') return value @@ -57,6 +62,54 @@ def run_settings(conn, run_id): return {**DEFAULTS, **json.loads(row[0]), 'resolution': 'report_only_v1'} +def budget_step(purpose): + for step, prefix in (('entry_profile', 'consolidation_entry_profile_'), + ('person_summary', 'consolidation_person_summary_'), ('persona', 'persona_')): + if purpose.startswith(prefix): + return step + return 'memory' + + +def step_call_counts(conn, run_id): + # Include interrupted reservations and every retry/repair. Never infer usage + # from attempts, which may contain several network calls or no call at all. + counts = dict.fromkeys(BUDGET_FIELDS, 0) + for purpose, count in conn.execute('SELECT purpose,COUNT(*) FROM consolidation_calls WHERE run_id=? GROUP BY purpose', (run_id,)): + counts[budget_step(purpose)] += count + return counts + + +def report_step_budget(conn, run_id, step, stamp, reason=None): + settings = run_settings(conn, run_id) + details = {'step': step, 'max_calls': settings[BUDGET_FIELDS[step]], + 'calls_used': step_call_counts(conn, run_id)[step]} + conn.execute("""INSERT INTO maintenance_items + (run_id,phase,item_key,object_id,outcome,reason,details_json,created_at) + VALUES(?,'consolidation',?,?,'model_step_budget',?,?,?) + ON CONFLICT(run_id,phase,item_key) DO UPDATE SET details_json=excluded.details_json, + reason=COALESCE(excluded.reason,maintenance_items.reason)""", + (run_id,'budget:'+step,run_id,reason,dumps(details),stamp)) + + +def check_step_admission(store, gateway, run_id, step, *, required=2, clock=utc_now, stop=None): + """Preflight only: common stops precede step limits, without reserving a call.""" + try: + if stop and stop(): + raise ModelError('paused', 'maintenance stopped', paused=True, reason='interrupted') + health = getattr(gateway, 'health', None) + if health: + health.check('chat', 'learning') + with store.read() as conn: + settings = run_settings(conn, run_id) + count = step_call_counts(conn, run_id)[step] + if count + required > settings[BUDGET_FIELDS[step]]: + raise ModelError('budget', 'call budget', reason='call_budget') + except ModelError as error: + with store.write() as conn: + report_step_budget(conn, run_id, step, clock().isoformat(), error.reason or error.category) + raise + + def pair_policy(settings): return {key: settings[key] for key in ('merge_enabled', 'conflict_enabled')} @@ -570,13 +623,22 @@ def merge_detail_decision(payload, result): class _CallHealth: """Purpose-local view; never mutate the shared Gateway or its health object.""" - def __init__(self, owner, health, work): + def __init__(self, owner, health, work, *, stop=None): self.owner,self.real,self.work=owner,health,work + self.stop=stop if stop is not None else owner.stop self.call_id=None def check(self, kind, purpose, *, probe=False): # learning's existing admission includes the common daily generation limit. - token=self.real.check('chat','learning') if self.real else None + try: + if self.stop and self.stop(): + raise ModelError('paused','maintenance stopped',paused=True,reason='interrupted') + token=self.real.check('chat','learning') if self.real else None + except ModelError as error: + with self.owner.store.write() as conn: + report_step_budget(conn,self.owner.run_id,budget_step(purpose),self.owner.clock().isoformat(), + error.reason or error.category) + raise self.call_id=self.owner.reserve(self.work,purpose) return token @@ -596,17 +658,25 @@ class Consolidation: def __init__(self, store, gateway=None, *, clock=utc_now): self.store,self.gateway,self.clock=store,gateway,clock self.run_id=None + self.stop=None def reserve(self, work, purpose): + step=budget_step(purpose) with self.store.write() as conn: - settings=json.loads(conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?',(self.run_id,)).fetchone()[0]) - count=conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?',(self.run_id,)).fetchone()[0] - # A generation retry/repair must still leave one call for the checker. + settings=run_settings(conn,self.run_id) + count=step_call_counts(conn,self.run_id)[step] + # Each generation retry/repair still leaves one call for its checker. required = 2 if purpose.startswith(('persona_generate', 'consolidation_entry_profile_generate', 'consolidation_person_summary_generate')) else 1 - if count+required>settings['max_calls']: - raise ModelError('budget','call budget',reason='call_budget') - return conn.execute('INSERT INTO consolidation_calls(run_id,work_id,purpose,created_at) VALUES(?,?,?,?)', - (self.run_id,work['id'],purpose,self.clock().isoformat())).lastrowid + exhausted=count+required>settings[BUDGET_FIELDS[step]] + cid=None + if not exhausted: + cid=conn.execute('INSERT INTO consolidation_calls(run_id,work_id,purpose,created_at) VALUES(?,?,?,?)', + (self.run_id,work['id'],purpose,self.clock().isoformat())).lastrowid + report_step_budget(conn,self.run_id,step,self.clock().isoformat(),'call_budget' if exhausted else None) + # Persist the stop reason without spending a call or rolling back its report. + if exhausted: + raise ModelError('budget','call budget',reason='call_budget') + return cid def finish_call(self,cid,result,duration,usage,finish=None,error=None,effort=None,raw=None): with self.store.write() as conn: @@ -877,6 +947,7 @@ def enqueue(kind,payload): scanned=[(mid,fp) for mid,fp in scanned if mid not in failed_ids] with self.store.write() as conn: conn.executemany('INSERT OR REPLACE INTO consolidation_scanned VALUES(?,?)',scanned) + self._trim_pair_backlog(conn,run['id']) # A bounded run snapshot; the durable pending queue remains ordered. conn.execute('''INSERT OR IGNORE INTO consolidation_run_work(run_id,work_id) SELECT ?,id FROM consolidation_work WHERE state IN ('pending','classified','decided') @@ -892,6 +963,24 @@ def enqueue(kind,payload): settings['conflict_enabled'],settings['merge_enabled'],settings['conflict_enabled'])) conn.execute('UPDATE consolidation_runs SET planned=1 WHERE run_id=?',(run['id'],)) + def _trim_pair_backlog(self,conn,run_id): + # A terminal queue state, not a semantic decision or receipt. Unchanged + # fingerprints stay trimmed; a changed scan can enqueue a new fingerprint. + trimmed=[row[0] for row in conn.execute("""SELECT id FROM consolidation_work + WHERE kind='pair' AND state IN ('pending','classified','decided') + ORDER BY importance DESC,julianday(changed_at) DESC,id LIMIT -1 OFFSET ?""",(MAX_PENDING_PAIRS,))] + if not trimmed: + return + conn.executemany("UPDATE consolidation_work SET state='backlog_trimmed',result_json=? WHERE id=?", + ((dumps({'reason':'backlog_trimmed'}),wid) for wid in trimmed)) + previous=conn.execute("SELECT details_json FROM maintenance_items WHERE run_id=? AND phase='consolidation' AND item_key='backlog:pair'",(run_id,)).fetchone() + count=len(trimmed)+(json.loads(previous[0])['count'] if previous else 0) + conn.execute("""INSERT INTO maintenance_items + (run_id,phase,item_key,object_id,outcome,reason,details_json,created_at) + VALUES(?,'consolidation','backlog:pair',?,'backlog_trimmed','backlog_trimmed',?,?) + ON CONFLICT(run_id,phase,item_key) DO UPDATE SET details_json=excluded.details_json""", + (run_id,run_id,dumps({'count':count,'limit':MAX_PENDING_PAIRS}),self.clock().isoformat())) + def _stale(self,conn,payload): for old in payload['memories']: current=snapshot(conn,old['id']) @@ -924,6 +1013,7 @@ def halt(self,reason): def run(self,run,*,stop=None): self.run_id=run['id'] + self.stop=stop with self.store.write() as conn: if not conn.execute('SELECT 1 FROM consolidation_runs WHERE run_id=?',(self.run_id,)).fetchone(): conn.execute('INSERT INTO consolidation_runs(run_id,settings_json) VALUES(?,?)', @@ -938,9 +1028,22 @@ def run(self,run,*,stop=None): return True # Entry profiles are a separate durable step using this run's call ledger. from .entry_profile import update_profiles_for_run - if not update_profiles_for_run(self.store, _PersonaGateway(self, stop=stop), run, config, - clock=self.clock, stop=stop): - return False + profile_gateway=_PersonaGateway(self, stop=stop) + with self.store.read() as conn: + counts=step_call_counts(conn,self.run_id) + # Legacy entry-profile preflight counts the whole ledger. Offset only + # other steps; per-HTTP admission below always uses the exact step cap. + profile_config={**config,'max_calls':config['entry_profile_max_calls']+sum(counts.values())-counts['entry_profile']} + try: + if config['entry_profile_enabled']: + check_step_admission(self.store,profile_gateway,self.run_id,'entry_profile', + required=0,clock=self.clock,stop=stop) + except ModelError as error: + if error.reason=='interrupted': + return False + else: + if not update_profiles_for_run(self.store,profile_gateway,run,profile_config,clock=self.clock,stop=stop): + return False if not any(config[key] for key in ('merge_enabled','conflict_enabled','dependency_enabled')): if not self._person_summaries(run,config,stop): return False @@ -962,6 +1065,8 @@ def run(self,run,*,stop=None): return False except ModelError as exc: reason=exc.reason or exc.category + if reason=='interrupted': + return False if reason in ('call_budget','usage_limit','paused','temporary_unavailable') or exc.paused: with self.store.write() as conn: self.record(conn,work,'skipped',reason,done=False) @@ -1019,7 +1124,7 @@ def run(self,run,*,stop=None): # The failure receipt is a new short transaction. self.fail_work(work,exc,'invalid_action') blocked.add(frozenset(payload.get('pair_ids',[payload.get('target_id')]))) - # Memory merge/conflict/dependency work has first use of the budget. + # Person summaries have their own cap even if memory work exhausted its cap. if not self._person_summaries(run,config,stop): return False self.halt(budget_reason) @@ -1234,7 +1339,7 @@ def __init__(self, owner, *, stop=None): inner = object.__new__(_CallGateway) inner.__dict__ = self.gateway.__dict__.copy() inner._co = owner - inner.health = _CallHealth(owner, self.health, {'id': None}) + inner.health = _CallHealth(owner, self.health, {'id': None}, stop=stop) self.gateway = inner def chat(self, messages, purpose, max_tokens=16000, **kwargs): @@ -1242,7 +1347,7 @@ def chat(self, messages, purpose, max_tokens=16000, **kwargs): raise ModelError('paused', 'maintenance stopped', paused=True, reason='interrupted') if isinstance(self.gateway, Gateway): return self.gateway.chat(messages, purpose, max_tokens, **kwargs) - gate = _CallHealth(self.owner, self.health, {'id': None}) + gate = _CallHealth(self.owner, self.health, {'id': None}, stop=self.stop) gate.check('chat', purpose) started = self.monotonic() try: @@ -1290,10 +1395,14 @@ def update_persona_for_run(store, gateway, run, *, clock=utc_now, stop=None): status, reason = 'not_due', due['reason'] elif not gateway or not callable(getattr(gateway, 'chat', None)): reason = 'unconfigured' + if reason is None: + admission_owner=Consolidation(store,gateway,clock=clock) + admission_owner.run_id=run_id + try: + check_step_admission(store,_PersonaGateway(admission_owner,stop=stop),run_id,'persona',clock=clock,stop=stop) + except ModelError as error: + reason=error.reason or error.category with store.write() as conn: - count = conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?', (run_id,)).fetchone()[0] - if reason is None and count+2 > settings['max_calls']: - reason = 'call_budget' attempt = None if reason is None: previous = conn.execute('SELECT id FROM persona_versions WHERE is_current=1').fetchone() diff --git a/src/iris/migrations/034_consolidation_step_budgets.sql b/src/iris/migrations/034_consolidation_step_budgets.sql new file mode 100644 index 00000000..c433ac12 --- /dev/null +++ b/src/iris/migrations/034_consolidation_step_budgets.sql @@ -0,0 +1,11 @@ +-- Upgrade only the untouched historical default. Any administrator save, +-- including a switch-only save, preserves the recorded choice. Run snapshots +-- and all other settings remain unchanged; new step limits use Python defaults. +UPDATE runtime_settings +SET value_json=json_set(value_json,'$.max_calls',300) +WHERE key='consolidation' + AND json_type(value_json,'$.max_calls')='integer' + AND json_extract(value_json,'$.max_calls')=50 + AND NOT EXISTS ( + SELECT 1 FROM admin_operations WHERE action='consolidation_settings_saved' + ); diff --git a/src/iris/person_summary.py b/src/iris/person_summary.py index e740fdae..6c9ffa52 100644 --- a/src/iris/person_summary.py +++ b/src/iris/person_summary.py @@ -1167,11 +1167,13 @@ def update_summaries_for_run(store, gateway, run, config, *, clock=utc_now, stop if not due['due']: continue expected=due['current_version'] - count=conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?',(run_id,)).fetchone()[0] engine=PersonSummaryEngine(store,gateway,clock=clock) try: - if count+2>config['max_calls']: - result=engine._skip(subject_id,expected,'periodic',due,'call_budget',run_id,stamp) + from .consolidation import check_step_admission + try: + check_step_admission(store,gateway,run_id,'person_summary',clock=clock,stop=stop) + except ModelError as error: + result=engine._skip(subject_id,expected,'periodic',due,error.reason or error.category,run_id,stamp) else: result=engine.update(subject_id,expected_version=expected,run_id=run_id) except PersonSummaryBusy: diff --git a/src/iris/settings_api.py b/src/iris/settings_api.py index 55397716..75e614b5 100644 --- a/src/iris/settings_api.py +++ b/src/iris/settings_api.py @@ -171,7 +171,10 @@ class Lifecycle(Input): class ConsolidationSettings(Input): - max_calls: int = Field(default=50, ge=0, le=50, strict=True) + max_calls: int = Field(default=300, ge=0, le=1000, strict=True) + entry_profile_max_calls: int = Field(default=18, ge=0, le=1000, strict=True) + person_summary_max_calls: int = Field(default=40, ge=0, le=1000, strict=True) + persona_max_calls: int = Field(default=6, ge=0, le=1000, strict=True) merge_enabled: bool = Field(default=True, strict=True) conflict_enabled: bool = Field(default=True, strict=True) dependency_enabled: bool = Field(default=True, strict=True) diff --git a/tests/test_consolidation_step_budgets.py b/tests/test_consolidation_step_budgets.py new file mode 100644 index 00000000..7e9fe06b --- /dev/null +++ b/tests/test_consolidation_step_budgets.py @@ -0,0 +1,362 @@ +"""CO-07/08 independent durable budgets and bounded pair planning; no model IO.""" +import json +from types import SimpleNamespace + +import pytest +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from iris import consolidation, db +from iris.api import create_app +from iris.consolidation import Consolidation, _PersonaGateway, consolidation_settings, run_settings +from iris.db import Store, dumps +from iris.maintenance import Maintenance +from iris.memory_ops import operation +from iris.models import ModelError +from iris.settings_api import ConsolidationSettings +from conftest import login_admin +from test_consolidation import Model, pair +from test_lifecycle import Clock + + +BUDGETS = ('max_calls', 'entry_profile_max_calls', 'person_summary_max_calls', 'persona_max_calls') +PURPOSES = ('consolidation_merge', 'consolidation_entry_profile_observe', + 'consolidation_person_summary_check', 'persona_check') +MIGRATION = '034_consolidation_step_budgets.sql' + + +def owner_for(store, **settings): + store.set_setting('consolidation', {**dict.fromkeys(BUDGETS, 2), **settings}) + clock = Clock() + maintenance = Maintenance(store, gateway=Model(store), clock=clock) + rid = maintenance.request() + owner = Consolidation(store, maintenance.gateway, clock=clock) + owner.run_id = rid + with store.read() as conn: + run = dict(conn.execute('SELECT * FROM maintenance_runs WHERE id=?', (rid,)).fetchone()) + return owner, maintenance, run + + +@pytest.mark.parametrize('precreated', [False, True]) +def test_new_database_memory_call_budget_defaults_to_300(tmp_path, precreated): + path = tmp_path / 'new.db' + if precreated: + path.touch() + store = Store(path) + try: + with store.read() as conn: + assert consolidation_settings(conn)['max_calls'] == 300 + assert store.setting('consolidation')['max_calls'] == 300 + finally: + store.close() + + +@pytest.mark.parametrize('saved,value,expected', [(False, 50, 300), (True, 50, 50), + (True, 7, 7), (False, 7, 7), (True, 0, 0)]) +def test_migration_preserves_admin_choices_and_existing_run_snapshots(tmp_path, monkeypatch, saved, value, expected): + path = tmp_path / 'old.db' + original = db.files + def old_resources(package): + root = original(package) + def joinpath(name): + target = root.joinpath(name) + if name == 'migrations': + return SimpleNamespace(iterdir=lambda: [p for p in target.iterdir() if p.name != MIGRATION]) + return target + return SimpleNamespace(joinpath=joinpath) + with monkeypatch.context() as patch: + patch.setattr(db, 'files', old_resources) + old = Store(path) + settings = {'max_calls': value, 'merge_enabled': False, 'method': 'strict', 'extra_legacy_field': 'keep'} + old.set_setting('consolidation', settings) + with old.write() as conn: + if saved: + operation(conn, 'consolidation_settings_saved', 'settings', 'consolidation', {'max_calls': value}) + rid = Maintenance(old, clock=Clock()).request() + with old.read() as conn: + snapshot = conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?', (rid,)).fetchone()[0] + old.close() + for _ in range(2): + upgraded = Store(path) + try: + assert upgraded.setting('consolidation') == {**settings, 'max_calls': expected} + with upgraded.read() as conn: + assert conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?', (rid,)).fetchone()[0] == snapshot + finally: + upgraded.close() + + +def test_any_prior_admin_save_preserves_50_even_when_only_switch_was_saved(tmp_path, monkeypatch): + # A real settings PATCH saves the complete effective object, even if the + # administrator only changes a switch. The migration must respect that act. + path = tmp_path / 'old.db' + old = Store(path) + old.set_setting('consolidation', {'max_calls': 50}) + with old.write() as conn: + conn.execute('DELETE FROM schema_migrations WHERE version=?', (MIGRATION,)) + operation(conn, 'consolidation_settings_saved', 'settings', 'consolidation', {'merge_enabled': False}) + old.close() + reopened = Store(path) + try: + assert reopened.setting('consolidation')['max_calls'] == 50 + finally: + reopened.close() + + +@pytest.mark.parametrize('key', BUDGETS) +@pytest.mark.parametrize('value', [0, 51, 1000]) +def test_budget_range_accepts_integer_boundaries(store, key, value): + assert getattr(ConsolidationSettings(**{key: value}), key) == value + store.set_setting('consolidation', {key: value}) + with store.read() as conn: + assert consolidation_settings(conn)[key] == value + + +@pytest.mark.parametrize('key', BUDGETS) +@pytest.mark.parametrize('value', [-1, 1001, True, 1.5, '2']) +def test_budget_range_rejects_invalid_values(store, key, value): + with pytest.raises(ValidationError): + ConsolidationSettings(**{key: value}) + store.set_setting('consolidation', {key: value}) + with store.read() as conn, pytest.raises(ValueError): + consolidation_settings(conn) + + +def test_old_settings_missing_new_keys_and_patch_roundtrip(store): + store.set_setting('consolidation', {'max_calls': 50}) + with TestClient(create_app(store=store), base_url='http://127.0.0.1', client=('127.0.0.1', 12345)) as client: + login_admin(client) + client.app.state.scheduler.stop() + value = client.get('/admin/api/settings').json()['consolidation'] + assert value['max_calls'] == 50 + for key in BUDGETS[1:]: + assert value[key] == consolidation.DEFAULTS[key] + response = client.patch('/admin/api/settings/consolidation', json={'persona_max_calls': 9}) + assert response.status_code == 200 + assert response.json()['consolidation']['persona_max_calls'] == 9 + assert response.json()['consolidation']['max_calls'] == 50 + assert store.setting('consolidation')['person_summary_max_calls'] == value['person_summary_max_calls'] + + +@pytest.mark.parametrize('exhausted', range(4)) +def test_every_step_has_independent_limit_and_report(store, exhausted): + owner, maintenance, run = owner_for(store) + for _ in range(2): + owner.reserve({'id': None}, PURPOSES[exhausted]) + with pytest.raises(ModelError) as error: + owner.reserve({'id': None}, PURPOSES[exhausted]) + assert error.value.reason == 'call_budget' + for step, purpose in enumerate(PURPOSES): + if step != exhausted: + owner.reserve({'id': None}, purpose) + owner.reserve({'id': None}, purpose) + with store.read() as conn: + assert conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?', (run['id'],)).fetchone()[0] == 8 + reports = [i for i in maintenance.report(run['id'])['items'] if i['outcome'] == 'model_step_budget'] + assert len(reports) == 4 + assert all(i['details']['calls_used'] == 2 and i['details']['max_calls'] == 2 for i in reports) + assert sum(i['reason'] == 'call_budget' for i in reports) == 1 + + +def test_resume_uses_snapshot_and_counts_interrupted_calls_without_reserving_twice(store): + owner, maintenance, run = owner_for(store) + owner.reserve({'id': None}, 'persona_generate') + owner.reserve({'id': None}, 'consolidation_merge') + store.set_setting('consolidation', dict.fromkeys(BUDGETS, 1000)) + reopened = Consolidation(store, owner.gateway, clock=owner.clock) + reopened.run_id = run['id'] + with store.read() as conn: + settings = run_settings(conn, run['id']) + assert all(settings[key] == 2 for key in BUDGETS) + assert conn.execute("SELECT COUNT(*) FROM consolidation_calls WHERE result='interrupted'").fetchone()[0] == 2 + with pytest.raises(ModelError): + reopened.reserve({'id': None}, 'persona_generate_repair') + reopened.reserve({'id': None}, 'persona_check') + reopened.reserve({'id': None}, 'consolidation_conflict_repair') + with pytest.raises(ModelError): + reopened.reserve({'id': None}, 'consolidation_dependency') + assert maintenance.report(run['id'])['summary']['model_calls']['count'] == 4 + + +@pytest.mark.parametrize('reason', ['usage_limit', 'paused']) +@pytest.mark.parametrize('purpose', PURPOSES) +def test_global_gate_precedes_empty_step_budget(store, reason, purpose): + owner, _, run = owner_for(store, **dict.fromkeys(BUDGETS, 0)) + class Health: + def check(self, kind, call_purpose, **kwargs): + assert (kind, call_purpose) == ('chat', 'learning') + raise ModelError('paused', reason, reason=reason, paused=True) + owner.gateway.health = Health() + gateway = _PersonaGateway(owner) + with pytest.raises(ModelError) as error: + gateway.chat([], purpose) + assert error.value.reason == reason + with store.read() as conn: + assert conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?', (run['id'],)).fetchone()[0] == 0 + + +def test_persona_global_gate_precedes_empty_preflight_budget(store): + from test_maintenance_cognition import persona_case, complete + engine, model, _ = persona_case(store, budget=0, persona_max_calls=0) + class Health: + def check(self, *args, **kwargs): + raise ModelError('paused', 'limit', reason='usage_limit', paused=True) + model.health = Health() + report = complete(engine) + assert report['persona']['reason'] == 'usage_limit' + assert not model.calls + + +def test_summary_global_gate_precedes_empty_preflight_budget(store): + from iris.person_summary import update_summaries_for_run + from test_person_summary_consolidation import prepare + clock, model, gateway, run, config = prepare(store, budget=0) + class Health: + def check(self, *args, **kwargs): + raise ModelError('paused', 'limit', reason='usage_limit', paused=True) + gateway.health = Health() + update_summaries_for_run(store, gateway, run, config, clock=clock) + with store.read() as conn: + assert conn.execute('SELECT reason FROM person_summary_attempts WHERE run_id=?', (run['id'],)).fetchone()[0] == 'usage_limit' + assert not model.calls + + +def test_stop_precedes_budget_and_does_not_consume_calls(store): + owner, _, _ = owner_for(store, **dict.fromkeys(BUDGETS, 0)) + with pytest.raises(ModelError) as error: + _PersonaGateway(owner, stop=lambda: True).chat([], 'persona_generate') + assert error.value.reason == 'interrupted' + with store.read() as conn: + assert conn.execute('SELECT COUNT(*) FROM consolidation_calls').fetchone()[0] == 0 + + +def test_pair_backlog_keeps_processing_order_and_excludes_dependencies(store): + owner, maintenance, run = owner_for(store) + payload = {'pair_ids': [], 'memories': [], 'resolution': 'report_only_v1', + 'report_version': consolidation.PROMPT_VERSION, 'merge_version': consolidation.MERGE_PROMPT_VERSION} + with store.write() as conn: + for i in range(2003): + conn.execute('''INSERT INTO consolidation_work(fingerprint,kind,payload_json,importance,changed_at,created_at,state) + VALUES(?,'pair',?,?,?,?,?)''', (str(i), dumps(payload), i % 4, + f'2026-10-{10 + i % 5:02}T00:00:00+00:00', Clock()().isoformat(), + ('pending', 'classified', 'decided')[i % 3])) + conn.execute("INSERT INTO consolidation_work(fingerprint,kind,payload_json,importance,changed_at,created_at) VALUES('dep','dependency','{}',0,'2020-01-01','2020-01-01')") + expected = [r[0] for r in conn.execute("SELECT id FROM consolidation_work WHERE kind='pair' ORDER BY importance DESC,julianday(changed_at) DESC,id LIMIT 2000")] + config = run_settings(conn, run['id']) + owner.plan(run, config) + with store.read() as conn: + kept = [r[0] for r in conn.execute("SELECT id FROM consolidation_work WHERE kind='pair' AND state IN ('pending','classified','decided') ORDER BY importance DESC,julianday(changed_at) DESC,id")] + assert kept == expected + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE state='backlog_trimmed'").fetchone()[0] == 3 + assert conn.execute("SELECT state FROM consolidation_work WHERE fingerprint='dep'").fetchone()[0] == 'pending' + assert conn.execute('SELECT COUNT(*) FROM consolidation_receipts').fetchone()[0] == 0 + items = [i for i in maintenance.report(run['id'])['items'] if i['reason'] == 'backlog_trimmed'] + assert len(items) == 1 and items[0]['details']['count'] == 3 + owner.plan(run, config) + items = [i for i in maintenance.report(run['id'])['items'] if i['reason'] == 'backlog_trimmed'] + assert len(items) == 1 and items[0]['details']['count'] == 3 + + +@pytest.mark.parametrize('changed_side', [0, 1]) +def test_trimmed_pair_reenters_only_after_semantic_change(store, monkeypatch, changed_side): + a, b = pair(store) + owner, _, run = owner_for(store) + with store.read() as conn: + config = run_settings(conn, run['id']) + monkeypatch.setattr(consolidation, 'MAX_PENDING_PAIRS', 0, raising=False) + owner.plan(run, config) + with store.read() as conn: + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE state='backlog_trimmed'").fetchone()[0] == 1 + monkeypatch.setattr(consolidation, 'MAX_PENDING_PAIRS', 2000) + second, _, next_run = owner_for(store) + second.plan(next_run, config) + with store.read() as conn: + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE state='pending'").fetchone()[0] == 0 + with store.write() as conn: + conn.execute("UPDATE memories SET content=content||'，补充细节',revision=revision+1,updated_at=? WHERE id=?", + (Clock()().isoformat(), (a, b)[changed_side])) + third, _, final_run = owner_for(store) + third.plan(final_run, config) + with store.read() as conn: + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE state='pending'").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE state='backlog_trimmed'").fetchone()[0] == 1 + assert conn.execute('SELECT COUNT(*) FROM consolidation_receipts').fetchone()[0] == 0 + + + +def test_legacy_run_snapshot_missing_step_keys_keeps_memory_cap_and_call_counts(store): + owner, maintenance, run = owner_for(store) + legacy = dumps({'max_calls': 50, 'merge_enabled': True}) + with store.write() as conn: + conn.execute('UPDATE consolidation_runs SET settings_json=? WHERE run_id=?', (legacy, run['id'])) + owner.reserve({'id': None}, 'persona_generate') + store.set_setting('consolidation', dict.fromkeys(BUDGETS, 0)) + resumed = Consolidation(store, owner.gateway, clock=owner.clock) + resumed.run_id = run['id'] + resumed.reserve({'id': None}, 'persona_check') + with store.read() as conn: + settings = run_settings(conn, run['id']) + assert settings['max_calls'] == 50 + assert all(settings[key] == consolidation.DEFAULTS[key] for key in BUDGETS[1:]) + assert conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?', (run['id'],)).fetchone()[0] == legacy + assert maintenance.report(run['id'])['summary']['model_calls']['count'] == 2 + + +def test_pair_backlog_does_not_starve_summary_or_persona_and_resume_does_not_repeat(store): + from test_maintenance_cognition import persona_case, complete + from test_person_summary import Model as SummaryModel, seed + engine, persona, clock = persona_case(store, budget=2, max_calls=1, + merge_enabled=True, conflict_enabled=True, person_summary_max_calls=2, entry_profile_enabled=False) + seed(store, clock, count=16) + memory, summary = Model(store), SummaryModel(store) + class Combined: + def chat(self, messages, purpose, max_tokens=16000, **kwargs): + model = persona if purpose.startswith('persona_') else summary if purpose.startswith('consolidation_person_summary_') else memory + return model.chat(messages, purpose, max_tokens, **kwargs) + engine.gateway = Combined() + report = complete(engine) + assert report['state'] == 'completed' + assert report['consolidation']['skip_reason'] == 'call_budget' + assert report['persona']['status'] == 'published' + assert any(i['outcome'] == 'person_summary' and i['details']['status'] == 'published' for i in report['items']) + assert len(memory.requests) == 1 and len(summary.calls) == len(persona.calls) == 2 + with store.read() as conn: + assert conn.execute("SELECT COUNT(*) FROM consolidation_work WHERE kind='pair' AND state='pending'").fetchone()[0] > 20 + assert consolidation.step_call_counts(conn, report['id']) == {'memory': 1, 'entry_profile': 0, 'person_summary': 2, 'persona': 2} + Maintenance(store, gateway=engine.gateway, clock=clock).run(report['id']) + assert len(memory.requests) == 1 and len(summary.calls) == len(persona.calls) == 2 + assert engine.report(report['id'])['summary']['model_calls']['count'] == 5 + + +@pytest.mark.parametrize('reason', ['usage_limit', 'paused', 'interrupted']) +@pytest.mark.parametrize('purpose', PURPOSES) +def test_global_stop_during_http_retry_precedes_step_budget_without_extra_reservation(store, reason, purpose): + import httpx + from iris.models import Gateway, ModelConfig + owner, _, run = owner_for(store, **dict.fromkeys(BUDGETS, 1)) + hits = [] + class Health: + def check(self, *args, **kwargs): + if hits and reason != 'interrupted': + raise ModelError('paused', reason, reason=reason, paused=True) + def observe(self, *args, **kwargs): + return False + def respond(request): + assert not store._writer.in_transaction + hits.append(1) + return httpx.Response(429, json={'error': {'code': 'RateLimitExceeded'}}, headers={'Retry-After': '0'}) + with httpx.Client(transport=httpx.MockTransport(respond)) as client: + gateway = Gateway({'chat': ModelConfig('https://example.invalid', '', 'fake')}, store, client, sleeper=lambda _: None) + gateway.health = Health() + owner.gateway = gateway + try: + with pytest.raises(ModelError) as error: + _PersonaGateway(owner, stop=lambda: reason == 'interrupted' and bool(hits)).chat([], purpose) + assert error.value.reason == reason + finally: + gateway.close() + assert len(hits) == 1 + with store.read() as conn: + assert conn.execute('SELECT COUNT(*) FROM consolidation_calls WHERE run_id=?', (run['id'],)).fetchone()[0] == 1 + assert conn.execute('SELECT COUNT(*) FROM model_calls').fetchone()[0] == 1 + assert conn.execute("SELECT reason FROM maintenance_items WHERE item_key=?", ('budget:'+consolidation.budget_step(purpose),)).fetchone()[0] == reason diff --git a/tests/test_entry_profile_consolidation.py b/tests/test_entry_profile_consolidation.py index 4f419996..9db80287 100644 --- a/tests/test_entry_profile_consolidation.py +++ b/tests/test_entry_profile_consolidation.py @@ -11,7 +11,7 @@ def dream(store, clock, **settings): - store.set_setting('consolidation',dict(enabled=True,max_calls=50,merge_enabled=False, + store.set_setting('consolidation',dict(enabled=True,entry_profile_max_calls=50,merge_enabled=False, conflict_enabled=False,dependency_enabled=False,persona_enabled=False,goal_review_enabled=False,**settings)) engine=Maintenance(store,gateway=Model(store),clock=clock) return engine,engine.request() @@ -33,7 +33,7 @@ def test_profiles_run_when_other_model_steps_are_disabled_and_not_repeated(store 'consolidation_entry_profile_generate','consolidation_entry_profile_check'] -@pytest.mark.parametrize('setting,value',[('enabled',False),('entry_profile_enabled',False),('max_calls',0),('max_calls',1)]) +@pytest.mark.parametrize('setting,value',[('enabled',False),('entry_profile_enabled',False),('entry_profile_max_calls',0),('entry_profile_max_calls',1)]) def test_switches_and_small_budgets_do_not_call(store,setting,value): clock=Clock(); messages(store,clock) engine,rid=dream(store,clock) @@ -51,7 +51,7 @@ def test_two_entries_share_run_cap_and_budget_skips_do_not_consume_daily_generat engine,rid=dream(store,clock) with store.write() as conn: config=json.loads(conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?',(rid,)).fetchone()[0]) - config['max_calls']=2 + config['entry_profile_max_calls']=2 conn.execute('UPDATE consolidation_runs SET settings_json=? WHERE run_id=?',(dumps(config),rid)) engine.run(rid) assert len(engine.gateway.calls)==2 @@ -64,7 +64,7 @@ def test_gateway_reserves_checker_on_each_generation_retry(store): clock=Clock(); messages(store,clock) _,rid=dream(store,clock) with store.write() as conn: - config=json.loads(conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?',(rid,)).fetchone()[0]);config['max_calls']=2 + config=json.loads(conn.execute('SELECT settings_json FROM consolidation_runs WHERE run_id=?',(rid,)).fetchone()[0]);config['entry_profile_max_calls']=2 conn.execute('UPDATE consolidation_runs SET settings_json=? WHERE run_id=?',(dumps(config),rid)) owner=Consolidation(store,Model(store),clock=clock);owner.run_id=rid owner.reserve({'id':None},'consolidation_entry_profile_generate') diff --git a/tests/test_maintenance_cognition.py b/tests/test_maintenance_cognition.py index 25db53b7..252e42dc 100644 --- a/tests/test_maintenance_cognition.py +++ b/tests/test_maintenance_cognition.py @@ -25,7 +25,7 @@ def persona_case(store, *, budget=50, degree='small', supported=True, **settings with store.write() as conn: conn.execute('UPDATE persona_versions SET created_at=?', ((clock()-timedelta(days=8)).isoformat(),)) add_self(store, '我在阅读时喜欢安静。') - store.set_setting('consolidation', {'max_calls':budget, 'merge_enabled':False, + store.set_setting('consolidation', {'persona_max_calls':budget, 'merge_enabled':False, 'conflict_enabled':False, 'dependency_enabled':False, **settings}) model = FakeGateway(store, {'sentences':[{'text':'初始设定中，我来自云城。','basis':['M1']}]}, degree=degree, supported=supported) @@ -62,7 +62,7 @@ def test_persona_insufficient_budget_does_not_start_and_next_run_can_update(stor assert report['persona']['reason'] == 'call_budget' with store.read() as conn: assert conn.execute('SELECT COUNT(*) FROM persona_attempts').fetchone()[0] == 0 - store.set_setting('consolidation', {**store.setting('consolidation'), 'max_calls':2}) + store.set_setting('consolidation', {**store.setting('consolidation'), 'persona_max_calls':2}) assert complete(engine)['persona']['status'] == 'published' assert len(model.calls) == 2 @@ -186,7 +186,7 @@ def test_settings_patch_is_audited_and_only_next_run_changes(store): assert response.json()['lifecycle']['maintenance_time']=='04:15' engine.run(rid) assert engine.report(rid)['summary']['merged']['count']==1 - assert engine.report(rid)['consolidation']['settings']['max_calls']==50 + assert engine.report(rid)['consolidation']['settings']['max_calls']==300 next_report = complete(engine) assert not next_report['consolidation']['settings']['merge_enabled'] assert next_report['persona']['reason']=='disabled' @@ -194,7 +194,7 @@ def test_settings_patch_is_audited_and_only_next_run_changes(store): assert conn.execute("SELECT 1 FROM admin_operations WHERE action='consolidation_settings_saved'").fetchone() -@pytest.mark.parametrize('payload', [{'max_calls':51},{'max_calls':-1},{'max_calls':True},{'merge_enabled':'false'}, +@pytest.mark.parametrize('payload', [{'max_calls':1001},{'max_calls':-1},{'max_calls':True},{'merge_enabled':'false'}, {'maintenance_time':'24:00'},{'method':'strict'},{'unrecognized':True}]) def test_settings_strict_validation(store, payload): with TestClient(create_app(store=store),base_url='http://127.0.0.1',client=('127.0.0.1',12345)) as client: @@ -227,8 +227,8 @@ def test_persona_writeback_rechecks_version_and_evidence(store, change): assert not [i for i in report['items'] if i['outcome']=='persona_published'] -def test_budget_is_shared_after_merge_and_defers_persona(store): - engine, persona, _ = persona_case(store, budget=2, merge_enabled=True, conflict_enabled=True) +def test_memory_budget_does_not_defer_persona_with_its_own_limit(store): + engine, persona, _ = persona_case(store, budget=2, max_calls=1, merge_enabled=True, conflict_enabled=True) pair(store) # Keep persona's self evidence from forming irrelevant candidate pairs. co = Model(store,merge_answer) @@ -239,8 +239,8 @@ def chat(self, messages, purpose, max_tokens=16000, **kwargs): engine.gateway = Combined() report = complete(engine) assert report['summary']['merged']['count']==1 - assert report['summary']['model_calls']['count']==1 - assert report['persona']['reason']=='call_budget' and not persona.calls + assert report['summary']['model_calls']['count']==3 + assert report['persona']['status']=='published' and len(persona.calls)==2 def test_D01_prepare_and_goal_operations_while_persona_is_running(store): @@ -282,7 +282,7 @@ def test_stop_between_persona_generation_and_check_never_publishes(store): (2,'repair',1,'failed'), (3,'repair',3,'published'), (3,'retry',3,'published'), (2,'retry',1,'failed'), ]) -def test_persona_http_retries_and_json_repairs_use_shared_budget(store,budget,first_response,expected_calls,status): +def test_persona_http_retries_and_json_repairs_use_persona_budget(store,budget,first_response,expected_calls,status): import httpx from iris.models import Gateway, ModelConfig from fake_openai import completion @@ -399,7 +399,7 @@ def test_consolidation_settings_csrf_and_time_alias(store): assert client.get('/admin/api/settings').json()['consolidation']['maintenance_time']=='05:30' client.headers.pop('X-Iris-CSRF') assert client.patch('/admin/api/settings/consolidation',json={'max_calls':1}).status_code==403 - assert client.get('/admin/api/settings').json()['consolidation']['max_calls']==50 + assert client.get('/admin/api/settings').json()['consolidation']['max_calls']==300 def test_complete_phase_order_merges_before_persona_then_reviews_goals(store): diff --git a/tests/test_merge_scoring_v2.py b/tests/test_merge_scoring_v2.py index 9a54cdfa..b309eb58 100644 --- a/tests/test_merge_scoring_v2.py +++ b/tests/test_merge_scoring_v2.py @@ -278,7 +278,7 @@ def test_revision_skip_after_an_earlier_merge_is_a_quality_miss_not_infrastructu def test_runner_budget_exhaustion_is_invalid_without_extra_calls(tmp_path): case = runner.load_corpus(runner.ROOT / 'evals/consolidation_v1.json')[0] def factory(store, clock): - store.set_setting('consolidation', {'max_calls': 1}) + store.set_setting('consolidation', {**store.setting('consolidation'), 'max_calls': 1}) return Model(store, {'decision': 'separate', 'reason': 'fake', 'evidence': [], 'updates': [], 'unknown': 1}) row = runner.case_run(case, tmp_path / 'budget.db', factory, 'broad') assert len(row['calls']) == 1 diff --git a/tests/test_person_summary_admin.py b/tests/test_person_summary_admin.py index 4f52c3de..0392bd99 100644 --- a/tests/test_person_summary_admin.py +++ b/tests/test_person_summary_admin.py @@ -119,6 +119,6 @@ def test_consolidation_setting_roundtrip_and_strict_type(client,store): response=client.patch('/admin/api/settings/consolidation',json={'person_summary_enabled':False}) assert response.status_code==200 and response.json()['consolidation']['person_summary_enabled'] is False assert client.patch('/admin/api/settings/consolidation',json={'person_summary_enabled':1}).status_code==400 - assert client.get('/admin/api/settings').json()['consolidation']['max_calls']==50 + assert client.get('/admin/api/settings').json()['consolidation']['max_calls']==300 with store.read() as conn: assert conn.execute("SELECT 1 FROM admin_operations WHERE action='consolidation_settings_saved'").fetchone() diff --git a/tests/test_person_summary_consolidation.py b/tests/test_person_summary_consolidation.py index cc39bbdb..cb635bfa 100644 --- a/tests/test_person_summary_consolidation.py +++ b/tests/test_person_summary_consolidation.py @@ -1,4 +1,4 @@ -"""Shared budgets and the PS2 automatic consolidation step.""" +"""Independent step budgets and the PS2 automatic consolidation step.""" import json import pytest @@ -13,7 +13,7 @@ def prepare(store, *, budget=50, limit=10, enabled=True): clock=Clock(); seed(store,clock) for i in range(8): memory(store,clock,f'乙同学测试{i}',speaker='b',days_ago=3) - config={'enabled':True,'max_calls':budget,'merge_enabled':False,'conflict_enabled':False, + config={'enabled':True,'person_summary_max_calls':budget,'merge_enabled':False,'conflict_enabled':False, 'dependency_enabled':False,'persona_enabled':False,'goal_review_enabled':False,'entry_profile_enabled':False, 'person_summary_enabled':enabled} store.set_setting('consolidation',config) @@ -235,9 +235,10 @@ def names(conn): assert built[0].unique_labels['b'] == ['乙同学'] -def test_memory_consolidation_uses_budget_before_person_summaries(store, monkeypatch): - clock, model, gateway, run, config = prepare(store, budget=3, limit=2) +def test_memory_consolidation_does_not_spend_person_summary_budget(store, monkeypatch): + clock, model, gateway, run, config = prepare(store, budget=4, limit=2) config['merge_enabled'] = True + config['max_calls'] = 1 store.set_setting('consolidation', config) with store.write() as conn: conn.execute('UPDATE consolidation_runs SET settings_json=? WHERE run_id=?', (dumps(config),run['id'])) @@ -264,10 +265,12 @@ def apply(owner, work, payload, result, config, blocked): maintenance = Maintenance(store, gateway=model, clock=clock) maintenance.run(run['id']) assert order == ['memory', 'applied', 'summary'] - assert len(model.calls) == 2 + assert len(model.calls) == 4 report = maintenance.report(run['id']) assert report['state'] == 'completed' - assert any(i['outcome']=='person_summary' and i['reason']=='call_budget' for i in report['items']) + assert sum(i['outcome']=='person_summary' and i['details']['status']=='published' for i in report['items']) == 2 with store.read() as conn: purposes = [r[0] for r in conn.execute('SELECT purpose FROM consolidation_calls ORDER BY id')] - assert purposes == ['consolidation_dependency','consolidation_person_summary_generate','consolidation_person_summary_check'] + assert purposes == ['consolidation_dependency', + 'consolidation_person_summary_generate','consolidation_person_summary_check', + 'consolidation_person_summary_generate','consolidation_person_summary_check'] From a17811bc8ca3e2cd558ae3c70997b53a3ddfe2bc Mon Sep 17 00:00:00 2001 From: Fiber Date: Sat, 17 Oct 2026 16:51:10 +0800 Subject: [PATCH 2/2] fix: increase entry profile and persona default call caps --- README.md | 14 +++++++++----- src/iris/consolidation.py | 2 +- src/iris/settings_api.py | 4 ++-- tests/test_consolidation_step_budgets.py | 6 ++++-- 4 files changed, 16 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index b1fd05a4..1c54ba03 100644 --- a/README.md +++ b/README.md @@ -716,9 +716,9 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 | `PATCH /admin/api/settings/consolidation` | 部分更新，未提供字段保持原值；写 `consolidation_settings_saved` 操作记录 | | `maintenance_time` | 默认 `03:00`，严格 `HH:MM`；与 `lifecycle.maintenance_time` 共用存储，两条设置接口保持同步 | | `max_calls` | 记忆步骤每轮上限，默认 300，整数 0—1000；0 只停止记忆步骤的模型调用 | -| `entry_profile_max_calls` | 入口画像每轮上限，默认 18，整数 0—1000 | +| `entry_profile_max_calls` | 入口画像每轮上限，默认 60，整数 0—1000 | | `person_summary_max_calls` | 人物摘要每轮上限，默认 40，整数 0—1000 | -| `persona_max_calls` | persona 每轮上限，默认 6，整数 0—1000 | +| `persona_max_calls` | persona 每轮上限，默认 30，整数 0—1000 | | `merge_enabled / conflict_enabled / dependency_enabled` | 合并／矛盾建议／模型依赖复核开关，默认全部 true；不关闭 M2 确定性维护 | | `persona_enabled / goal_review_enabled` | 定期 persona／目标依据复核开关，默认全部 true | | `person_summary_enabled` | 人物摘要整理开关，默认 true；只影响后续整理运行，不关闭已发布摘要的宿主读取 | @@ -729,7 +729,11 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 新数据库的记忆上限为 300；升级时，仅当已保存的 `max_calls` 为整数 50，且操作记录中从未出现 `consolidation_settings_saved`，迁移才将其改为 300，其余字段不变。管理员保存过整理设置（包括只改开关、显式保存 50）则保留原值。首次设置不写整理设置，数据库导入／备份恢复保留设置及操作记录；直接 SQL 或内部 `Store.set_setting` 写入且未留管理员记录的 50 无法与历史默认值区分，也会升级。 -其他三项默认值按已有公开评测／确定性测试中单轮 HTTP 调用数的 P95 × 2 向上取整（线性分位数，包含重试／修正）：入口画像用 [EF-15](evals/entry_profile_eval/EF-15-report.md) 六个入口在同一时间点的合计，14 轮共 62 次（含一次 JSON 修正），P95=9，取 18；人物摘要用 [EF-02](evals/person_summary_eval/EF-02-20261013-report.md) 两次公开集完整运行的 16、16 次及 [PS1](evals/person_summary_eval/PS1-report.md) 已有十人定期步骤测试的 20 次，P95=19.6，取 40；persona 用 [维护集成测试](tests/test_maintenance_cognition.py) 三个正常生成／检查用例和四个 HTTP 重试／JSON 修正用例的 2、2、2、1、3、3、1 次，P95=3，取 6。这是小样本参考值；公开入口按时间点汇总，人物摘要包括完整评测批次，不能视为上线流量的 P95。管理员可按负载调整；0 停止对应步骤的模型调用，其他步骤及目标复核照常。 +入口画像默认 60：公开 [EF-15](evals/entry_profile_eval/EF-15-report.md) 六个入口在同一时间点的合计，14 轮共 62 次（含一次 JSON 修正），单轮 P95=9；上线实例已有 12 个群入口，入口数还会增加，原先按小样本取两倍得到的 18 可能让部分入口每天轮不到。规划者据此将上限设为 60，给更多入口留出额度。 + +persona 默认 30：现有[维护集成测试](tests/test_maintenance_cognition.py) 的单轮 P95=3，不能代表分句依据校验的调用量。规划者汇总的尚未合并的 [#61](https://github.com/AirObject/iris-core/pull/61) 在公开 persona_v1／v2 上分别为 133 次／23 个生成候选、265 次／30 个生成候选，平均约 5.8／8.8 次（含检查与复查）；原上限 6 会使该实现合并后的 persona 再次因预算被跳过，故预留到 30。 + +人物摘要默认 40 保持不变：依据 [EF-02](evals/person_summary_eval/EF-02-20261013-report.md) 两次公开集完整运行的 16、16 次及 [PS1](evals/person_summary_eval/PS1-report.md) 已有十人定期步骤测试的 20 次，线性 P95=19.6，两倍向上取整为 40。上述数值是每轮调用上限，不代表常规用量；HTTP 重试和 JSON 修正均计入所属步骤。公开小样本不能视为上线流量的 P95。管理员可按负载调整；0 停止对应步骤的模型调用，其他步骤及目标复核照常。 每轮规划仍至多扫描 128 个新种子。待处理 pair（`pending / classified / decided`）超过内部上限 2000 时，按既有重要度降序、变化时间降序、ID 升序保留前 2000 项，其余记为终态 `backlog_trimmed`，不生成 `consolidation_receipts`，不改变记忆。相同指纹不反复入队；任一侧记忆以后发生语义变化、扫描指纹改变时可重新规划。依赖任务不参与修剪。维护报告 `items` 中同名 outcome／reason 的 `details.count` 记录该轮修剪数量，`details.limit` 为 2000。 @@ -750,7 +754,7 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 - **代码累计证据，决定是否展示**：依据作为 sources 累积，独立性按确定性规则计算（转引、复述、接龙、刷屏、同一件事的跨日后续都不算新讨论）。达到门槛才送检查：话题至少两次独立讨论且跨两个本地日期；说法至少两次独立使用（保守地合并自然变体）；群规需要一次面向全群的明确公告（个人请求、一次性安排、外部规定都不算）。代码只硬拦能确定判断的情形：具体实体（专名、作品名、地点、型号、数字、日期）必须出现在依据中；明显的个人请求、一次性安排、外部规定不能成为群规；回复或引用链、同一发起人对同一对象的跟进合为同一讨论。其余语义判断交给逐条检查。 - **逐条检查**：候选刚达到门槛、或需要修正时，只带这一条和它自己的依据检查一次（提示词 entry_profile_admit_v3）：核对依据、性别代词、引文隐私、含义依据、局部做法不能写成群体习惯、不写容易过时的描述。检查只能删减或缩短依据，不改写正文；失败只影响这一条。 - **稳定性靠反复出现**：候选 importance 30、准入后提为 50（仍走普通衰减）；话题和说法准入时，保留强度补足到 importance 50 的初始值并重置衰减（#117；更高的保留强度不降低，同一已准入正文的复查和重放不重复补足）；不再被确认的主题按生命周期淡出，被确认会增加保留强度。同一讨论只加一次。 -- **预算与恢复**：观察和检查各自 180 秒（含 HTTP 重试和一次 JSON 修正），在事务外调用；观察、检查分别持久保存，重启可续跑，不重复加分。使用独立的 `entry_profile_max_calls`（默认 18，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。 +- **预算与恢复**：观察和检查各自 180 秒（含 HTTP 重试和一次 JSON 修正），在事务外调用；观察、检查分别持久保存，重启可续跑，不重复加分。使用独立的 `entry_profile_max_calls`（默认 60，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。 - **宿主**：`prepare.entry_profile` 的 `{text, version, generated_at, possibly_stale}` 不变，兼容新增 `memory_ids`（与 text 中展示的条目一一对应）。version 是每个入口的单调整数，只在可展示的画像集合变化时加一，读取不改变它。依据失效的条目直接不返回。 - **管理**：入口画像面板按条显示画像记忆（子类、正文、依据原话、作者、准入状态、独立天数／次数、保留强度、最近确认时间、修订历史）和宿主投影预览。逐条编辑、删除、恢复、人工新建走通用记忆接口（修订号校验）。旧的整版编辑、回滚、确认、拒绝接口返回 410，旧版本保留为只读历史。全自动，没有“待确认候选”。 - **迁移**（032、033）：管理员手写的群规逐条转为人工创建的画像记忆（无法分类的显示为“管理员说明”）；旧的模型条目不直接导入，从原消息按新方式重建。 @@ -771,7 +775,7 @@ persona 在记忆整理后调用既有到期判断：自上一版至少五条自 `entry_profile.py` 确定性选择本入口最近 14 天的实际发言（message／self_output，不含行动结果、事件、未来消息），按角色时区分天，各天平分 6000 token 预算；每一天依次取最早、最晚和区间中点，摘录最多约 300 token，连同元数据计入预算。有效、在本入口可见且有本入口直接来源的记忆按重要度降序、ID 升序选取，合计 2000 token；每条最多附最早一条及最近三条本入口原话。材料使用 S／M 编号，入库换为消息 ID 与指纹，或记忆 ID、修订号、来源指纹。 -梦境整理的入口画像步骤使用独立的 `entry_profile_max_calls`（默认 18，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。首版要求窗口内至少 50 条消息；更新要求自上一版以来至少 50 条新消息，或自生成时间经过至少 168 小时且有新消息。依据被清理、修改、删除、遗忘或可见范围变化时只读计算 `possibly_stale` 和 `stale_basis`，下一次整理重新生成；全部生成路径每入口每个角色本地日期最多一次。手动重新生成可跳过更新数量／七天条件，首版门槛、每日上限、开关和用途暂停仍有效。停用不删除历史。 +梦境整理的入口画像步骤使用独立的 `entry_profile_max_calls`（默认 60，含 HTTP 重试和 JSON 修正），与其他步骤共用每日 token 限额。首版要求窗口内至少 50 条消息；更新要求自上一版以来至少 50 条新消息，或自生成时间经过至少 168 小时且有新消息。依据被清理、修改、删除、遗忘或可见范围变化时只读计算 `possibly_stale` 和 `stale_basis`，下一次整理重新生成；全部生成路径每入口每个角色本地日期最多一次。手动重新生成可跳过更新数量／七天条件，首版门槛、每日上限、开关和用途暂停仍有效。停用不删除历史。 生成和逐条检查各共享 120 秒总预算（含调用内重试），在数据库事务外执行，沿用原用途和整理调用上限。代码将空白连续段规整为一个空格，除此之外不改大小写、标点或字词；引语必须是材料中本入口消息正文或其 `quote.content` 的连续子串，也可来自入选记忆附带的本入口原始消息。材料摘录不包含人工添加的截断提示。只引用记忆编号而无消息原话不能通过。明确的账号号段、账号标签、邮箱、平台提及语法以及已知不透明账号 ID 作确定性过滤，其余禁止内容由检查调用判断。 diff --git a/src/iris/consolidation.py b/src/iris/consolidation.py index 07271fb1..03e212f8 100644 --- a/src/iris/consolidation.py +++ b/src/iris/consolidation.py @@ -21,7 +21,7 @@ METHODS = {'balanced': 0.40, 'strict': 0.65, 'broad': 0.20} RESOLUTIONS = ('report_only_v1',) DEFAULTS = {'enabled': True, 'max_calls': 300, - 'entry_profile_max_calls': 18, 'person_summary_max_calls': 40, 'persona_max_calls': 6, + 'entry_profile_max_calls': 60, 'person_summary_max_calls': 40, 'persona_max_calls': 30, 'method': 'broad', 'resolution': 'report_only_v1', 'merge_enabled': True, 'conflict_enabled': True, 'dependency_enabled': True, 'persona_enabled': True, 'goal_review_enabled': True, 'entry_profile_enabled': True, 'person_summary_enabled': True} diff --git a/src/iris/settings_api.py b/src/iris/settings_api.py index 75e614b5..e92c4ba2 100644 --- a/src/iris/settings_api.py +++ b/src/iris/settings_api.py @@ -172,9 +172,9 @@ class Lifecycle(Input): class ConsolidationSettings(Input): max_calls: int = Field(default=300, ge=0, le=1000, strict=True) - entry_profile_max_calls: int = Field(default=18, ge=0, le=1000, strict=True) + entry_profile_max_calls: int = Field(default=60, ge=0, le=1000, strict=True) person_summary_max_calls: int = Field(default=40, ge=0, le=1000, strict=True) - persona_max_calls: int = Field(default=6, ge=0, le=1000, strict=True) + persona_max_calls: int = Field(default=30, ge=0, le=1000, strict=True) merge_enabled: bool = Field(default=True, strict=True) conflict_enabled: bool = Field(default=True, strict=True) dependency_enabled: bool = Field(default=True, strict=True) diff --git a/tests/test_consolidation_step_budgets.py b/tests/test_consolidation_step_budgets.py index 7e9fe06b..be73bb90 100644 --- a/tests/test_consolidation_step_budgets.py +++ b/tests/test_consolidation_step_budgets.py @@ -129,8 +129,10 @@ def test_old_settings_missing_new_keys_and_patch_roundtrip(store): client.app.state.scheduler.stop() value = client.get('/admin/api/settings').json()['consolidation'] assert value['max_calls'] == 50 - for key in BUDGETS[1:]: - assert value[key] == consolidation.DEFAULTS[key] + expected = {'entry_profile_max_calls': 60, 'person_summary_max_calls': 40, 'persona_max_calls': 30} + for key, limit in expected.items(): + assert value[key] == limit + assert getattr(ConsolidationSettings(), key) == limit response = client.patch('/admin/api/settings/consolidation', json={'persona_max_calls': 9}) assert response.status_code == 200 assert response.json()['consolidation']['persona_max_calls'] == 9