fix(qa): harness reliability — 4 gate-fidelity + #623 recovery fixes

qa/assert_behavioral.py

@@ -520,9 +520,30 @@ def _has_spells(c: dict) -> bool:
             f"(advance_time / travel_to(advance_time=True) / long_rest)")
         locs = state.get("locations", {}) or {}
         visited = sum(1 for l in locs.values() if isinstance(l, dict) and l.get("visited"))
-        chk("party_traveled", visited >= 2,
+        # IN-PLACE PROGRESSION EXCEPTION (#623 false-cap): a multi-beat arc that genuinely
+        # RESOLVED in a single location (e.g. a 9-beat tavern negotiation) is a SUCCESS, not a
+        # frozen stall — but the bare `visited >= 2` rule false-REDs it (RED-capping every lens
+        # ≤ 2.5). Distinguish the two with signals already in scope: a COMPLETE single-scene
+        # drama advanced the clock AND resolved its arc; a FROZEN opening did neither. The AND
+        # keeps a frozen stall RED (day==1/morning ⇒ clock_advanced False, no completed quest ⇒
+        # arc_resolved False — it fails ≥2 conjuncts). Deliberately NOT broadened to clock-only
+        # or beats-only.
+        clock_advanced = day > 1 or (tod not in ("", "morning"))
+        # arc_resolved requires an ACTUAL completed quest in the snapshot — NOT the status-blind
+        # quest_resolved tool-count (coverage_from_tool_counts counts set_quest_status(...,"active"
+        # /"failed") too, which would let a FROZEN DM game this exception with one cheap call:
+        # advance_time + set_quest_status(status="active") on a dead scene. Adversarial-verified.
+        arc_resolved = any(
+            isinstance(q, dict) and q.get("status") == "completed" for q in quest_iter)
+        SINGLE_SCENE_MIN_BEATS = 8  # strictly above MIN_BEATS(6): a real arc, not a smoke test
+        in_place_progression = (visited >= 1 and clock_advanced and arc_resolved
+                                and session_beats >= SINGLE_SCENE_MIN_BEATS)
+        chk("party_traveled", visited >= 2 or in_place_progression,
             f"visited {visited}/{len(locs)} location(s) after {session_beats} beats — the party never "
-            f"left the opening scene (travel_to / add_location make_current=True)")
+            f"left the opening scene (travel_to / add_location make_current=True); "
+            f"in-place-progression exception NOT met "
+            f"(clock_advanced={clock_advanced} arc_resolved={arc_resolved} "
+            f"beats>={SINGLE_SCENE_MIN_BEATS}? {session_beats >= SINGLE_SCENE_MIN_BEATS})")
         # WARN (the metric is softer): did the world gain/engage faces, or just sit in the seed?
         npcs_met = sum(1 for c in chars.values()
                        if isinstance(c, dict) and c.get("kind") == "npc" and c.get("met"))

qa/dm_beat_mark.py

@@ -118,11 +118,18 @@ def _is_new_prose(row):
     return True
 
 
-def cmd_mark(state_dir, mark_file):
+def cmd_mark(state_dir, mark_file, first=None):
     snap = _snapshot_path(state_dir)
     log_path = _session_log_path(snap) if snap else ""
     lines = _line_count(log_path) if log_path and os.path.isfile(log_path) else 0
     payload = {"session": os.path.abspath(log_path) if log_path else "", "lines": lines}
+    # FIX 2(a) (#623): record the beat's first/cold-open signal so cmd_check can tell a TRUE
+    # cold open (first=1: no prior session legitimately existed) from a CONTINUING beat whose
+    # mark came back EMPTY (first=0: a mark-write bug — the baseline was lost, so any "later"
+    # prose the #357 fallback recovers is the PREVIOUS beat's, recycled). Only "0"/"1" are
+    # recorded; anything else (or absent) leaves the legacy fail-open behavior in cmd_check.
+    if first in ("0", "1"):
+        payload["first"] = first
     with open(mark_file, "w", encoding="utf-8") as f:
         json.dump(payload, f)
     return 0
@@ -134,6 +141,7 @@ def cmd_check(state_dir, mark_file):
             mark = json.load(f)
         marked_session = str(mark.get("session") or "")
         marked_lines = int(mark.get("lines") or 0)
+        marked_first = str(mark.get("first") or "")
     except Exception:
         return 0  # unreadable mark -> fail OPEN (assume genuine; legacy behavior)
     snap = _snapshot_path(state_dir)
@@ -142,6 +150,15 @@ def cmd_check(state_dir, mark_file):
     cur = _session_log_path(snap)
     if not cur or not os.path.isfile(cur):
         return 1
