fix verbatim-repeat false positive on auth-retry; add subagent thrash detail

engine/skills/reflect/scripts/tests/test_token_audit.py

@@ -1042,13 +1042,16 @@ class TestFrustrationSignals(unittest.TestCase):
  synthesis ranked root causes by frustration caused, not tokens burned."""
 
  def test_detects_each_signal_kind_and_counts_interruptions(self):
- usage = {"input_tokens": 1, "output_tokens": 1}
+ usage = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}
  lines = [
  claude_user_text_line("thanks, looks good", ts="2026-08-18T02:00:00Z"),
+ claude_assistant_line("m0", "u0", [{"type": "text", "text": "great"}], usage),
  claude_user_text_line("WHERE IS MY DIGITAL TWIN? WHAT THE FUCK IS GOING ON", ts="2026-08-18T02:30:00Z"),
+ claude_assistant_line("m0b", "u0b", [{"type": "text", "text": "checking"}], usage),
  claude_user_text_line("i told you to have it ready", ts="2026-08-18T02:31:00Z"),
  claude_user_text_line("am i in a zoom meeting at all???", ts="2026-08-18T02:32:00Z"),
  claude_user_text_line("please fix the audio now", ts="2026-08-18T02:33:00Z"),
+ claude_assistant_line("m0c", "u0c", [{"type": "text", "text": "working on it"}], usage),
  claude_user_text_line("please fix the audio now", ts="2026-08-18T02:35:00Z"),
  {"type": "user", "message": {"role": "user", "content": [
  {"type": "text", "text": "[Request interrupted by user]"}]}},
@@ -1151,6 +1154,55 @@ def test_repeat_outside_ten_minute_window_not_flagged(self):
  finally:
  os.unlink(path)
 
+ def test_verbatim_repeat_suppressed_when_assistant_responded_between_sends(self):
+ """Real session: user prompt hit an OAuth 401 (API error, no assistant
+ response generated), user interrupted, ran /login, and re-sent the
+ same prompt. The verbatim-repeat detector must suppress this because a
+ successful assistant turn DID NOT occur between the two sends — but
+ when the assistant DID respond between two identical sends, that is
+ still a real verbatim-repeat (the user re-sent because the response
+ was inadequate)."""
+ usage = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}
+ lines_no_assistant_between = [
+ claude_user_text_line("please search for auth tokens in the codebase", ts="2026-09-01T10:00:00Z"),
+ claude_user_text_line("please search for auth tokens in the codebase", ts="2026-09-01T10:02:00Z"),
+ claude_assistant_line("m1", "u1", [{"type": "text", "text": "ok"}], usage),
+ ]
+ path = write_jsonl(lines_no_assistant_between)
+ try:
+ with redirect_stdout(io.StringIO()):
+ result = token_audit.audit_claude(path)
+ kinds = {k for f in result["frustration"]["flagged"] for k in f["kinds"]}
+ self.assertNotIn("verbatim-repeat", kinds)
+ finally:
+ os.unlink(path)
+
+ lines_with_assistant_between = [
+ claude_user_text_line("please search for auth tokens in the codebase", ts="2026-09-01T10:00:00Z"),
+ claude_assistant_line("m1", "u1", [{"type": "text", "text": "searching now"}], usage),
+ claude_user_text_line("please search for auth tokens in the codebase", ts="2026-09-01T10:02:00Z"),
+ claude_assistant_line("m2", "u2", [{"type": "text", "text": "ok done"}], usage),
+ ]
+ path = write_jsonl(lines_with_assistant_between)
+ try:
+ with redirect_stdout(io.StringIO()):
+ result = token_audit.audit_claude(path)
+ kinds = {k for f in result["frustration"]["flagged"] for k in f["kinds"]}
+ self.assertIn("verbatim-repeat", kinds)
+ finally:
+ os.unlink(path)
+
+ def test_verbatim_repeat_still_works_for_non_claude_modes(self):
+ """Codex/OMP/Cursor don't pass assistant_turn_indices, so
+ verbatim-repeat detection must still work (no suppression)."""
+ msgs = [
+ (0, "2026-09-01T10:00:00Z", "please fix the build"),
+ (1, "2026-09-01T10:02:00Z", "please fix the build"),
+ ]
+ result = token_audit.frustration_signals(msgs, interruptions=0)
+ kinds = {k for f in result["flagged"] for k in f["kinds"]}
+ self.assertIn("verbatim-repeat", kinds)
+
  def test_agent_blame_and_same_type_must_automate(self):
  """You-messed-up is agent-blame; product 'the ui is messed up' is not.
  Two told-yous fire intervention-must-automate; one told-you does not."""
@@ -1437,6 +1489,27 @@ def test_thrash_inside_subagents_is_reported(self):
  self.assertEqual(flags["subagent-thrash"]["value"], "yes")
  self.assertEqual(flags["subagent-thrash"]["count"], 1)
 
+ def test_subagent_thrash_detail_includes_files_and_failure_signatures(self):
+ """B2: the --out JSON report must include which files were redundantly
+ read and what the recurring failure error messages were, not just
+ counts and flag names."""
+ u = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}
+ agent_c = os.path.join(self.subagents_dir, "agent-c.jsonl")
+ self._write(agent_c, [
+ _sidechain_user_line("run the tests", "c"),
+ claude_assistant_line("c1", "cu1", [{"type": "tool_use", "id": "tc1", "name": "Bash", "input": {"command": "pytest"}}], u),
+ claude_error_line("tc1", "ModuleNotFoundError: No module named 'foo'"),
+ claude_assistant_line("c2", "cu2", [{"type": "tool_use", "id": "tc2", "name": "Bash", "input": {"command": "pytest"}}], u),
+ claude_error_line("tc2", "ModuleNotFoundError: No module named 'foo'"),
+ ])
+ res, _ = self._audit()
+ thrash = res["subagents"]["thrash"]
+ self.assertIn("agent-a.jsonl:/x.py", thrash["redundant_read_files"])
+ self.assertTrue(any(
+ d["agent"] == "agent-c.jsonl" and d["tool"] == "Bash"
+ for d in thrash["recurring_failure_details"]
+ ))
+
  def test_subagent_first_user_turn_is_never_a_human_intervention(self):
  res, _ = self._audit()
  self.assertEqual(res["frustration"]["n_user_messages"], 1)

engine/skills/reflect/scripts/token_audit.py

@@ -182,16 +182,33 @@ def _is_allcaps(text):
  return sum(c.isupper() for c in letters) / len(letters) > 0.6
 
 
-def frustration_signals(user_msgs, interruptions=0):
+def _has_assistant_between(sorted_asst_indices, prev_user_idx, curr_user_idx):
+ """True if any element of sorted_asst_indices falls strictly between
+ prev_user_idx and curr_user_idx (exclusive on both ends). Uses bisect
+ for O(log n) on large sessions."""
+ import bisect
+ lo = bisect.bisect_right(sorted_asst_indices, prev_user_idx)
+ return lo < len(sorted_asst_indices) and sorted_asst_indices[lo] < curr_user_idx
+
+
+def frustration_signals(user_msgs, interruptions=0, assistant_turn_indices=None):
  """user_msgs: [(ordinal, iso_timestamp_or_None, text)] — HUMAN-authored
  messages only (never tool_results, never interruption markers).
  Returns flagged messages with their signal kinds, plus a verbatim-repeat
  check: the same normalized text re-sent within 10 minutes is the single
  strongest frustration signal (the user re-sent it because nothing visibly
  changed). `interruptions` is counted by the caller (tool-specific shape).
