Harden coverage honesty, diff GONE gating, and key redaction

skills/ir-search/scripts/diff_surveys.py

@@ -56,6 +56,17 @@
 
 COMPARE_FIELDS = ["title", "apply_start", "apply_end", "status", "content_hash"]
 
+
+def _reject_dup_keys(pairs):
+    """object_pairs_hook — 한 레코드에 중복 키가 있으면 거부(Codex #12).
+    기본 로더는 뒤값만 남겨 위조 필드가 검사를 우회한다."""
+    d = {}
+    for k, v in pairs:
+        if k in d:
+            raise ValueError(f"중복 JSON 키: {k!r}")
+        d[k] = v
+    return d
+
 # ir-search-profile.md bullets that define the judgment axes. If any of these
 # change, previous A/B/C verdicts can no longer be carried over.
 PROFILE_AXES = ["창업 단계", "지역 연고", "대표자", "필요한 것"]
@@ -83,8 +94,8 @@ def load_dir(d: Path):
             if not line:
                 continue
             try:
-                rec = json.loads(line)
-            except json.JSONDecodeError as e:
+                rec = json.loads(line, object_pairs_hook=_reject_dup_keys)
+            except (json.JSONDecodeError, ValueError) as e:
                 sys.exit(f"ERROR: broken JSON at {f}:{ln} — {e}")
             if "kind" in rec and "record" in rec:
                 continue  # stray diff artifact record — not a raw crawl row
@@ -135,6 +146,44 @@ def profile_fingerprint(fields):
     return hashlib.sha256(payload.encode()).hexdigest()[:16]
 
 
+def gone_eligible_from_manifest(curr_dir):
+    """Sources that MAY yield GONE/CLOSED — only those with a proven-complete
+    current run. Coverage honesty: an announcement can be declared gone only when
+    the current run demonstrably covered that source in full.
+
+    Reads run_manifest.json in *curr_dir* and returns (eligible, note):
+      - eligible: set of source names whose run is status=="ok" AND exit_code==0.
+        A source that is partial (api-window/page-cap/manual/inactive) OR absent
+        from the manifest is NOT eligible — its records must not be reported GONE.
+      - eligible is None when the manifest is absent OR unreadable — the caller
+        then suppresses ALL removals unless --assume-complete is given
+        (fail-closed: a partial/unknown run must never look like everything
+        closed — README ir §K-Startup API).
+    note is a human string for the summary, "" when a clean manifest was read.
+    """
+    mpath = curr_dir / "run_manifest.json"
+    if not mpath.exists():
+        return None, ("no run_manifest.json — coverage unproven; GONE suppressed "
+                      "(pass --assume-complete if these were full crawls)")
+    try:
+        data = json.loads(mpath.read_text(encoding="utf-8"),
+                          object_pairs_hook=_reject_dup_keys)
+        runs = data.get("runs", [])
+        if not isinstance(runs, list):
+            raise ValueError('"runs" not a list')
+    except (OSError, json.JSONDecodeError, ValueError, AttributeError) as e:
+        return None, f"run_manifest.json unreadable ({e}) — GONE suppressed"
+    eligible = set()
+    for r in runs:
+        if not isinstance(r, dict):
+            continue
+        if r.get("status") == "ok" and r.get("exit_code") == 0:
+            s = r.get("source")
+            if s:
+                eligible.add(s)
+    return eligible, ""
+
+
 def changed_fields(old, new):
     return [f for f in COMPARE_FIELDS if (old.get(f) or None) != (new.get(f) or None)]
 
@@ -164,6 +213,11 @@ def classify(old, new):
                 "changed_fields": ["hash_version(산식 전환 — 1회 상세 재검증)"]}
     if old_h and not new_h:
         return {"kind": "NEEDS_REHASH", "changed_fields": []}
+    if new_h and not old_h:
+        # 직전엔 해시가 없었는데 이번에 상세를 처음 수집했다 — 목록 필드가 같아도
+        # 직전 판정은 상세 없이 내려졌을 수 있으므로 1회 재검토(Codex #6).
+        return {"kind": "CHANGED",
+                "changed_fields": ["content_hash(최초 상세수집 — 재검토)"]}
     return {"kind": "UNCHANGED", "changed_fields": []}
 
 
@@ -187,6 +241,11 @@ def main():
     ap.add_argument("--out", type=Path, help="write items needing review as jsonl")
     ap.add_argument("--old-profile", help="profile snapshot used for prev_dir")
     ap.add_argument("--new-profile", help="profile used for curr_dir")
+    ap.add_argument("--assume-complete", action="store_true",
+                    help="treat the current run as a full crawl of every source "
+                         "even without a run_manifest.json (legacy/manual dirs). "
+                         "Without this AND without a proven manifest run, GONE is "
+                         "suppressed — a partial crawl must not read as all-closed.")
     args = ap.parse_args()
 
     for d in (args.prev_dir, args.curr_dir):
