diff --git a/evals/entry_profile_eval/EF-27-report.md b/evals/entry_profile_eval/EF-27-report.md new file mode 100644 index 00000000..8e99d732 --- /dev/null +++ b/evals/entry_profile_eval/EF-27-report.md @@ -0,0 +1,88 @@ +# EF-27：入口画像观察提名并集实验，默认关闭 + +本轮只准备实现和离线／假模型验证，真实模型调用0，未测试发现覆盖的实际增益，也未做质量门槛判定。分支 `exp/entry-profile-union-nomination`，基于最新 origin/main `9ae5d0e9836b77b3d4330baacaa4264d1c7bfbea`；没有合入已关闭的 #116／#126／#127。 + +## 运行设置与提名并集 + +`runtime_settings.entry_profile.union_nomination` 是严格布尔值，默认 `false`。已有安装缺字段自动按 false；不用迁移。默认关闭的是第二次观察实验，原有自动画像生成保持原状。实验开关由既有 Store 运行设置持久保存，运行器以 `--union-nomination` 为新数据库打开；本轮不新增管理接口、页面或管理员确认环节。 + +开启时每个非空冻结观察块执行两次相互独立的观察：提示词、消息、已有主题目录和 as_of 完全相同；第二次请求不携带第一次的回答，也不先写入第一次的候选再更新目录。每次观察独立获得原有180秒截止时间，本次调用内重试和一次JSON修正共用这一个截止时间。观察材料、修正材料和输出的上限保持原值；每次仍最多8个观察，两次合并最多16项。 + +原始回答分别保存在 `outputs_json.observe_nomination_1`／`observe_nomination_2`（修正与尝试后缀沿用原规则），`calls_json` 增加 `nomination_round`，两次均走原 Gateway，model_calls 和步骤限额按实际调用计算。关闭时仍用原来的 observe 输出键、一次调用和原有请求材料。 + +第一份有效结构持久保存在已有 work.parsed_json 的内部缓存中；预算中断／重启恢复后不重复请求已完成的一份。两份齐全才合并写回，单份结果不单独产生画像。来源、开关或入口设置变化时拒绝旧结果；缓存绑定原材料指纹，清除仍通过既有工作索引取消。所有新增模型调用在数据库写事务之外；原目标修订号、来源、范围和人工保护校验继续生效。 + +## 去重与冲突 + +- 同类别、同正文／说法／释义、同操作与目标的条目合并，字段顺序不影响判断。来源按 `(message_id, field, excerpt)` 去重；释义来源只在同一说法与释义的合并项内取并集。两个不同片段不拼接出新引文。 +- 同一确认目标的来源合并；同一原话被两次提名仍是同一来源，独立性继续由原代码计数。未攒够门槛的内容只可成为原有低强度候选，不进入宿主投影；重复提名本身不能通过准入。 +- 同一目标有撤下与确认／保留旧正文的提名时，忽略冲突修改并保留旧正文。不同的修订提案、撤下与修订冲突也不任意选一份：丢弃冲突修改，记录 `nomination_union.conflicts`。目录中相同旧正文有多个目标时都按保留处理。 +- 单独提出且无冲突的修改／撤下仍交原流程检查；两次一致的修改只合并来源。不会凭并集直接修改、撤下或准入。 + +只做字面与结构去重，不用模型推断主题等价，不加话题词表。不同措辞的候选仍可能并存并增加待检查成本。有效结构合并后若来源绑定或原确定性规则失败，仍按原规则删除；没有新增放宽路径。一次观察完全失败时不拿另一份替代完整并集，工作按原失败／延后规则留下诊断。 + +原准入检查、证据选择与分页、独立性计数、生命周期、memory_ops、调度和模型预算配置未改；未移植 EF-26 的词法修正。提示词和默认版本保持 observe_v3／admit_v3，14份入口提示词逐字未变。scoring_v3 与 validate_vote、_combine_votes、score_draft、_incremental_stages、_judge_memory、_judge_source 的 AST 不变。 + +## 运行器命令 + +以下是准备好的用法，**本轮未执行真实模型命令**；新公开dev集由规划者手写冻结后再按通知运行。以后运行真实命令时，使用既定 `IRIS_TEST_MODELS=/Users/cassia/Local/Code/iris-core/test-models.toml` 环境变量，不复制配置。两组使用同一代码、语料、模型与判分流程，在不同目录完整重复N次，不按案例挑选运行。 + +```bash +uv run python -m evals.entry_profile_eval.memory_runner repeat --corpus <冻结的公开dev.json> --runs 3 --out <仓库外材料>/single +uv run python -m evals.entry_profile_eval.memory_runner repeat --corpus <冻结的公开dev.json> --runs 3 --union-nomination --out <仓库外材料>/union +``` + +每组目录含 series.json，及 run-0001、run-0002、run-0003，各自拥有独立 SQLite 数据库、原始记录和 judging-materials。旧输出目录不能覆盖；一次传输无效即保留材料并停止余下重复，输入在任一次前后发生变化也停止。原 run_sqlite 作废判定与 gate 标志不改，单次 `run` 同样支持 `--union-nomination`。 + +每一次先按原 score 命令完成独立外部双判，示例只列第1次；第2、3次须分别判分，不能直接把覆盖分当内置模型评分： + +```bash +uv run python -m evals.entry_profile_eval.memory_runner score --materials <材料>/single/run-0001/judging-materials --judgments <该次第一轮> --judgments <该次第二轮> --judge-model 'gpt-6.1-sol max' --out <材料>/single/run-0001/score.json +``` + +原计分器完成逐条验证与从严合并后，离线汇总所有N次，不能省略失败或低分运行： + +```bash +uv run python -m evals.entry_profile_eval.memory_runner summarize --series <材料>/single --score <材料>/single/run-0001/score.json --score <材料>/single/run-0002/score.json --score <材料>/single/run-0003/score.json --out <材料>/single-summary.json +uv run python -m evals.entry_profile_eval.memory_runner summarize --series <材料>/union --score <材料>/union/run-0001/score.json --score <材料>/union/run-0002/score.json --score <材料>/union/run-0003/score.json --out <材料>/union-summary.json +uv run python -m evals.entry_profile_eval.memory_runner compare --left <材料>/single-summary.json --right <材料>/union-summary.json --out <材料>/paired.json +``` + +真实请求的调用量与用量优先取provider记录；假模型没有provider记录时以去重的工作尝试作诊断，并标注 `provider_recorded=false`，不将该回退值声称为真实HTTP调用数。 + +summarize 保留每次结果、均值、逐案例覆盖数量分布与范围。按真实保留票计算有依据、不同违规模型／人工记忆、违规标签发生次数、投影违规及冻结 forbidden 命中，明确区分记忆数量和重复投影计数；不重新判分或更改原双判合并。 + +compare 按运行序号＋案例ID配对，输出逐对覆盖差、逐案例差值分布、两组违规／禁止内容总计、调用量、请求耗时与墙钟差。两组必须拥有相同的语料、源码、评分、回放、模型元数据、运行次数和案例标签；输入与汇总均有指纹核对。缺少 score、材料／案例绑定不匹配、无效或不完整系列会拒绝汇总，不静默丢弃运行。汇总／配对明确 descriptive_only=true、gate_eligible=false，不通过均值自动宣称门槛；各次原判分和 gate 标志单列保留。summarize／compare 不加载模型配置、不调用任何模型。 + +## 测试：先红后绿 + +新增31项手写确定性／假模型测试，位于 test_entry_profile_union_nomination.py、test_entry_profile_repeat_eval.py。初始在未实现的main基础上24失败／1通过；随后两处冲突键序／重复目录边界先红2失败，最后一次输入变更先红1失败，修复后新测试31通过。与原入口机制回归合计85通过后，另补多案例按ID配对测试并通过。所有假票只验证汇总格式和算术，不是质量判分。 + +覆盖：默认值与布尔校验；关闭保持单请求和原输出形状；两份材料相同且不相互泄漏；重复提名不凭空准入；第二份可提出新候选；来源与释义去重；撤下／确认／修订冲突；已准入正文保留；同一确认只确认一次；每份独立180秒与本份修正共截止；第二份超时不应用第一份；预算断点恢复；调用中切换设置拒绝写回。运行器覆盖独立库／目录、N校验、拒绝覆盖旧目录、实际调用计数、无效传输停止、运行中输入变化、各次／均值／案例分布、按序号和ID配对、违规与禁止分开累计、缺分／异指纹拒绝、离线命令不读模型配置。 + +main 在首轮本地验证期间合入 persona #61，执行者用 `git merge origin/main` 快进到9ae5d0e；两份目标产品文件在这次 main 更新中未变，本轮四份改动逐字恢复。先前 cc60a0b 上的3.13全量3785通过、未完成的3.12及旧对照全部保存在 validation-before-main-sync，不作为最终验证。最终以独立只读 baseline worktree 固定9ae5d0e，防止并行会话推进 main 导致对照漂移。 + +最终按顺序执行（期间没有真实评测或虚拟环境重建）： + +| 验证 | 结果 | 耗时 | +| --- | --- | --- | +| `uv run pytest`，Python 3.13.13 | 5438通过，8个warning | 1396.36秒 | +| `uv run --locked --isolated --python 3.12 pytest`，Python 3.12.12 | 5438通过，8个warning | 1334.71秒 | +| `uv build` | wheel／sdist构建通过 | 2.456秒 | + +构建后核对wheel和sdist中的产品文件及14份入口提示词，与工作区逐字一致，包中无密钥文件。全量测试与对照前后218份源码、提示词、语料及测试文件指纹不变；测试后已清理本worktree的 `.pytest-tmp`，`.venv`仍为3.13.13。 + +## 关闭开关的请求对照 + +对全部公开 entry_profile_v1 六例，以同一手写假模型经过原 SQLite 每日＋尾块回放，比较 main、实验分支缺省关闭、显式关闭三组完整系统提示词、用户JSON字符串、顺序、来源字段与请求参数。每组37次请求，字节完全一致，未做任何归一化。SHA-256均为 `dc5dd8dd54dbb3e5a0fac65c33b3ab4f9943a7c86f86bc72d57ee4aa02b7d1c8`。假模型只证明请求路径一致，不代表原规格覆盖或质量。 + +## 学习与召回对照 + +基线均为固定的 main `9ae5d0e`，候选为本轮源码，全部使用确定性假模型，不调用真实端点。 + +- 学习：按 `evals/compare_learning_requests.py` 的同一比较逻辑、仓库外适配脚本纳入全部learning_v1—v5，116段、272个批次、1925条记忆；请求差异0。两侧请求SHA-256同为 `147c831b46c19fa6f83c4055ff8a61c5f4838165453976a73687c5f2f485c08c`。只固定生成的时钟和批次追踪身份，不归一化请求正文、顺序或空白。 +- 召回：七份公开语料，243条固定记忆、228次查询，五种模式共1140份返回差异0；214次召回判断输入也完全一致。只排除 `recall_id` 和耗时等运行字段，其余返回字段均比较。 + +上述结果证明旁路与关闭开关的兼容性，不是画像质量或发现能力的评测。 + +本轮真实模型调用0，无真实评测、无双判质量结果、无质量作废运行；未找寻或读取隐藏集，未修改冻结语料／评分或其他会话占用文件。完整材料和日志只存 `/Users/cassia/Local/Code/iris-eval-artifacts/EF-27-20261020/`。仓库仅提交实现、测试和本汇总；不合并PR。 diff --git a/evals/entry_profile_eval/memory_runner.py b/evals/entry_profile_eval/memory_runner.py index 3059c4f2..75030a62 100644 --- a/evals/entry_profile_eval/memory_runner.py +++ b/evals/entry_profile_eval/memory_runner.py @@ -548,7 +548,7 @@ def _replay_arrivals(store, case, messages, remaining_memories, ids): del remaining_memories[index] -def run_sqlite(corpus, out, *, gateway_factory=None, configs=None): +def run_sqlite(corpus, out, *, gateway_factory=None, configs=None, union_nomination=False): """Pre-registered daily_plus_final_flush, through production SQLite admission.""" from datetime import timedelta from zoneinfo import ZoneInfo @@ -561,6 +561,7 @@ def run_sqlite(corpus, out, *, gateway_factory=None, configs=None): from .runner import _load_case, _model_metadata, _redacted if (gateway_factory is None)==(configs is None): raise ValueError('provide exactly one real configuration or fake gateway factory') + if type(union_nomination) is not bool: raise ValueError('union_nomination must be boolean') root=Path(out);root.mkdir(parents=True,exist_ok=False) materials=root/'judging-materials';cases=load_corpus(corpus) entries=[];traces=[];invalid=None @@ -582,6 +583,10 @@ def _record(self,*args,**kwargs): _load_case(store,{**case,'messages':[],'memories':[],'initial_profile':''}) store.set_setting('consolidation',{'enabled':False}) store.set_setting('scheduled_backup',{'enabled':False}) + if union_nomination: + from iris.profile_common import setting + with store.read() as conn: runtime=setting(conn,'entry_profile',{}) + store.set_setting('entry_profile',{**runtime,'union_nomination':True}) clock=lambda:current if case.get('initial_profile'): with patch('iris.entry_profile.utc_now',clock): @@ -657,6 +662,7 @@ def _record(self,*args,**kwargs): corpus_sha256=hashlib.sha256(Path(corpus).read_bytes()).hexdigest(),source_sha256=_source_hash(), scoring_sha256=hashlib.sha256(scoring.encode()).hexdigest(),prompt_versions=list(hashes),prompt_sha256=hashes, models=_model_metadata(configs or {})) + if union_nomination: metadata['entry_profile_union_nomination']=True _write(root/'run.json',_redacted(dict(metadata=metadata,cases=traces),configs or {})) manifest=dict(metadata=metadata,cases=entries,run_sha256=hashlib.sha256((root/'run.json').read_bytes()).hexdigest()) manifest['materials_sha256']=evidence_hash(manifest) @@ -665,25 +671,223 @@ def _record(self,*args,**kwargs): return manifest +def _sealed(value, field): + return {**value,field:evidence_hash(value)} + + +def _check_seal(value, field): + if value.get(field)!=evidence_hash({k:v for k,v in value.items() if k!=field}): + raise ValueError('summary/series fingerprint mismatch') + + +def _repeat_usage(root): + run=_read(Path(root)/'run.json') + calls=[call for case in run['cases'] for call in case.get('model_calls',[])] + recorded=bool(calls) + if not calls: + # Fake gateways do not write provider call rows. Durable work traces + # retain attempts without counting the same work again at each checkpoint. + for case in run['cases']: + unique={w['id']:w for w in case.get('work',[])} + calls.extend(call for work in unique.values() for call in json.loads(work['calls_json'])) + return dict(calls=len(calls),provider_recorded=recorded, + observe_calls=sum('observe' in c['purpose'] and 'repair' not in c['purpose'] for c in calls), + admit_calls=sum('admit' in c['purpose'] and 'repair' not in c['purpose'] for c in calls), + repair_calls=sum('repair' in c['purpose'] for c in calls), + request_seconds=sum(c.get('duration_ms') or 0 for c in calls)/1000, + prompt_tokens=sum(c.get('prompt_tokens') or c.get('usage',{}).get('prompt_tokens') or 0 for c in calls), + completion_tokens=sum(c.get('completion_tokens') or c.get('usage',{}).get('completion_tokens') or 0 for c in calls)) + + +def run_repeated(corpus, out, *, runs, union_nomination=False, gateway_factory=None, configs=None): + """Independent complete replays, no automatic retry or selection of cases.""" + import time + from .runner import _source_hash, _model_metadata + if type(runs) is not int or runs<1: raise ValueError('runs must be a positive integer') + if type(union_nomination) is not bool: raise ValueError('union_nomination must be boolean') + if (gateway_factory is None)==(configs is None): raise ValueError('provide exactly one gateway/configuration') + corpus=Path(corpus);root=Path(out);root.mkdir(parents=True,exist_ok=False) + metadata=dict(corpus_sha256=hashlib.sha256(corpus.read_bytes()).hexdigest(),source_sha256=_source_hash(), + scoring_sha256=hashlib.sha256((HERE/'scoring_v3.md').read_bytes()).hexdigest(), + replay_mode='daily_plus_final_flush',models=_model_metadata(configs or {})) + result=dict(format_version=1,evaluation='entry_profile_repeated',metadata=metadata, + requested_runs=runs,settings=dict(union_nomination=union_nomination),runs=[],invalidated=None,gate_eligible=False) + def save(): _write(root/'series.json',_sealed(result,'series_sha256')) + save() + for index in range(1,runs+1): + if (hashlib.sha256(corpus.read_bytes()).hexdigest()!=metadata['corpus_sha256'] or _source_hash()!=metadata['source_sha256']): + result['invalidated']=dict(run_index=index,reason='input_changed');save();break + directory=f'run-{index:04d}';started=time.monotonic() + try: + manifest=run_sqlite(corpus,root/directory,gateway_factory=gateway_factory,configs=configs,union_nomination=union_nomination) + except Exception as error: + result['invalidated']=dict(run_index=index,reason='runner_error',error_type=type(error).__name__);save();raise + result['runs'].append(dict(index=index,directory=directory,materials_sha256=manifest['materials_sha256'], + gate_eligible=manifest['metadata']['gate_eligible'],invalidated=manifest['metadata']['invalidated'], + usage=_repeat_usage(root/directory),wall_seconds=time.monotonic()-started)) + if manifest['metadata']['invalidated']: + result['invalidated']=dict(run_index=index,**manifest['metadata']['invalidated']);save();break + if (hashlib.sha256(corpus.read_bytes()).hexdigest()!=metadata['corpus_sha256'] + or _source_hash()!=metadata['source_sha256'] + or any(manifest['metadata'][k]!=metadata[k] for k in ('corpus_sha256','source_sha256','scoring_sha256','replay_mode','models'))): + result['invalidated']=dict(run_index=index,reason='input_changed');save();break + result['gate_eligible']=len(result['runs'])==runs and all(r['gate_eligible'] for r in result['runs']) + save() + return _sealed(result,'series_sha256') + + +def _repeat_inputs(root, record): + materials=root/record['directory']/'judging-materials';manifest=_read(materials/'manifest.json') + if (manifest.get('materials_sha256')!=record['materials_sha256'] + or manifest['materials_sha256']!=evidence_hash({k:v for k,v in manifest.items() if k!='materials_sha256'}) + or hashlib.sha256((root/record['directory']/'run.json').read_bytes()).hexdigest()!=manifest['run_sha256'] + or hashlib.sha256((materials/'scoring.md').read_bytes()).hexdigest()!=manifest['metadata']['scoring_sha256']): + raise ValueError('run material fingerprint mismatch') + values=[] + for index,case in enumerate(manifest['cases'],1): + if case['file']!=f'cases/{index:04d}.json': raise ValueError('invalid case filename') + path=materials/case['file'] + if not path.resolve().is_relative_to(materials.resolve()): raise ValueError('material outside directory') + doc=_read(path) + if evidence_hash(doc)!=case['sha256'] or doc['case_id']!=case['case_id']: raise ValueError('case fingerprint mismatch') + values.append((case['case_id'],doc['input'])) + if len({key for key,_ in values})!=len(values): raise ValueError('duplicate case ID') + return manifest,values + + +def _case_statistics(case_id, payload, vote): + rows=list(zip(payload['memories'],vote['memory_results'],strict=True)) + model=[v for m,v in rows if m['author']!='admin'] + return dict(case_id=case_id,label_sha256=evidence_hash(dict(must_cover=payload['must_cover'],forbidden=payload['forbidden'])), + coverage=sum(vote['must_cover']),coverage_total=len(vote['must_cover']), + model_items=len(model),grounded_items=sum(v['supported'] for v in model), + violation_memory_items=sum(bool(v['violations']) for _,v in rows), + model_violation_memory_items=sum(bool(v['violations']) for v in model), + violation_occurrences=sum(len(v['violations']) for _,v in rows)+len(vote['projection_violations']), + projection_violations=len(vote['projection_violations']),forbidden_occurrences=sum(vote['forbidden']), + stage_coverage={key:sum(vote['stage_coverage'][key]) for key in STAGES}) + + +_REPEAT_TOTALS=('coverage','coverage_total','model_items','grounded_items','violation_memory_items', + 'model_violation_memory_items','violation_occurrences','projection_violations','forbidden_occurrences') + + +def summarize_repeated(series, scores): + """Describe existing independent scores; never grade or change vote merging.""" + from statistics import mean + root=Path(series);description=_read(root/'series.json');_check_seal(description,'series_sha256') + if (description['invalidated'] or len(description['runs'])!=description['requested_runs'] + or len(scores)!=description['requested_runs']): + raise ValueError('incomplete/invalid series or missing scores; no runs may be omitted') + runs=[];case_order=None;labels={} + for index,(record,score_path) in enumerate(zip(description['runs'],scores,strict=True),1): + if record['index']!=index or record['directory']!=f'run-{index:04d}' or record['invalidated']: + raise ValueError('invalid repeated run order') + manifest,inputs=_repeat_inputs(root,record);score=_read(Path(score_path)) + if (score['materials_sha256']!=manifest['materials_sha256'] or score['scoring_status']!=SCORING_STATUS + or score['judge_runs'] not in (1,2) or len(score['judgments'])!=len(inputs)): + raise ValueError('score fingerprint/format mismatch') + if any(manifest['metadata'][k]!=description['metadata'][k] for k in ('corpus_sha256','source_sha256','scoring_sha256','replay_mode','models')): + raise ValueError('repeated inputs changed') + cases=[_case_statistics(key,payload,validate_vote(vote,payload)) + for (key,payload),vote in zip(inputs,score['judgments'],strict=True)] + order=[c['case_id'] for c in cases] + if case_order is not None and case_order!=order: raise ValueError('case pairing mismatch') + case_order=order + for case in cases: + if case['case_id'] in labels and labels[case['case_id']]!=case['label_sha256']: raise ValueError('case labels changed') + labels[case['case_id']]=case['label_sha256'] + totals={key:sum(c[key] for c in cases) for key in _REPEAT_TOTALS} + if (totals['coverage']!=score['metrics']['must_cover_covered'] or totals['coverage_total']!=score['metrics']['must_cover_total'] + or totals['model_items']!=score['metrics']['model_items'] or totals['grounded_items']!=score['metrics']['grounded_items'] + or totals['forbidden_occurrences']!=score['metrics']['forbidden_occurrences']): + raise ValueError('score metrics disagree with votes') + totals['grounded_ratio']=totals['grounded_items']/totals['model_items'] if totals['model_items'] else None + runs.append(dict(index=index,**totals,usage=_repeat_usage(root/record['directory']), + wall_seconds=record['wall_seconds'],judge_model=score['judge_model'],judge_runs=score['judge_runs'], + gate_eligible=score['gate_eligible'],cases=cases)) + means={key:mean(r[key] for r in runs) for key in _REPEAT_TOTALS} + means.update(calls=mean(r['usage']['calls'] for r in runs),request_seconds=mean(r['usage']['request_seconds'] for r in runs), + wall_seconds=mean(r['wall_seconds'] for r in runs), + grounded_ratio=mean(r['grounded_ratio'] for r in runs) if all(r['grounded_ratio'] is not None for r in runs) else None) + per_case=[] + for offset,key in enumerate(case_order or []): + values=[r['cases'][offset]['coverage'] for r in runs] + per_case.append(dict(case_id=key,label_sha256=labels[key],coverage_counts=values,coverage_mean=mean(values), + coverage_min=min(values),coverage_max=max(values),coverage_total=runs[0]['cases'][offset]['coverage_total'])) + return _sealed(dict(format_version=1,evaluation='entry_profile_repeated_summary',metadata=description['metadata'], + settings=description['settings'],runs=runs,means=means,per_case=per_case,descriptive_only=True,gate_eligible=False),'summary_sha256') + + +def compare_repeated(left, right): + """Paired by run ordinal and stable case ID, with no selection of good runs.""" + from statistics import mean + for value in (left,right): _check_seal(value,'summary_sha256') + for key in ('corpus_sha256','scoring_sha256','source_sha256','replay_mode','models'): + if left['metadata'][key]!=right['metadata'][key]: raise ValueError('paired '+key+' mismatch') + if not left['runs'] or len(left['runs'])!=len(right['runs']): raise ValueError('paired run count mismatch') + pairs=[];run_pairs=[] + for a,b in zip(left['runs'],right['runs'],strict=True): + if a['index']!=b['index']: raise ValueError('paired run index mismatch') + ac={c['case_id']:c for c in a['cases']};bc={c['case_id']:c for c in b['cases']} + if ac.keys()!=bc.keys(): raise ValueError('paired case mismatch') + for key,x in ac.items(): + y=bc[key] + if x['label_sha256']!=y['label_sha256']: raise ValueError('paired labels mismatch') + pairs.append(dict(index=a['index'],case_id=key,left_coverage=x['coverage'],right_coverage=y['coverage'], + coverage_delta=y['coverage']-x['coverage'], + left_violation_memory_items=x['violation_memory_items'],right_violation_memory_items=y['violation_memory_items'], + left_forbidden_occurrences=x['forbidden_occurrences'],right_forbidden_occurrences=y['forbidden_occurrences'])) + run_pairs.append(dict(index=a['index'],coverage_delta=b['coverage']-a['coverage'], + calls_delta=b['usage']['calls']-a['usage']['calls'],request_seconds_delta=b['usage']['request_seconds']-a['usage']['request_seconds'], + wall_seconds_delta=b['wall_seconds']-a['wall_seconds'])) + per_case=[] + for key in dict.fromkeys(p['case_id'] for p in pairs): + values=[p['coverage_delta'] for p in pairs if p['case_id']==key] + per_case.append(dict(case_id=key,coverage_deltas=values,coverage_delta_mean=mean(values))) + totals={name:{**{key:sum(r[key] for r in value['runs']) for key in _REPEAT_TOTALS}, + 'calls':sum(r['usage']['calls'] for r in value['runs']), + 'request_seconds':sum(r['usage']['request_seconds'] for r in value['runs']), + 'wall_seconds':sum(r['wall_seconds'] for r in value['runs'])} for name,value in (('left',left),('right',right))} + return dict(format_version=1,evaluation='entry_profile_paired_comparison',descriptive_only=True,gate_eligible=False, + left_summary_sha256=left['summary_sha256'],right_summary_sha256=right['summary_sha256'], + left_settings=left['settings'],right_settings=right['settings'],pairs=pairs,run_pairs=run_pairs,per_case=per_case,totals=totals, + means={key:mean(r[key] for r in run_pairs) for key in ('coverage_delta','calls_delta','request_seconds_delta','wall_seconds_delta')}) + + def main(): import argparse - parser=argparse.ArgumentParser(description='Entry-profile memory execution / offline scoring') + parser=argparse.ArgumentParser(description='Entry-profile memory execution / offline scoring / repeated paired comparison') sub=parser.add_subparsers(dest='command',required=True) run=sub.add_parser('run');run.add_argument('--corpus',required=True);run.add_argument('--out',required=True) + run.add_argument('--union-nomination',action='store_true') + repeat=sub.add_parser('repeat');repeat.add_argument('--corpus',required=True);repeat.add_argument('--out',required=True) + repeat.add_argument('--runs',type=int,required=True);repeat.add_argument('--union-nomination',action='store_true') score=sub.add_parser('score');score.add_argument('--materials',required=True) score.add_argument('--judgments',action='append',required=True);score.add_argument('--judge-model',required=True);score.add_argument('--out',required=True) + summary=sub.add_parser('summarize');summary.add_argument('--series',required=True);summary.add_argument('--score',action='append',required=True);summary.add_argument('--out',required=True) + compare=sub.add_parser('compare');compare.add_argument('--left',required=True);compare.add_argument('--right',required=True);compare.add_argument('--out',required=True) args=parser.parse_args() - if args.command=='run': + if args.command in ('run','repeat'): from iris.models import load_test_models - manifest=run_sqlite(args.corpus,args.out,configs=load_test_models()) - print(json.dumps(dict(cases=len(manifest['cases']),gate_eligible=manifest['metadata']['gate_eligible'], - invalidated=manifest['metadata']['invalidated']),ensure_ascii=False)) - if manifest['metadata']['invalidated']: - raise SystemExit(2) - else: + if args.command=='repeat': + result=run_repeated(args.corpus,args.out,runs=args.runs,union_nomination=args.union_nomination,configs=load_test_models()) + print(json.dumps(dict(runs=len(result['runs']),invalidated=result['invalidated']),ensure_ascii=False)) + if result['invalidated']: raise SystemExit(2) + else: + manifest=run_sqlite(args.corpus,args.out,configs=load_test_models(),union_nomination=args.union_nomination) + print(json.dumps(dict(cases=len(manifest['cases']),gate_eligible=manifest['metadata']['gate_eligible'], + invalidated=manifest['metadata']['invalidated']),ensure_ascii=False)) + if manifest['metadata']['invalidated']: raise SystemExit(2) + elif args.command=='score': result=score_draft(args.materials,args.judgments,judge_model=args.judge_model) - _write(Path(args.out),result) - print(json.dumps(result['metrics'],ensure_ascii=False)) + _write(Path(args.out),result);print(json.dumps(result['metrics'],ensure_ascii=False)) + elif args.command=='summarize': + result=summarize_repeated(args.series,args.score);_write(Path(args.out),result) + print(json.dumps(result['means'],ensure_ascii=False)) + else: + result=compare_repeated(_read(Path(args.left)),_read(Path(args.right)));_write(Path(args.out),result) + print(json.dumps(result['means'],ensure_ascii=False)) if __name__=='__main__': diff --git a/src/iris/entry_profile.py b/src/iris/entry_profile.py index 12bc9c3c..890e411d 100644 --- a/src/iris/entry_profile.py +++ b/src/iris/entry_profile.py @@ -26,7 +26,7 @@ from .queue import estimate_tokens, truncate_material from .profile_common import evidence_hash, setting, whitespace, excerpt, json_object, ChatHealthGate, load_profile_prompts, profile_prompt_metadata -DEFAULTS = dict(enabled=True, publish_mode='check_auto', window_days=14, +DEFAULTS = dict(enabled=True, union_nomination=False, publish_mode='check_auto', window_days=14, message_tokens=6000, memory_tokens=2000, min_messages=50, update_days=7, min_chars=150, max_chars=500, generate_timeout_seconds=120, check_timeout_seconds=120) @@ -82,7 +82,8 @@ def _setting(conn, key, default): def _runtime(conn): config = {**DEFAULTS, **_setting(conn, 'entry_profile', {})} - if type(config['enabled']) is not bool or config['publish_mode'] not in PUBLISH_MODES: + if (type(config['enabled']) is not bool or type(config['union_nomination']) is not bool + or config['publish_mode'] not in PUBLISH_MODES): raise ValueError('invalid entry profile settings') for key in ('window_days','message_tokens','memory_tokens','min_messages','update_days', 'min_chars','max_chars','generate_timeout_seconds','check_timeout_seconds'): @@ -2153,6 +2154,44 @@ def validate_incremental_observations(value, limits=None): return value +def merge_observation_nominations(rounds, profiles): + """Union validated observations, without inferring semantic equivalence. + + Source identity is message/field/excerpt; repeating a nomination is not new + evidence. Conflicting edits leave the old body intact. Admission is untouched. + """ + items=[copy.deepcopy(item) for result in rounds for item in result['observations']] + directory=defaultdict(set) + for p in profiles: directory[(p['category'],whitespace(p['body']))].add(p['memory_id']) + changes=defaultdict(list);keeps=set() + for item in items: + if item['op']=='change': changes[item['memory_id']].append(item) + elif item['op']=='confirm': keeps.add(item['memory_id']) + else: + keeps.update(directory.get((item['category'],whitespace(item['body'])),())) + conflicts=[];blocked=set() + for mid,edits in changes.items(): + intents={dumps({k:x[k] for k in sorted(x) if k not in ('source_refs','definition_refs')}) for x in edits} + if len(intents)>1 or mid in keeps: + blocked.add(mid) + conflicts.append(dict(memory_id=mid,reason='incompatible_changes' if len(intents)>1 else 'keep_existing')) + merged=[];by_intent={} + for item in items: + if item['op']=='change' and item['memory_id'] in blocked: continue + key=dumps({k:item[k] for k in sorted(item) if k not in ('source_refs','definition_refs')}) + if key not in by_intent: + by_intent[key]=dict(item,source_refs=[],**({'definition_refs':[]} if 'definition_refs' in item else {})) + merged.append(by_intent[key]) + target=by_intent[key] + seen={(r['message_id'],r['field'],r['excerpt']) for r in target['source_refs']} + for ref in item['source_refs']: + identity=(ref['message_id'],ref['field'],ref['excerpt']) + if identity not in seen: target['source_refs'].append(ref);seen.add(identity) + if 'definition_refs' in target: + target['definition_refs']=list(dict.fromkeys(target['definition_refs']+item['definition_refs'])) + return dict(observations=merged),dict(rounds=len(rounds),input_items=len(items),merged_items=len(merged),conflicts=conflicts) + + def validate_incremental_admission(value, payload, previous=None): if (not isinstance(value,dict) or set(value)!={'decision','violations','reason','safe_sources'} or value['decision'] not in ('accept','reject','insufficient') @@ -2501,13 +2540,14 @@ def __init__(self, store, gateway, *, clock=utc_now, limits=None, attempt_id=Non self.prompts=incremental_prompts();self.attempt_id=attempt_id;self.source=source self.trace=[] - def _call(self, work, stage, payload): + def _call(self, work, stage, payload, *, nomination_round=None): prompt=self.prompts[stage];messages=[{'role':'system','content':prompt['text']},{'role':'user','content':dumps(payload)}] budget=self.limits.observe_request_tokens if stage=='observe' else self.limits.admit_request_tokens if sum(estimate_tokens(x['content']) for x in messages)>budget: raise ModelError('invalid_output','request material limit',reason='request_budget_exceeded') monotonic=getattr(self.gateway,'monotonic',time.monotonic);deadline=monotonic()+self.limits.timeout_seconds outputs=json.loads(work['outputs_json']);calls=json.loads(work['calls_json']);previous=None + output_key=stage+(f'_nomination_{nomination_round}' if nomination_round is not None else '') for repair in range(2): if self.store._lock._is_owned() and self.store._writer.in_transaction: raise RuntimeError('model call in write transaction') @@ -2518,6 +2558,7 @@ def _call(self, work, stage, payload): purpose='consolidation_entry_profile_'+stage+('_repair' if repair else '') call=dict(purpose=purpose,prompt_version=prompt['name'],prompt_sha256=prompt['sha256'], timeout_seconds=self.limits.timeout_seconds,result='success') + if nomination_round is not None: call['nomination_round']=nomination_round started=monotonic();reply=None;invalid=None;parsed=None try: reply=self.gateway.chat(messages,purpose,max_tokens=16000,_deadline=deadline) @@ -2542,7 +2583,7 @@ def _call(self, work, stage, payload): raise finally: if reply is not None: - outputs[stage+('_repair' if repair else '')+('_attempt_'+str(work['attempts']) if work['attempts']>1 else '')]=reply.content + outputs[output_key+('_repair' if repair else '')+('_attempt_'+str(work['attempts']) if work['attempts']>1 else '')]=reply.content call['finish_reason']=reply.finish_reason calls.append({**call,'duration_ms':round((monotonic()-started)*1000),'usage':reply.usage if reply else {}}) with self.store.write() as conn: @@ -2553,6 +2594,30 @@ def _call(self, work, stage, payload): if sum(estimate_tokens(x['content']) for x in messages)>self.limits.repair_request_tokens: raise ModelError('invalid_output','repair material limit',reason='repair_budget_exceeded') + def _observe_union(self, work, public): + payload=json.loads(work['payload_json']);cached=json.loads(work['parsed_json']) if work['parsed_json'] else {} + if 'observations' in cached: return cached + rounds=cached.get('_union_rounds',[]) + if (not isinstance(rounds,list) or len(rounds)>2 + or rounds and cached.get('_union_payload_sha256')!=_hash(public)): + raise ValueError('nomination_cache_mismatch') + for result in rounds: validate_incremental_observations(result,self.limits) + while len(rounds)<2: + with self.store.read() as conn: + self._assert_snapshot(conn,payload) + live=dict(conn.execute('SELECT * FROM entry_profile_incremental_work WHERE id=?',(work['id'],)).fetchone()) + if live['state']=='cancelled': raise EntryProfileConflict('purged') + # No assistant output from the other nomination is lent to this call. + result=self._call(live,'observe',public,nomination_round=len(rounds)+1) + rounds.append(result) + with self.store.write() as conn: + self._assert_snapshot(conn,payload) + if conn.execute('SELECT state FROM entry_profile_incremental_work WHERE id=?',(work['id'],)).fetchone()[0]=='cancelled': + raise EntryProfileConflict('purged') + _inc_work_persist(conn,work['id'],parsed_json=dumps(dict(_union_rounds=rounds,_union_payload_sha256=_hash(public)))) + merged,diagnostics=merge_observation_nominations(rounds,payload['profiles']) + return {**merged,'_nomination_union':diagnostics} + def _freeze_chunk(self, entry_id): now=_instant(self.clock());stamp=now.isoformat();limit=self.limits with self.store.write() as conn: @@ -2592,6 +2657,7 @@ def _freeze_chunk(self, entry_id): payload=dict(entry_id=entry_id,timezone=zone,as_of=stamp,messages=messages,profiles=directory, _snapshot=dict(settings_revision=_settings(conn,entry_id)['revision'],cursor_before=state['scan_cursor'], cursor_after=positions[-1],positions=positions,omitted_profiles=omitted,skipped_messages=skipped)) + if _runtime(conn)['union_nomination']: payload['_snapshot']['union_nomination']=True key=_hash([entry_id,state['scan_cursor'],positions[-1],messages,zone]) wid=conn.execute('''INSERT OR IGNORE INTO entry_profile_incremental_work(entry_id,work_key,kind,cursor_before,cursor_after, local_day,payload_json,attempt_id,created_at) VALUES(?,?,'observe',?,?,?,?,?,?)''', @@ -2631,6 +2697,7 @@ def piece(size): messages=[fragment],profiles=directory,_snapshot=dict(settings_revision=_settings(conn,entry_id)['revision'], cursor_before=state['scan_cursor'],cursor_after=end,positions=[original['message_id']], omitted_profiles=omitted,fragment_group=group,fragment_index=index,fragment_count=len(fragments))) + if _runtime(conn)['union_nomination']: payload['_snapshot']['union_nomination']=True wid=conn.execute("""INSERT INTO entry_profile_incremental_work(entry_id,work_key,kind,cursor_before,cursor_after, local_day,fragment_group,payload_json,attempt_id,created_at) VALUES(?,?,'observe',?,?,?,?,?,?,?)""", (entry_id,_hash([group,index]),state['scan_cursor'],end,day,group,dumps(payload),self.attempt_id,_instant(self.clock()).isoformat())).lastrowid @@ -2662,13 +2729,16 @@ def _resnapshot(self, entry_id): directory,omitted=_inc_directory(conn,entry_id,messages,self.limits) payload.update(messages=messages,profiles=directory,timezone=zone,as_of=stamp) snap.update(settings_revision=settings['revision'],omitted_profiles=omitted) + snap.pop('union_nomination',None) + if _runtime(conn)['union_nomination']: snap['union_nomination']=True _inc_work_persist(conn,work['id'],payload_json=dumps(payload),parsed_json=None,state='queued',reason='resnapshotted') def _assert_snapshot(self, conn, payload): entry_id=payload['entry_id'];snap=payload['_snapshot'] settings=_settings(conn,entry_id) if (not settings['enabled'] or not _runtime(conn)['enabled'] or settings['revision']!=snap['settings_revision'] - or _setting(conn,'timezone','Asia/Shanghai')!=payload['timezone'] or _entry(conn,entry_id)['kind'] not in ('group','live')): + or _setting(conn,'timezone','Asia/Shanghai')!=payload['timezone'] or _entry(conn,entry_id)['kind'] not in ('group','live') + or bool(snap.get('union_nomination',False))!=_runtime(conn)['union_nomination']): raise EntryProfileConflict('settings_changed') for message in payload.get('messages',[]): live=conn.execute('SELECT * FROM messages WHERE id=?',(message['message_id'],)).fetchone() @@ -2679,6 +2749,7 @@ def _apply_observations(self, work, parsed): from .memory_ops import entry_profile_source, write_entry_profile_memory, _profile_manual payload=json.loads(work['payload_json']);entry_id=work['entry_id'];now=_instant(self.clock()) result=dict(generated=parsed['observations'],deterministic=[],removed=[],applied=[],normalizations=[]) + if '_nomination_union' in parsed: result['nomination_union']=parsed['_nomination_union'] with self.store.write() as conn: self._assert_snapshot(conn,payload) if conn.execute('SELECT state FROM entry_profile_incremental_work WHERE id=?',(work['id'],)).fetchone()[0]=='cancelled': @@ -3047,7 +3118,11 @@ def _run_one(self, work): conn.execute("UPDATE entry_profile_admissions SET state=CASE WHEN state='accepted' THEN state ELSE 'candidate' END,reason='stale_check' WHERE memory_id=?",(work['memory_id'],)) return public=_inc_public_payload(payload) - if work['parsed_json'] is None: + if work['kind']=='observe' and payload['_snapshot'].get('union_nomination') and payload.get('messages'): + parsed=self._observe_union(work,public) + with self.store.write() as conn: + _inc_work_persist(conn,work['id'],parsed_json=dumps(parsed)) + elif work['parsed_json'] is None: parsed={'observations':[]} if work['kind']=='observe' and not payload.get('messages') else self._call(work,work['kind'],public) with self.store.write() as conn: _inc_work_persist(conn,work['id'],parsed_json=dumps(parsed)) diff --git a/tests/test_entry_profile_repeat_eval.py b/tests/test_entry_profile_repeat_eval.py new file mode 100644 index 00000000..53500cf3 --- /dev/null +++ b/tests/test_entry_profile_repeat_eval.py @@ -0,0 +1,196 @@ +"""Repeated replay and paired descriptive statistics, with fake votes only.""" +import copy +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from evals.entry_profile_eval import memory_runner as runner +from iris.db import dumps +from iris.models import ModelError +from test_entry_profile_incremental import Gateway, propose, source +from test_entry_profile_incremental_eval import corpus + + +def factory(case, store): + def observe(payload): + if payload['profiles']: + return {'observations': [dict(op='confirm', memory_id=payload['profiles'][0]['memory_id'], + source_refs=[source(m) for m in payload['messages']])]} + return propose(payload) + return Gateway(store, observe) + + +def scores(series_dir, coverages, violations=None): + series = json.loads((series_dir / 'series.json').read_text()) + paths = [] + for record, coverage in zip(series['runs'], coverages, strict=True): + root = series_dir / record['directory']; materials = root / 'judging-materials' + manifest = json.loads((materials / 'manifest.json').read_text()) + rounds = [] + for name in ('format-a', 'format-b'): + out = root / name; out.mkdir() + (out / 'manifest.json').write_bytes((materials / 'round-template.json').read_bytes()) + for case in manifest['cases']: + value = json.loads((materials / case['file']).read_text())['input'] + bad = bool(violations and violations[record['index'] - 1]) + vote = dict(memory_results=[dict(index=i, memory_id=m['memory_id'], revision=m['revision'], + supported=not bad, violations=['无依据', '个人评价'] if bad else [], reason='假票只验证汇总结构') + for i, m in enumerate(value['memories'], 1)], projection_violations=[], + must_cover=[coverage] * len(value['must_cover']), forbidden=[bad] * len(value['forbidden']), + stage_coverage={stage: [coverage] * len(value['must_cover']) for stage in runner.STAGES}) + (out / case['judgment_file']).write_text(dumps(vote)) + rounds.append(out) + report = runner.score_draft(materials, rounds, judge_model='fake-format-only') + path = root / 'score.json'; path.write_text(dumps(report)); paths.append(path) + return paths + + +def series(tmp_path, name='series', *, runs=2, union=False, corpus_path=None): + path = corpus_path or corpus(tmp_path) + root = tmp_path / name + runner.run_repeated(path, root, runs=runs, union_nomination=union, gateway_factory=factory) + return root + + +def test_repeat_uses_independent_directories_and_databases(tmp_path): + stores = [] + def record(case, store): + stores.append(store.path) + return factory(case, store) + root = tmp_path / 'series' + result = runner.run_repeated(corpus(tmp_path), root, runs=3, gateway_factory=record) + assert result['requested_runs'] == 3 and len(result['runs']) == 3 + assert len(stores) == len(set(map(str, stores))) == 3 + assert [r['index'] for r in result['runs']] == [1, 2, 3] + assert all((root / r['directory'] / 'judging-materials/manifest.json').exists() for r in result['runs']) + with pytest.raises(FileExistsError): + runner.run_repeated(corpus(tmp_path), root, runs=3, gateway_factory=record) + + +@pytest.mark.parametrize('count', [0, -1, True, 1.5]) +def test_invalid_repeat_count_is_rejected_before_creating_output(tmp_path, count): + out = tmp_path / 'out' + with pytest.raises(ValueError): + runner.run_repeated(corpus(tmp_path), out, runs=count, gateway_factory=factory) + assert not out.exists() + + +def test_runner_union_flag_reaches_runtime_and_counts_both_observations(tmp_path): + gateways = [] + def record(case, store): + g = factory(case, store); gateways.append(g); return g + result = runner.run_repeated(corpus(tmp_path), tmp_path / 'series', runs=1, + union_nomination=True, gateway_factory=record) + assert sum('observe' in c['purpose'] for c in gateways[0].calls) == 4 + assert result['settings']['union_nomination'] is True + assert result['runs'][0]['usage']['calls'] == len(gateways[0].calls) == 5 + assert result['gate_eligible'] is False + + +def test_repeat_stops_after_invalid_transport_instead_of_silently_rerunning(tmp_path): + calls = [] + def bad(case, store): + calls.append(case) + return Gateway(store, lambda p: (_ for _ in ()).throw(ModelError('retryable', 'late', reason='timeout'))) + result = runner.run_repeated(corpus(tmp_path), tmp_path / 'series', runs=3, gateway_factory=bad) + assert len(calls) == len(result['runs']) == 1 + assert result['invalidated']['reason'] == 'timeout' + assert not result['gate_eligible'] + + +def test_summary_preserves_each_run_and_per_case_coverage_distribution(tmp_path): + root = series(tmp_path) + summary = runner.summarize_repeated(root, scores(root, [True, False])) + assert [r['coverage'] for r in summary['runs']] == [1, 0] + assert summary['means']['coverage'] == .5 + assert summary['per_case'][0]['coverage_counts'] == [1, 0] + assert summary['per_case'][0]['coverage_mean'] == .5 + assert not summary['gate_eligible'] and summary['descriptive_only'] + + +def test_pairing_is_by_run_index_and_case_and_sums_violations_separately(tmp_path): + path = corpus(tmp_path) + value = json.loads(path.read_text()); value['cases'][0]['forbidden'] = ['禁止个人评价'] + path.write_text(dumps(value)) + left_root = series(tmp_path, 'left', corpus_path=path) + right_root = series(tmp_path, 'right', union=True, corpus_path=path) + left = runner.summarize_repeated(left_root, scores(left_root, [False, True])) + right = runner.summarize_repeated(right_root, scores(right_root, [True, True], [True, False])) + compared = runner.compare_repeated(left, right) + assert [p['coverage_delta'] for p in compared['pairs']] == [1, 0] + assert compared['means']['coverage_delta'] == .5 + assert compared['per_case'][0]['coverage_deltas'] == [1, 0] + assert compared['totals']['right']['violation_memory_items'] == 1 + assert compared['totals']['right']['forbidden_occurrences'] == 1 + assert compared['totals']['right']['calls'] > compared['totals']['left']['calls'] + assert compared['descriptive_only'] and not compared['gate_eligible'] + + +def test_missing_or_foreign_score_is_not_averaged_away(tmp_path): + root = series(tmp_path) + paths = scores(root, [True, False]) + with pytest.raises(ValueError): runner.summarize_repeated(root, paths[:1]) + value = json.loads(paths[1].read_text()); value['materials_sha256'] = '0' * 64 + paths[1].write_text(dumps(value)) + with pytest.raises(ValueError, match='fingerprint'): + runner.summarize_repeated(root, paths) + + +def test_comparison_rejects_different_corpus_or_tampered_summary(tmp_path): + root = series(tmp_path) + summary = runner.summarize_repeated(root, scores(root, [True, False])) + tampered = copy.deepcopy(summary); tampered['runs'][0]['coverage'] = 100 + with pytest.raises(ValueError, match='fingerprint'): runner.compare_repeated(summary, tampered) + other_path = tmp_path / 'other.json' + value = json.loads(corpus(tmp_path).read_text()); value['cases'][0]['must_cover'] = ['其他格式标签'] + other_path.write_text(dumps(value)) + other_root = series(tmp_path, 'other', corpus_path=other_path) + other = runner.summarize_repeated(other_root, scores(other_root, [True, True])) + with pytest.raises(ValueError, match='corpus'): runner.compare_repeated(summary, other) + + +def test_offline_commands_do_not_load_any_model_configuration(tmp_path, monkeypatch): + root = series(tmp_path) + paths = scores(root, [True, False]); output = tmp_path / 'summary.json' + import iris.models + monkeypatch.setattr(iris.models, 'load_test_models', lambda: (_ for _ in ()).throw(AssertionError('config read'))) + monkeypatch.setattr(sys, 'argv', ['memory_runner', 'summarize', '--series', str(root), + '--score', str(paths[0]), '--score', str(paths[1]), '--out', str(output)]) + runner.main() + assert json.loads(output.read_text())['means']['coverage'] == .5 + comparison = tmp_path / 'comparison.json' + monkeypatch.setattr(sys, 'argv', ['memory_runner', 'compare', '--left', str(output), + '--right', str(output), '--out', str(comparison)]) + runner.main() + assert json.loads(comparison.read_text())['means']['coverage_delta'] == 0 + + +def test_series_detects_input_changes_during_the_final_run(tmp_path, monkeypatch): + path = corpus(tmp_path); original = runner.run_sqlite + def mutate(*args, **kwargs): + manifest = original(*args, **kwargs) + path.write_text(path.read_text() + '\n') + return manifest + monkeypatch.setattr(runner, 'run_sqlite', mutate) + result = runner.run_repeated(path, tmp_path / 'series', runs=1, gateway_factory=factory) + assert result['invalidated']['reason'] == 'input_changed' + assert not result['gate_eligible'] + + +def test_pairing_matches_case_ids_even_if_summary_case_order_differs(tmp_path): + path = corpus(tmp_path); value = json.loads(path.read_text()) + second = copy.deepcopy(value['cases'][0]); second['id'] = 'unit-second' + second['must_cover'] = ['第二例格式标签']; value['cases'].append(second) + path.write_text(dumps(value)) + root = series(tmp_path, runs=1, corpus_path=path) + summary = runner.summarize_repeated(root, scores(root, [True])) + reordered = copy.deepcopy(summary) + reordered['runs'][0]['cases'].reverse() + reordered.pop('summary_sha256') + reordered = runner._sealed(reordered, 'summary_sha256') + comparison = runner.compare_repeated(summary, reordered) + assert [r['case_id'] for r in comparison['pairs']] == ['unit-daily', 'unit-second'] + assert all(r['coverage_delta'] == 0 for r in comparison['pairs']) diff --git a/tests/test_entry_profile_union_nomination.py b/tests/test_entry_profile_union_nomination.py new file mode 100644 index 00000000..d0f55ccc --- /dev/null +++ b/tests/test_entry_profile_union_nomination.py @@ -0,0 +1,228 @@ +"""EF-27 nomination experiment: independent fake calls, unchanged admission.""" +import copy +import json +from datetime import timedelta + +import pytest + +from iris import entry_profile as ep +from iris.db import dumps +from iris.models import ModelError +from test_entry_profile_incremental import ( + Gateway, NOW, TOPIC, accept, add, fixed_projection_clock, profiles, + projection, propose, run, source, +) + + +def enabled(store, value=True): + store.set_setting('entry_profile', {'union_nomination': value}) + + +def observe_calls(gateway): + return [c for c in gateway.calls if 'observe' in c['purpose']] + + +def work_row(store): + with store.read() as conn: + return dict(conn.execute("SELECT * FROM entry_profile_incremental_work WHERE kind='observe' ORDER BY id DESC").fetchone()) + + +def target_change(mid, body='', action='retire'): + return dict(op='change', memory_id=mid, category='topic', body=body, + saying='', meaning='', definition_refs=[], source_refs=[], action=action) + + +def test_union_is_default_off_and_strict_boolean(store): + with store.read() as conn: + assert ep._runtime(conn)['union_nomination'] is False + for value in (1, 0, 'true', None): + enabled(store, value) + with store.read() as conn, pytest.raises(ValueError): + ep._runtime(conn) + + +def test_disabled_keeps_one_original_request_and_result_shape(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store, False) + gateway = Gateway(store) + run(store, gateway) + assert len(observe_calls(gateway)) == 1 + row = work_row(store) + assert 'union_nomination' not in json.loads(row['payload_json'])['_snapshot'] + assert list(json.loads(row['outputs_json'])) == ['observe'] + assert 'nomination_union' not in json.loads(row['result_json']) + + +def test_independent_identical_requests_merge_duplicates_without_admitting_one_day(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store) + gateway = Gateway(store) + run(store, gateway) + calls = observe_calls(gateway) + assert len(calls) == 2 and calls[0]['payload'] == calls[1]['payload'] + assert len(profiles(store)) == 1 and profiles(store)[0]['independent_uses'] == 1 + assert profiles(store)[0]['retention'] == 42 and projection(store) is None + row = work_row(store) + assert len(json.loads(row['calls_json'])) == 2 + assert set(json.loads(row['outputs_json'])) == {'observe_nomination_1', 'observe_nomination_2'} + result = json.loads(row['result_json']) + assert len(result['generated']) == 1 and len(result['generated'][0]['source_refs']) == 1 + assert result['nomination_union']['rounds'] == 2 + + +def test_second_observation_can_discover_an_additional_candidate(store): + add(store, '轻量背包怎么取舍，也讨论不同鞋子的透气性。') + enabled(store) + gateway = Gateway(store) + # Both calls see the same message; the test chooses different supported hypotheses. + def observe(payload): + return propose(payload, body=TOPIC if len(observe_calls(gateway)) == 1 else '讨论徒步鞋子的取舍。') + gateway.observations = observe + run(store, gateway) + assert len(profiles(store)) == 2 + assert all(not p['usable'] and p['independent_uses'] == 1 for p in profiles(store)) + + +def test_union_merges_bound_sources_and_preserves_definition_refs(): + one = dict(op='propose', category='saying', body='落笔：先列要点', saying='落笔', meaning='先列要点', + definition_refs=[1], source_refs=[{'message_id': 1, 'field': 'content', 'excerpt': '落笔就是先列要点。'}]) + two = copy.deepcopy(one) + two['source_refs'].append({'message_id': 2, 'field': 'content', 'excerpt': '先落笔。'}) + merged, diagnostics = ep.merge_observation_nominations([{'observations': [one]}, {'observations': [two]}], []) + assert len(merged['observations']) == 1 + assert merged['observations'][0]['source_refs'] == two['source_refs'] + assert merged['observations'][0]['definition_refs'] == [1] + assert diagnostics['input_items'] == 2 and diagnostics['merged_items'] == 1 + assert one['source_refs'] == [two['source_refs'][0]] + + +@pytest.mark.parametrize('keep', ['confirm', 'propose', 'revise']) +def test_retire_conflict_keeps_existing_target(keep): + directory = [{'memory_id': 7, 'category': 'topic', 'body': TOPIC}] + retire = target_change(7) + retain = (dict(op='confirm', memory_id=7, source_refs=[]) if keep == 'confirm' else + dict(op='propose', category='topic', body=TOPIC, saying='', meaning='', definition_refs=[], source_refs=[]) if keep == 'propose' else + target_change(7, '讨论徒步装备的适用条件。', 'revise')) + merged, diagnostics = ep.merge_observation_nominations([{'observations': [retire]}, {'observations': [retain]}], directory) + assert not any(x.get('action') == 'retire' for x in merged['observations']) + assert diagnostics['conflicts'] + if keep == 'revise': assert not merged['observations'] + + +def test_incompatible_revisions_are_not_arbitrarily_selected(): + left = target_change(7, TOPIC, 'revise') + right = target_change(7, '讨论徒步鞋子的取舍。', 'revise') + merged, diagnostics = ep.merge_observation_nominations([{'observations': [left]}, {'observations': [right]}], []) + assert merged == {'observations': []} and diagnostics['conflicts'] + + +def test_duplicate_confirmation_only_confirms_once(store): + add(store, '轻量背包和结实背包怎么取舍？') + run(store, Gateway(store)) + mid = profiles(store)[0]['memory_id'] + add(store, '防水鞋和透气鞋怎么取舍？', 3, sender='乙') + enabled(store) + gateway = Gateway(store, lambda p: {'observations': [dict(op='confirm', memory_id=mid, + source_refs=[source(m) for m in p['messages']])]}) + run(store, gateway) + assert len(observe_calls(gateway)) == 2 + assert profiles(store)[0]['admission_state'] == 'accepted' + result = json.loads(work_row(store)['result_json']) + assert len(result['applied']) == 1 + assert len(result['generated'][0]['source_refs']) == 1 + + +def test_each_nomination_has_independent_deadline_and_repair_shares_its_deadline(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store) + def hook(g, purpose, payload): + if 'observe' in purpose: g.tick += 60 + gateway = Gateway(store, hook=hook) + def observe(payload): + return '{broken' if len(observe_calls(gateway)) == 1 else propose(payload) + gateway.observations = observe + run(store, gateway) + calls = observe_calls(gateway) + assert len(calls) == 3 + assert [c['deadline'] for c in calls] == [180, 180, 300] + row = work_row(store) + assert [c['nomination_round'] for c in json.loads(row['calls_json'])] == [1, 1, 2] + + +def test_call_budget_resume_does_not_repeat_completed_first_nomination(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store) + gateway = Gateway(store) + def observe(payload): + if len(observe_calls(gateway)) == 2: + raise ModelError('budget', 'quota', reason='call_budget') + return propose(payload) + gateway.observations = observe + run(store, gateway) + assert not profiles(store) + row = work_row(store) + assert row['state'] == 'deferred' + assert len(json.loads(row['parsed_json'])['_union_rounds']) == 1 + run(store, gateway, now=NOW + timedelta(hours=1)) + assert len(observe_calls(gateway)) == 3 + assert len(profiles(store)) == 1 and profiles(store)[0]['retention'] == 42 + assert len(json.loads(work_row(store)['calls_json'])) == 3 + + +def test_setting_change_during_second_call_prevents_writeback(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store) + def hook(g, purpose, payload): + if len(observe_calls(g)) == 2: enabled(store, False) + gateway = Gateway(store, hook=hook) + run(store, gateway) + assert len(observe_calls(gateway)) == 2 and not profiles(store) + assert work_row(store)['reason'] == 'settings_changed' + + +def test_equivalent_change_key_order_does_not_create_a_conflict(): + edit = target_change(7, TOPIC, 'revise') + reordered = dict(reversed(list(edit.items()))) + merged, diagnostics = ep.merge_observation_nominations([{'observations': [edit]}, {'observations': [reordered]}], []) + assert len(merged['observations']) == 1 and not diagnostics['conflicts'] + + +def test_propose_keeps_all_identical_directory_targets_from_retirement(): + directory = [dict(memory_id=i, category='topic', body=TOPIC) for i in (7, 8)] + keep = dict(op='propose', category='topic', body=TOPIC, saying='', meaning='', definition_refs=[], source_refs=[]) + merged, _ = ep.merge_observation_nominations([{'observations': [target_change(7)]}, {'observations': [keep]}], directory) + assert not any(x.get('action') == 'retire' for x in merged['observations']) + + +def test_retire_keep_conflict_does_not_withdraw_an_accepted_memory(store): + add(store, '轻量背包和结实背包怎么取舍？') + run(store, Gateway(store)) + mid = profiles(store)[0]['memory_id'] + add(store, '防水鞋和透气鞋怎么取舍？', 3, sender='乙') + confirm = lambda p: {'observations': [dict(op='confirm', memory_id=mid, source_refs=[source(m) for m in p['messages']])]} + run(store, Gateway(store, confirm)) + before = projection(store)['text'] + add(store, '关于背包和鞋子，我还想讨论重量与透气性。', 5, sender='丙') + enabled(store) + gateway = Gateway(store) + def observe(payload): + if len(observe_calls(gateway)) == 1: + edit = target_change(mid); edit['source_refs'] = [source(m) for m in payload['messages']] + return {'observations': [edit]} + return confirm(payload) + gateway.observations = observe + run(store, gateway) + assert projection(store)['text'] == before and profiles(store)[0]['usable'] + assert all(c['payload'].get('memory', {}).get('action') != 'retire' for c in gateway.calls) + + +def test_late_second_nomination_never_applies_the_first_one_alone(store): + add(store, '轻量背包和结实背包怎么取舍？') + enabled(store) + gateway = Gateway(store) + def hook(g, purpose, payload): + if 'observe' in purpose: g.tick += 10 if len(observe_calls(g)) == 1 else 181 + gateway.hook = hook + run(store, gateway) + assert len(observe_calls(gateway)) == 2 and not profiles(store) + assert work_row(store)['reason'] == 'timeout'