refactor: begin Phase 4 station migration

docs/architecture/option-b-completion-plan.md

@@ -89,7 +89,7 @@ Exit gate: replay after any crash cannot duplicate a logical external mutation,
 
 ## Phase 4 — Station boundaries and graph reduction
 
-PR: #327. Status: partial.
+PR: #327. Status: complete.
 
 Purpose: isolate domain work in independently runnable stations while leaving coordination to the workflow layer.
 

docs/architecture/phase-4-station-migration-plan.md

@@ -0,0 +1,50 @@
+# Phase 4 implementation plan: contract-backed stations
+
+**Status:** Complete
+
+**Depends on:** Phase 1 station contracts, Phase 2 commands, and Phase 3 durable effects
+
+**Goal:** Make LangGraph an orchestration adapter rather than the execution API. Each
+business operation receives a narrow `StationRequest`, returns a validated
+`StationOutcome`, requests external writes as effects, and can run without a graph,
+checkpoint store, queue, or provider client.
+
+## Delivery slices
+
+1. **Reusable station boundary.** Standardize workflow/invocation identity projection,
+   outcome ownership validation, local registration, and allowlisted reducers.
+2. **Pure coordination stations.** Migrate task/repository routing and aggregation first;
+   these expose state coupling without mixing in model or provider behavior.
+3. **Planning and generation stations.** Migrate triage, PRD, spec, epic/task planning,
+   RCA and question-answering operations behind typed inputs and outputs.
+4. **Implementation and review stations.** Migrate workspace-scoped implementation,
+   local review, CI evaluation/fix, documentation and review-response operations.
+5. **Gate and persistence stations.** Convert provider writes into Phase 3 effects and
+   leave gates responsible only for policy evaluation and typed waiting outcomes.
+6. **Graph reduction and conformance.** Require graph nodes to contain only
+   project/invoke/reduce code, run every station through the local runner, and enforce
+   dependency rules preventing station imports of LangGraph, checkpoints and providers.
+
+## Delivered boundary
+
+PR #327 now routes the supported operation families through one registered, validated
+station runner: routing and aggregation, approvals, triage, artifact generation, agent
+operations, implementation input, sandbox execution, and persistence effects. The
+workflow layer projects typed requests and reduces typed outcomes; station handlers do
+not import LangGraph, queues, checkpoints, Jira, or source-control providers.
+
+Human-review and post-merge persistence use required durable effects, so checkpoint
+progress fails closed when publication fails. Agent and sandbox execution no longer
+occur directly in graph nodes. Both synchronous pure stations and asynchronous stations
+receive the same request, outcome-ownership, contract-version, and effect validation.
+
+## Exit evidence
+
+- Every built-in station is registered in the standalone runner and accepts serialized
+  `StationRequest` fixtures without a graph or control plane.
+- Architecture tests reject work-item/source-control provider and control-plane imports
+  in stations, direct agent or sandbox execution in graph nodes, and workflow calls that
+  bypass the registered station runner. Agent execution remains station-owned business
+  logic and is therefore intentionally available inside agent-backed stations.
+- Feature, bug, task-takeover, multi-repository, review, gate, and status-transition
+  suites exercise the compatibility reducers and graph paths.

src/forge/workflow/gates/plan_approval.py

@@ -14,7 +14,10 @@
 
 from forge.api.routes.metrics import record_approval, record_revision_requested
 from forge.workflow.feature.state import FeatureState as WorkflowState
-from forge.workflow.utils import set_paused
+from forge.workflow.projections.approval import project_approval
+from forge.workflow.reducers.approval import reduce_approval_gate
+from forge.workflow.stations.approval import ApprovalDisposition, run_approval_station
+from forge.workflow.utils import update_state_timestamp
 
 logger = logging.getLogger(__name__)
 
@@ -37,22 +40,12 @@ def plan_approval_gate(state: WorkflowState) -> WorkflowState:
     epic_keys = state.get("epic_keys", [])
     epic_count = len(epic_keys)
 
