fix: set completed status on workflow cache hit

A cache hit reusing a step's result left stale status: a step skipped
on an earlier run kept 'skipped' even after the gate reopened and a
cached result was served. Mark both step.status and step_statuses as
'completed' on cache hits (execution callbacks stay disabled since
nothing runs). Addresses CodeRabbit review on PR #5421.

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/praisonaiagents/workflows/workflows.py

@@ -1613,40 +1613,6 @@ class AgentFlow:
                 variables=all_variables.copy()
             )
 
-            # Step-result cache (opt-in via cache=). This linear path invokes
-            # handlers INLINE rather than through _execute_single_step_internal,
-            # which the pattern paths use -- so both are wrapped. Caching only
-            # one would make the feature work or not depending on whether a step
-            # happened to sit inside a Parallel or an If, which is worse than
-            # not having it.
-            _step_cache = getattr(self, "_step_cache", None)
-            _cache_key = None
-            if _step_cache is not None:
-                from .step_cache import make_step_key
-                _cache_key = make_step_key(step, previous_output, input, all_variables)
-                _cached = _step_cache.get(_cache_key)
-                if _cached is not None:
-                    if verbose:
-                        print(f"↩︎  cache hit: {step.name}")
-                    previous_output = _cached.get("output")
-                    results.append({"step": step.name, "output": previous_output})
-                    if _cached.get("variables"):
-                        all_variables.update(_cached["variables"])
-                    i += 1
-                    continue
-            
-            # Update step status
-            if hasattr(step, 'status'):
-                step.status = "running"
-            self.step_statuses[step.name] = "running"
-            
-            # Call on_step_start callback
-            if self.on_step_start:
-                try:
-                    self.on_step_start(step.name, context)
-                except Exception as e:
-                    logger.error(f"on_step_start callback failed: {e}")
-            
             # Check should_run condition
             if step.should_run:
                 try:
@@ -1660,7 +1626,45 @@ class AgentFlow:
                         continue
                 except Exception as e:
                     logger.error(f"should_run failed for {step.name}: {e}")
+
+            # Cache only the result, never the decision to run. A condition can
+            # observe external state that is absent from the cache key, so it
+            # must be evaluated even when identical inputs have a cached result.
+            _step_cache = getattr(self, "_step_cache", None)
+            _cache_key = None
+            if _step_cache is not None:
+                from .step_cache import make_step_key
+                _cache_key = make_step_key(step, previous_output, input, all_variables)
+                _cached = _step_cache.get(_cache_key)
+                if _cached is not None:
+                    if verbose:
+                        print(f"↩︎  cache hit: {step.name}")
+                    previous_output = _cached.get("output")
+                    # A cache hit is a successful (re)use of the step's result,
+                    # so record it as completed. Without this, a step that was
+                    # skipped on an earlier run keeps its stale "skipped" status
+                    # even though the gate reopened and a cached result was
+                    # served. Execution callbacks stay disabled: nothing ran.
+                    if hasattr(step, 'status'):
+                        step.status = "completed"
+                    self.step_statuses[step.name] = "completed"
+                    results.append({"step": step.name, "output": previous_output})
+                    if _cached.get("variables"):
+                        all_variables.update(_cached["variables"])
+                    i += 1
+                    continue
             
+            # Mark actual execution only after the condition and cache lookup.
+            if hasattr(step, 'status'):
+                step.status = "running"
+            self.step_statuses[step.name] = "running"
+
+            if self.on_step_start:
+                try:
+                    self.on_step_start(step.name, context)
+                except Exception as e:
+                    logger.error(f"on_step_start callback failed: {e}")
+
             # Gap 3c: Check for cross-step handoff cycles
             self._check_handoff_cycle(step)
             

src/praisonai-agents/tests/integration/test_workflow_cache_condition_real.py

@@ -0,0 +1,38 @@
+"""Opt-in real-agent validation of cached conditional workflow execution."""
+
+import os
+
+import pytest
+
+
+@pytest.mark.skipif(
+    os.getenv("RUN_REAL_KEY_TESTS") != "1" or not os.getenv("OPENAI_API_KEY"),
+    reason="requires an explicitly enabled real test provider",
+)
+def test_real_agent_cached_result_respects_current_condition():
+    from praisonaiagents import Agent, Task
+    from praisonaiagents.workflows.workflows import AgentFlow
+
+    enabled = {"value": True}
+    calls = []
+    agent = Agent(
+        name="conditional", instructions="Answer briefly.",
+        llm=os.getenv("PRAISONAI_TEST_MODEL", "gpt-4o-mini"),
+    )
+
+    def handler(ctx):
+        calls.append("run")
+        response = agent.start("Say hello in one short sentence.")
+        print(response)
+        assert response
+        return response
+
+    flow = AgentFlow(steps=[Task(
+        name="conditional", handler=handler,
+        should_run=lambda ctx: enabled["value"],
+    )], cache=True)
+    flow.run("same", verbose=False)
+    flow.run("same", verbose=False)
+    enabled["value"] = False
+    assert flow.run("same", verbose=False)["steps"] == []
+    assert calls == ["run"]

src/praisonai-agents/tests/unit/workflows/test_cache_run_condition.py

@@ -0,0 +1,55 @@
+"""A cached step result must still respect its current execution gate."""
+
+from praisonaiagents import Task
+from praisonaiagents.workflows.workflows import AgentFlow
+
+
+def test_cached_step_rechecks_gate_without_rerunning_handler():
+    enabled = {"value": True}
+    checks = []
+    calls = []
+
+    def should_run(ctx):
+        checks.append(enabled["value"])
+        return enabled["value"]
+
+    def handler(ctx):
+        calls.append("run")
+        return "done"
+
+    step = Task(name="conditional", handler=handler, should_run=should_run)
+    flow = AgentFlow(steps=[step], cache=True)
+    flow.run("same", verbose=False)
+    flow.run("same", verbose=False)
+    enabled["value"] = False
+    skipped = flow.run("same", verbose=False)
+
+    assert checks == [True, True, False]
+    assert calls == ["run"]
+    assert skipped["steps"] == []
+    assert flow.step_statuses["conditional"] == "skipped"
+    assert "conditional_output" not in skipped["variables"]
+
+    enabled["value"] = True
+    resumed = flow.run("same", verbose=False)
+    assert checks == [True, True, False, True]
+    assert calls == ["run"]
+    assert resumed["output"] == "done"
+    # The reopened gate serves a cached result: the step's status must reflect
+    # that terminal success, not the "skipped" state left by the closed-gate run.
+    assert flow.step_statuses["conditional"] == "completed"
+    assert step.status == "completed"
+
+
+def test_skipped_step_is_not_cached_when_gate_later_opens():
+    enabled = {"value": False}
+    calls = []
+    step = Task(
+        name="conditional", should_run=lambda ctx: enabled["value"],
+        handler=lambda ctx: calls.append("run") or "done",
+    )
+    flow = AgentFlow(steps=[step], cache=True)
+    assert flow.run("same", verbose=False)["steps"] == []
+    enabled["value"] = True
+    assert flow.run("same", verbose=False)["output"] == "done"
+    assert calls == ["run"]