feat(reporting): flag large CF operating reconciling plug

robosystems/operations/roboledger/reports/fact_grid.py

@@ -1247,6 +1247,14 @@ def _synthesize_ppe_net_facts(
 # the adjustment both foots the subtotal and renders as a line.
 _CF_NET_CHANGE_QNAME = "rs-gaap:CashAndCashEquivalentsPeriodIncreaseDecrease"
 _CF_RECONCILING_LEAF_QNAME = "rs-gaap:IncreaseDecreaseInOtherOperatingCapitalNet"
+_CF_OPERATING_SUBTOTAL_QNAME = "rs-gaap:NetCashProvidedByUsedInOperatingActivities"
+
+# A reconciling plug larger than this fraction of operating cash is the signal
+# that a material non-cash item is un-itemized (gain/loss on disposal, unrealized
+# MTM, …) — or that an investing/financing flow was misclassified and silently
+# absorbed by `_reconcile_operating_to_cash`. Shared with guard_rails so the
+# render-time validator and the fact-bundle log threshold stay in lockstep.
+_CF_PLUG_WARN_RATIO = 0.25
 
 
 def _emit_flow_facts(
@@ -1614,11 +1622,19 @@ def _reconcile_operating_to_cash(
 
   id_rows = session.execute(
     text("SELECT id, qname, name, balance_type FROM elements WHERE qname = ANY(:q)"),
-    {"q": [_CF_NET_CHANGE_QNAME, _CF_RECONCILING_LEAF_QNAME]},
+    {
+      "q": [
+        _CF_NET_CHANGE_QNAME,
+        _CF_RECONCILING_LEAF_QNAME,
+        _CF_OPERATING_SUBTOTAL_QNAME,
+      ]
+    },
   ).fetchall()
   by_qname = {r.qname: r for r in id_rows}
   net_change = by_qname.get(_CF_NET_CHANGE_QNAME)
   recon = by_qname.get(_CF_RECONCILING_LEAF_QNAME)
+  operating = by_qname.get(_CF_OPERATING_SUBTOTAL_QNAME)
+  operating_id = operating.id if operating is not None else None
   if net_change is None or recon is None:
     # Framework missing the CF net-change or the reconciling leaf — nothing to
     # foot against. Leave the CF as derived; the tie-out check surfaces the gap.
@@ -1680,15 +1696,32 @@ def _reconcile_operating_to_cash(
     residual = cash_delta - derived_net_change
     if abs(residual) <= 0.005:
       continue
-    logger.info(
+    # Elevate to a warning when the plug dwarfs operating cash — the signal that
+    # a material non-cash item is un-itemized or a flow was misclassified and
+    # silently absorbed here (see the TRADEOFF note above). Basis is post-plug
+    # operating cash, floored by ΔCash so a near-zero operating section can't
+    # divide-by-zero.
+    derived_operating = computed.get(operating_id, 0.0) if operating_id else 0.0
+    true_operating = derived_operating + residual
+    denom = max(abs(true_operating), abs(cash_delta))
+    large = denom > 0.005 and abs(residual) > _CF_PLUG_WARN_RATIO * denom
+    suffix = (
+      " — LARGE relative to operating cash (%.0f%%); review for an un-itemized "
+      "non-cash item (gain/loss on disposal, …) or a flow misclassification."
+      % (100 * abs(residual) / denom)
+      if large
+      else "; itemizing its components is a future enrichment."
+    )
+    (logger.warning if large else logger.info)(
       "CF reconciled to cash for period ending %s: non-cash operating "
       "adjustment %.2f (derived net change %.2f, actual cash movement %.2f). "
-      "Booked to %s; itemizing its components is a future enrichment.",
+      "Booked to %s%s",
       current.end,
       residual,
       derived_net_change,
       cash_delta,
       _CF_RECONCILING_LEAF_QNAME,
+      suffix,
     )
     facts.append(
       ReportFact(

robosystems/operations/roboledger/reports/guard_rails.py

@@ -8,7 +8,13 @@
 
 from dataclasses import dataclass, field
 
-from .fact_grid import FactRow, _infer_classification
+from .fact_grid import (
+  _CF_OPERATING_SUBTOTAL_QNAME,
+  _CF_PLUG_WARN_RATIO,
+  _CF_RECONCILING_LEAF_QNAME,
+  FactRow,
+  _infer_classification,
+)
 
 # Rounding tolerance for balance checks (dollars)
 _TOLERANCE = 0.01
@@ -314,6 +320,7 @@ def _validate_cash_flow(rows: list[FactRow]) -> ValidationResult:
   _check_totals_foot(rows, result)
   _check_zero_subtotals(rows, result)
   _check_comparative_data(rows, result)
+  _check_operating_plug(rows, result)
 
   return result
 
@@ -379,3 +386,39 @@ def _check_comparative_data(rows: list[FactRow], result: ValidationResult) -> No
       result.warnings.append(
         f"Period column {col_idx + 1} has no data — column will be empty"
       )
+
+
+def _check_operating_plug(rows: list[FactRow], result: ValidationResult) -> None:
+  """Warn when the operating-CF reconciling plug is large vs operating cash.
+
+  ``fact_grid._reconcile_operating_to_cash`` foots the indirect CF to actual
+  cash by booking the aggregate non-cash operating adjustment (gain/loss on
+  disposal, unrealized MTM, write-offs, …) onto
+  ``IncreaseDecreaseInOtherOperatingCapitalNet``. That makes the statement
+  articulate *by construction* — and so silently absorbs any investing/financing
+  misclassification too. A plug that dwarfs operating cash is the signal that a
+  material item is un-itemized or mis-tagged; surface it so it isn't invisible.
+
+  Row-level approximation: that line also carries any tenant-mapped "other
+  operating capital" content, but it's a system catch-all rarely mapped
+  directly, so the row value ≈ the plug in practice. Warning-only — the CF still
+  foots and renders.
+  """
+  result.checks.append("operating_plug")
+  plug = next((r for r in rows if r.element_qname == _CF_RECONCILING_LEAF_QNAME), None)
+  op = next((r for r in rows if r.element_qname == _CF_OPERATING_SUBTOTAL_QNAME), None)
+  if plug is None or op is None:
+    return
+  for col in range(len(plug.values)):
+    plug_val = (plug.values[col] or 0.0) if col < len(plug.values) else 0.0
+    if abs(plug_val) <= _TOLERANCE:
+      continue
+    op_val = (op.values[col] or 0.0) if col < len(op.values) else 0.0
+    if abs(op_val) < _TOLERANCE or abs(plug_val) > _CF_PLUG_WARN_RATIO * abs(op_val):
+      result.warnings.append(
+        f"Operating cash flow (period column {col + 1}) carries a large "
+        f"unattributed reconciling adjustment in 'Other operating capital, net' "
+        f"({plug_val:.2f} vs operating cash {op_val:.2f}) — likely an "
+        f"un-itemized non-cash item (gain/loss on disposal, etc.) or a flow "
+        f"misclassification; review."
+      )

tests/operations/roboledger/reports/test_guard_rails.py

@@ -430,3 +430,41 @@ def test_unknown_report_type(self):
     result = validate_report("unknown_type", [])
     assert result.passed is True
     assert "no_validation_rules" in result.checks
+
+
+class TestCashFlowOperatingPlug:
+  """`_check_operating_plug` — warn when the reconciling plug dwarfs operating cash."""
+
+  _OP = "rs-gaap:NetCashProvidedByUsedInOperatingActivities"
+  _PLUG = "rs-gaap:IncreaseDecreaseInOtherOperatingCapitalNet"
+
+  def _cf(self, op_value: float, plug_value: float) -> list[FactRow]:
+    # Plain (non-subtotal, single-period) rows so the other CF checks stay quiet
+    # and only _check_operating_plug can speak.
+    return [
+      _row("Operating", op_value, qname=self._OP),
+      _row("Other operating capital net", plug_value, qname=self._PLUG),
+    ]
+
+  def test_large_plug_warns(self):
+    # 400 / 1000 = 40% > 25% threshold
+    result = validate_report("cash_flow_statement", self._cf(1000.0, 400.0))
+    assert "operating_plug" in result.checks
+    assert any("Other operating capital" in w for w in result.warnings)
+
+  def test_small_plug_is_silent(self):
+    # 100 / 1000 = 10% < 25% threshold
+    result = validate_report("cash_flow_statement", self._cf(1000.0, 100.0))
+    assert "operating_plug" in result.checks
+    assert not any("Other operating capital" in w for w in result.warnings)
+
+  def test_material_plug_with_zero_operating_warns(self):
+    # operating ≈ 0 but a material plug is the worst case — must surface.
+    result = validate_report("cash_flow_statement", self._cf(0.0, 500.0))
+    assert any("Other operating capital" in w for w in result.warnings)
+
+  def test_no_plug_row_is_silent(self):
+    result = validate_report(
+      "cash_flow_statement", [_row("Operating", 1000.0, qname=self._OP)]
+    )
+    assert not any("Other operating capital" in w for w in result.warnings)