[AISOS-1977] [forge] take over tasks

CLAUDE.md

@@ -109,11 +109,17 @@ podman rm $(podman ps -a --filter name=forge- -q)
 | `forge:spec-pending` | Awaiting spec approval |
 | `forge:plan-pending` | Awaiting plan approval |
 | `forge:task-pending` | Awaiting task approval |
+| `forge:task-takeover` | Standalone task/epic takeover trigger |
+| `forge:task-triage-pending` | Task takeover awaiting triage completion |
+| `forge:task-plan-pending` | Task takeover awaiting plan approval |
+| `forge:task-plan-approved` | Task takeover plan approved |
+| `forge:managed:task` | Task identity preservation label |
+| `forge:managed:task-takeover` | Task takeover identity preservation label |
 | `forge:blocked` | Workflow blocked, needs intervention |
 | `forge:retry` | Trigger retry of failed step |
 | `forge:yolo` | Autonomous mode — skip all artifact approval gates (see warning below) |
 
-> **⚠️ Warning — `forge:yolo`:** This label removes all human checkpoints for PRD, spec, plan, and task approval. Forge will proceed autonomously from ticket creation to implementation without pausing for review. Only use this on tickets where you are confident in the requirements and comfortable with Forge making all planning decisions. It does not bypass code review (the human review gate on the implementation PR is always required).
+> **⚠️ Warning — `forge:yolo`:** This label removes all human checkpoints for PRD, spec, plan, task, and task plan approval. Forge will proceed autonomously from ticket creation to implementation without pausing for review. Only use this on tickets where you are confident in the requirements and comfortable with Forge making all planning decisions. It does not bypass code review (the human review gate on the implementation PR is always required).
 
 ## Jira Comment Syntax
 

docs/guide/labels.md

@@ -24,17 +24,29 @@ These labels advance the pipeline. Forge watches for label changes via Jira webh
 | Plan Approval Gate | `forge:plan-pending` | Forge | Plan posted; waiting for approval |
 | Plan Approval Gate | `forge:plan-approved` | Human | Approve plan and trigger task decomposition + implementation |
 
+### Task Takeover Workflow
+
+Standalone Tasks and Epics can be processed using Task Takeover trigger labels. These tickets bypass the standard parent Feature validation.
+
+| Stage | Pending Label | Approved Label | Purpose |
+|-------|--------------|----------------|---------|
+| Triage | `forge:task-triage-pending` | _N/A_ | Standalone ticket is missing required fields; waiting for update |
+| Plan Approval | `forge:task-plan-pending` | `forge:task-plan-approved` | Plan is posted; waiting for approval |
+
 ## Control Labels
 
 | Label | Purpose |
 |-------|---------|
 | `forge:managed` | Marks the ticket for Forge automation. Add this when creating a ticket to start the workflow. |
+| `forge:task-takeover` | Triggers the Task Takeover workflow for standalone Tasks or Epics. |
+| `forge:managed:task` | Identity preservation label used during Task Takeover transitions. |
+| `forge:managed:task-takeover` | Identity preservation label used during Task Takeover transitions. |
 | `forge:blocked` | Set by Forge when a stage fails. Forge posts a comment with the error. |
 | `forge:retry` | Add this to resume from the exact node that failed. Forge removes it after resuming. |
 
 ## How to Use Labels
 
-**Starting a workflow:** Create a Jira issue and add `forge:managed`. Forge detects the issue type (Feature or Bug) and begins the appropriate pipeline.
+**Starting a workflow:** Create a Jira issue and add `forge:managed`. Forge detects the issue type (Feature or Bug) and begins the appropriate pipeline. For standalone Tasks or Epics, add `forge:task-takeover` (or another configured trigger label) to initiate the Task Takeover workflow.
 
 **Approving a stage:** When Forge posts a PRD, spec, or other artifact, it sets the `forge:*-pending` label. Change it to `forge:*-approved` to advance the workflow. Do not add the approved label manually before Forge posts — it won't be recognized until the pending state is set.
 

docs/reference/config.md