+
+ `assistant_turn_indices` (optional): sorted list of JSONL line indices
+ where a successful assistant response occurred. When provided and the
+ ordinals in `user_msgs` are JSONL line indices (not enumerate counters),
+ the verbatim-repeat detector suppresses a match when a successful
+ assistant turn occurred between the two sends — that means the user
+ re-sent after an API error / auth failure, not out of frustration.
  """
  flagged = []
- seen = [] # (epoch_seconds_or_None, normalized_text)
+ seen = []
+ _asst = sorted(assistant_turn_indices) if assistant_turn_indices else None
  for idx, ts, text in user_msgs:
  t = (text or "").strip()
  if not t:
@@ -205,11 +222,13 @@ def frustration_signals(user_msgs, interruptions=0):
  secs = _ts_seconds(ts)
  norm = re.sub(r"\s+", " ", t).casefold()
  if len(norm) >= 12:
- for prev_secs, prev_norm in seen:
+ for prev_secs, prev_norm, prev_idx in seen:
  if prev_norm == norm and (secs is None or prev_secs is None or 0 <= secs - prev_secs <= 600):
+ if _asst is not None and not _has_assistant_between(_asst, prev_idx, idx):
+ continue
  kinds.append("verbatim-repeat")
  break
- seen.append((secs, norm))
+ seen.append((secs, norm, idx))
  if kinds:
  flagged.append({"index": idx, "ts": ts, "kinds": sorted(set(kinds)), "excerpt": t[:100]})
  peak = None
@@ -387,8 +406,10 @@ def audit_subagents(path):
  totals = {"input": 0, "output": 0, "cache_read": 0, "cache_creation": 0, "total": 0}
  thrash = {
  "redundant_reads": 0,
+ "redundant_read_files": [],
  "tool_errors": 0,
  "recurring_failure_signatures": 0,
+ "recurring_failure_details": [],
  "longest_edit_streak_no_verify": 0,
  "self_retraction": 0,
  "by_agent": {},
@@ -410,8 +431,15 @@ def audit_subagents(path):
  if fired:
  thrash["by_agent"][fname] = fired
  thrash["redundant_reads"] += flags["redundant-reads"]["count"]
+ for fp in stats.get("redundant_read_files", []):
+ thrash["redundant_read_files"].append(f"{fname}:{fp}")
  thrash["tool_errors"] += stats["n_errors"]
  thrash["recurring_failure_signatures"] += stats["n_recurring_failures"]
+ for detail in stats.get("recurring_failure_details", []):
+ thrash["recurring_failure_details"].append(
+ {"agent": fname, "tool": detail["tool"], "signature": detail["signature"],
+ "occurrences": detail["occurrences"]}
+ )
  thrash["longest_edit_streak_no_verify"] = max(
  thrash["longest_edit_streak_no_verify"], stats["longest_edit_streak_no_verify"]
  )
@@ -480,23 +508,24 @@ def audit_claude(path, out_path=None, include_subagents=True):
  # different lines of the same message — found by e2e sample fixtures.
  msg_tool_names = {} # mid -> [tool name, ...]
  msg_output_tokens = {} # mid -> output_tokens (from first line of that message)
+ _human_utterances = transcript_provenance.direct_human_utterances(path, "claude")
  user_msgs = [
- (index, utterance.timestamp, utterance.text)
- for index, utterance in enumerate(
- transcript_provenance.direct_human_utterances(path, "claude")
- )
+ (utterance.index, utterance.timestamp, utterance.text)
+ for utterance in _human_utterances
  ]
- n_interruptions = 0 # "[Request interrupted by user" markers
- assistant_texts = [] # text blocks for self-retraction scan
+ n_interruptions = 0
+ assistant_texts = []
+ assistant_line_indices = []
 
- for d in lines:
+ for line_idx, d in enumerate(lines):
  if d.get("type") == "assistant":
  msg = d.get("message", {})
  mid = msg.get("id")
  u = msg.get("usage", {})
  is_new_msg = mid not in counted_msg_ids
  if is_new_msg:
  counted_msg_ids.add(mid)
+ assistant_line_indices.append(line_idx)
  total_input += u.get("input_tokens", 0)
  total_output += u.get("output_tokens", 0)
  total_cache_read += u.get("cache_read_input_tokens", 0)
@@ -625,7 +654,7 @@ def audit_claude(path, out_path=None, include_subagents=True):
  seen.add(c)
  spikes.append((s, c, r))
 
- frustration = frustration_signals(user_msgs, n_interruptions)
+ frustration = frustration_signals(user_msgs, n_interruptions, assistant_turn_indices=assistant_line_indices)
 
  flags = [
  _flag(
@@ -682,6 +711,12 @@ def audit_claude(path, out_path=None, include_subagents=True):
  flags.append(_subagent_thrash_flag(subagents))
  combined_total = grand + (subagents["totals"]["total"] if subagents else 0)
 
+ redundant_read_files = sorted({fp for fp, _, _, _ in redundant})
+ recurring_failure_details = [
+ {"tool": name, "signature": norm, "occurrences": len(seqs)}
+ for (name, norm), seqs in sorted(recurring.items(), key=lambda x: -len(x[1]))
+ ]
+
  result = {
  "input": total_input,
  "output": total_output,
@@ -699,6 +734,8 @@ def audit_claude(path, out_path=None, include_subagents=True):
  "self_retraction": retraction_hits,
  "subagents": subagents,
  "combined_total": combined_total,
+ "redundant_read_files": redundant_read_files,
+ "recurring_failure_details": recurring_failure_details,
  }
 
  if out_path: