feat: persist structured workflow handoffs across workers

src/forge/workflow/base.py

@@ -43,6 +43,14 @@ class BaseState(TypedDict, total=False):
     context: dict[str, Any]
 
 
+class HandoffState(TypedDict):
+    """Durable task-continuity summary for one repository."""
+
+    content: str
+    task_key: str
+    captured_at: str
+
+
 class PRIntegrationState(TypedDict, total=False):
     """Mixin for workflows that create PRs."""
 
@@ -67,6 +75,7 @@ class PRIntegrationState(TypedDict, total=False):
     review_push_pending: bool
     review_push_pending_updates: dict[str, Any]
     review_exhaustion_report: Annotated[dict[str, Any], operator.or_]
+    handoffs: dict[str, HandoffState]
 
 
 class CIIntegrationState(TypedDict, total=False):

src/forge/workflow/bug/state.py

@@ -95,6 +95,7 @@ def create_initial_bug_state(ticket_key: str, **kwargs: Any) -> BugState:
         "persistence_retry_count": 0,
         "review_push_pending": False,
         "review_push_pending_updates": {},
+        "handoffs": {},
         "tdd_approach": False,
         "ci_status": None,
         "current_pr_url": None,

src/forge/workflow/feature/state.py

@@ -105,6 +105,7 @@ def create_initial_feature_state(ticket_key: str, **kwargs: Any) -> FeatureState
         "persistence_retry_count": 0,
         "review_push_pending": False,
         "review_push_pending_updates": {},
+        "handoffs": {},
         "ci_status": None,
         "current_pr_url": None,
         "current_pr_number": None,

src/forge/workflow/nodes/ci_evaluator.py

@@ -25,6 +25,7 @@
     set_review_pending_label,
 )
 from forge.workspace.git_ops import GitOperations
+from forge.workspace.handoff import capture_handoff
 from forge.workspace.manager import Workspace
 
 logger = logging.getLogger(__name__)
@@ -285,6 +286,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState:
     ci_fix_max = state.get("ci_fix_max_attempts", 5)
 
     jira = JiraClient()
+    fix_started = False
     try:
         message = f"🔧 CI checks failed. Analyzing failure and attempting fix ({ci_fix_attempt}/{ci_fix_max})."
         await post_status_comment(jira, ticket_key, message)
@@ -366,6 +368,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState:
         fix_prompt = load_prompt("fix-ci", fix_plan=fix_plan)
 
         runner = ContainerRunner(settings)
+        fix_started = True
         result = await runner.run(
             workspace_path=Path(workspace_path),
             task_summary=f"Apply CI fix plan (attempt {attempt})",
@@ -441,6 +444,14 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState:
                 attempt=attempt,
             )
 
