1#!/usr/bin/env python32"""drop_block_probe.py -- replay a captured conversation against the Messages API with the3preserved thinking controls turned on, and report what the API drops and why.45Dependency-free (Python 3.8+, urllib only). Credentials come from the environment and are never6printed: ANTHROPIC_API_KEY (sent as x-api-key) or ANTHROPIC_AUTH_TOKEN (sent as7"Authorization: Bearer ..."; for example the short-lived token an OAuth login provides). When both8are set only the token is sent -- the API rejects a request carrying both headers. A bearer token9also gets the oauth-2025-04-20 beta value, which the API requires for OAuth and workload-identity10tokens.1112 python3 drop_block_probe.py capture.jsonl --yes # replay, mode drop_block (without --yes: plan only)13 python3 drop_block_probe.py capture.jsonl --yes --mode error # the loud arm (expect 400s)14
295 """Return (shaped body, skip reason or None, notes). Only fields OUTSIDE the compared prefix are touched:
296 the model (on request), the thinking configuration, block_binding, max_tokens, stream, tool_choice and
297 the sampling parameters a thinking request rejects. system, tools and messages go out exactly as captured."""
298 notes = []
299 b = json.loads(json.dumps(body)) # deep copy
300 if model:
301 b["model"] = model
302 th = b.get("thinking")
303 if not isinstance(th, dict):
304 # On models with preserved thinking a request with no thinking key already runs with adaptive thinking,
305 # and the thinking configuration is outside the compared prefix, so adding it changes no verdict.
306 th = {"type": "adaptive"}
307 notes.append("no thinking configuration in the capture: sent with {type: adaptive} (outside the compared prefix)")
308 if th.get("type") == "disabled":
309 return None, "thinking is disabled in this request; nothing to check", notes
310 bb = th.get("block_binding")
311 if th.get("type") == "enabled":
312 # The budget form is rejected by models that only take adaptive thinking; adaptive is accepted everywhere the
313 # check runs, and the configuration is outside the compared prefix.
314 th = {"type": "adaptive"}
315 notes.append("thinking {type: enabled} rewritten to {type: adaptive} (budget_tokens dropped; outside the compared prefix)")
316 if th.get("type") == "between_tools":
317 # block_binding is a 400 with this thinking type (it is accepted only with adaptive), and the thinking
318 # configuration is outside the compared prefix, so the replay goes out as adaptive.
319 th = {"type": "adaptive"}
320 notes.append("thinking {type: between_tools} rewritten to {type: adaptive} (block_binding is a 400 with between_tools; outside the compared prefix)")
321 if not isinstance(bb, dict):
322 bb = {}
323 removed = [k for k in bb if k != "prefix_mismatch_behavior"]
664 msg, transformations = parse_body(status, raw, args.stream and not args.count_tokens)
665 # A 200 with no input_transformations key was never checked (the count endpoint aside,
666 # which carries none by design): reading it as clean would be a false pass, so the turn
667 # is not evaluated and the conversation goes inconclusive.
668 missing_transformations = transformations is None and status == 200 and not args.count_tokens
669 if transformations is None:
670 transformations = []
671 if not args.quiet and shaping_notes:
672 for n in shaping_notes:
673 print(" shaped: " + n)
674 prefix_drops = [t for t in transformations if t.get("type") == DROPPED_ENTRY["type"] and t.get("reason") == "prefix_binding_mismatch"]
675 # A model_binding_mismatch drop comes from the separate model check (the conversation switched to a
676 # model that cannot read the block). It is not a prefix edit, so it is counted apart from the prefix
677 # breaks this script measures; it is still reported per turn and per conversation.
678 model_drops = [t for t in transformations if t.get("type") == DROPPED_ENTRY["type"] and t.get("reason") == "model_binding_mismatch"]
679 # A thinking_mismatch_allowed entry on a replay means thinking.block_binding never reached the
680 # check (the probe sets it on every request; a request that sets it never receives one): the check
681 # ran record-only, so the turn measured nothing and reading it as clean would be a false pass.
682 allowed_entries = [t for t in transformations if t.get("type") == ALLOWED_ENTRY["type"]]
683 # An organization_binding_mismatch drop means another account produced the block. It is removed before
684 # the prefix check judges it, so a turn with only these measured nothing: reading it as clean would be
685 # a false pass (the usual cause is a replay key from a different account than the capture's).
686 account_drops = [t for t in transformations if t.get("type") == DROPPED_ENTRY["type"] and t.get("reason") == "organization_binding_mismatch"]
687 other_drops = [t for t in transformations if t not in prefix_drops and t not in model_drops and t not in allowed_entries and t not in account_drops]
688 keys = {}
689 for t in prefix_drops:
690 keys[t.get("path")] = block_key(resolve_block(shaped, t.get("path") or ""), t.get("path") or "")
763 "usage": {k: usage.get(k) for k in ("input_tokens", "output_tokens", "cache_read_input_tokens", "cache_creation_input_tokens")} if isinstance(usage, dict) else None,
764 }
765 record(rec)
766 if not args.quiet:
767 print_turn(rec, error_text)
768 if status == 400 and args.mode == "error" and not args.continue_after_error:
769 if not args.quiet:
770 print(" (error mode: stopping this conversation at the first 400; --continue-after-error to keep going)")
771 break
772 sent = [r for r in results if "evaluated" in r]
780 "replayed_thinking_blocks_max": max([r["replayed_thinking_blocks"] for r in results] or [0]),
781 "statuses": sorted(set(str(r["status"]) for r in sent)),
782 "unevaluated": [{"turn": t, "status": s, "error": e} for t, s, e in unevaluated],
783 "inconclusive": bool(unevaluated) or not sent,
784 "_results": sent}
785 return results, summary
786
787
788def print_turn(r, full_error=None):
789 if not r.get("evaluated"):
790 flag = "N/A " if r["status"] is None else "HTTP %s" % r["status"]
791 else:
792 if r["prefix_drops"] or (r["status"] == 400 and r["error"] and ("different conversation" in r["error"] or "cannot be modified" in r["error"] or ("thinking" in r["error"] and "modified" in r["error"]))):
793 flag = "BREAK"
794 elif r["replayed_thinking_blocks"] == 0:
795 flag = "none " # nothing replayed: this request could not fail the check
796 elif r.get("other_drops"):
797 flag = "WARN " # a drop reason this probe does not classify: read the entries
798 elif r.get("model_drops"):
799 flag = "model" # only model-check drops: reasoning lost by routing, not a prefix break
804 line += " dropped=%d new_blocks=%d new_paths=%s" % (r["prefix_drops"], r["new_dropped_blocks"], r["new_dropped_paths"] or "[]")
805 if r.get("model_drops"):
806 line += " model_drops=%d paths=%s (model_binding_mismatch: this model cannot read those blocks -- a model switch, not a prefix edit)" % (r["model_drops"], r["model_dropped_paths"])
807 if r["other_drops"]:
808 line += " other=%s" % [(t.get("type"), t.get("reason")) for t in r["other_drops"]]
809 if r.get("account_drops"):
810 line += " account_drops=%d paths=%s" % (r["account_drops"], r["account_dropped_paths"])
840 s["replayed_thinking_blocks_max"], ",".join(s["patterns"]) or "-", s["statuses"],
841 (" models=%s" % ",".join(s["models"])) if s.get("models") else "",
842 (" model_drop_turns=%s" % s["model_drop_turns"]) if s.get("model_drop_turns") else "",
843 " INCONCLUSIVE" if s["inconclusive"] else ""))
844 conclusive = [s for s in summaries if not s["inconclusive"]]
845 nothing_sent = [s for s in summaries if s["inconclusive"] and not s["unevaluated"]]
846 n = len(conclusive)
847 broken = sum(1 for s in conclusive if s["first_break_turn"] is not None)
848 never_replayed = sum(1 for s in conclusive if s["replayed_thinking_blocks_max"] == 0)
849 lost = sum(s["distinct_dropped_blocks"] for s in conclusive)
850 unevald = [u for s in summaries for u in s["unevaluated"]]
851 print("\n%d conversation(s) evaluated: %d with a prefix break (%.0f%%); %d distinct thinking block(s) dropped in total (turns of reasoning lost)" % (n, broken, 100.0 * broken / n if n else 0, lost))
852 unclassified = sum(1 for s in summaries for r in s.get("_results", []) if r.get("other_drops"))
853 if unclassified:
854 print("%d request(s) reported a drop reason this probe does not classify -- read their input_transformations entries" % unclassified)
855 model_switched = sum(1 for s in conclusive if s.get("model_drop_turns"))
856 if model_switched:
857 print("%d conversation(s) also had model_binding_mismatch drops (a model that cannot read earlier blocks) -- counted apart from prefix breaks; see each turn's model_drops" % model_switched)
858 if unevald:
859 by = {}
860 for u in unevald:
861 by[u["status"]] = by.get(u["status"], 0) + 1
862 print("!! %d request(s) not evaluated (%s) -- %d conversation(s) marked inconclusive and left out of the share" % (
863 len(unevald), ", ".join("HTTP %s x%d" % (k, v) if k is not None else "network error x%d" % v for k, v in by.items()),
864 len(summaries) - n))
865 if nothing_sent:
866 print("!! %d conversation(s) had no request to send (every request was skipped: thinking disabled, or no replayed thinking) -- inconclusive, left out of the share" % len(nothing_sent))
867 if never_replayed:
868 print("!! %d conversation(s) never replayed a thinking block -- a clean result on those proves nothing about binding" % never_replayed)
869
870
871def self_test(args, api_key, betas):
872 """Three tiny live requests: one mint, then two replays of its thinking block - an honest replay
873 (expect []) and an edited replay (expect a drop)."""
874 if not args.model:
875 raise SystemExit("--self-test needs --model <id of a model with preserved thinking>")
876 q1 = "What is 17*23? Think it through, then answer with just the number."
903 print(" %s: %s" % (PREFIX_HEADER, hdr or "(absent)"))
904 err = ((msg.get("error") or {}).get("message") or "") if (status != 200 and isinstance(msg, dict)) else ""
905 if err:
906 print(" error: %s" % err[:400])
907 print(" request %s, %.1fs" % (headers.get("request-id") or headers.get("x-request-id"), secs))
908 outcomes[name] = (status, tr, err)
909 h_status, h_tr, _ = outcomes["honest replay"]
910 e_status, e_tr, e_err = outcomes["edited first message"]
911 honest_ok = h_status == 200 and h_tr == []
912 if args.mode == "error":
913 edited_ok = e_status == 400 and "bound to a different conversation" in e_err
914 else:
915 edited_ok = e_status == 200 and any(t.get("type") == DROPPED_ENTRY["type"] and t.get("reason") == "prefix_binding_mismatch" for t in (e_tr or []))
916 if honest_ok and edited_ok:
917 print("WIRED: the honest replay reported no drop and the edited replay was caught (%s)." % ("a 400, as the loud arm should" if args.mode == "error" else "reported as a dropped block"))
918 return 0
919 why = []
920 if not honest_ok:
921 if h_status == 200 and h_tr is None:
922 why.append("the honest replay's 200 carried no input_transformations key at all -- the beta header did not"
923 " reach the API (a gateway or proxy stripping anthropic-beta, or a surface without the controls)")
924 else:
925 why.append("the honest replay did not come back 200 with an empty input_transformations (something in the path already edits the history)")
926 if not edited_ok:
927 if e_status == 200 and any(t.get("type") == ALLOWED_ENTRY["type"] for t in (e_tr or [])):
928 why.append("the edited replay came back 200 with thinking_mismatch_allowed entries although the request set"
929 " thinking.block_binding: anthropic-beta reached the API but the field did not (a gateway or"
930 " proxy rebuilding the body and dropping thinking.block_binding), so the check records instead"
931 " of enforcing")
932 else:
933 why.append("the edited replay was not caught (wrong model id, the header missing, a platform without the controls, or the field misspelled)")
934 print("NOT WIRED: " + "; ".join(why))
935 return 1
936
937
938def main(argv=None):
939 ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
940 ap.add_argument("capture", nargs="?", help="capture.jsonl or a directory of *.jsonl (one conversation each)")
949 ap.add_argument("--max-tokens", type=int, default=16, help="max_tokens on every replayed request (0 = keep the capture's); the verdict is decided before the first output token")
950 ap.add_argument("--count-tokens", action="store_true", help="replay against /v1/messages/count_tokens instead of /v1/messages: free, runs the same conversation check; implies --mode error because a 400 is its only break signal (no input_transformations on a 200, so it cannot count dropped blocks)")
951 ap.add_argument("--yes", action="store_true", help="send the replay; without it the plan is printed and nothing is sent (same as --dry-run), so a live run is always an explicit decision")
952 ap.add_argument("--include-no-thinking", action="store_true", help="also send requests that replay no thinking block (skipped by default: they cannot fail the check and only add cost)")
953 ap.add_argument("--retries", type=int, default=3, help="retries of the same request on 408/409/429/529/5xx (x-should-retry: false is honoured)")
956 ap.add_argument("--json", help="write full per-request results to this file")
957 ap.add_argument("--dry-run", action="store_true", help="validate the capture (structure, signatures, consecutive-pair diff via prefix_diff.py) and print the plan with an input-token estimate; send nothing")
958 ap.add_argument("--self-test", action="store_true", help="three tiny live requests to check the wiring (needs --model and --yes); exits 1 if the check is not running")
998 if not skip and count_thinking(b) == 0 and not args.include_no_thinking:
999 skip = "replays no thinking block"
1000 plan.append((label, b, skip, notes))
1001 n_send = sum(1 for _l, _b, skip, _n in plan if not skip)
1002 n_tokens_send = sum(estimate_input_tokens(b) for _l, b, skip, _n in plan if not skip)
1003 shaping = {}
1004 for _l, _b, skip, notes in plan:
1005 for n in notes:
1006 shaping[n] = shaping.get(n, 0) + 1
1007 print("capture: %d conversation(s), %d request(s) of which %d replay thinking and will be sent, %d replayed thinking block(s) in total; mode=%s; betas=%s; base_url=%s"
1009 capture_models = sorted({b.get("model") for v in convs.values() for _l, b, _h in v if b.get("model")})
1010 print("model(s) in the capture: %s%s -- a block is judged by the model that minted it, so a capture whose blocks were"
1011 " minted by a model without preserved thinking cannot show a prefix break (see 'Suspect the test slice' in the guide)"
1012 % (", ".join(capture_models) or "(none)", ("; every request replayed as %s" % args.model) if args.model else ""))
1013 print("estimated input tokens for the %d request(s) to be sent: about %s (JSON bytes / 4, images excluded; all %d requests would be about %s; output capped at max_tokens=%d) -- price it at the model's current input rate for the approval"
1019 skipped_plan = [(l, s) for l, _b, s, _n in plan if s]
1020 if skipped_plan and not args.quiet:
1021 print(" %d request(s) will be skipped: %s" % (len(skipped_plan), "; ".join("%s (%s)" % ls for ls in skipped_plan[:8]) + (" ..." if len(skipped_plan) > 8 else "")))
1022 if n_think == 0:
1023 print("!! no request in this capture replays a thinking block: the prefix check has nothing to verify. "
1024 "Capture turns AFTER the first assistant reply, with its thinking blocks included.")
1025 if args.dry_run or not args.yes:
1026 print("dry run: nothing sent." if args.dry_run else "nothing sent: re-run with --yes to send the replay above (it spends real money).")