+    # FIX 2(a) (#623): an EMPTY marked_session on a CONTINUING beat (first=0) is a mark-write
+    # bug — no baseline was captured, so scanning from line 0 would match the PREVIOUS beat's
+    # prose as "new" and stamp a recycled (dead) beat fallback_recovered:true. Force-fail (1 =
+    # NOT genuine) here. We require BOTH an empty mark AND the recorded first=0 signal so a TRUE
+    # first-prose-then-die cold open (first=1, where no session legitimately existed at mark
+    # time) is NOT wrongly failed — it keeps the legacy scan-from-0 path below. A mark WITHOUT a
+    # recorded first signal (legacy/external callers) also keeps the legacy fail-open path.
+    if not marked_session and cur and marked_first == "0":
+        return 1
     # A DIFFERENT session file than the marked one (the beat started a new session, or no
     # session existed at mark time) means every row in it is new — scan from line 0.
     skip = marked_lines if os.path.abspath(cur) == marked_session else 0
@@ -166,10 +183,14 @@ def cmd_check(state_dir, mark_file):
 
 def main(argv):
     if len(argv) < 4 or argv[1] not in ("mark", "check"):
-        print("usage: dm_beat_mark.py mark|check <state_dir> <mark_file>", file=sys.stderr)
+        print("usage: dm_beat_mark.py mark|check <state_dir> <mark_file> [first]", file=sys.stderr)
         return 0  # never fail a beat over a usage error
     try:
-        return (cmd_mark if argv[1] == "mark" else cmd_check)(argv[2], argv[3])
+        if argv[1] == "mark":
+            # FIX 2(a) (#623): optional 5th arg = the beat's first/cold-open signal ("1"|"0").
+            first = argv[4] if len(argv) > 4 else None
+            return cmd_mark(argv[2], argv[3], first)
+        return cmd_check(argv[2], argv[3])
     except Exception:
         return 0  # any internal failure fails OPEN
 

qa/lib_beat_driver.sh

@@ -433,8 +433,15 @@ clawdnd_dm_final_text() {
 # beat's. Best-effort (never fails a beat); standalone python — no heredoc-in-$() (bash 3.2).
 clawdnd_dm_prebeat_mark() {
   local state_dir="$1"
+  # FIX 2(a) (#623): pass the beat's first/cold-open signal ("1"=cold open, "0"=continuing) so
+  # dm_beat_mark.py can force-fail a CONTINUING beat whose mark came back empty (a mark-write bug
+  # that else recycles the previous beat's prose). Default "1" (treat as cold open / fail-open)
+  # when the caller did not pass it — never tightens an unknown caller's behavior.
+  local first="${2:-1}"
   local mark_py="${CLAWDND_LIB_DIR:-$(dirname "${BASH_SOURCE[0]}")}/dm_beat_mark.py"
-  python3 "$mark_py" mark "$state_dir" "$state_dir/.dm_prebeat_mark" 2>/dev/null || true
+  # Drop the 2>/dev/null swallow (FIX 2(a)): an empty/failed mark must surface on stderr so a
+  # mark-write bug is visible in the beat log instead of silently producing a recycled beat.
+  python3 "$mark_py" mark "$state_dir" "$state_dir/.dm_prebeat_mark" "$first" || true
   return 0
 }
 

qa/release_readiness.py

@@ -817,6 +817,13 @@ def main() -> int:
             "run_build_sha": rj.get("build_sha") or "",
             "part_b_result": (rj.get("part_b") or {}).get("persona_loop") or "n/a",
             "part_b_score_pass": bool((rj.get("part_b") or {}).get("score_pass")),
+            # FIX 1 (#623 false-cap): a NON-quota player_rc!=0 CRASH is INCONCLUSIVE evidence
+            # (re-measure), not a product-quality fail. ui_playtest_app.sh stamps this true on
+            # such a crash. We EXCLUDE these personas from score_pass_failed_personas / the
+            # cross_persona_sat quality gate and instead surface them as an evidence gap (RED-cap
+            # as harness_contaminated). The discriminator is the PROCESS exit, never the score —
+            # a played low-score run exits rc=0, harness_error=False, and stays a clean quality RED.
+            "part_b_harness_error": bool((rj.get("part_b") or {}).get("harness_error")),
             "part_a_result": (rj.get("part_a") or {}).get("result") or "n/a",
             "part_a_failure_bucket": (rj.get("part_a") or {}).get("failure_bucket") or "",
             "part_a_failure_detail": (rj.get("part_a") or {}).get("failure_detail") or "",
@@ -835,7 +842,13 @@ def main() -> int:
     avg_sat = sum(sats) / len(sats) if sats else 0.0
     any_gave_up = any(p["gave_up"] for p in persona_scores)
     any_completed = any(p["completed_intro_flow"] for p in persona_scores)
-    score_pass_failed_personas = [str(p["persona"]) for p in persona_scores if not p.get("part_b_score_pass")]
+    # FIX 1 (#623 false-cap): a persona whose player process CRASHED (part_b_harness_error) is
+    # INCONCLUSIVE, not a quality fail — exclude it from the score_pass quality gate and route it
+    # through the evidence_gaps/harness_contaminated machinery below (RED-cap as a re-measure).
+    # A played low-score run has harness_error=False and STILL fails the gate (a clean quality RED).
+    harness_error_personas = [p for p in persona_scores if p.get("part_b_harness_error")]
+    score_pass_failed_personas = [str(p["persona"]) for p in persona_scores
+                                  if not p.get("part_b_score_pass") and not p.get("part_b_harness_error")]
     score_pass_complete = bool(persona_scores) and not score_pass_failed_personas
     total_critical = sum(p["critical"] for p in persona_scores)
     total_console_errors = sum(p["console_errors"] for p in persona_scores)
@@ -947,7 +960,23 @@ def main() -> int:
             "missing": f"{h['run']}/{h.get('missing') or 'score.json'}",
             "detail": f"persona={h.get('persona') or 'unknown'} detail={h.get('detail') or ''} part_a={h.get('part_a') or 'n/a'} part_b={h.get('part_b') or 'n/a'} {buckets}".strip(),
         })