+        state = capture_handoff(
+            workspace_path,
+            state.get("current_repo", ""),
+            f"{ticket_key}-ci-fix-{attempt}",
+            state,
+        )
+        fix_started = False
+
         return update_state_timestamp(
             {
                 **state,
@@ -451,6 +462,13 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState:
 
     except Exception as e:
         logger.error(f"CI fix failed for {ticket_key}: {e}")
+        if fix_started:
+            state = capture_handoff(
+                workspace_path,
+                state.get("current_repo", ""),
+                f"{ticket_key}-ci-fix-{attempt}",
+                state,
+            )
         return {
             **state,
             "last_error": str(e),

src/forge/workflow/nodes/implement_review.py

@@ -21,6 +21,7 @@
     merge_review_decisions,
     reply_to_review_decisions,
 )
+from forge.workspace.handoff import capture_handoff
 
 logger = logging.getLogger(__name__)
 
@@ -202,6 +203,7 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
     logger.info(f"Implementing PR review feedback for {ticket_key}")
 
     settings = get_settings()
+    fix_started = False
 
     try:
         try:
@@ -310,6 +312,7 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
             fix_prompt = load_prompt("implement-review-fix", ticket_key=ticket_key)
 
             runner = ContainerRunner(settings)
+            fix_started = True
             result = await runner.run(
                 workspace_path=Path(workspace_path),
                 task_summary=f"Implement PR review plan for {ticket_key}",
@@ -322,6 +325,9 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
             )
             state = merge_review_exhaustion(state, result, ticket_key, "implement_review_fix")
 
+            state = capture_handoff(workspace_path, current_repo, f"{ticket_key}-review-fix", state)
+            fix_started = False
+
             # Commit any uncommitted changes the container left
             if git.has_uncommitted_changes():
                 git.stage_all()
@@ -409,6 +415,8 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
 
     except Exception as e:
         logger.error(f"implement_review failed for {ticket_key}: {e}")
+        if fix_started and workspace_path:
+            state = capture_handoff(workspace_path, current_repo, f"{ticket_key}-review-fix", state)
         return {
             **state,
             "last_error": str(e),

src/forge/workflow/nodes/implementation.py

@@ -28,6 +28,7 @@
 from forge.workflow.utils import merge_review_exhaustion, update_state_timestamp
 from forge.workflow.utils.jira_status import post_status_comment
 from forge.workspace.git_ops import GitOperations
+from forge.workspace.handoff import capture_handoff
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +54,8 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
     current_task = state.get("current_task_key")
     task_keys = state.get("task_keys", [])
     implementation_node = _implementation_node_name(state)
+    current_repo = state.get("current_repo", "")
+    container_started = False
     recorded_workspace = state.get("workspace_path")
     local_workspace_survived = bool(recorded_workspace and Path(recorded_workspace).exists())
 
@@ -197,9 +200,9 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
         # Run implementation in container sandbox
         runner = ContainerRunner(settings)
 
-        current_repo = state.get("current_repo", "")
         # Copy list to avoid mutation after passing to runner
         implemented_tasks = list(state.get("implemented_tasks", []))
+        container_started = True
         result = await runner.run(
             workspace_path=Path(workspace_path),
             task_summary=task_summary,
@@ -219,6 +222,8 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
 
         # Collect review exhaustion data (if auto-review ran and exhausted)
         state = merge_review_exhaustion(state, result, current_task, "implement_task")
+        state = capture_handoff(workspace_path, current_repo, current_task, state)
+        container_started = False
 
         if result.success:
             logger.info(f"Container completed successfully for {current_task}")
@@ -274,6 +279,8 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
 
     except Exception as e:
         logger.error(f"Implementation failed for {current_task}: {e}")
+        if container_started:
+            state = capture_handoff(workspace_path, current_repo, current_task or ticket_key, state)
         return {
             **state,
             "last_error": str(e),

src/forge/workflow/nodes/workspace_setup.py

@@ -19,6 +19,7 @@
 )
 from forge.workspace.git_ops import GitOperations
 from forge.workspace.guardrails import GuardrailsLoader
+from forge.workspace.handoff import materialize_handoff
 from forge.workspace.manager import Workspace, WorkspaceManager
 
 WorkflowState = dict[str, Any]
@@ -44,6 +45,7 @@ def _recreate_workspace_from_fork(
     branch_name: str,
     fork_owner: str,
     fork_repo: str,
+    state: WorkflowState,
     stale_workspace_path: str | None = None,
 ) -> tuple[str, GitOperations]:
     if not branch_name or not current_repo or not fork_owner or not fork_repo:
@@ -106,6 +108,7 @@ def _recreate_workspace_from_fork(
     git.workspace.path = target_path
     git.workspace_recreated = True
     write_workspace_identity(target_path, ticket_key=ticket_key, repo_name=current_repo)
+    materialize_handoff(target_path, current_repo, state)
     logger.info(f"Workspace recreated at {target_path} for {ticket_key}")
     return str(target_path), git
 
@@ -165,6 +168,7 @@ def prepare_workspace(
                 branch_name=branch_name,
                 fork_owner=fork_owner,
                 fork_repo=fork_repo,
+                state=state,
                 stale_workspace_path=workspace_path,
             )
         return workspace_path, git
@@ -176,6 +180,7 @@ def prepare_workspace(
         branch_name=branch_name,
         fork_owner=fork_owner,
         fork_repo=fork_repo,
+        state=state,
     )
 
 
@@ -340,6 +345,7 @@ async def setup_workspace(state: WorkflowState) -> WorkflowState:
             ticket_key=ticket_key,
             repo_name=current_repo,
         )
+        materialize_handoff(workspace.path, current_repo, state)
 
         # Keep Forge handoff files local to this clone without modifying the
         # target repository's tracked .gitignore.

src/forge/workspace/handoff.py

@@ -0,0 +1,106 @@
+"""Persist the semantic task handoff across ephemeral workspaces."""
+
+import logging
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_HANDOFF_PATH = Path(".forge/handoff.md")
+MAX_HANDOFF_BYTES = 64 * 1024
+
+
+def capture_handoff(
+    workspace_path: str | Path,
+    repo: str,
+    task_key: str,
+    state: dict[str, Any],
+) -> dict[str, Any]:
+    """Capture the current repository handoff as bounded, structured state.
+
+    A missing handoff removes any saved value for the repository. This mirrors
+    the workspace exactly and prevents deleted content from being resurrected
+    after a later workspace recreation.
+    """
+    saved_handoffs = state.get("handoffs", {})
+    if not isinstance(saved_handoffs, dict):
+        logger.warning("Ignoring malformed handoff state while capturing %s", repo)
+        saved_handoffs = {}
+    handoffs = dict(saved_handoffs)
+    handoff_path = Path(workspace_path) / _HANDOFF_PATH
+
+    try:
+        size = handoff_path.stat().st_size
+    except FileNotFoundError:
+        handoffs.pop(repo, None)
+        return {**state, "handoffs": handoffs}
+    except OSError as exc:
+        logger.warning("Failed to inspect handoff for %s: %s", repo, exc)
+        return state
+
+    if handoff_path.is_symlink() or not handoff_path.is_file():
+        logger.warning("Ignoring non-regular handoff for %s", repo)
+        handoffs.pop(repo, None)
+        return {**state, "handoffs": handoffs}
+    if size > MAX_HANDOFF_BYTES:
+        logger.warning(
+            "Ignoring oversized handoff for %s (%d bytes; limit %d)",
+            repo,
+            size,
+            MAX_HANDOFF_BYTES,
+        )
+        handoffs.pop(repo, None)
+        return {**state, "handoffs": handoffs}
+
+    try:
+        content = handoff_path.read_text()
+    except (OSError, UnicodeError) as exc:
+        logger.warning("Failed to capture handoff for %s: %s", repo, exc)
+        return state
+
+    handoffs[repo] = {
+        "content": content,
+        "task_key": task_key,
+        "captured_at": datetime.now(UTC).isoformat(),
+    }
+    return {**state, "handoffs": handoffs}
+
+
+def materialize_handoff(
+    workspace_path: str | Path,
+    repo: str,
+    state: dict[str, Any],
+) -> None:
+    """Materialize the saved repository handoff at its fixed workspace path."""
+    handoffs = state.get("handoffs", {})
+    if not isinstance(handoffs, dict):
+        logger.warning("Ignoring malformed handoff state for %s", repo)
+        return
+    handoff = handoffs.get(repo)
+    if not isinstance(handoff, dict):
+        return
+
+    content = handoff.get("content")
+    try:
+        content_size = len(content.encode()) if isinstance(content, str) else -1
+    except UnicodeError:
+        content_size = -1
+    if content_size < 0 or content_size > MAX_HANDOFF_BYTES:
+        logger.warning("Ignoring invalid saved handoff for %s", repo)
+        return
+
+    handoff_path = Path(workspace_path) / _HANDOFF_PATH
+    temporary_path = handoff_path.with_suffix(".md.tmp")
+    try:
+        forge_dir = handoff_path.parent
+        if forge_dir.is_symlink():
+            logger.warning("Refusing to materialize handoff through symlink for %s", repo)
+            return
+        forge_dir.mkdir(parents=True, exist_ok=True)
+        temporary_path.unlink(missing_ok=True)
+        temporary_path.write_text(content)
+        temporary_path.replace(handoff_path)
+    except (OSError, UnicodeError) as exc:
+        logger.warning("Failed to materialize handoff for %s: %s", repo, exc)
+        temporary_path.unlink(missing_ok=True)

tests/unit/workflow/feature/test_state.py

@@ -54,6 +54,7 @@ def test_create_initial_feature_state(self):
         assert state["ticket_key"] == "TEST-123"
         assert state["prd_content"] == ""
         assert state["epic_keys"] == []
+        assert state["handoffs"] == {}
 
 
 class TestQAStateFields:
@@ -95,6 +96,7 @@ def test_feature_state_qa_defaults(self):
         assert state["qa_history"] == []
         assert state["generation_context"] == {}
         assert state["is_question"] is False
+        assert state["handoffs"] == {}
 
     def test_bug_state_has_qa_history(self):
         """BugState includes qa_history field."""