Separate paid fallback usage in operations reports

ops/report.py

@@ -122,7 +122,9 @@ def scheduled_codex_usage(directory, start, end):
     if not directory or not Path(directory).is_dir():
         return {'available': False, 'reason': '定时运营汇总目录未配置或不可用'}
     totals = {k: 0 for k in ('input_tokens', 'cached_input_tokens', 'output_tokens', 'reasoning_output_tokens')}
+    api_totals = dict.fromkeys(totals, 0)
     runs = failed = 0
+    api_runs = api_uncertain = api_usage_events = 0
     for path in Path(directory).glob('*.summary.json'):
         try:
             record = json.loads(path.read_text())
@@ -132,13 +134,21 @@ def scheduled_codex_usage(directory, start, end):
                 continue
             runs += 1
             failed += record.get('status') != 'complete'
+            if record.get('auth') == 'api_key':
+                api_runs += record.get('api_fallback_used') is True
+                api_uncertain += record.get('api_fallback_used') is None
+                for usage in record.get('usage', []):
+                    api_usage_events += 1
+                    for k in api_totals:
+                        api_totals[k] += max(0, int(usage.get(k, 0)))
+                continue
             for usage in record.get('usage', []):
                 for k in totals:
                     totals[k] += max(0, int(usage.get(k, 0)))
         except (ValueError, OSError, KeyError, TypeError, AttributeError):
             continue
-    return {'available': True, 'runs': runs, 'failed_runs': failed, **totals,
+    return {'available': True, 'runs': runs, 'failed_runs': failed, 'api_runs_separate': api_runs, 'api_runs_uncertain': api_uncertain, 'api_observed_usage': api_totals if api_usage_events else None, **totals,
-            'measurement_note': '来自独立定时会话的运行汇总，与交互会话分开统计；失败或中断可能没有完整用量。'}
+            'measurement_note': '来自独立定时会话的运行汇总，与交互会话分开统计；API 备用调用不计入订阅 token，实际成本单独核算。失败或中断可能没有完整用量。'}
 
 
 def harness_run_usage(path, start, end):

ops/test_report.py

@@ -59,9 +59,11 @@ class AccountingTests(unittest.TestCase):
             root = Path(directory)
             (root / 'one.summary.json').write_text(json.dumps({'id':'20261103T141341Z-a','status':'complete','usage':[{'input_tokens':100,'cached_input_tokens':80,'output_tokens':10}]}))
             (root / 'two.summary.json').write_text(json.dumps({'id':'20261104T141341Z-b','status':'failed','usage':[]}))
+            (root / 'paid.summary.json').write_text(json.dumps({'id':'20261104T151341Z-d','status':'complete','auth':'api_key','api_fallback_used':True,'usage':[{'input_tokens':33,'output_tokens':5}]}))
             (root / 'old.summary.json').write_text(json.dumps({'id':'20261003T141341Z-c','status':'complete','usage':[{'input_tokens':900}]}))
             value=report.scheduled_codex_usage(root,dt.datetime(2026,11,3,tzinfo=report.TZ),dt.datetime(2026,11,5,tzinfo=report.TZ))
-            self.assertEqual((value['runs'],value['failed_runs'],value['input_tokens'],value['cached_input_tokens']), (2,1,100,80))
+            self.assertEqual((value['runs'],value['failed_runs'],value['input_tokens'],value['cached_input_tokens']), (3,1,100,80))
+            self.assertEqual(value['api_runs_separate'],1);self.assertEqual(value['api_observed_usage']['input_tokens'],33)
 
     def test_harness_ledger_counts_only_selected_period(self):
         import json