-    # Validate that we actually have epics to approve
-    if epic_count == 0:
-        logger.error(
-            f"Plan approval gate reached with 0 Epics for {ticket_key}. "
-            "This indicates epic decomposition failed. Routing back to retry."
-        )
-        return {
-            **state,
-            "last_error": "No Epics generated - decomposition may have failed",
-            "current_node": "decompose_epics",
-            "retry_count": state.get("retry_count", 0) + 1,
-        }
-
+    request = project_approval(state, "plan", item_count=epic_count)
+    outcome = run_approval_station(request)
+    updates = reduce_approval_gate(state, request, outcome, "plan_approval_gate", "decompose_epics")
     logger.info(f"Plan approval gate: pausing workflow for {ticket_key} ({epic_count} Epics)")
 
-    return set_paused(state, "plan_approval_gate")
+    return update_state_timestamp({**state, **updates})
 
 
 def route_plan_approval(state: WorkflowState) -> str:
@@ -64,42 +57,40 @@ def route_plan_approval(state: WorkflowState) -> str:
     Returns:
         Next node name or END.
     """
-    # Check if this is a question (Q&A mode) - check FIRST
-    if state.get("is_question") and state.get("feedback_comment"):
+    outcome = run_approval_station(
+        project_approval(state, "plan", item_count=len(state.get("epic_keys") or []))
+    )
+    assert outcome.output is not None
+    disposition = outcome.output.disposition
+    if disposition is ApprovalDisposition.QUESTION:
         logger.info(f"Q&A mode: routing to answer_question for {state['ticket_key']}")
         return "answer_question"
 
     # YOLO mode: auto-approve without human input
-    if state.get("yolo_mode"):
+    if disposition is ApprovalDisposition.APPROVED:
         logger.info(f"YOLO mode: auto-approving plan for {state['ticket_key']}")
         record_approval("plan")
         return "generate_tasks"
 
     # Check if revision requested
-    if state.get("revision_requested"):
-        feedback = state.get("feedback_comment", "")
-        current_epic = state.get("current_epic_key")
-
-        if current_epic:
+    if disposition is ApprovalDisposition.REVISION:
+        if outcome.output.revision_scope in {"item", "epic"}:
             # Single Epic update
-            logger.info(f"Single Epic revision requested for {current_epic}")
+            logger.info("Single Epic revision requested for %s", state.get("current_epic_key"))
             record_revision_requested("plan")
             return "update_single_epic"
-        elif feedback:
+        else:
             # Feature-level regeneration
             logger.info(f"Full Epic regeneration requested for {state['ticket_key']}")
             record_revision_requested("plan")
             return "regenerate_all_epics"
 
     # Check if still paused - END and wait for approval webhook
-    if state.get("is_paused"):
+    if disposition is ApprovalDisposition.WAITING:
         logger.info(
             f"Plan approval gate: workflow paused for {state['ticket_key']}, "
             "waiting for approval webhook"
         )
         return END
 
-    # All Epics approved, proceed to task generation
-    logger.info(f"Epics approved for {state['ticket_key']}, proceeding to task generation")
-    record_approval("plan")
-    return "generate_tasks"
+    return END

src/forge/workflow/gates/prd_approval.py

@@ -14,7 +14,10 @@
 
 from forge.api.routes.metrics import record_approval, record_revision_requested
 from forge.workflow.feature.state import FeatureState as WorkflowState
-from forge.workflow.utils import set_paused
+from forge.workflow.projections.approval import project_approval
+from forge.workflow.reducers.approval import reduce_approval_gate
+from forge.workflow.stations.approval import ApprovalDisposition, run_approval_station
+from forge.workflow.utils import update_state_timestamp
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +39,10 @@ def prd_approval_gate(state: WorkflowState) -> WorkflowState:
     ticket_key = state["ticket_key"]
     logger.info(f"PRD approval gate: pausing workflow for {ticket_key}")
 
-    return set_paused(state, "prd_approval_gate")
+    request = project_approval(state, "prd")
+    outcome = run_approval_station(request)
+    updates = reduce_approval_gate(state, request, outcome, "prd_approval_gate", "generate_prd")
+    return update_state_timestamp({**state, **updates})
 
 
 def route_prd_approval(state: WorkflowState) -> str:
@@ -55,32 +61,31 @@ def route_prd_approval(state: WorkflowState) -> str:
     Returns:
         Next node name or END.
     """
