ci: sweep the Db2 cases in Evidence Refresh and pin the Db2 image by digest

docs/language_status/oracle_assumptions.md

@@ -42,6 +42,27 @@ makes every proof's evidence stale). If Debian drops the pinned package from `de
 the pinned build fails (the weekly CICS crucible run notices); restore it from `snapshot.debian.org` rather than
 taking whatever version is current.
 
+## The Db2 image is pinned (#4733)
+
+A Db2 case runs its SQL on one real Db2 (IBM Db2 Community Edition, in a container), the COBOL side through IBM's CLI
+driver and the Java side through JDBC (`tests/tools/equivalence_db2.py`). The image was `icr.io/db2_community/db2:latest`,
+a tag that moves: a new Db2 could change a code page, a collation or a SQLCODE with no change of ours. It is now named
+by content digest:
+
+| pinned | value |
+|---|---|
+| Db2 image | `icr.io/db2_community/db2@sha256:2de8151713c261843868c5c3411b57be6ae79d99d70a5b3022337836776bfda6` |
+| built | 2026-07-13 (the image every Db2 proof so far ran on; it was `:latest` on 2026-10-08) |
+
+`equivalence_db2.IMAGE` is the pin. `tests/cobol_mainframe/test_db2_image_pin.py` fails when it is a tag, or when this
+page does not name its digest. A container left running from another image is refused (`docker rm -f gitgalaxy-db2`),
+and every Db2 case's proof report records the image under `oracle.db2`. `equivalence_db2.py` is part of the harness
+fingerprint, so moving the pin makes every evidence record stale (re-proved by the Evidence Refresh bot).
+
+Moving it is a deliberate change: resolve the new digest (`curl -sI -H 'Accept: application/vnd.oci.image.index.v1+json'
+https://icr.io/v2/db2_community/db2/manifests/latest`, header `docker-content-digest`), set `IMAGE`, update this table,
+and re-sweep the Db2 cases.
+
 ## Status key
 
 | status | meaning |

tests/cobol_mainframe/test_db2_image_pin.py

@@ -0,0 +1,42 @@
+"""#4733: the Db2 image a Db2 case runs its SQL on is pinned by digest (like the GnuCOBOL oracle, #4309), the pin is
+written down in docs/language_status/oracle_assumptions.md, and a Db2 case's recorded oracle names it."""
+
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
+import equivalence_db2 as db2  # noqa: E402
+import equivalence_oracle as eo  # noqa: E402
+
+REPO = Path(__file__).resolve().parents[2]
+DIGEST = re.compile(r"@(sha256:[0-9a-f]{64})$")
+
+
+def test_the_db2_image_is_pinned_by_digest_not_a_tag():
+    assert db2.pinned_by_digest(db2.IMAGE), f"{db2.IMAGE} must be name@sha256:<digest>"
+    assert ":latest" not in db2.IMAGE
+
+
+def test_a_tag_is_not_a_pin():
+    assert not db2.pinned_by_digest("icr.io/db2_community/db2:latest")
+    assert not db2.pinned_by_digest("icr.io/db2_community/db2:12.1.0.0")
+    assert not db2.pinned_by_digest("icr.io/db2_community/db2@sha256:abc")
+
+
+def test_the_pinned_digest_is_written_in_the_oracle_assumptions():
+    text = (REPO / "docs" / "language_status" / "oracle_assumptions.md").read_text(encoding="utf-8")
+    digest = DIGEST.search(db2.IMAGE).group(1)
+    assert digest in text, "docs/language_status/oracle_assumptions.md must name the pinned Db2 image digest"
+
+
+def test_every_container_start_uses_the_pinned_image():
+    src = (REPO / "tests" / "tools" / "equivalence_db2.py").read_text(encoding="utf-8")
+    assert "db2:latest" not in src.replace("`:latest`", "")
+
+
+def test_a_db2_cases_oracle_records_the_db2_image(monkeypatch):
+    monkeypatch.setattr(eo, "checked", lambda image=eo.IMAGE: {"image": {"name": image}, "matches_pin": True})
+    fp = eo.for_case({"db2": {"ddl": []}})
+    assert fp["db2"] == {"image": db2.IMAGE, "pinned_by_digest": True}
+    assert "db2" not in eo.for_case({})

tests/cobol_mainframe/test_proof_sweep.py

@@ -51,3 +51,11 @@ def test_update_baseline_only_drops_entries_that_now_prove(tmp_path):
     ok, bad = {"proved": True, "coverage": ""}, {"proved": False, "coverage": ""}
     assert ps.prune_baseline({"det": {"x": ok, "y": bad, "z": ok}}, path) == ["det x"]
     assert list(ps.load_baseline(path)) == [("det", "y")]
+
+
+def test_db2_only_and_skip_db2_split_the_det_cases():
+    """#4733: Evidence Refresh's Db2 job sweeps exactly the cases the det-sweep workflow leaves out."""
+    every, plain, db2 = ps.det_cases(), ps.det_cases(skip_db2=True), ps.det_cases(only_db2=True)
+    assert db2 and plain
+    assert set(plain) | set(db2) == set(every) and not set(plain) & set(db2)
+    assert all(ps.is_db2(c) for c in db2) and not any(ps.is_db2(c) for c in plain)

tests/tools/equivalence_db2.py

@@ -53,7 +53,10 @@
 
 CONTAINER = "gitgalaxy-db2"
 NETWORK = "gitgalaxy-db2"
-IMAGE = "icr.io/db2_community/db2:latest"
+# #4733: pinned by digest, never a tag -- `:latest` moves under the proofs (the Db2 version, its code pages, its SQL
+# behaviour) and a Db2 case's verdict could change with no change of ours. Moving it is a deliberate change: see "The Db2
+# image is pinned" in docs/language_status/oracle_assumptions.md. tests/cobol_mainframe/test_db2_image_pin.py enforces it.
+IMAGE = "icr.io/db2_community/db2@sha256:2de8151713c261843868c5c3411b57be6ae79d99d70a5b3022337836776bfda6"
 COBOL_IMAGE = "gitgalaxy-gnucobol-db2:3"
 DATABASE0 = "GGDB"  # the container's own database: the pool's first
 DATABASE = DATABASE0  # this process's database (hold_lock picks it from the pool)
@@ -158,6 +161,11 @@ def _ensure_database(database: str) -> None:
         if database in listed():
             return
         ref = _db_cfg(DATABASE0)
+        for _ in range(60):  # #4733: a fresh container (CI) is still creating GGDB when the first pool cases start
+            if "Database code set" in ref:
+                break
+            time.sleep(10)
+            ref = _db_cfg(DATABASE0)
         out = _instance(f"db2 create database {database} using codeset {ref['Database code set']} territory "
                         f"{ref['Database territory']} collate using {ref['Database collating sequence']} "
                         f"pagesize {ref['Database page size']}")  # fmt: skip
@@ -168,26 +176,50 @@ def _ensure_database(database: str) -> None:
 def _ensure_container() -> None:
     if not _docker("images", "-q", COBOL_IMAGE).strip():
         _docker("build", "-q", "-t", COBOL_IMAGE, "-f", str(DOCKERFILE), str(common.CASES), timeout=1800)
-    state = _docker("inspect", "-f", "{{.State.Running}}", CONTAINER, check=False).strip()
-    if state == "true":
-        return
-    if not _docker("network", "ls", "-q", "-f", f"name=^{NETWORK}$").strip():
-        _docker("network", "create", NETWORK)
-    if state == "false":
-        _docker("start", CONTAINER)
-    else:
-        _docker("run", "-d", "--name", CONTAINER, "--network", NETWORK, "--privileged", "-p", f"127.0.0.1:{PORT}:50000",
-                "-e", "LICENSE=accept", "-e", f"DB2INSTANCE={USER}", "-e", f"DB2INST1_PASSWORD={PASSWORD}",
-                "-e", f"DBNAME={DATABASE0}", "-e", "BLU=false", "-e", "ENABLE_ORACLE_COMPATIBILITY=false",
-                "-e", "UPDATEAVAIL=NO", "-e", "TO_CREATE_SAMPLEDB=false", "-e", "REPODB=false",
-                "-e", "IS_OSXFS=false", "-e", "PERSISTENT_HOME=false", "-e", "HADR_ENABLED=false", IMAGE)  # fmt: skip
-    for _ in range(120):  # first start: several minutes
-        if "Setup has completed" in _docker("logs", CONTAINER, check=False) and _clp("VALUES 1", check=False)[0] == 0:
+    import fcntl
+
+    # #4733: several processes start together on a fresh container (CI): one starts it, the others wait for it
+    lock = Path.home() / ".cache" / "gitgalaxy-db2-start.lock"
+    lock.parent.mkdir(parents=True, exist_ok=True)
+    with lock.open("w") as fh:
+        fcntl.flock(fh, fcntl.LOCK_EX)
+        state = _docker("inspect", "-f", "{{.State.Running}}", CONTAINER, check=False).strip()
+        if state in ("true", "false"):
+            _check_container_image()
+        if not _docker("network", "ls", "-q", "-f", f"name=^{NETWORK}$").strip():
+            _docker("network", "create", NETWORK)
+        if state == "false":
+            _docker("start", CONTAINER)
+        elif state != "true":
+            _docker("run", "-d", "--name", CONTAINER, "--network", NETWORK, "--privileged", "-p", f"127.0.0.1:{PORT}:50000",
+                    "-e", "LICENSE=accept", "-e", f"DB2INSTANCE={USER}", "-e", f"DB2INST1_PASSWORD={PASSWORD}",
+                    "-e", f"DBNAME={DATABASE0}", "-e", "BLU=false", "-e", "ENABLE_ORACLE_COMPATIBILITY=false",
+                    "-e", "UPDATEAVAIL=NO", "-e", "TO_CREATE_SAMPLEDB=false", "-e", "REPODB=false",
+                    "-e", "IS_OSXFS=false", "-e", "PERSISTENT_HOME=false", "-e", "HADR_ENABLED=false", IMAGE)  # fmt: skip
+    for _ in range(120):  # first start: several minutes. Ready = set up, and GGDB (the pool's reference) exists
+        if "Setup has completed" in _docker("logs", CONTAINER, check=False) and "Database code set" in _db_cfg(
+            DATABASE0
+        ):
             return
         time.sleep(10)
     raise RuntimeError("Db2 did not come up")
 
 
