Agent/knowhub/scripts/version_step3_verify.py

#!/usr/bin/env python3
"""Step 3: 回归验证 dev_abstract / dev_dedup 双版本查询是否对称且各自正确。"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from knowhub.knowhub_db.pg_capability_store import PostgreSQLCapabilityStore


def q(cur, sql, params=()):
    cur.execute(sql, params)
    return cur.fetchall()


def qc(cur, sql, params=()):
    cur.execute(sql, params)
    return cur.fetchone()['c']


def main():
    s = PostgreSQLCapabilityStore()
    cur = s._get_cursor()
    try:
        cur.execute("SET statement_timeout = '30s'")

        # 1) 挑一个 req，两版各自看到的 strategy
        print('\n=== Test 1: REQ_003 两版的 strategy（应分别看到 abstract 和 concrete）===', flush=True)
        for v in ('dev_abstract', 'dev_dedup'):
            rows = q(cur, """SELECT s.id, s.name, s.version, rs.is_selected
                             FROM requirement_strategy rs
                             JOIN strategy s ON s.id=rs.strategy_id
                             WHERE rs.requirement_id='REQ_003' AND s.version=%s""", (v,))
            print(f'  [{v}]: {len(rows)} strategies', flush=True)
            for r in rows:
                print(f'    - {r["id"]}: {r["name"][:40]}  selected={r["is_selected"]}', flush=True)

        # 2) 每版的 strategy → cap 路径是否自洽
        print('\n=== Test 2: cap 过滤是否隔离（dev_dedup 的 strat 不能指向 dev_abstract cap）===', flush=True)
        for v in ('dev_abstract', 'dev_dedup'):
            leak = qc(cur, """SELECT COUNT(*) c FROM strategy_capability sc
                              JOIN strategy s ON s.id=sc.strategy_id
                              JOIN capability c ON c.id=sc.capability_id
                              WHERE s.version=%s AND c.version!=%s""", (v, v))
            print(f'  [{v}] strat.ver={v} 但 cap.ver 不等于 {v}: {leak}  {"❌" if leak else "✓"}', flush=True)

        # 3) 每版 strategy_resource 是否自洽
        for v in ('dev_abstract', 'dev_dedup'):
            leak = qc(cur, """SELECT COUNT(*) c FROM strategy_resource sr
                              JOIN strategy s ON s.id=sr.strategy_id
                              JOIN resource r ON r.id=sr.resource_id
                              WHERE s.version=%s AND r.version!=%s""", (v, v))
            print(f'  [{v}] strat_resource 跨版: {leak}  {"❌" if leak else "✓"}', flush=True)

        # 4) req 覆盖度（每 req 在每版都有 strategy？）
        print('\n=== Test 3: 每版 strategy 覆盖的 req 数量 ===', flush=True)
        for v in ('dev_abstract', 'dev_dedup'):
            n = qc(cur, """SELECT COUNT(DISTINCT rs.requirement_id) c
                           FROM requirement_strategy rs
                           JOIN strategy s ON s.id=rs.strategy_id
                           WHERE s.version=%s""", (v,))
            print(f'  [{v}]: {n}/99 reqs 有 strategy 链接', flush=True)

        # 5) 约束验证：knowledge_cap ⊆ req_cap（dev_abstract 的核心约束）
        print('\n=== Test 4: knowledge_cap ⊆ req_cap（dev_abstract 约束）===', flush=True)
        viol = qc(cur, """SELECT COUNT(*) c FROM knowledge_capability kc
                          JOIN requirement_knowledge rk ON rk.knowledge_id=kc.knowledge_id
                          JOIN knowledge k ON k.id=kc.knowledge_id
                          LEFT JOIN requirement_capability rc
                            ON rc.requirement_id=rk.requirement_id
                            AND rc.capability_id=kc.capability_id
                          WHERE k.version='dev_abstract' AND rc.capability_id IS NULL""")
        print(f'  违规: {viol}  {"❌" if viol else "✓"}', flush=True)

        # 6) dev_dedup 约束：strat_cap ⊆ req_cap（dev_dedup 里 req 是每 cap 研究全集）
        print('\n=== Test 5: dev_dedup: strat_cap ⊆ req_cap（每 req 的 strat 用的 cap ⊆ 该 req 的 cap）===', flush=True)
        viol = qc(cur, """SELECT COUNT(*) c FROM strategy_capability sc
                          JOIN strategy s ON s.id=sc.strategy_id
                          JOIN requirement_strategy rs ON rs.strategy_id=sc.strategy_id
                          LEFT JOIN requirement_capability rc
                            ON rc.requirement_id=rs.requirement_id
                            AND rc.capability_id=sc.capability_id
                          JOIN capability c ON c.id=sc.capability_id
                          WHERE s.version='dev_dedup' AND c.version='dev_dedup'
                            AND rc.capability_id IS NULL""")
        print(f'  违规: {viol}  {"❌" if viol else "✓"}', flush=True)

        # 7) 交叉版本是否完全隔离（不会误查到另一版）
        print('\n=== Test 6: 交叉版本隔离（dev_abstract 查询不应看到 dev_dedup 的 strat）===', flush=True)
        n = qc(cur, """SELECT COUNT(*) c FROM strategy WHERE version='dev_abstract'""")
        print(f'  dev_abstract strat: {n}  (预期 26)  {"✓" if n==26 else "❌"}', flush=True)
        n = qc(cur, """SELECT COUNT(*) c FROM strategy WHERE version='dev_dedup'""")
        print(f'  dev_dedup strat: {n}  (预期 99)  {"✓" if n==99 else "❌"}', flush=True)

    finally:
        cur.close(); s.close()


if __name__ == '__main__':
    main()