fix: skip unknown transform ops instead of silently running base64_decode

RELEASE_NOTES.md

@@ -1,3 +1,18 @@
+# Release Notes: FlowRunner CLI (Unreleased)
+
+## ⚠️ Behavior fix — severity: HIGH — unknown transform ops no longer silently mis-execute
+
+**What changed.** An unknown/unsupported transform `op` was previously **silently rewritten to `base64_decode`** and executed (`_normalize_transform_op`), so a flow authored with a newer/unrecognized op ran the WRONG operation against live traffic and produced plausible-but-wrong output. It is now **skipped with a machine-readable warning** rather than executed:
+- `execute_transform_ops` logs a machine-readable `TRANSFORM_OP_UNSUPPORTED op=... set=... index=...` marker at ERROR level and records a structured entry in its returned `warnings[]` (`{"type": "unsupported_transform_op", "op": <name>, "set": <var>, "index": <i>, "status": "skipped"}`), which the step handler surfaces to the run log. The op's output variable is left unset and the remaining ops run; the skip itself does not halt the flow. (If a *later* op or step hard-references the unset variable via `{{var}}`, that still resolves as an undefined reference and halts as usual, exactly as before — the skip neither invents new failures nor papers over genuinely missing data.)
+- `_normalize_transform_op` now **raises** on an unknown op (defense-in-depth) instead of downgrading to `base64_decode`.
+- `_execute_transform_step` surfaces each warning at WARNING level in the run log.
+
+**Audit note.** Run artifacts produced by earlier CLI versions may contain values from an unintended `base64_decode` of an unknown op's first argument. To audit, review flows using transform ops outside the supported set (`base64_encode/decode`, `jwt_encode/decode`, `json_set`, `math_add/sub/mul/div`, `to_number/string/boolean`, `boolean_not`); any output variable whose value looks like a base64 decode of an unexpected input may have been affected.
+
+This is part of the cross-app FlowMap graceful-degradation strategy (unknown step type / transform op / operator ⇒ skip-with-warning, never crash, never mis-execute). See the FlowRunner UI repo's `docs/flowmap-evolution.md` and `gotchas.md` "Cross-app FlowMap contract".
+
+---
+
 # Release Notes: FlowRunner CLI v1.2.0
 
 ## Highlights

flow_runner.py

@@ -733,7 +733,10 @@ def _normalize_transform_op(op: Any) -> Dict[str, Any]:
 
     op_name = safe_op.get("op")
     if op_name not in _TRANSFORM_OP_DEFS:
-        op_name = "base64_decode"
+        # Never silently substitute a different op (historically base64_decode). Callers
+        # must skip unknown ops with a warning; this raise is defense-in-depth so no code
+        # path can execute the wrong operation against live traffic.
+        raise ValueError(f'Unsupported transform op "{op_name}".')
 
     defn = _TRANSFORM_OP_DEFS[op_name]
     raw_args = safe_op.get("args")