+def pinned_by_digest(image: str = IMAGE) -> bool:
+    """#4733: the Db2 image is named by content digest (`name@sha256:<64 hex>`), not by a tag that can move."""
+    return re.fullmatch(r"[a-z0-9][a-z0-9./_-]*@sha256:[0-9a-f]{64}", image) is not None
+
+
+def _check_container_image() -> None:
+    """An existing container must have been made from the pinned image: one left from `:latest` of another day would
+    prove against a Db2 the pin does not name. Compared by image id, when the pinned image is on this host."""
+    want = _docker("image", "inspect", "-f", "{{.Id}}", IMAGE, check=False).strip()
+    have = _docker("inspect", "-f", "{{.Image}}", CONTAINER, check=False).strip()
+    if want.startswith("sha256:") and have.startswith("sha256:") and want != have:
+        raise RuntimeError(f"Db2: container {CONTAINER} was made from another image ({have[:19]}) than the pin "
+                           f"{IMAGE} ({want[:19]}); `docker rm -f {CONTAINER}` and run again (#4733)")  # fmt: skip
+
+
 def _clp(script: str, check: bool = True) -> tuple[int, str]:
     """Run SQL (`;`-terminated statements) with Db2's command line processor, connected to the case database."""
     proc = subprocess.run(["docker", "exec", "-i", CONTAINER, "su", "-", USER, "-c",  # noqa: S603, S607

tests/tools/equivalence_oracle.py

@@ -119,7 +119,12 @@ def image_for(case: dict[str, Any]) -> str:
 
 def for_case(case: dict[str, Any]) -> Optional[dict[str, Any]]:
     """What a proof report records under `oracle` for `case` (checked against the pin)."""
-    return checked(image_for(case))
+    fp = checked(image_for(case))
+    if fp is not None and case.get("db2"):  # #4733: a Db2 case's oracle is also the Db2 it ran SQL on
+        import equivalence_db2  # noqa: PLC0415
+
+        fp = {**fp, "db2": {"image": equivalence_db2.IMAGE, "pinned_by_digest": equivalence_db2.pinned_by_digest()}}
+    return fp
 
 
 def adopt_from(case: dict[str, Any], earlier: Path) -> None:

tests/tools/evidence.py

@@ -983,6 +983,7 @@ def main(argv: Optional[list[str]] = None) -> int:
         p.add_argument("--crucible", type=Path, help="the cics-crucible checkout (at the pin)")
         p.add_argument("--offline", action="store_true")
         p.add_argument("--skip-db2", action="store_true", help="leave the Db2 cases (they need a Db2 container)")
+        p.add_argument("--only-db2", action="store_true", help="only the Db2 cases (#4733: Evidence Refresh's Db2 job)")
     sub.add_parser("mutation")
     r = sub.add_parser("render")
     r.add_argument("--check", action="store_true", help="exit 1 when a page is not current")
@@ -1008,12 +1009,16 @@ def main(argv: Optional[list[str]] = None) -> int:
             return 1
         return 0
     if args.cmd in ("prove", "refresh"):
+        if args.only_db2 and args.skip_db2:
+            ap.error("--only-db2 and --skip-db2 leave out each other's cases")
         work = (args.work or Path(tempfile.mkdtemp(prefix="evidence_"))).resolve()
         failed = 0
         for t in _pick(args.keys, args.all):
             if args.skip_db2 and t.db2:
                 print(f"{t.key}: skipped (Db2)")
                 continue
+            if args.only_db2 and not t.db2:
+                continue
             if args.cmd == "refresh" and args.stale:
                 st = status(load(t), t)
                 if st["status"] != "no-record" and not st["stale"]:

tests/tools/proof_sweep.py

@@ -17,6 +17,7 @@
 tests/tools/det_coverage_ledger.py, #4270). Every det sweep also fails when a case's proof reached an outcome
 tests/equivalence/infeasible_outcomes.json states no input can reach (tests/tools/infeasible_outcomes.py, #4602).
 --skip-db2 leaves out the cases with a "db2" section (IBM's Db2 container is slow to start): CI's det-sweep workflow (#4463).
+--db2-only is the other half (#4733): only those cases, which the Db2 job of Evidence Refresh sweeps; `--aggregate --expect db2` expects them.
 """
 
 from __future__ import annotations
@@ -115,9 +116,13 @@ def shard_cases(
     return split_cases(cases, n, durations)[i - 1]
 
 
-def det_cases(skip_db2: bool = False) -> list[str]:
-    """Every case a det sweep proves."""
-    return sorted(p.parent.name for p in CASES.glob("*/case.json") if not (skip_db2 and is_db2(p.parent.name)))
+def det_cases(skip_db2: bool = False, only_db2: bool = False) -> list[str]:
+    """Every case a det sweep proves (only the Db2 cases with `only_db2`, #4733)."""
+    return sorted(
+        p.parent.name
+        for p in CASES.glob("*/case.json")
+        if not (skip_db2 and is_db2(p.parent.name)) and (not only_db2 or is_db2(p.parent.name))
+    )
 
 
 def merge_sweeps(dirs: list[Path]) -> dict[str, dict[str, dict]]:
@@ -228,6 +233,8 @@ def main() -> int:
     which.add_argument("--det-only", action="store_true")
     which.add_argument("--model-only", action="store_true")
     ap.add_argument("--skip-db2", action="store_true", help='leave out the cases with a "db2" section')
+    ap.add_argument("--db2-only", action="store_true",
+                    help="only the cases with a \"db2\" section (#4733: Evidence Refresh's Db2 job; the det-sweep workflow uses --skip-db2)")  # fmt: skip
     ap.add_argument("--shard", type=parse_shard, metavar="I/N", help="only the I-th of N balanced slices of the cases")
     ap.add_argument("--cases", help="only these cases (comma-separated; default all)")
     ap.add_argument("--no-ratchet", action="store_true",
@@ -240,18 +247,22 @@ def main() -> int:
         help="no sweep: merge these shards' sweep.json (directories or files) and apply the ratchet",
     )
     ap.add_argument("--expect", default="all", help="with --aggregate: the cases that must be in the merge: all "
-                    "(every det case bar Db2's, as --skip-db2) | none | NAME,NAME")  # fmt: skip
+                    "(every det case bar Db2's, as --skip-db2) | db2 (every Db2 case, as --db2-only) | none | NAME,NAME")  # fmt: skip
     ap.add_argument(
         "--update-baseline", action="store_true", help="after the sweep, drop the baseline entries that now prove"
     )
     ap.add_argument("--update-durations", action="store_true",
                     help="after the sweep, record each det case's seconds in det_sweep_durations.json (what --shard balances by)")  # fmt: skip
     args = ap.parse_args()
+    if args.db2_only and args.skip_db2:
+        ap.error("--db2-only and --skip-db2 leave out each other's cases")
     if args.aggregate:
         results = merge_sweeps(args.aggregate)
         expect = (
             det_cases(skip_db2=True)
             if args.expect == "all"
+            else det_cases(only_db2=True)
+            if args.expect == "db2"
             else []
             if args.expect == "none"
             else args.expect.split(",")
@@ -265,7 +276,7 @@ def main() -> int:
     if only and (unknown := sorted(set(only) - known)):
         ap.error(f"--cases: no such case {', '.join(unknown)}")
     if args.shard:  # the slice of what this sweep would prove, balanced by what each case took last time
-        det_all = [c for c in det_cases(args.skip_db2) if only is None or c in only]
+        det_all = [c for c in det_cases(args.skip_db2, args.db2_only) if only is None or c in only]
         model_all = [p.parent.name for p in CASES.glob("*/port") if (only is None or p.parent.name in only)
                      and not (args.skip_db2 and is_db2(p.parent.name))]  # fmt: skip
         det_only = shard_cases(det_all, args.shard, load_durations())
@@ -274,6 +285,8 @@ def main() -> int:
         det_only, model_only = only, only
     results: dict[str, dict[str, dict]] = {}
     if not args.model_only:
+        if args.db2_only:  # whatever --cases says, only Db2 cases
+            det_only = [c for c in (det_only if det_only is not None else det_cases()) if is_db2(c)]
         results["det"] = det_sweep(args.work / "det", args.jobs, args.faults, args.skip_db2, det_only)
     if not args.det_only:
         (args.work / "model").mkdir(exist_ok=True)