@@ -128,3 +128,21 @@ These variables are used by `docker-compose.yml`, `devtools/docker-compose.dev.y
 ### MCP Servers
 
 MCP server configuration lives in `mcp-servers.json`, not `.env`. See the [MCP servers section](https://github.com/forge-sdlc/forge/blob/main/mcp-servers.json) of the repository.
+
+## Task Takeover Configuration
+
+Task Takeover allows Forge to process standalone Task and Epic issues directly from Jira. When a standalone Task/Epic issue is labeled with a task takeover trigger label, Forge bypasses the parent validation check and executes the task directly.
+
+Configuration settings can be defined in `Settings` under the `task_takeover` key (which can also be configured using environment variables as a JSON string under `TASK_TAKEOVER` or within the application config).
+
+### Settings Schema
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `enabled` | `bool` | `False` | Whether Task Takeover is enabled. |
+| `issue_types` | `list[str]` | `[]` | List of Jira issue types that support task takeover (e.g., `["Task", "Epic"]`). |
+| `require_tests` | `bool` | `True` | Whether tests are required to pass before merging the code. |
+| `review_max_attempts` | `int` | `2` | Maximum number of PR review fix attempts. |
+| `labels.trigger` | `str` | `"forge:task-takeover"` | Label that triggers the Task Takeover workflow. |
+| `labels.pending` | `str` | `"forge:task-plan-pending"` | Label set by Forge when a task plan is pending approval. |
+| `labels.approved` | `str` | `"forge:task-plan-approved"` | Label used by humans to approve the task plan. |

src/forge/api/routes/jira.py

@@ -109,7 +109,7 @@ async def receive_jira_webhook(
         # Record webhook received metric
         record_webhook_received(source="jira", event_type=webhook_data.event_type)
 
-        # Filter: only process issues with forge:managed label
+        # Filter: only process issues with forge:managed label or task-takeover triggers
         issue_labels = payload.get("issue", {}).get("fields", {}).get("labels", [])
         has_forge_managed = "forge:managed" in issue_labels
 
@@ -122,7 +122,27 @@ async def receive_jira_webhook(
                     has_forge_managed = True
                     break
 
-        if not has_forge_managed:
+        # Detect task-takeover trigger labels
+        has_takeover_trigger = False
+        if settings.task_takeover and settings.task_takeover.enabled:
+            takeover_triggers = {
+                "forge:task-takeover",
+                "forge:managed:task",
+                "forge:managed:task-takeover",
+            }
+            if settings.task_takeover.labels and settings.task_takeover.labels.trigger:
+                takeover_triggers.add(settings.task_takeover.labels.trigger)
+
+            has_takeover_trigger = any(label in issue_labels for label in takeover_triggers)
+            for item in changelog_items:
+                if item.get("field") == "labels":
+                    to_labels = item.get("toString", "") or ""
+                    updated_labels = to_labels.split()
+                    if any(label in updated_labels for label in takeover_triggers):
+                        has_takeover_trigger = True
+                        break
+
+        if not (has_forge_managed or has_takeover_trigger):
             span.set_attribute("forge.skipped", True)
             span.set_attribute("forge.skip_reason", "missing forge:managed label")
             logger.debug(f"Skipping {webhook_data.ticket_key}: missing forge:managed label")
@@ -163,6 +183,13 @@ async def receive_jira_webhook(
                     f"Routing {issue_type} {source_ticket_key} webhook "
                     f"to parent Feature {routing_ticket_key}"
                 )
+            elif has_takeover_trigger and issue_type in ("Epic", "Task"):
+                # Bypass parent validation for Epic/Task if takeover trigger label is present.
+                # routing_ticket_key remains webhook_data.ticket_key, source_ticket_key remains None.
+                logger.info(
+                    f"Bypassing parent checks for standalone {issue_type} "
+                    f"{webhook_data.ticket_key} due to task-takeover trigger label."
+                )
             else:
                 # Epics/Tasks without forge:parent are invalid - reject
                 span.set_attribute("forge.skipped", True)

src/forge/config.py

@@ -4,7 +4,7 @@
 from functools import cached_property, lru_cache
 from typing import TYPE_CHECKING, Literal
 
-from pydantic import Field, SecretStr
+from pydantic import BaseModel, Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 if TYPE_CHECKING:
@@ -13,6 +13,24 @@
 logger = logging.getLogger(__name__)
 
 
+class TaskTakeoverLabels(BaseModel):
+    """Labels used for task takeover workflow."""
+
+    trigger: str = "forge:task-takeover"
+    pending: str = "forge:task-plan-pending"
+    approved: str = "forge:task-plan-approved"
+
+
+class TaskTakeoverSettings(BaseModel):
+    """Settings configuration for task takeover."""
+
+    enabled: bool = False
+    issue_types: list[str] = Field(default_factory=list)
+    labels: TaskTakeoverLabels = Field(default_factory=TaskTakeoverLabels)
+    require_tests: bool = True
+    review_max_attempts: int = 2
+
+
 class Settings(BaseSettings):
     """Application settings loaded from environment variables."""
 
@@ -360,6 +378,12 @@ def ignored_ci_checks(self) -> list[str]:
         description="Enable distributed tracing",
     )
 
+    # Task Takeover Configuration
+    task_takeover: TaskTakeoverSettings = Field(
+        default_factory=TaskTakeoverSettings,
+        description="Configuration settings for Task Takeover feature",
+    )
+
     @property
     def langfuse_enabled(self) -> bool:
         """Check if Langfuse tracing is enabled and configured."""

src/forge/integrations/agents/agent.py

@@ -1159,17 +1159,40 @@ async def answer_question(
         generation_context = context.get("generation_context", {})
         raw_requirements = generation_context.get("raw_requirements", "Not available")
 
-        prompt = load_prompt(
-            "answer-question",
-            artifact_type=artifact_type,
-            artifact_content=artifact_content,
-            raw_requirements=raw_requirements,
-            question=question,
-        )
+        ticket_type = context.get("ticket_type")
+        ticket_type_str = ""
+        if ticket_type is not None:
+            ticket_type_str = (
+                ticket_type.value if hasattr(ticket_type, "value") else str(ticket_type)
+            )
+
+        if (
+            artifact_type == "plan"
+            and context.get("current_node") == "task_plan_approval_gate"
+            and ticket_type_str == "task"
+        ):
+            prompt = load_prompt(
+                "task-takeover-qa",
+                ticket_key=context.get("ticket_key", ""),
+                summary=context.get("summary", ""),
+                description=context.get("description", ""),
+                plan_content=artifact_content,
+                question=question,
+            )
+            task_name = "task-takeover-qa"
+        else:
+            prompt = load_prompt(
+                "answer-question",
+                artifact_type=artifact_type,
+                artifact_content=artifact_content,
+                raw_requirements=raw_requirements,
+                question=question,
+            )
+            task_name = "answer-question"
 
-        logger.info(f"Answering question about {artifact_type}")
+        logger.info(f"Answering question about {artifact_type} using task={task_name}")
         result = await self.run_task(
-            task="answer-question",
+            task=task_name,
             prompt=prompt,
             context={
                 "artifact_type": artifact_type,

src/forge/integrations/jira/client.py

@@ -742,13 +742,15 @@ async def set_workflow_label(
         # Get current labels
         current_labels = await self.get_labels(issue_key)
 
-        # Find forge: labels to remove (except the new one and forge:managed)
+        # Find forge: labels to remove (except the new one, forge:managed, and identity preservation labels)
         labels_to_remove = [
             label
             for label in current_labels
             if label.startswith(remove_prefix)
             and label != new_label.value
             and label != ForgeLabel.FORGE_MANAGED.value
+            and label != "forge:managed:task"
+            and label != "forge:managed:task-takeover"
         ]
 
         # Build update operations

src/forge/integrations/jira/models.py

@@ -183,6 +183,8 @@ def extract_children(nodes: list[dict[str, Any]]) -> list[str]:
             return [block for block in blocks if block]
 
         blocks = extract_blocks(adf)
+        if adf.get("type") == "doc" and not blocks:
+            return ""
         return "\n\n".join(blocks) if blocks else str(adf)
 
 

src/forge/models/workflow.py

@@ -124,6 +124,12 @@ class ForgeLabel(StrEnum):
     RCA_APPROVED = "forge:rca-approved"
     TRIAGE_PENDING = "forge:triage-pending"
 
+    # Task Takeover workflow
+    TASK_TAKEOVER = "forge:task-takeover"
+    TASK_TRIAGE_PENDING = "forge:task-triage-pending"
+    TASK_PLAN_PENDING = "forge:task-plan-pending"
+    TASK_PLAN_APPROVED = "forge:task-plan-approved"
+
     # General
     FORGE_MANAGED = "forge:managed"
     BLOCKED = "forge:blocked"

src/forge/orchestrator/worker.py

@@ -61,6 +61,7 @@ def _is_workflow_errored(state: dict) -> bool:
     "prd_approval_gate",
     "spec_approval_gate",
     "plan_approval_gate",
+    "task_plan_approval_gate",
     "task_approval_gate",
     "rca_option_gate",
 }
@@ -258,9 +259,10 @@ async def _process_workflow(self, message: QueueMessage) -> None:
                 )
             else:
                 # Use router to resolve which workflow to use
+                labels = message.payload.get("issue", {}).get("fields", {}).get("labels", []) or []
                 workflow_instance = self.router.resolve(
                     ticket_type=ticket_type,
-                    labels=[],  # TODO: Extract labels from message payload
+                    labels=labels,
                     event=message.payload,
                 )
 
@@ -580,6 +582,8 @@ async def _handle_resume_event(
                     approval_stage = "prd"
                 elif "spec-approved" in to_labels.lower():
                     approval_stage = "spec"
+                elif "task-plan-approved" in to_labels.lower():
+                    approval_stage = "task_plan"
                 elif "plan-approved" in to_labels.lower():
                     approval_stage = "plan"
                 elif "task-approved" in to_labels.lower():
@@ -597,10 +601,18 @@ async def _handle_resume_event(
                     "decompose_epics": "plan",
                     "regenerate_all_epics": "plan",
                     "update_single_epic": "plan",
+                    "task_plan_approval_gate": "task_plan",
                     "task_approval_gate": "task",
                     "generate_tasks": "task",
                 }
                 expected_stage = node_to_stage.get(current_node)
+                if current_node == "plan_approval_gate" and current_state.get("ticket_type") in (
+                    "Task",
+                    "Epic",
+                    TicketType.TASK,
+                    TicketType.EPIC,
+                ):
+                    expected_stage = "task_plan"
 
                 if approval_stage and expected_stage and approval_stage == expected_stage:
                     is_approved = True
@@ -622,7 +634,11 @@ async def _handle_resume_event(
             gate_to_approved_label = {
                 "prd_approval_gate": "forge:prd-approved",
                 "spec_approval_gate": "forge:spec-approved",
-                "plan_approval_gate": "forge:plan-approved",
+                "plan_approval_gate": "forge:task-plan-approved"
+                if current_state.get("ticket_type")
+                in ("Task", "Epic", TicketType.TASK, TicketType.EPIC)
+                else "forge:plan-approved",
+                "task_plan_approval_gate": "forge:task-plan-approved",
                 "task_approval_gate": "forge:task-approved",
             }
             expected_label = gate_to_approved_label.get(current_node)
@@ -1121,6 +1137,7 @@ async def _handle_resume_event(
                 "plan_approval_gate",
                 "task_approval_gate",
                 "plan_approval_gate_bug",
+                "task_plan_approval_gate",
             }
             prev_error = current_state.get("last_error")
             is_paused_at_gate = current_state.get("is_paused") and current_node in approval_gates
@@ -1314,6 +1331,7 @@ def _stage_label_for_node(current_node: str) -> str:
             "update_single_epic": "the plan",
             "rca_option_gate": "the RCA",
             "plan_approval_gate_bug": "the plan",
+            "task_plan_approval_gate": "the task plan",
             "task_approval_gate": "the tasks",
             "generate_tasks": "the tasks",
             "regenerate_all_tasks": "the tasks",
@@ -1535,8 +1553,23 @@ def _extract_ticket_type(self, message: QueueMessage) -> TicketType:
             # by the Jira webhook handler. The payload still carries the child's
             # issue type, which won't match any workflow. Fall through to UNKNOWN
             # so _find_workflow_by_state resolves it from checkpoint.
+            # stand-alone task takeover events (which have trigger labels) bypass child checks.
+            labels = fields.get("labels", []) or []
+            takeover_triggers = {
+                "forge:task-takeover",
+                "forge:managed:task",
+                "forge:managed:task-takeover",
+            }
+            if (
+                self.settings.task_takeover
+                and self.settings.task_takeover.labels
+                and self.settings.task_takeover.labels.trigger
+            ):
+                takeover_triggers.add(self.settings.task_takeover.labels.trigger)
+            is_takeover = any(label in labels for label in takeover_triggers)
+
             child_types = {"Epic", "Task", "Sub-task"}
-            if ticket_type_str in child_types:
+            if ticket_type_str in child_types and not is_takeover:
                 return TicketType.UNKNOWN
 
             # Map string to TicketType enum