@@ -1168,6 +1171,40 @@ def execute_transform_ops(ops: Any, context: Dict[str, Any], evaluate_path: Opti
     output = {"updatedVars": [], "warnings": []}
     ops_list = ops if isinstance(ops, list) else []
     for index, op in enumerate(ops_list):
+        # Graceful degradation: an unknown/newer transform op is SKIPPED with a
+        # machine-readable warning rather than silently downgraded to base64_decode
+        # (which would run the wrong operation against live traffic) or raised (which
+        # would set flow_error and halt the whole flow). Known ops proceed normally.
+        if isinstance(op, TransformOp):
+            raw_op = op.model_dump()
+        elif isinstance(op, dict):
+            raw_op = op
+        else:
+            raw_op = {}
+        op_name = raw_op.get("op")
+        # Unknown-op detection mirrors _normalize_transform_op's check below; keep the
+        # two in sync if op-name normalization/aliasing is ever introduced.
+        if op_name not in _TRANSFORM_OP_DEFS:
+            set_hint = raw_op.get("set") if isinstance(raw_op.get("set"), str) else None
+            var_phrase = f'output variable "{set_hint}" left unset' if set_hint else "no output variable set"
+            output["warnings"].append({
+                "type": "unsupported_transform_op",
+                "op": op_name,
+                "set": set_hint,
+                "index": index,
+                "status": "skipped",
+                "message": (
+                    f'Unsupported transform op "{op_name}" at position {index + 1}; '
+                    f'skipped ({var_phrase}). This runner may be older than the flow '
+                    f'that produced it.'
+                ),
+            })
+            logger.error(
+                "TRANSFORM_OP_UNSUPPORTED op=%r set=%r index=%d - skipped, not executed "
+                "(refusing to silently substitute base64_decode).",
+                op_name, set_hint, index,
+            )
+            continue
         normalized = _normalize_transform_op(op)
         set_name = normalized.get("set")
         if not isinstance(set_name, str) or not set_name:
@@ -2809,6 +2846,10 @@ async def _execute_transform_step(
             ops = step.ops or []
             logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Executing {len(ops)} ops.")
             output = execute_transform_ops(ops, context, evaluate_path=get_value_from_context)
+            for warning in output.get("warnings", []):
+                logger.warning(
+                    f"{indent}User {user_id_log}: Transform {step_identifier}: {warning.get('message')}"
+                )
             logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Updated vars {output.get('updatedVars', [])}.")
         except Exception as exc:
             logger.error(

tests/unit/test_flow_runner.py

@@ -54,12 +54,15 @@ def decorator(fn):
     LoopStep,
     ConditionStep,
     TransformStep,
+    TransformOp,
     ConditionData,
     Metrics,
     StartRequest,
     get_value_from_context,
     _MISSING,
     set_value_in_context,
+    execute_transform_ops,
+    _normalize_transform_op,
     logger as fr_logger,
 )
 
@@ -205,6 +208,105 @@ async def test_transform_step_updates_context(base_config, empty_flow):
     assert context["payload"]["exp"] == 110
 
 
+def test_execute_transform_ops_skips_unknown_op_without_downgrade():
+    # An unknown/newer transform op must NOT be silently rewritten to base64_decode.
+    # It is skipped with a machine-readable warning; later known ops still run.
+    context: Dict[str, Any] = {}
+    ops = [
+        # "SGVsbG8" base64url-decodes to "Hello". If the old bug downgrades this to
+        # base64_decode, context["decoded"] would become "Hello". It must stay unset.
+        {"op": "totally_unknown_future_op", "set": "decoded", "args": ["SGVsbG8"]},
+        {"op": "math_add", "set": "sum", "args": [1, 2]},
+    ]
+    output = execute_transform_ops(ops, context)
+    # unknown op skipped: variable never set (definitely not base64-decoded to "Hello")
+    assert "decoded" not in context
+    # subsequent known op still executed
+    assert context["sum"] == 3
+    assert "sum" in output["updatedVars"]
+    assert "decoded" not in output["updatedVars"]
+    # a machine-readable warning was recorded for the skipped op
+    warnings = output["warnings"]
+    assert len(warnings) == 1
+    w = warnings[0]
+    assert w["op"] == "totally_unknown_future_op"
+    assert w["set"] == "decoded"
+    assert w["status"] == "skipped"
+
+
+def test_normalize_transform_op_raises_on_unknown_op():
+    # Defense in depth: normalization must never silently substitute base64_decode.
+    with pytest.raises(ValueError):
+        _normalize_transform_op({"op": "nope_not_real", "set": "x", "args": ["SGVsbG8"]})
+
+
+@pytest.mark.asyncio
+async def test_transform_step_skips_unknown_op_without_halting(base_config, empty_flow):
+    # Graceful degradation: an unknown op does not halt the flow (no flow_error is set)
+    # and known ops in the same step still apply.
+    runner = make_runner(base_config, empty_flow)
+    context: Dict[str, Any] = {}
+    step = TransformStep(
+        id="t2",
+        name="Transform",
+        type="transform",
+        ops=[
+            {"op": "totally_unknown_future_op", "set": "decoded", "args": ["SGVsbG8"]},
+            {"op": "math_add", "set": "sum", "args": [2, 3]},
+        ],
+    )
+    await runner._execute_transform_step(step, context, depth=0, user_id_log="test")
+    # no crash / no halt
+    assert get_value_from_context(context, "flow_error") is _MISSING
+    # unknown op did not run (not downgraded to base64_decode)
+    assert "decoded" not in context
+    # known op still ran
+    assert context["sum"] == 5
+
+
+def test_execute_transform_ops_skips_unknown_transformop_instance():
+    # Exercises the model_dump() branch of the guard directly: an unknown op passed as a
+    # TransformOp model instance (the shape ops actually take after step validation).
+    context: Dict[str, Any] = {}
+    op = TransformOp.model_validate({"op": "unknown_model_op", "set": "decoded", "args": ["SGVsbG8"]})
+    output = execute_transform_ops([op], context)
+    assert "decoded" not in context
+    assert output["updatedVars"] == []
+    assert len(output["warnings"]) == 1
+    assert output["warnings"][0]["op"] == "unknown_model_op"
+    assert output["warnings"][0]["status"] == "skipped"
+
+
+def test_execute_transform_ops_handles_missing_op_name_and_set():
+    # op is None (missing) and set is missing -> each skipped with a warning, no crash.
+    context: Dict[str, Any] = {}
+    output = execute_transform_ops([{"args": []}, {"op": "still_unknown"}], context)
+    assert output["updatedVars"] == []
+    assert len(output["warnings"]) == 2
+    assert output["warnings"][0]["op"] is None
+    assert output["warnings"][0]["set"] is None
+
+
+@pytest.mark.asyncio
+async def test_transform_step_all_ops_unknown_does_not_halt(base_config, empty_flow):
+    # Every op unknown: the whole step degrades to a no-op and the flow is not halted.
+    runner = make_runner(base_config, empty_flow)
+    context: Dict[str, Any] = {}
+    step = TransformStep(
+        id="t3",
+        name="Transform",
+        type="transform",
+        ops=[
+            {"op": "unknown_a", "set": "a", "args": []},
+            {"op": "unknown_b", "set": "b", "args": []},
+        ],
+    )
+    await runner._execute_transform_step(step, context, depth=0, user_id_log="test")
+    assert get_value_from_context(context, "flow_error") is _MISSING
+    assert "a" not in context
+    assert "b" not in context
+
+
 def test_extract_data_status_headers_and_body(base_config, empty_flow):
     runner = make_runner(base_config, empty_flow)
     ctx: Dict[str, Any] = {}