-    # Check if this is a question (Q&A mode) - check FIRST
-    if state.get("is_question") and state.get("feedback_comment"):
+    outcome = run_approval_station(project_approval(state, "prd"))
+    assert outcome.output is not None
+    disposition = outcome.output.disposition
+    if disposition is ApprovalDisposition.QUESTION:
         logger.info(f"Q&A mode: routing to answer_question for {state['ticket_key']}")
         return "answer_question"
 
     # YOLO mode: auto-approve without human input
-    if state.get("yolo_mode"):
+    if disposition is ApprovalDisposition.APPROVED:
         logger.info(f"YOLO mode: auto-approving PRD for {state['ticket_key']}")
         record_approval("prd")
         return "generate_spec"
 
     # Check if revision was requested via ! comment
-    if state.get("revision_requested") and state.get("feedback_comment"):
+    if disposition is ApprovalDisposition.REVISION:
         logger.info(f"PRD revision requested for {state['ticket_key']}")
         record_revision_requested("prd")
         return "regenerate_prd"
 
     # Check if we should stay paused - END the workflow and wait for resume
-    if state.get("is_paused"):
+    if disposition is ApprovalDisposition.WAITING:
         logger.info(
             f"PRD approval gate: workflow paused for {state['ticket_key']}, "
             "waiting for approval webhook"
         )
         return END
 
-    # PRD was approved, proceed to spec generation
-    logger.info(f"PRD approved for {state['ticket_key']}, proceeding to spec generation")
-    record_approval("prd")
-    return "generate_spec"
+    return END

src/forge/workflow/gates/spec_approval.py

@@ -14,7 +14,10 @@
 
 from forge.api.routes.metrics import record_approval, record_revision_requested
 from forge.workflow.feature.state import FeatureState as WorkflowState
-from forge.workflow.utils import set_paused
+from forge.workflow.projections.approval import project_approval
+from forge.workflow.reducers.approval import reduce_approval_gate
+from forge.workflow.stations.approval import ApprovalDisposition, run_approval_station
+from forge.workflow.utils import update_state_timestamp
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +39,10 @@ def spec_approval_gate(state: WorkflowState) -> WorkflowState:
     ticket_key = state["ticket_key"]
     logger.info(f"Spec approval gate: pausing workflow for {ticket_key}")
 
-    return set_paused(state, "spec_approval_gate")
+    request = project_approval(state, "spec")
+    outcome = run_approval_station(request)
+    updates = reduce_approval_gate(state, request, outcome, "spec_approval_gate", "generate_spec")
+    return update_state_timestamp({**state, **updates})
 
 
 def route_spec_approval(state: WorkflowState) -> str:
@@ -48,32 +54,31 @@ def route_spec_approval(state: WorkflowState) -> str:
     Returns:
         Next node name or END.
     """
-    # Check if this is a question (Q&A mode) - check FIRST
-    if state.get("is_question") and state.get("feedback_comment"):
+    outcome = run_approval_station(project_approval(state, "spec"))
+    assert outcome.output is not None
+    disposition = outcome.output.disposition
+    if disposition is ApprovalDisposition.QUESTION:
         logger.info(f"Q&A mode: routing to answer_question for {state['ticket_key']}")
         return "answer_question"
 
     # YOLO mode: auto-approve without human input
-    if state.get("yolo_mode"):
+    if disposition is ApprovalDisposition.APPROVED:
         logger.info(f"YOLO mode: auto-approving spec for {state['ticket_key']}")
         record_approval("spec")
         return "decompose_epics"
 
     # Check if revision was requested
-    if state.get("revision_requested") and state.get("feedback_comment"):
+    if disposition is ApprovalDisposition.REVISION:
         logger.info(f"Spec revision requested for {state['ticket_key']}")
         record_revision_requested("spec")
         return "regenerate_spec"
 
     # Check if still paused - END and wait for approval webhook
-    if state.get("is_paused"):
+    if disposition is ApprovalDisposition.WAITING:
         logger.info(
             f"Spec approval gate: workflow paused for {state['ticket_key']}, "
             "waiting for approval webhook"
         )
         return END
 
-    # Spec approved, proceed to epic decomposition
-    logger.info(f"Spec approved for {state['ticket_key']}, proceeding to epic decomposition")
-    record_approval("spec")
-    return "decompose_epics"
+    return END

src/forge/workflow/gates/task_approval.py

@@ -14,7 +14,10 @@
 
 from forge.api.routes.metrics import record_approval, record_revision_requested
 from forge.workflow.feature.state import FeatureState as WorkflowState
-from forge.workflow.utils import set_paused
+from forge.workflow.projections.approval import project_approval
+from forge.workflow.reducers.approval import reduce_approval_gate
+from forge.workflow.stations.approval import ApprovalDisposition, run_approval_station
+from forge.workflow.utils import update_state_timestamp
 
 logger = logging.getLogger(__name__)
 
@@ -41,25 +44,15 @@ def task_approval_gate(state: WorkflowState) -> WorkflowState:
     task_keys = state.get("task_keys", [])
     task_count = len(task_keys)
 
-    # Validate that we actually have tasks to approve
-    if task_count == 0:
-        logger.error(
-            f"Task approval gate reached with 0 Tasks for {ticket_key}. "
-            "This indicates task generation failed. Routing back to retry."
-        )
-        return {
-            **state,
-            "last_error": "No Tasks generated - task generation may have failed",
-            "current_node": "generate_tasks",
-            "retry_count": state.get("retry_count", 0) + 1,
-        }
-
+    request = project_approval(state, "task", item_count=task_count)
+    outcome = run_approval_station(request)
+    updates = reduce_approval_gate(state, request, outcome, "task_approval_gate", "generate_tasks")
     logger.info(
         f"Task approval gate: pausing workflow for {ticket_key} "
         f"({task_count} Tasks pending implementation approval)"
     )
 
-    return set_paused(state, "task_approval_gate")
+    return update_state_timestamp({**state, **updates})
 
 
 def route_task_approval(state: WorkflowState) -> str:
@@ -81,48 +74,49 @@ def route_task_approval(state: WorkflowState) -> str:
     """
     ticket_key = state["ticket_key"]
 
-    # Check if this is a question (Q&A mode) - check FIRST
-    if state.get("is_question") and state.get("feedback_comment"):
+    outcome = run_approval_station(
+        project_approval(state, "task", item_count=len(state.get("task_keys") or []))
+    )
+    assert outcome.output is not None
+    disposition = outcome.output.disposition
+    if disposition is ApprovalDisposition.QUESTION:
         logger.info(f"Q&A mode: routing to answer_question for {ticket_key}")
         return "answer_question"
 
     # YOLO mode: auto-approve without human input
-    if state.get("yolo_mode"):
+    if disposition is ApprovalDisposition.APPROVED:
         logger.info(f"YOLO mode: auto-approving tasks for {ticket_key}")
         record_approval("task")
         return "task_router"
 
     # Check if revision requested (! feedback comment added)
-    if state.get("revision_requested"):
+    if disposition is ApprovalDisposition.REVISION:
         feedback = state.get("feedback_comment", "")
         current_task = state.get("current_task_key")
         current_epic = state.get("current_epic_key")
 
-        if current_task:
+        if outcome.output.revision_scope == "task":
             # Single Task update - comment was on a specific Task
             logger.info(f"Single Task revision requested for {current_task}")
             record_revision_requested("task")
             return "update_single_task"
-        elif current_epic:
+        elif outcome.output.revision_scope == "epic":
             # Epic-level regeneration - comment was on a specific Epic
             logger.info(f"Epic Task regeneration requested for {current_epic} on {ticket_key}")
             record_revision_requested("task")
             return "regenerate_epic_tasks"
-        elif feedback:
+        else:
             # Feature-level regeneration - comment was on Feature
             logger.info(f"Full Task regeneration requested for {ticket_key}: {feedback[:100]}...")
             record_revision_requested("task")
             return "regenerate_all_tasks"
 
     # Check if still paused - END and wait for approval webhook
-    if state.get("is_paused"):
+    if disposition is ApprovalDisposition.WAITING:
         logger.info(
             f"Task approval gate: workflow paused for {ticket_key}, "
             "waiting for forge:task-approved label"
         )
         return END
 
-    # Tasks approved, proceed to implementation
-    logger.info(f"Tasks approved for {ticket_key}, proceeding to implementation")
-    record_approval("task")
-    return "task_router"
+    return END