@@ -210,12 +269,15 @@ def main():
         new_fields = parse_profile_bullets(args.new_profile)
         # 판정 축(PROFILE_AXES)이 하나도 없는 프로필은 파싱 실패와 같다 — 무관한
         # 불릿만 있는 파일 두 개가 "동일 fingerprint"로 승계를 통과하면 안 된다.
-        old_axes = any(old_fields.get(k) for k in PROFILE_AXES)
-        new_axes = any(new_fields.get(k) for k in PROFILE_AXES)
+        # 판정 축은 하나라도 빠지면(지역만·단계만 등) 승계 근거가 불완전하다 —
+        # any가 아니라 ALL을 요구한다(Codex #14). 미완성 프로필은 승계 무효.
+        old_axes = all(old_fields.get(k) for k in PROFILE_AXES)
+        new_axes = all(new_fields.get(k) for k in PROFILE_AXES)
         if not old_fields or not new_fields or not old_axes or not new_axes:
             invalidate = True
-            print("WARNING: 프로필 파일을 읽지 못했거나 판정 축(창업 단계·지역 등)이 "
-                  "비어 있다 — 승계 무효(fail-closed), 전체 재검토", file=sys.stderr)
+            print("WARNING: 프로필 파일을 읽지 못했거나 판정 축(창업 단계·지역·대표자·"
+                  "필요한 것)이 하나라도 비어 있다 — 승계 무효(fail-closed), 전체 재검토",
+                  file=sys.stderr)
         elif profile_fingerprint(old_fields) != profile_fingerprint(new_fields):
             invalidate = True
             print("WARNING: profile changed — 전체 재판정 필요 "
@@ -228,8 +290,23 @@ def main():
     curr_sources = {k[0] for k in curr}
     common = prev_sources & curr_sources
 
+    # Coverage guard (fail-closed): an announcement may be declared GONE/CLOSED
+    # only for a source the CURRENT run proved it covered in full (manifest
+    # status=ok, exit 0). A partial run (api-window/page-cap) or a source with no
+    # proven run cannot distinguish "closed" from "outside the collected window",
+    # so its removals are suppressed rather than reported as false expirations.
+    eligible, manifest_note = gone_eligible_from_manifest(args.curr_dir)
+    if eligible is None:  # no/unreadable manifest → nothing proven
+        gone_eligible = set(common) if args.assume_complete else set()
+    else:
+        gone_eligible = {s for s in eligible if s in common}
+    suppressed_sources = {s for s in common if s not in gone_eligible}
+
     new = [curr[k] for k in curr if k not in prev and k[0] in common]
-    closed = [prev[k] for k in prev if k not in curr and k[0] in common]
+    closed = [prev[k] for k in prev if k not in curr and k[0] in common
+              and k[0] not in suppressed_sources]
+    suppressed_closed = [prev[k] for k in prev if k not in curr and k[0] in common
+                         and k[0] in suppressed_sources]
     results = {k: classify(prev[k], curr[k]) for k in curr if k in prev}
     changed = [
         (prev[k], curr[k], r["changed_fields"])
@@ -269,6 +346,16 @@ def main():
     for r in closed:
         print(f"  - [{r.get('source')}] {r.get('title', '(no title)')}")
 
+    if manifest_note:
+        print(f"\n## COVERAGE NOTE — {manifest_note}")
+    if suppressed_closed:
+        srcs = ", ".join(sorted({r.get('source') for r in suppressed_closed}))
+        print(f"\n## CLOSED SUPPRESSED ({len(suppressed_closed)}) — current run for "
+              f"[{srcs}] was partial (api-window/page-cap/manual): absence is NOT "
+              "concluded GONE. Re-run a full crawl to confirm expirations.")
+        for r in suppressed_closed:
+            print(f"  · [{r.get('source')}] {r.get('title', '(no title)')}")
+
     if invalidate:
         print(f"\n## UNCHANGED: {unchanged} items — 승계 불가(프로필 변경), 전건 재검토")
     else:

skills/ir-search/scripts/kstartup_api.py

@@ -415,6 +415,7 @@ def list_announcements(key, min_expected=1, per_page=PER_PAGE, max_pages=MAX_PAG
     """
     redact = _make_redactor(key)
     out = {}
+    seen_ids = set()   # every id seen across pages — overlap breaks the position proof
     total = None
     zero_open_streak = 0
     page = 1
@@ -444,6 +445,21 @@ def list_announcements(key, min_expected=1, per_page=PER_PAGE, max_pages=MAX_PAG
             # a mixed container (valid dicts + junk) is a schema change, and the
             # junk still counts toward totalCount — fail closed to the crawl.
             raise ApiError("mixed unparseable records; using crawl")
+        # Exhaustion is proved by scanned ROW POSITIONS reaching totalCount, so
+        # every scanned position must map to a *distinct real id*. If a row has
+        # no id, or an id repeats within this page, or overlaps an id already
+        # seen on a prior page (duplicate/overlapping pagination), the position
+        # counter overshoots while unique coverage stalls — we'd declare `proven`
+        # early and silently drop the real tail. Any of these fails closed to the
+        # crawl (the exhaustive-coverage authority).
+        page_ids = [_pick(r, ("pbanc_sn", "biz_pbanc_sn", "pbancSn")) for r in recs]
+        if not all(page_ids):
+            raise ApiError("record without id; cannot prove coverage; using crawl")
+        if len(set(page_ids)) != len(page_ids):
+            raise ApiError("duplicate ids within a page; using crawl")
+        if seen_ids.intersection(page_ids):
+            raise ApiError("overlapping pages (duplicate ids); using crawl")
+        seen_ids.update(page_ids)
         open_on_page = 0
         for r in recs:
             if not _is_open(r):

skills/ir-search/scripts/run_manifest.py

@@ -43,6 +43,17 @@
 
 MANIFEST_SCHEMA_VERSION = 1
 MANIFEST_NAME = "run_manifest.json"
+
+
+def _reject_dup_keys(pairs):
+    """object_pairs_hook — 중복 키 거부(Codex #12). 중복 manifest_schema_version/
+    status가 #10 검증을 우회하지 못하게 매니페스트 로드에도 적용한다."""
+    d = {}
+    for k, v in pairs:
+        if k in d:
+            raise ValueError(f"duplicate JSON key: {k!r}")
+        d[k] = v
+    return d
 KST = timezone(timedelta(hours=9))
 VALID_STATUS = ("ok", "partial", "manual", "inactive")
 
@@ -52,6 +63,20 @@ def make_run(source, status, exit_code, pages_fetched, collected, stop_reason,
     """Build one schema-v1 run entry. Counts/status only — no content."""
     if status not in VALID_STATUS:
         raise ValueError(f"invalid status {status!r} (expected one of {VALID_STATUS})")
+    # 정합성 계약(Codex #10): 상태와 종료코드는 모순될 수 없다. ok는 성공(0),
+    # 그 외(partial/manual/inactive)는 반드시 비-0이어야 한다. 음수 카운트도 거부.
+    exit_code = int(exit_code)
+    if status == "ok" and exit_code != 0:
+        raise ValueError(f"status=ok인데 exit_code={exit_code} (0이어야 함)")
+    if status != "ok" and exit_code == 0:
+        raise ValueError(f"status={status!r}인데 exit_code=0 (비-0이어야 함)")
+    for label, val in (("pages_fetched", pages_fetched), ("collected", collected)):
+        if int(val) < 0:
+            raise ValueError(f"{label}가 음수({val}) — 불가")
+    if reported_total is not None and int(reported_total) < 0:
+        raise ValueError(f"reported_total 음수({reported_total}) — 불가")
+    if duplicates is not None and int(duplicates) < 0:
+        raise ValueError(f"duplicates 음수({duplicates}) — 불가")
     run = {
         "source": source,
         "status": status,
@@ -84,9 +109,16 @@ def update_manifest(output_path, new_runs):
     if os.path.exists(path):
         try:
             with open(path, encoding="utf-8") as f:
-                old = json.load(f)
+                old = json.load(f, object_pairs_hook=_reject_dup_keys)
             if not isinstance(old, dict):
                 raise ValueError("manifest top level is not a JSON object")
+            # 알 수 없는 스키마 버전을 조용히 v1로 덮어쓰지 않는다(Codex #10) —
+            # 미래/손상 버전은 corrupt 경로로 보존한 뒤 새로 시작한다.
+            old_ver = old.get("manifest_schema_version")
+            if old_ver != MANIFEST_SCHEMA_VERSION:
+                raise ValueError(
+                    f"unsupported manifest_schema_version {old_ver!r} "
+                    f"(this writer emits v{MANIFEST_SCHEMA_VERSION})")
             old_runs = old.get("runs", [])
             if not isinstance(old_runs, list):
                 raise ValueError('"runs" is not a list')

tests/test_diff_record_schema.py

@@ -166,7 +166,7 @@ def biz(i, title, **extra):
     write_jsonl(curr / "bizinfo.jsonl", [biz(1, "그대로")])
     out = tmp_path / "new_items.jsonl"
     monkeypatch.setattr("sys.argv", ["diff_surveys.py", str(prev), str(curr),
-                                     "--out", str(out)])
+                                     "--out", str(out), "--assume-complete"])
     diff_surveys.main()
 
     lines = []