reflect: detect Codex no-verify thrash

engine/hooks/reflect-on-thrash/tests/test_hooks.py

@@ -70,6 +70,48 @@ def no_verify_streak_path(directory: str) -> str:
     return path
 
 
+def codex_no_verify_streak_path(directory: str) -> str:
+    path = os.path.join(directory, "codex-no-verify.jsonl")
+
+    def patch_call(call_id: str) -> dict:
+        return {
+            "type": "response_item",
+            "payload": {
+                "type": "custom_tool_call",
+                "call_id": call_id,
+                "name": "exec",
+                "input": (
+                    'const patch = "*** Begin Patch\\n'
+                    "*** Update File: /repo/packages/data-store/src/__tests__/scale.test.ts\\n"
+                    "@@\\n-old\\n+new\\n*** End Patch\";\n"
+                    "text(await tools.apply_patch(patch));"
+                ),
+            },
+        }
+
+    def output(call_id: str) -> dict:
+        return {
+            "type": "response_item",
+            "payload": {
+                "type": "custom_tool_call_output",
+                "call_id": call_id,
+                "output": "Success. Updated the following files:\nM /repo/packages/data-store/src/__tests__/scale.test.ts",
+            },
+        }
+
+    lines = [
+        {"type": "response_item", "payload": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "fix the workspace test"}]}},
+    ]
+    for i in range(3):
+        call_id = f"patch-{i}"
+        lines.append(patch_call(call_id))
+        lines.append(output(call_id))
+    with open(path, "w", encoding="utf-8") as handle:
+        for line in lines:
+            handle.write(json.dumps(line) + "\n")
+    return path
+
+
 def run_claude(payload: dict):
     err = io.StringIO()
     with patch.object(sys, "stdin", io.StringIO(json.dumps(payload))):
@@ -121,6 +163,15 @@ def test_codex_fixture_is_sniffed_and_routed_through_audit_codex(self):
         self.assertIn("intervention-must-automate", joined)
         self.assertTrue(detect.intervention_hit(hits))
 