-    failed_part_b = [p for p in persona_scores if p.get("part_b_result") != "PASS"]
+    # FIX 1 (#623 false-cap): a persona whose player process CRASHED (non-quota player_rc!=0) is
+    # INCONCLUSIVE — surface it as a cross_persona_sat EVIDENCE GAP (re-measure), exactly like a
+    # harness_failure, so it RED-caps as harness_contaminated rather than a quality FAIL. It is
+    # already excluded from score_pass_failed_personas (the quality gate) above.
+    for p in harness_error_personas:
+        evidence_gaps.append({
+            "gate": "cross_persona_sat",
+            "missing": f"{p['run']}/run.json part_b.harness_error",
+            "detail": f"persona={p.get('persona') or 'unknown'} part_b player process crashed (non-quota player_rc!=0) "
+                      f"— INCONCLUSIVE, re-measure (not a quality fail) "
+                      f"failure_bucket={p.get('part_b_failure_bucket') or ''} failure_detail={p.get('part_b_failure_detail') or ''}".strip(),
+        })
+    # A crashed (harness_error) persona's part_b is "FAIL", but it is INCONCLUSIVE, not a dropped
+    # product arc — exclude it from the arc_completed product-failure attribution (it's already a
+    # cross_persona_sat evidence gap above). Genuine non-PASS played runs still flow through here.
+    failed_part_b = [p for p in persona_scores
+                     if p.get("part_b_result") != "PASS" and not p.get("part_b_harness_error")]
     for p in failed_part_b:
         bucket = p.get("part_b_failure_bucket") or ""
         detail = p.get("part_b_failure_detail") or ""
@@ -1242,6 +1271,9 @@ def main() -> int:
             "arc_completed": any_completed,
             "cross_persona_satisfaction": round(avg_sat, 1),
             "score_pass_failed_personas": score_pass_failed_personas,
+            # FIX 1 (#623 false-cap): personas reclassified as INCONCLUSIVE (player-process crash,
+            # non-quota) — excluded from score_pass_failed_personas, surfaced as evidence gaps.
+            "harness_error_personas": [str(p["persona"]) for p in harness_error_personas],
             "any_gave_up": any_gave_up,
             "total_critical_bugs": total_critical,
             "total_console_errors": total_console_errors,

qa/test_assert_behavioral.py

@@ -661,3 +661,90 @@ def test_structural_completeness_silent_in_combat_sprint(tmp_path):
     rc, out = _run_gate(tmp_path, events, state, env=env)
     assert rc == 0, out
     assert "structural_completeness" not in out
+
+
+# ── FIX 4: party_traveled in-place-progression exception (#623 false-cap) ─────────
+# A multi-beat arc that RESOLVED in a single location (clock advanced + quest completed,
+# beats >= 8) is a SUCCESS — it must PASS party_traveled. A frozen opening (day 1/morning,
+# no resolved quest) must STILL fail (RED). The AND keeps the frozen stall red.
+
+def _single_scene_state(*, day, tod=None, quest_completed, visited_count=1):
+    """A single-location final state (visited_count locations visited). `day`/`tod` drive
+    clock_advanced; `quest_completed` drives arc_resolved. No companion → the structural
+    floor (>=10 beats + companion) stays silent so we isolate party_traveled."""
+    locs = {"loc_a": {"visited": True}}
+    if visited_count >= 2:
+        locs["loc_b"] = {"visited": True}
+    state = {
+        "leveling_mode": "milestone",
+        "day": day,
+        "party": ["pc1"],
+        "locations": locs,
+        "characters": {"pc1": {"name": "Dal", "kind": "player", "location_id": "loc_a"}},
+    }
+    if tod is not None:
+        state["time_of_day"] = tod
+    if quest_completed:
+        state["quests"] = {"q1": {"title": "Tavern Negotiation", "status": "completed",
+                                  "objectives": ["strike the bargain"],
+                                  "completed_objectives": ["strike the bargain"]}}
+    else:
+        state["quests"] = {"q1": {"title": "Tavern Negotiation", "status": "active",
+                                  "objectives": ["strike the bargain"],
+                                  "completed_objectives": []}}
+    return state
+
+
+def test_party_traveled_passes_single_scene_arc_that_progressed_in_place(tmp_path):
+    # visited=1, but day advanced (day 2) AND the quest completed AND beats>=8 → the in-place
+    # progression exception fires → party_traveled PASSES (a resolved one-location drama).
+    events = _dm_text_turns(9)  # session_beats=9 (>= SINGLE_SCENE_MIN_BEATS 8)
+    state = _single_scene_state(day=2, quest_completed=True, visited_count=1)
+    rc, out = _run_gate(tmp_path, events, state)
+    assert "[PASS] party_traveled" in out, out
+    assert "[FAIL] party_traveled" not in out, out
+    assert rc == 0, out  # clock advanced too, so world_advanced_time also passes
+
+
+def test_party_traveled_still_red_on_frozen_run(tmp_path):
+    # visited=1, day==1/morning (clock never moved), no completed quest, beats>=8 → the
+    # exception's AND fails (clock_advanced False, arc_resolved False) → party_traveled RED.
+    events = _dm_text_turns(9)
+    state = _single_scene_state(day=1, tod="morning", quest_completed=False, visited_count=1)
+    rc, out = _run_gate(tmp_path, events, state)
+    assert "[FAIL] party_traveled" in out, out
+    assert rc == 1, out
+
+
+def test_party_traveled_still_red_when_clock_advanced_but_arc_unresolved(tmp_path):
+    # Guard against broadening to clock-only: day advanced but NO completed quest → the AND
+    # still fails (arc_resolved False) → party_traveled stays RED.
+    events = _dm_text_turns(9)
+    state = _single_scene_state(day=3, quest_completed=False, visited_count=1)
+    rc, out = _run_gate(tmp_path, events, state)
+    assert "[FAIL] party_traveled" in out, out
+    assert rc == 1, out
+
+
+def test_party_traveled_red_despite_status_blind_quest_tool_count(tmp_path):
+    # GATE-WEAKENING REGRESSION (adversarial-verified): a FROZEN single scene where the DM
+    # advanced the clock AND called set_quest_status — the OLD status-blind quest_resolved
+    # tool-count would have flipped arc_resolved True and let this DEAD scene PASS via the
+    # in-place exception. arc_resolved now requires a snapshot quest at status=="completed";
+    # the quest stays "active" (quest_completed=False) → arc_resolved False → party_traveled
+    # stays RED even though set_quest_status was called.
+    events = _dm_text_turns(9) + _toolcall("set_quest_status")
+    state = _single_scene_state(day=3, quest_completed=False, visited_count=1)
+    rc, out = _run_gate(tmp_path, events, state)
+    assert "[FAIL] party_traveled" in out, out
+    assert rc == 1, out
+
+
+def test_party_traveled_still_red_when_arc_resolved_but_too_few_beats(tmp_path):
+    # Guard against broadening to beats-only / arc-only: clock advanced + quest completed but
+    # only 7 beats (< SINGLE_SCENE_MIN_BEATS 8) → exception not met → party_traveled RED.
+    events = _dm_text_turns(7)
+    state = _single_scene_state(day=2, quest_completed=True, visited_count=1)
+    rc, out = _run_gate(tmp_path, events, state)
+    assert "[FAIL] party_traveled" in out, out
+    assert rc == 1, out