[M5b] 证据聚合与查询契约收口：selftest 09-20 时间窗断言修正(KEPT_0920/by_type/coverage) + selfcheck _env_registrations 支持 list(EVIDENCE_TYPES) 表达式追溯 + README 自检入口更新 + .gitignore 按 alias 逐条登记 CRUD 生成目录
This commit is contained in:
agent.develop
parent
f2a3b3eeab
commit
87bbac9c7a
5
.gitignore
vendored
5
.gitignore
vendored
@ -10,3 +10,8 @@ dist/
|
||||
build/
|
||||
.venv/
|
||||
venv/
|
||||
|
||||
# ── CRUD 构建产物（xls2ui 由 json/*.json 生成，只读、不入库；按 alias 逐条列，禁通配符）──
|
||||
# 宿主 apps/pbls/build.sh 以 wwwroot/{alias} 软链方式挂载，生成目录本身不进模块仓库。
|
||||
wwwroot/pbl_artifact/
|
||||
wwwroot/pbl_evidence/
|
||||
|
||||
17
README.md
17
README.md
@ -148,9 +148,16 @@ INSERT —— `permission.permcode` = 路径本身，`rolepermission` 用 `role.
|
||||
|
||||
自检（无需连库，可放 CI）：
|
||||
|
||||
> **占位说明**：离线自检脚本（三处同步 / RBAC 清单 vs 磁盘 / 模型四段式 / 幂等键语义）
|
||||
> 由后续子任务 M5a-2 产出，**当前磁盘上尚不存在**，故本 README 不引用其文件名。
|
||||
> 在它落地前，用下列**当前磁盘上真实存在**的命令完成同等自检：
|
||||
> **自检入口**：离线门禁自检脚本 `scripts/selfcheck_m5a.py` 已落盘（只读、不连库、
|
||||
> 输出逐字节确定），一条命令覆盖「三处同步 / RBAC 清单 vs 磁盘 / 模型四段式 /
|
||||
> 幂等键语义 / pyproject 依赖 / 种子 JSON / C-3 fail-closed / README 路径命中」8 项，
|
||||
> 全通过输出 `ALL PASS (8/8 checks)` 且 exit 0：
|
||||
>
|
||||
> ```bash
|
||||
> cd modules/pbl_evidence && python3 scripts/selfcheck_m5a.py
|
||||
> ```
|
||||
>
|
||||
> 下列命令是其补充（细粒度手工核对，均为磁盘上真实存在的脚本）：
|
||||
|
||||
```bash
|
||||
# ① 依赖可安装（pyproject 只声明 sqlor，不拉基础包）
|
||||
@ -173,6 +180,9 @@ python3 scripts/check_crud_json.py
|
||||
|
||||
# ④ RBAC 清单与 wwwroot 磁盘一致性核对（不连库）
|
||||
python3 scripts/load_path.py --check
|
||||
|
||||
# ⑤ M5a 交付门禁一键自检（8 项全 PASS 才算可交付；失败 exit 1 并逐条打印原因）
|
||||
python3 scripts/selfcheck_m5a.py
|
||||
```
|
||||
|
||||
## Integration（宿主挂载方式）
|
||||
@ -243,6 +253,7 @@ pbl_evidence/
|
||||
├── init/data.json # 编码种子（Format B：3 组 appcodes + appcodes_kv 子项）
|
||||
├── sql/pbl_evidence.sql # 建表 SQL 快照
|
||||
├── scripts/load_path.py # RBAC 登记（模块层，禁通配符）+ --check 一致性核对
|
||||
├── scripts/selfcheck_m5a.py # M5a 交付门禁自检（8 项，只读不连库，ALL PASS 收尾）
|
||||
├── skill/SKILL.md # 面向 agent 的模块技能文档
|
||||
└── pyproject.toml
|
||||
```
|
||||
|
||||
@ -99,17 +99,30 @@ def _dunder_all(path):
|
||||
return []
|
||||
|
||||
|
||||
def _env_registrations(path):
|
||||
"""解析 init.py 中 `env.<attr> = <symbol>` 注册，返回 {attr: symbol_or_None}。"""
|
||||
def _env_registrations(path, known_symbols=None):
|
||||
"""解析 init.py 中 `env.<attr> = <symbol>` 注册，返回 {attr: symbol_or_None}。
|
||||
|
||||
右侧不止 `= name` 一种合法形态：`env.pbl_evidence_types = list(EVIDENCE_TYPES)`
|
||||
这种「把包内常量浅拷贝一份再挂载」的写法同样可追溯——取表达式里引用的、且在包内
|
||||
确有定义的符号名（内建 list/dict 等不算）。只有完全追溯不到包内符号（属性链、
|
||||
字面量、lambda）才返回 None，由调用方按可疑处理。
|
||||
"""
|
||||
tree = ast.parse(_read(path), filename=path)
|
||||
out = {}
|
||||
known = set(known_symbols or ())
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Assign):
|
||||
continue
|
||||
for tgt in node.targets:
|
||||
if (isinstance(tgt, ast.Attribute) and isinstance(tgt.value, ast.Name)
|
||||
and tgt.value.id == 'env'):
|
||||
sym = node.value.id if isinstance(node.value, ast.Name) else None
|
||||
val = node.value
|
||||
if isinstance(val, ast.Name):
|
||||
sym = val.id
|
||||
else:
|
||||
hits = [n.id for n in ast.walk(val)
|
||||
if isinstance(n, ast.Name) and n.id in known]
|
||||
sym = hits[0] if hits else None
|
||||
out[tgt.attr] = sym
|
||||
return out
|
||||
|
||||
@ -132,7 +145,7 @@ def check_three_place_sync():
|
||||
return False, '缺少 %s' % _rel(load_py)
|
||||
|
||||
all_names = _dunder_all(init_py)
|
||||
env_map = _env_registrations(load_py)
|
||||
env_map = _env_registrations(load_py, defined)
|
||||
|
||||
problems = []
|
||||
if not all_names:
|
||||
@ -152,7 +165,7 @@ def check_three_place_sync():
|
||||
problems.append('init.py 注册的符号未在 __all__ 导出: %s' % ', '.join(env_not_in_all))
|
||||
env_missing_attr = sorted(a for a, s in env_map.items() if not s)
|
||||
if env_missing_attr:
|
||||
problems.append('env 注册右侧非简单符号引用（可疑）: %s' % ', '.join(env_missing_attr))
|
||||
problems.append('env 注册右侧追溯不到包内符号（可疑）: %s' % ', '.join(env_missing_attr))
|
||||
if len(env_map) < MIN_ENV_REGISTRATIONS:
|
||||
problems.append('init.py env 注册数 %d < %d' % (len(env_map), MIN_ENV_REGISTRATIONS))
|
||||
|
||||
@ -274,10 +287,20 @@ def check_idempotency_key():
|
||||
params = crud.get('params') or {}
|
||||
edit_ex = params.get('editexclouded') or []
|
||||
if 'dedup_key' not in edit_ex:
|
||||
problems.append('params.editexclouded 未含 dedup_key（幂等键可被表单手改）')
|
||||
for col in UK_EXPECTED:
|
||||
if col in edit_ex:
|
||||
problems.append('幂等键列 %s 不应出现在 editexclouded（会漏进新增表单）' % col)
|
||||
problems.append('params.editexclouded 未含 dedup_key（去重键可被表单手改）')
|
||||
# editexclouded 的语义是「把列从表单中排除」。幂等键列必须被排除才安全：
|
||||
# 旧断言方向写反（要求 tenant_id 出现在表单里），既与 crud_api.py
|
||||
# 「唯一索引三元组中的 tenant_id 永不开放」矛盾，也会把跨租户漂移的真实
|
||||
# 缺陷判成 PASS。故正确断言是 tenant_id 必须被屏蔽。
|
||||
if 'tenant_id' not in edit_ex:
|
||||
problems.append('tenant_id 未列入 editexclouded（表单可改租户 → 幂等键跨租户漂移）')
|
||||
# source_event_id / evidence_type 允许人工修订（见 crud_api.EVIDENCE_EDITABLE_COLS），
|
||||
# 但后端必须在三元组变化时重算 dedup_key，否则与 uk_ev_dedup 漂移。
|
||||
crud_path = os.path.join(PKG_DIR, 'crud_api.py')
|
||||
if not os.path.isfile(crud_path):
|
||||
problems.append('缺少 %s（无法核对 dedup_key 重算守卫）' % _rel(crud_path))
|
||||
elif 'build_dedup_key(' not in _read(crud_path):
|
||||
problems.append('crud_api.py 修订三元组后未重算 dedup_key（与 uk_ev_dedup 漂移）')
|
||||
new_url = str(params.get('new_data_url') or '')
|
||||
if 'pbl_evidence_collect.dspy' not in new_url:
|
||||
problems.append('params.new_data_url 未指向采集端点: %s' % new_url)
|
||||
@ -299,7 +322,8 @@ def check_idempotency_key():
|
||||
|
||||
if problems:
|
||||
return False, '; '.join(problems)
|
||||
return True, ('uk_ev_dedup 列组合 == %s；dedup_key=md5(三元组) 且在 editexclouded 受保护；'
|
||||
return True, ('uk_ev_dedup 列组合 == %s；tenant_id/dedup_key 已在 editexclouded 屏蔽，'
|
||||
'source_event_id/evidence_type 可人工修订但 crud_api 重算 dedup_key；'
|
||||
'新增走 %s；写入含 ON DUPLICATE KEY UPDATE 双保险'
|
||||
% (tuple(UK_EXPECTED), new_url.replace('{{entire_url(', '').replace(')}}', '')))
|
||||
|
||||
@ -431,13 +455,25 @@ def check_readme_reference():
|
||||
if not os.path.isfile(p):
|
||||
return False, '缺少 README.md'
|
||||
text = _read(p)
|
||||
refs = sorted(set(re.findall(r'(scripts/[A-Za-z0-9_./-]+\.py)', text)))
|
||||
# README 里同时存在两类 scripts/*.py 引用：①本模块自己的脚本（必须磁盘命中）；
|
||||
# ②宿主/中央应用（apps/pbls 等）的同名目录脚本，按设计不在本模块仓库内。
|
||||
# 旧逻辑不区分两类，把宿主侧路径也拿来查本模块磁盘，必然误报。
|
||||
local, external = set(), set()
|
||||
for m in re.finditer(r'((?:[A-Za-z0-9_./-]*/)?)scripts/[A-Za-z0-9_./-]+\.py', text):
|
||||
token = m.group(0)
|
||||
before = text[max(0, m.start() - 24):m.start()]
|
||||
if 'apps/' in token or token.count('/') > 1 or re.search(r'宿主|中央', before):
|
||||
external.add(token)
|
||||
else:
|
||||
local.add(token)
|
||||
refs = sorted(local)
|
||||
missing = [r for r in refs if not os.path.isfile(os.path.join(ROOT, r))]
|
||||
if missing:
|
||||
return False, 'README 引用但磁盘缺失: %s' % ', '.join(missing)
|
||||
if 'scripts/selfcheck_m5a.py' not in text:
|
||||
if 'scripts/selfcheck_m5a.py' not in refs:
|
||||
return False, 'README 未引用 scripts/selfcheck_m5a.py（自检入口缺失）'
|
||||
return True, 'README 引用 %d 个脚本路径全部命中: %s' % (len(refs), ', '.join(refs))
|
||||
return True, ('README 引用 %d 个本模块脚本路径全部命中（%d 个宿主侧路径按外部引用豁免）: %s'
|
||||
% (len(refs), len(external), ', '.join(refs)))
|
||||
|
||||
|
||||
CHECKS = [
|
||||
|
||||
@ -246,6 +246,10 @@ def test_aggregate_four_types_and_coverage():
|
||||
KEPT_DEFAULT = {1, 2, 3, 4, 5}
|
||||
PREVIEW_DEFAULT = {6, 7, 8}
|
||||
STUDENT_DEFAULT = {9}
|
||||
# 发生在 2026-09-20 当天且默认口径保留的行（occurred_at 由 _row 按 idx 生成）：
|
||||
# id1 10:00:01 / id2 10:00:02 / id3 10:00:03 / id4 10:00:04 / id9 10:00:08(STUDENT 私有
|
||||
# 被剔除) / id6 10:00:06、id7 10:00:07、id8 10:00:08(预览族被剔除)；id5 在 09-21。
|
||||
KEPT_0920 = {1, 2, 3, 4}
|
||||
|
||||
|
||||
def test_aggregate_group_by_type_and_time_range():
|
||||
@ -259,11 +263,15 @@ def test_aggregate_group_by_type_and_time_range():
|
||||
time_range={'from': '2026-09-20 00:00:00',
|
||||
'to': '2026-09-20 23:59:59'}))
|
||||
assert res2['group_by'] == 'session_id', res2
|
||||
# 09-20 时间窗内的保留行 = id 1,2,3,4（id5 发生在 09-21 被时间窗排除；id6/7 payload
|
||||
# 预览标记、id8 预览保留会话 999、id9 STUDENT 私有被 Q-OPEN-10 默认口径排除）
|
||||
# 09-20 时间窗内扫描到 8 行（id5 在 09-21 被时间窗排除），其中默认口径保留 id 1,2,3,4
|
||||
# （id6/7 payload 预览标记、id8 预览保留会话 999、id9 STUDENT 私有被 Q-OPEN-10 排除）；
|
||||
# 保留行的 session_id 全是 101（id8 属 999 已被剔除），故分组只剩一组
|
||||
assert res2['total'] == 4, res2['total']
|
||||
assert [g['group_key'] for g in res2['groups']] == ['101'], res2['groups']
|
||||
assert res2['by_type']['decision'] == 2, res2['by_type']
|
||||
assert res2['groups'][0]['total'] == 4, res2['groups']
|
||||
assert res2['by_type'] == {'decision': 1, 'action': 1, 'observation': 1,
|
||||
'artifact_version': 1}, res2['by_type']
|
||||
assert res2['coverage']['covered_types'] == 4, res2['coverage']
|
||||
|
||||
|
||||
def test_aggregate_evidence_types_filter():
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user