[AISOS-2159] Workflow continues to PR creation after container execution failure

Pipfile

@@ -0,0 +1,11 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+
+[dev-packages]
+
+[requires]
+python_version = "3.14"

docs/guide/task-workflow.md

@@ -117,7 +117,7 @@ After the container run, Forge stages and commits workspace changes on the host.
 
 Forge reviews the implemented changes before opening a PR. If the verdict is adequate, the workflow proceeds to PR creation.
 
-If the review finds incomplete tests or a symptom-only implementation, Forge routes back to implementation with the review feedback. The Task workflow allows up to 2 qualitative review retries. If the retry cap is reached, Forge proceeds to PR creation with the failed review state retained so reviewers can see the risk.
+If the review finds incomplete tests or a symptom-only implementation, Forge routes back to implementation with the review feedback. The Task workflow allows up to 2 qualitative review retries. If the retry cap is reached, as long as there is no active error and a successful commit was made, Forge proceeds to PR creation with the failed review state retained. If there is an active error (e.g., container execution failed) or no changes were committed, the workflow escalates to blocked (`forge:blocked`) instead of opening an empty or broken PR.
 
 Infrastructure errors that prevent review from producing a verdict route to `forge:blocked`.
 

src/forge/workflow/nodes/task_takeover_execution.py

@@ -137,6 +137,15 @@ async def execute_task_changes(state: TaskTakeoverState) -> TaskTakeoverState:
             git.stage_all()
             committed = git.commit(commit_message)
 