+    def test_codex_no_verify_fixture_is_routed_through_audit_codex(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = codex_no_verify_streak_path(directory)
+            self.assertEqual(detect.sniff_mode(path), "codex")
+            hits = detect.thrash_hits(path)
+            self.assertTrue(hits)
+            self.assertIn("no-verify-edit-streak", " ".join(hits))
+            self.assertFalse(detect.intervention_hit(hits))
+
     def test_codex_clean_input_is_not_thrash(self):
         lines = [
             {"timestamp": "2026-08-27T10:00:00.000Z", "type": "session_meta", "payload": {}},

engine/skills/reflect/references/cost-audit.md

@@ -16,7 +16,7 @@ python3 skills/reflect/scripts/token_audit.py cursor <path-to-agent-transcript.j
 python3 skills/reflect/scripts/token_audit.py remotes   # names only, from ~/.invoker/config.json if present
 ```
 
-Prefer `--out <path>` when feeding lenses: it writes a JSON report of named yes/no flags with rationales (supported for `claude` and `omp` modes). Codex `--out` is totals only (no thrash flags). Stdout stays a short summary (path + flag lines). Progress/errors go to stderr. Without `--out`, stdout is the full prose report (legacy; existing tests use this).
+Prefer `--out <path>` when feeding lenses: it writes a JSON report of named yes/no flags with rationales. Codex output includes human-intervention flags plus same-problem thrash flags that can be recovered from rollout tool-call history; it does not yet include Claude's redundant-read or model-tier candidates. Stdout stays a short summary (path + flag lines). Progress/errors go to stderr. Without `--out`, stdout is the full prose report (legacy; existing tests use this).
 
 It reports, per session: total tokens by category and cache-read share, turns whose only tool calls were Read/Grep/Glob (model-tier downgrade candidates), redundant re-reads of an unchanged file, tool errors, cache-creation spikes (a fresh multi-hundred-KB cache write mid-session, instead of a cache read, usually means context got dropped/rebuilt rather than genuinely new information arriving — worth checking what preceded it), and per-turn token growth (a session where each successive turn costs more than the last, because the whole growing history gets resent every turn, burns quota fast even at a high cache-hit rate — this is the main thing to check when a session "ran out" quickly).
 

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

@@ -38,6 +38,72 @@ def codex_response_item(role, text, ts=None, ptype="message"):
     return d
 
 
+def codex_tool_call(call_id, source):
+    return {
+        "type": "response_item",
+        "payload": {
+            "type": "custom_tool_call",
+            "call_id": call_id,
+            "name": "exec",
+            "input": source,
+        },
+    }
+
+
+def codex_tool_output(call_id, text):
+    return {
+        "type": "response_item",
+        "payload": {
+            "type": "custom_tool_call_output",
+            "call_id": call_id,
+            "output": text,
+        },
+    }
+
+
+def codex_patch_call(call_id, path):
+    return codex_tool_call(
+        call_id,
+        (
+            'const patch = "*** Begin Patch\\n'
+            f"*** Update File: {path}\\n"
+            "@@\\n-old\\n+new\\n*** End Patch\";\n"
+            "text(await tools.apply_patch(patch));"
+        ),
+    )
+
+
+def codex_exec_call(call_id, cmd):
+    escaped = json.dumps({"cmd": cmd})[1:-1]
+    return codex_tool_call(
+        call_id,
+        f"const r = await tools.exec_command({{{escaped}}});\ntext(r.output);",
+    )
+
+
+def codex_token_count(total=100):
+    return {
+        "type": "event_msg",
+        "payload": {
+            "type": "token_count",
+            "info": {
+                "total_token_usage": {
+                    "input_tokens": total - 10,
+                    "cached_input_tokens": 0,
+                    "output_tokens": 10,
+                    "total_tokens": total,
+                },
+                "last_token_usage": {
+                    "input_tokens": total - 10,
+                    "cached_input_tokens": 0,
+                    "output_tokens": 10,
+                    "total_tokens": total,
+                },
+            },
+        },
+    }
+
+
 def write_jsonl(lines):
     f = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False)
     for d in lines:
@@ -587,6 +653,67 @@ def test_clean_codex_session_has_no_frustration_flags(self):
         finally:
             os.unlink(path)
 
+    def test_codex_no_verify_patch_streak_flags_same_problem_thrash(self):
+        lines = [
+            codex_response_item("user", "fix the failing workspace test", ts="2026-08-27T10:00:00.000Z"),
+            codex_patch_call("patch-1", "/repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_tool_output("patch-1", "Success. Updated the following files:\nM /repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_patch_call("patch-2", "/repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_tool_output("patch-2", "Success. Updated the following files:\nM /repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_patch_call("patch-3", "/repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_tool_output("patch-3", "Success. Updated the following files:\nM /repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_token_count(200),
+        ]
+        path = write_jsonl(lines)
+        try:
+            with redirect_stdout(io.StringIO()):
+                result = token_audit.audit_codex(path)
+            flags = {fl["name"]: fl for fl in result["flags"]}
+            self.assertEqual(flags["no-verify-edit-streak"]["value"], "yes")
+            self.assertEqual(flags["no-verify-edit-streak"]["count"], 3)
+            self.assertEqual(result["longest_edit_streak_no_verify"], 3)
+        finally:
+            os.unlink(path)
+
+    def test_codex_verified_patch_stays_silent_for_same_problem_thrash(self):
+        lines = [
+            codex_response_item("user", "fix the failing workspace test", ts="2026-08-27T10:00:00.000Z"),
+            codex_patch_call("patch-1", "/repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_tool_output("patch-1", "Success. Updated the following files:\nM /repo/packages/data-store/src/__tests__/scale.test.ts"),
+            codex_exec_call("test-1", "pnpm --filter @invoker/data-store test --run src/__tests__/scale.test.ts"),
+            codex_tool_output("test-1", "Process exited with code 0\n1 passed"),
+            codex_token_count(200),
+        ]
+        path = write_jsonl(lines)
+        try:
+            with redirect_stdout(io.StringIO()):
+                result = token_audit.audit_codex(path)
+            flags = {fl["name"]: fl for fl in result["flags"]}
+            self.assertEqual(flags["no-verify-edit-streak"]["value"], "no")
+            self.assertEqual(flags["recurring-failure-signatures"]["value"], "no")
+        finally:
+            os.unlink(path)
+
+    def test_codex_recurring_failed_tool_output_flags(self):
+        lines = [
+            codex_response_item("user", "keep the workspace test green", ts="2026-08-27T10:00:00.000Z"),
+            codex_exec_call("test-1", "pnpm test -- filter attempt 1"),
+            codex_tool_output("test-1", "Process exited with code 1\nAssertionError: expected 10001 rows"),
+            codex_exec_call("test-2", "pnpm test -- filter attempt 2"),
+            codex_tool_output("test-2", "Process exited with code 1\nAssertionError: expected 10002 rows"),
+            codex_token_count(200),
+        ]
+        path = write_jsonl(lines)
+        try:
+            with redirect_stdout(io.StringIO()):
+                result = token_audit.audit_codex(path)
+            flags = {fl["name"]: fl for fl in result["flags"]}
+            self.assertEqual(flags["recurring-failure-signatures"]["value"], "yes")
+            self.assertEqual(flags["recurring-failure-signatures"]["count"], 1)
+            self.assertEqual(result["n_recurring_failures"], 1)
+        finally:
+            os.unlink(path)
+
     def test_malformed_codex_lines_fail_open(self):
         path = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False).name
         with open(path, "w") as f:

engine/skills/reflect/scripts/token_audit.py

@@ -157,6 +157,8 @@ def _direct_run_targets(command):
 # "Exit code: 1" for apply_patch) - verified against a real rollout file,
 # not guessed. No structured is_error field exists on this transcript shape.
 _CODEX_EXIT_CODE_RE = re.compile(r"(?:Process exited with code|Exit code:)\s*(-?\d+)")
+_CODEX_CMD_RE = re.compile(r'"cmd"\s*:\s*"((?:\\.|[^"\\])*)"')
+_PATCH_PATH_RE = re.compile(r"\*\*\* (?:Update|Add|Delete) File: ([^\\\r\n\"]+)")
 
 
 def _codex_output_is_error(text):
@@ -169,6 +171,41 @@ def _codex_output_is_error(text):
         return False
 
 
+def _decode_json_string(value):
+    try:
+        return json.loads(f'"{value}"')
+    except json.JSONDecodeError:
+        return value
+
+
+def _codex_bash_command(payload):
+    name = str(payload.get("name") or "")
+    raw = payload.get("input")
+    if isinstance(raw, dict):
+        return raw.get("cmd") or raw.get("command")
+    if not isinstance(raw, str):
+        return None
+    if name in {"exec_command", "functions.exec_command"}:
+        match = _CODEX_CMD_RE.search(raw)
+        return _decode_json_string(match.group(1)) if match else raw
+    if name == "exec" and "exec_command" in raw:
+        match = _CODEX_CMD_RE.search(raw)
+        return _decode_json_string(match.group(1)) if match else raw
+    return None
+
+
+def _codex_patch_paths(payload):
+    name = str(payload.get("name") or "")
+    raw = payload.get("input")
+    if isinstance(raw, dict):
+        raw = raw.get("patch") or raw.get("input") or raw.get("cmd") or ""
+    if not isinstance(raw, str):
+        return []
+    if "apply_patch" not in raw and name not in {"apply_patch", "functions.apply_patch"}:
+        return []
+    return [path.strip() for path in _PATCH_PATH_RE.findall(raw)]
+
+
 def _codex_message_text(payload):
     parts = []
     for block in payload.get("content", []) or []:
@@ -967,7 +1004,11 @@ def audit_codex(path, out_path=None):
     assistant_texts = []
     n_interruptions = 0
     call_id_to_name = {}
+    call_id_to_seq = {}
+    tool_calls_seq = []
+    errors_detail = []
     n_errors = 0
+    seq = 0
 
     for d in lines:
         dtype = d.get("type")
@@ -1001,13 +1042,32 @@ def audit_codex(path, out_path=None):
                 if role == "assistant" and text.strip():
                     assistant_texts.append(text)
             elif ptype in ("function_call", "custom_tool_call"):
-                call_id_to_name[payload.get("call_id")] = payload.get("name")
+                call_id = payload.get("call_id")
+                call_id_to_name[call_id] = payload.get("name")
+                command = _codex_bash_command(payload)
+                patch_paths = _codex_patch_paths(payload)
+                if command:
+                    seq += 1
+                    call_id_to_seq[call_id] = seq
+                    tool_calls_seq.append((seq, "Bash", {"command": command}, call_id))
+                for file_path in patch_paths:
+                    seq += 1
+                    call_id_to_seq.setdefault(call_id, seq)
+                    tool_calls_seq.append((seq, "Edit", {"file_path": file_path}, call_id))
             elif ptype in ("function_call_output", "custom_tool_call_output"):
                 out_text = payload.get("output")
                 if isinstance(out_text, dict):
                     out_text = out_text.get("content")
                 if _codex_output_is_error(str(out_text or "")):
                     n_errors += 1
+                    call_id = payload.get("call_id")
+                    errors_detail.append(
+                        (
+                            call_id_to_seq.get(call_id),
+                            call_id_to_name.get(call_id) or "?",
+                            str(out_text or ""),
+                        )
+                    )
 
     cached_share = 0.0
     if turn_ends:
@@ -1020,13 +1080,61 @@ def audit_codex(path, out_path=None):
     retraction_hits = self_retraction_hits(assistant_texts)
     flags.append(_self_retraction_flag(retraction_hits))
 
+    sig_groups = {}
+    for s, name, text in errors_detail:
+        norm = re.sub(r"\d+", "#", text.strip())[:120]
+        sig_groups.setdefault((name, norm), []).append(s)
+    recurring = {k: v for k, v in sig_groups.items() if len(v) > 1}
+
+    edits_since_verify, file_streak_max = {}, {}
+    global_streak = global_streak_max = verify_count = direct_run_verify_count = 0
+    for s, name, inp, ident in sorted(tool_calls_seq, key=lambda x: x[0] or 0):
+        if name == "Bash" and isinstance(inp, dict) and VERIFY_RE.search(inp.get("command") or ""):
+            verify_count += 1
+            edits_since_verify.clear()
+            global_streak = 0
+        elif name == "Bash" and isinstance(inp, dict):
+            targets = _direct_run_targets(inp.get("command") or "")
+            edited_basenames = {os.path.basename(fp) for fp in edits_since_verify if fp}
+            if targets & edited_basenames:
+                direct_run_verify_count += 1
+                edits_since_verify.clear()
+                global_streak = 0
+        elif name == "Edit" and isinstance(inp, dict):
+            fp = inp.get("file_path")
+            edits_since_verify[fp] = edits_since_verify.get(fp, 0) + 1
+            file_streak_max[fp] = max(file_streak_max.get(fp, 0), edits_since_verify[fp])
+            global_streak += 1
+            global_streak_max = max(global_streak_max, global_streak)
+    flagged_files = {fp: n for fp, n in file_streak_max.items() if n >= 3}
+    flags.extend([
+        _flag(
+            "recurring-failure-signatures",
+            "yes" if recurring else "no",
+            len(recurring),
+            f"{len(recurring)} recurring failure signature(s) (same Codex tool-output error shape repeating across attempts)",
+        ),
+        _flag(
+            "no-verify-edit-streak",
+            "yes" if flagged_files or global_streak_max >= 3 else "no",
+            global_streak_max,
+            (
+                f"longest edit streak with zero verification: {global_streak_max}; "
+                f"{len(flagged_files)} file(s) at or above threshold 3; "
+                f"verify Bash calls={verify_count}, direct-run verifies={direct_run_verify_count}"
+            ),
+        ),
+    ])
+
     result = {
         "models": dict(models),
         "last_usage": last_usage,
         "n_turns": len(turn_ends),
         "cache_read_share": cached_share,
         "total": (last_usage or {}).get("total_tokens", 0),
         "n_errors": n_errors,
+        "n_recurring_failures": len(recurring),
+        "longest_edit_streak_no_verify": global_streak_max,
         "flags": flags,
         "frustration": frustration,
         "self_retraction": retraction_hits,
@@ -1045,6 +1153,8 @@ def audit_codex(path, out_path=None):
                 "models": dict(models),
                 "cache_read_share": cached_share,
                 "n_errors": n_errors,
+                "n_recurring_failures": len(recurring),
+                "longest_edit_streak_no_verify": global_streak_max,
             },
             "flags": flags,
             "frustration": frustration,