rehearsal: grab_json returns a reason, manifest fallback, single-line VERDICT

build/smoke_hub_resume.py

@@ -37,22 +37,42 @@ def sh(argv, label):
         print("   |", line[:220], flush=True)
     if p.returncode != 0:
         print("   STDERR:", (p.stderr or "")[-1500:], flush=True)
-    return {"rc": p.returncode, "out": out[-6000:], "err": (p.stderr or "")[-1500:]}
+    # Keep enough of the tail that the JSON block can never be the thing that got truncated.
+    return {"rc": p.returncode, "out": out[-120000:], "err": (p.stderr or "")[-1500:]}
 
 
 def grab_json(text, begin, end):
-    if begin in text and end in text:
+    """Pull the JSON object out of the LAST marker pair. Returns (obj, why_null).
+
+    The reason matters: v4 of this rehearsal reported COLD_RESUME_PASSED=false on a pipeline that had
+    actually worked, because one status line printed inside the block made json.loads fail and the
+    helper folded that into a bare None (E-019).
+    """
+    if begin not in text:
+        return None, "begin marker absent from captured stdout"
+    span = text.rsplit(begin, 1)[1]
+    if end not in span:
+        return None, "end marker absent -- stdout truncated mid-block"
+    body = span.split(end, 1)[0]
+    try:
+        return json.loads(body), None
+    except Exception as e:
+        strict = f"{type(e).__name__}: {str(e)[:120]}"
+    i = body.find("{")
+    if i >= 0:
         try:
-            return json.loads(text.split(begin, 1)[1].split(end, 1)[0])
-        except Exception:
-            return None
-    return None
+            obj = json.JSONDecoder().raw_decode(body[i:])[0]
+            return obj, f"lenient parse used (junk after the object): {strict}"
+        except Exception as e:
+            strict = f"lenient also failed {type(e).__name__}: {str(e)[:120]}"
+    return None, strict
 
 
 def main():
     R = {"repo": REPO, "sources": TWO}
     base = ["build_mix.py", "--root", ROOT, "--only", TWO, "--cap-tokens-per-source", "1200000",
             "--hub-repo", REPO]
+    notes = []
 
     # Start from an empty stage repo. v3 of this rehearsal failed and, by design, KEPT the repo as
     # evidence; without this the next run would restore those sources and skip the staging step it is
@@ -70,7 +90,10 @@ def main():
     R["preclean"] = preclean()
     shutil.rmtree(ROOT, ignore_errors=True)
     r1 = sh(base + ["--stage-only"], "1. stage two sources and publish")
-    R["stage"] = {"rc": r1["rc"], "json": grab_json(r1["out"], "BUILD_JSON_BEGIN", "BUILD_JSON_END")}
+    sj, why = grab_json(r1["out"], "BUILD_JSON_BEGIN", "BUILD_JSON_END")
+    R["stage"] = {"rc": r1["rc"], "json": sj}
+    if why:
+        notes.append(f"stage json: {why}")
 
     # the wipe IS the test: after this, nothing local survives except this script and the code
     shutil.rmtree(ROOT, ignore_errors=True)
@@ -78,17 +101,39 @@ def main():
                   "stage_dir_exists": os.path.exists(os.path.join(ROOT, "stage"))}
 
     r2 = sh(base + ["--merge-only"], "2. cold resume: merge from the Hub with an empty local tree")
-    R["merge"] = {"rc": r2["rc"], "json": grab_json(r2["out"], "BUILD_JSON_BEGIN", "BUILD_JSON_END")}
+    mj, why = grab_json(r2["out"], "BUILD_JSON_BEGIN", "BUILD_JSON_END")
+    R["merge"] = {"rc": r2["rc"], "json": mj}
+    if why:
+        notes.append(f"merge json: {why}")
 
     r3 = sh(["verify_mix.py", "--root", ROOT, "--min-sources-per-shard", "2",
              "--sample-decodes", "3"], "3. verify the restored mix")
-    R["verify"] = grab_json(r3["out"], "VERIFY_JSON_BEGIN", "VERIFY_JSON_END")
+    vj, why = grab_json(r3["out"], "VERIFY_JSON_BEGIN", "VERIFY_JSON_END")
+    R["verify"] = vj
+    if why:
+        notes.append(f"verify json: {why}")
+
+    # The manifest on disk is a second, independent source for the same number. If the stdout parse
+    # fails but the manifest is good, the pipeline still passes and the harness is what is broken --
+    # distinguishing those two outcomes is the whole point of recording `notes`.
+    mpath = os.path.join(ROOT, "manifest.json")
+    man = None
+    if os.path.exists(mpath):
+        try:
+            man = json.load(open(mpath))
+        except Exception as e:
+            notes.append(f"manifest unreadable: {type(e).__name__}: {str(e)[:120]}")
+    else:
+        notes.append(f"no manifest at {mpath}")
+    R["manifest_total_tokens"] = (man or {}).get("total_tokens")
+    R["manifest_n_shards"] = (man or {}).get("n_shards")
 
-    staged = (R["merge"].get("json") or {}).get("total_tokens", 0)
+    staged = (mj or {}).get("total_tokens") or (man or {}).get("total_tokens") or 0
     R["COLD_RESUME_PASSED"] = bool(
         r1["rc"] == 0 and r2["rc"] == 0 and staged > 1_000_000
         and (R["verify"] or {}).get("PASS") is True
         and (R["verify"] or {}).get("distinct_sources_in_mix") == 2)
+    R["notes"] = notes
 
     # cleanup: the throwaway stage repo must not outlive the test, or a later real build could restore
     # capped 1.2M-token sources and believe they were complete
@@ -111,8 +156,12 @@ def main():
         R["cleanup"] = {"error": f"{type(e).__name__}: {str(e)[:200]}",
                         "action_needed": f"delete the throwaway dataset {REPO} manually"}
 
+    blob = json.dumps(R, default=str, sort_keys=True)
+    # A single-line verdict survives log truncation and cannot be misread the way a nested dict can.
+    print(f"VERDICT COLD_RESUME_PASSED={R['COLD_RESUME_PASSED']} staged={staged} "
+          f"cleanup={str(R['cleanup'])[:40]} notes={len(notes)}")
     print("SMOKE_JSON_BEGIN")
-    print(json.dumps(R, indent=1, default=str)[:6000])
+    print(blob[:12000])
     print("SMOKE_JSON_END")