+        # Preserve the cumulative committed state if we've already committed in a previous attempt
+        prev_commit_info = state.get("commit_info") or {}
+        prev_committed = (
+            prev_commit_info.get("committed")
+            if isinstance(prev_commit_info, dict)
+            else getattr(prev_commit_info, "committed", False)
+        )
+        has_ever_committed = prev_committed or committed
+
         current_sha = git.get_current_sha()
         execution_state = {
             **state,
@@ -186,6 +195,11 @@ async def execute_task_changes(state: TaskTakeoverState) -> TaskTakeoverState:
             update_state_timestamp(
                 {
                     **execution_state,
+                    "commit_info": {
+                        "sha": current_sha,
+                        "message": commit_message,
+                        "committed": has_ever_committed,
+                    },
                     "implementation_push_pending": False,
                     "implementation_push_pending_task": None,
                     "persistence_retry_count": 0,

src/forge/workflow/task_takeover/graph.py

@@ -144,48 +144,59 @@ def _route_after_execution(state: TaskTakeoverState) -> str:
 def _route_after_qualitative_review(state: TaskTakeoverState) -> str:
     """Route after run_qualitative_review considering qualitative verdict and retry count.
 
-    If the review is adequate (success), proceed to create_pr.
-    If the review is failed or incomplete:
-      - Check if we've reached the configured retry limit.
-      - If limit reached: proceed to PR creation with the failed-review state retained.
-      - Otherwise: transition back to execute_task_changes.
-    If the node hit an unrecoverable error (no workspace), escalate.
+    The routing logic is state-driven:
+      - If review is adequate, proceed to create_pr.
+      - If there is an active error (last_error is set), always route to escalate_blocked if we've reached or exceeded the retry cap limit.
+      - Under the retry cap, if there is an execution error (last_error without verdict), retry the review.
+      - If we reached the retry cap and there are no active errors, we can proceed to create_pr only if commits were successfully made (commit_info.committed is True).
+      - Otherwise, escalate or loop back to execute_task_changes.
     """
     verdict = state.get("review_verdict")
     retry_count = state.get("qualitative_review_retry_count", 0)
     last_error = state.get("last_error")
 
+    limit = QUALITATIVE_REVIEW_MAX_ATTEMPTS
+
     if verdict == "adequate":
         return "create_pr"
 
-    limit = QUALITATIVE_REVIEW_MAX_ATTEMPTS
-
-    # Review infrastructure failures are bounded like negative verdicts. Retry
-    # the review itself (not implementation), then proceed with the failure
-    # retained in state so the PR warns human reviewers.
-    if last_error and not verdict:
-        if retry_count >= limit:
+    if retry_count >= limit:
+        if last_error:
             logger.error(
-                "Qualitative review failed %s times; skipping review and proceeding to PR: %s",
-                retry_count,
+                "Qualitative review retry limit reached with active error: %s. Escalating.",
                 last_error,
             )
-            return "create_pr"
-        logger.warning(
-            "Qualitative review execution failed; retrying review (%s/%s): %s",
-            retry_count,
-            limit,
-            last_error,
+            return "escalate_blocked"
+
+        commit_info = state.get("commit_info") or {}
+        committed = (
+            commit_info.get("committed")
+            if isinstance(commit_info, dict)
+            else getattr(commit_info, "committed", False)
         )
-        return "run_qualitative_review"
 
-    if retry_count >= limit:
+        if not committed:
+            logger.warning(
+                "Qualitative review retry limit reached with no committed changes. Escalating."
+            )
+            return "escalate_blocked"
+
         logger.warning(
             f"Qualitative review cap ({limit}) reached on task takeover workflow, "
             "proceeding to PR creation with review state retained"
         )
         return "create_pr"
 
+    # Under the limit
+    if last_error and not verdict:
+        logger.warning(
+            "Qualitative review execution failed; retrying review (%s/%s): %s",
+            retry_count,
+            limit,
+            last_error,
+        )
+        return "run_qualitative_review"
+
     logger.info(
         f"Qualitative review verdict is {verdict!r}, retry attempt {retry_count}/{limit}, "
         "routing back to execute_task_changes"

src/forge/workflow/task_takeover/state.py

@@ -1,7 +1,7 @@
 """Task Takeover workflow state definition."""
 
 from datetime import datetime
-from typing import Any, cast
+from typing import Any, TypedDict, cast
 
 from forge.models.workflow import TicketType
 from forge.workflow.base import (
@@ -12,6 +12,14 @@
 )
 
 
+class CommitInfo(TypedDict, total=False):
+    """Information about the implementation commit."""
+
+    sha: str | None
+    message: str | None
+    committed: bool
+
+
 class TaskTakeoverState(
     BaseState, PRIntegrationState, CIIntegrationState, ReviewIntegrationState, total=False
 ):
@@ -25,6 +33,7 @@ class TaskTakeoverState(
     review_feedback: str | None
     qualitative_review_retry_count: int
     qualitative_review_failed: bool
+    commit_info: CommitInfo
 
 
 def create_initial_task_takeover_state(ticket_key: str, **kwargs: Any) -> TaskTakeoverState:
@@ -62,6 +71,11 @@ def create_initial_task_takeover_state(ticket_key: str, **kwargs: Any) -> TaskTa
         "review_feedback": None,
         "qualitative_review_retry_count": 0,
         "qualitative_review_failed": False,
+        "commit_info": {
+            "sha": None,
+            "message": None,
+            "committed": False,
+        },
         "implementation_push_pending": False,
         "implementation_push_pending_task": None,
         "persistence_retry_count": 0,

tests/sandbox/test_task_execution.py

@@ -135,8 +135,8 @@ async def test_execute_task_changes_successful_workflow(
                     return_value=mock_jira,
                 ),
                 patch(
-                    "forge.workflow.nodes.task_takeover_execution.GitOperations",
-                    return_value=mock_git,
+                    "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                    return_value=(str(workspace_path), mock_git),
                 ),
                 patch("forge.workflow.nodes.task_takeover_execution.get_settings"),
             ):
@@ -198,8 +198,8 @@ async def test_build_and_test_recovery_workflow_iterative_self_correction(
                     return_value=mock_jira,
                 ),
                 patch(
-                    "forge.workflow.nodes.task_takeover_execution.GitOperations",
-                    return_value=mock_git_fail,
+                    "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                    return_value=(str(workspace_path), mock_git_fail),
                 ),
                 patch("forge.workflow.nodes.task_takeover_execution.get_settings"),
             ):
@@ -233,8 +233,8 @@ async def test_build_and_test_recovery_workflow_iterative_self_correction(
                     return_value=mock_jira,
                 ),
                 patch(
-                    "forge.workflow.nodes.task_takeover_execution.GitOperations",
-                    return_value=mock_git_success,
+                    "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                    return_value=(str(workspace_path), mock_git_success),
                 ),
                 patch("forge.workflow.nodes.task_takeover_execution.get_settings"),
             ):

tests/unit/workflow/nodes/test_task_takeover_execution.py

@@ -180,6 +180,38 @@ async def test_missing_workspace_path(self) -> None:
         assert result_state["last_error"] == "Workspace not set up"
         assert result_state["current_node"] == "execute_task_changes"
 
+    @pytest.mark.asyncio
+    async def test_preserves_earlier_successful_commit_on_retry(self) -> None:
+        """Verify that an earlier successful commit is preserved during a subsequent no-op feedback retry."""
+        state = _make_state()
+        state["commit_info"] = {
+            "sha": "previous_sha_123",
+            "message": "previous_commit",
+            "committed": True,
+        }
+        mock_jira = _make_mock_jira()
+        mock_runner = _make_mock_runner()
+        mock_git = _make_mock_git(has_changes=False, sha="current_sha_456")
+
+        with (
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.JiraClient", return_value=mock_jira
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.ContainerRunner",
+                return_value=mock_runner,
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                return_value=("/tmp/ws", mock_git),
+            ),
+            patch("forge.workflow.nodes.task_takeover_execution.get_settings"),
+        ):
+            result_state = await execute_task_changes(state)
+
+        assert result_state["commit_info"]["committed"] is True
+        assert result_state["commit_info"]["sha"] == "current_sha_456"
+
     @pytest.mark.asyncio
     async def test_unexpected_exception(self) -> None:
         """Test that unexpected exceptions are caught, logged, and updated in state."""

tests/unit/workflow/task_takeover/test_graph.py

@@ -7,6 +7,7 @@
 
 from forge.models.workflow import TicketType
 from forge.workflow.task_takeover.graph import (
+    _route_after_qualitative_review,
     _route_after_triage_check,
     build_task_takeover_graph,
     route_entry,
@@ -118,3 +119,64 @@ def test_build_task_takeover_graph(self) -> None:
         # Compile the graph to verify correctness
         compiled_graph = graph.compile()
         assert compiled_graph is not None
+
+
+class TestRouteAfterQualitativeReview:
+    """Tests for _route_after_qualitative_review routing logic."""
+
+    def test_route_after_qualitative_review_no_changes_escalates(self) -> None:
+        """Verifies: When qualitative_review_retry_count reaches the maximum attempts limit (e.g., 2) and commit_info.committed is False (or commit_info is missing/empty), _route_after_qualitative_review returns 'escalate_blocked'."""
+        state = _task_state(
+            qualitative_review_retry_count=2,
+            review_verdict="tests_incomplete",
+            commit_info={"committed": False},
+        )
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
+
+        state_missing = _task_state(
+            qualitative_review_retry_count=2,
+            review_verdict="tests_incomplete",
+        )
+        assert _route_after_qualitative_review(state_missing) == "escalate_blocked"
+
+    def test_route_after_qualitative_review_with_last_error_escalates(self) -> None:
+        """Verifies: When qualitative_review_retry_count reaches the maximum attempts limit (e.g., 2) and state.last_error is set, _route_after_qualitative_review returns 'escalate_blocked'."""
+        state = _task_state(
+            qualitative_review_retry_count=2,
+            review_verdict="tests_incomplete",
+            last_error="Container execution failed",
+            commit_info={"committed": True},
+        )
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
+
+    def test_route_after_qualitative_review_with_changes_creates_pr(self) -> None:
+        """Verifies: When qualitative_review_retry_count reaches/exceeds the limit, state.last_error is None / empty, and commit_info.committed is True, _route_after_qualitative_review still successfully proceeds to 'create_pr'."""
+        state = _task_state(
+            qualitative_review_retry_count=2,
+            review_verdict="tests_incomplete",
+            last_error=None,
+            commit_info={"committed": True},
+        )
+        assert _route_after_qualitative_review(state) == "create_pr"
+
+    def test_route_after_qualitative_review_infrastructure_failure_routes_to_escalate_blocked_at_limit(
+        self,
+    ) -> None:
+        """Verifies: When qualitative review has an active error and no verdict, it escalates to escalate_blocked if limit is reached."""
+        state = _task_state(
+            qualitative_review_retry_count=2,
+            review_verdict=None,
+            last_error="Docker/Podman daemon not running",
+        )
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
+
+    def test_route_after_qualitative_review_infrastructure_failure_retries_under_limit(
+        self,
+    ) -> None:
+        """Verifies: When qualitative review has an active error and no verdict, it retries review under the limit."""
+        state = _task_state(
+            qualitative_review_retry_count=1,
+            review_verdict=None,
+            last_error="Docker/Podman daemon not running",
+        )
+        assert _route_after_qualitative_review(state) == "run_qualitative_review"

tests/workflow/test_task_takeover_graph.py

@@ -196,16 +196,40 @@ def test_route_after_qualitative_review_failed_under_limit(self) -> None:
         assert _route_after_qualitative_review(state) == "execute_task_changes"
 
     def test_route_after_qualitative_review_failed_at_or_above_limit(self) -> None:
-        """If review is failed or incomplete and at/above the limit, proceed to PR creation."""
+        """If review is failed or incomplete and at/above the limit, proceed to PR creation if changes exist."""
         from forge.workflow.task_takeover.graph import _route_after_qualitative_review
 
         state = make_task_state(
             review_verdict="tests_incomplete",
             qualitative_review_retry_count=2,
+            commit_info={"committed": True},
         )
         # retry_count of 2 is at/above the limit of 2, so stop retrying but keep Jira silent.
         assert _route_after_qualitative_review(state) == "create_pr"
 
+    def test_route_after_qualitative_review_no_changes_escalates(self) -> None:
+        """When qualitative_review_retry_count reaches max and commit_info.committed is False, escalate."""
+        from forge.workflow.task_takeover.graph import _route_after_qualitative_review
+
+        state = make_task_state(
+            review_verdict="tests_incomplete",
+            qualitative_review_retry_count=2,
+            commit_info={"committed": False},
+        )
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
+
+    def test_route_after_qualitative_review_with_last_error_escalates(self) -> None:
+        """When qualitative_review_retry_count reaches max and state.last_error is set, escalate."""
+        from forge.workflow.task_takeover.graph import _route_after_qualitative_review
+
+        state = make_task_state(
+            review_verdict="tests_incomplete",
+            qualitative_review_retry_count=2,
+            commit_info={"committed": True},
+            last_error="Container execution failed",
+        )
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
+
     def test_route_after_qualitative_review_error_without_verdict_retries(self) -> None:
         """Review execution errors retry the review without rerunning implementation."""
         from forge.workflow.task_takeover.graph import _route_after_qualitative_review
@@ -216,16 +240,16 @@ def test_route_after_qualitative_review_error_without_verdict_retries(self) -> N
         )
         assert _route_after_qualitative_review(state) == "run_qualitative_review"
 
-    def test_route_after_qualitative_review_error_at_cap_skips(self) -> None:
-        """Review execution errors proceed to PR after the bounded retry cap."""
+    def test_route_after_qualitative_review_error_at_cap_escalates(self) -> None:
+        """Review execution errors escalate after retry limit when error is present with review_verdict=None."""
         from forge.workflow.task_takeover.graph import _route_after_qualitative_review
 
         state = make_task_state(
             last_error="Review container unavailable",
             qualitative_review_retry_count=2,
-            qualitative_review_failed=True,
+            review_verdict=None,
         )
-        assert _route_after_qualitative_review(state) == "create_pr"
+        assert _route_after_qualitative_review(state) == "escalate_blocked"
 
 
 class TestPostPrRouting: