feat: persist .forge/ artifacts in workflow state for cross-worker continuity

src/forge/workflow/base.py

@@ -67,6 +67,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_]
+    forge_artifacts: dict[str, dict[str, str]]
 
 
 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": {},
+        "forge_artifacts": {},
         "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": {},
+        "forge_artifacts": {},
         "ci_status": None,
         "current_pr_url": None,
         "current_pr_number": None,

src/forge/workflow/nodes/ci_evaluator.py

@@ -24,6 +24,7 @@
     set_ci_pending_label,
     set_review_pending_label,
 )
+from forge.workspace.artifacts import harvest_forge_artifacts
 from forge.workspace.git_ops import GitOperations
 from forge.workspace.manager import Workspace
 
@@ -443,6 +444,10 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState:
                 attempt=attempt,
             )
 
+        state = harvest_forge_artifacts(
+            workspace_path, state.get("current_repo", ""), ["handoff.md", "fix-plan.md"], state
+        )
+
         return update_state_timestamp(
             {
                 **state,

src/forge/workflow/nodes/implement_review.py

@@ -21,6 +21,7 @@
     merge_review_decisions,
     reply_to_review_decisions,
 )
+from forge.workspace.artifacts import harvest_forge_artifacts
 
 logger = logging.getLogger(__name__)
 
@@ -284,6 +285,11 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
             decisions=response_decisions,
         )
 
+        # Harvest Phase 1 outputs so they survive if the workspace is recreated later.
+        state = harvest_forge_artifacts(
+            workspace_path, current_repo, ["review-plan.md", "review-objections.md"], state
+        )
+
         # Backward-compatible fallback if an older analysis prompt writes only
         # the legacy objections file. New analysis never blocks accepted work.
         objections_path = Path(workspace_path) / _REVIEW_OBJECTIONS_FILE
@@ -324,6 +330,8 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
             )
             state = merge_review_exhaustion(state, result, ticket_key, "implement_review_fix")
 
+            state = harvest_forge_artifacts(workspace_path, current_repo, ["handoff.md"], state)
+
             # Commit any uncommitted changes the container left
             if git.has_uncommitted_changes():
                 git.stage_all()

src/forge/workflow/nodes/implementation.py

@@ -27,6 +27,7 @@
 from forge.workflow.nodes.workspace_setup import prepare_workspace
 from forge.workflow.utils import merge_review_exhaustion, update_state_timestamp
 from forge.workflow.utils.jira_status import post_status_comment
+from forge.workspace.artifacts import harvest_forge_artifacts
 from forge.workspace.git_ops import GitOperations
 
 logger = logging.getLogger(__name__)
@@ -224,6 +225,8 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
         if result.success:
             logger.info(f"Container completed successfully for {current_task}")
 
+            state = harvest_forge_artifacts(workspace_path, current_repo, ["handoff.md"], state)
+
             # Persist each task commit before checkpointing. A subsequent task
             # or local review may resume on a worker with a different filesystem.
             try:

src/forge/workflow/nodes/workspace_setup.py

@@ -17,6 +17,7 @@
     set_implementing_label,
     transition_tasks_to_in_progress,
 )
+from forge.workspace.artifacts import restore_forge_artifacts
 from forge.workspace.git_ops import GitOperations
 from forge.workspace.guardrails import GuardrailsLoader
 from forge.workspace.manager import Workspace, WorkspaceManager
@@ -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)
+    restore_forge_artifacts(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,
         )
+        restore_forge_artifacts(workspace.path, current_repo, state)
 
         # Keep Forge handoff files local to this clone without modifying the
         # target repository's tracked .gitignore.

src/forge/workspace/artifacts.py

@@ -0,0 +1,81 @@
+"""Utilities for persisting .forge/ artifacts in workflow state across workspace recreations.
+
+Harvest: after a container run, call harvest_forge_artifacts to read container-written
+files from .forge/ into state keyed by repo name.
+
+Restore: at workspace setup or recreation, call restore_forge_artifacts to write
+those files back to .forge/ before the next container starts.
+"""
+
+import logging
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def harvest_forge_artifacts(
+    workspace_path: str | Path,
+    repo: str,
+    files: list[str],
+    state: dict[str, Any],
+) -> dict[str, Any]:
+    """Read named files from .forge/ and merge their contents into state.
+
+    Files that do not exist are silently skipped. Existing artifacts for
+    other repos or other filenames in this repo are preserved.
+
+    Args:
+        workspace_path: Path to the workspace root (parent of .forge/).
+        repo: Repository name e.g. "org/repo", used as the outer state key.
+        files: Relative paths within .forge/ to harvest e.g. ["handoff.md"].
+        state: Current workflow state dict.
+
+    Returns:
+        New state dict with forge_artifacts updated for this repo.
+    """
+    forge_dir = Path(workspace_path) / ".forge"
+    all_artifacts: dict[str, dict[str, str]] = dict(state.get("forge_artifacts", {}))
+    repo_artifacts: dict[str, str] = dict(all_artifacts.get(repo, {}))
+
+    for filename in files:
+        file_path = forge_dir / filename
+        if file_path.exists():
+            try:
+                repo_artifacts[filename] = file_path.read_text()
+                logger.debug(f"Harvested .forge/{filename} for {repo}")
+            except Exception as e:
+                logger.warning(f"Failed to harvest .forge/{filename} for {repo}: {e}")
+
+    all_artifacts[repo] = repo_artifacts
+    return {**state, "forge_artifacts": all_artifacts}
+
+
+def restore_forge_artifacts(
+    workspace_path: str | Path,
+    repo: str,
+    state: dict[str, Any],
+) -> None:
+    """Write all harvested artifacts for this repo back to .forge/.
+
+    Creates parent directories as needed. No-op when no artifacts exist
+    for this repo in state.
+
+    Args:
+        workspace_path: Path to the workspace root (parent of .forge/).
+        repo: Repository name e.g. "org/repo".
+        state: Current workflow state dict.
+    """
+    artifacts = state.get("forge_artifacts", {}).get(repo, {})
+    if not artifacts:
+        return
+
+    forge_dir = Path(workspace_path) / ".forge"
+    for filename, content in artifacts.items():
+        file_path = forge_dir / filename
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            file_path.write_text(content)
+            logger.debug(f"Restored .forge/{filename} for {repo}")
+        except Exception as e:
+            logger.warning(f"Failed to restore .forge/{filename} for {repo}: {e}")

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["forge_artifacts"] == {}
 
 
 class TestQAStateFields:
@@ -132,3 +133,4 @@ def test_bug_state_qa_defaults(self):
         assert state["qa_history"] == []
         assert state["generation_context"] == {}
         assert state["is_question"] is False
+        assert state["forge_artifacts"] == {}

tests/unit/workflow/test_base.py

@@ -63,6 +63,7 @@ def test_pr_state_has_required_fields(self):
         assert "repos_completed" in hints
         assert "implemented_tasks" in hints
         assert "current_task_key" in hints
+        assert "forge_artifacts" in hints
 
 
 class TestCIIntegrationState:

tests/unit/workflow/test_ci_fix_harvest.py

@@ -0,0 +1,66 @@
+"""attempt_ci_fix harvests fix-plan.md and handoff.md into state."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def ci_state(tmp_path):
+    forge_dir = tmp_path / ".forge"
+    forge_dir.mkdir()
+    (forge_dir / "fix-plan.md").write_text("fix plan content")
+    (forge_dir / "handoff.md").write_text("ci fix handoff")
+    (forge_dir / "ci-failures.md").write_text("failures")
+    return {
+        "ticket_key": "TEST-1",
+        "workspace_path": str(tmp_path),
+        "current_repo": "org/repo",
+        "context": {"branch_name": "forge/test-1", "guardrails": ""},
+        "fork_owner": "fork-org",
+        "fork_repo": "repo",
+        "ci_fix_attempts": 0,
+        "ci_failed_checks": [{"name": "lint", "conclusion": "failure"}],
+        "ci_skipped_checks": [],
+        "pr_urls": ["https://github.com/org/repo/pull/1"],
+        "current_pr_number": 1,
+        "current_pr_url": "https://github.com/org/repo/pull/1",
+        "is_paused": False,
+        "retry_count": 0,
+        "forge_artifacts": {},
+        "spec_content": "",
+    }
+
+
+@pytest.mark.asyncio
+async def test_attempt_ci_fix_harvests_fix_plan_and_handoff(ci_state, tmp_path):
+    from forge.workflow.nodes.ci_evaluator import attempt_ci_fix
+
+    mock_git = MagicMock()
+    mock_git._run_git.return_value = MagicMock(stdout="abc123\n", returncode=0)
+    mock_git.has_uncommitted_changes.return_value = False
+    mock_git.push_to_fork = MagicMock()
+    mock_git.add_fork_remote = MagicMock()
+
+    mock_runner = AsyncMock()
+    mock_runner.run.return_value = MagicMock(success=True)
+
+    mock_github = AsyncMock()
+
+    with patch("forge.workflow.nodes.ci_evaluator.prepare_workspace",
+               return_value=(str(tmp_path), mock_git)), \
+         patch("forge.workflow.nodes.ci_evaluator.ContainerRunner", return_value=mock_runner), \
+         patch("forge.workflow.nodes.ci_evaluator.GitHubClient", return_value=mock_github), \
+         patch("forge.workflow.nodes.ci_evaluator.Workspace", return_value=MagicMock()), \
+         patch("forge.workflow.nodes.ci_evaluator.GitOperations", return_value=mock_git), \
+         patch("forge.workflow.nodes.ci_evaluator.run_post_change_review",
+               new=AsyncMock(return_value=(None, None))), \
+         patch("forge.workflow.nodes.ci_evaluator.sync_pr_description",
+               new_callable=AsyncMock), \
+         patch("forge.workflow.nodes.ci_evaluator._fetch_ci_logs_and_artifacts",
+               new_callable=AsyncMock), \
+         patch("forge.workflow.nodes.ci_evaluator._collect_error_info", return_value="errors"):
+        result = await attempt_ci_fix(ci_state)
+
+    artifacts = result.get("forge_artifacts", {}).get("org/repo", {})
+    assert artifacts.get("fix-plan.md") == "fix plan content"
+    assert artifacts.get("handoff.md") == "ci fix handoff"

tests/unit/workflow/test_implement_review_harvest.py

@@ -0,0 +1,85 @@
+"""implement_review harvests review artifacts into state."""
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def review_state(tmp_path):
+    forge_dir = tmp_path / ".forge"
+    forge_dir.mkdir()
+    (forge_dir / "handoff.md").write_text("review fix done")
+    return {
+        "ticket_key": "TEST-1",
+        "workspace_path": str(tmp_path),
+        "current_repo": "org/repo",
+        "context": {"branch_name": "forge/test-1", "guardrails": ""},
+        "fork_owner": "fork-org",
+        "fork_repo": "repo",
+        "current_pr_number": 1,
+        "current_pr_url": "https://github.com/org/repo/pull/1",
+        "feedback_comment": "please fix the style",
+        "revision_requested": True,
+        "review_response_posted": False,
+        "contested_comments": [],
+        "is_paused": False,
+        "retry_count": 0,
+        "forge_artifacts": {},
+        "spec_content": "",
+        "implemented_tasks": [],
+    }
+
+
+@pytest.mark.asyncio
+async def test_implement_review_harvests_plan_and_handoff(review_state, tmp_path):
+    from forge.workflow.nodes.implement_review import implement_review
+
+    forge_dir = tmp_path / ".forge"
+
+    def phase1_side_effect(**kwargs):
+        # Simulate Phase 1 container writing review-plan.md and review-objections.md
+        (forge_dir / "review-plan.md").write_text("## Fix foo\n- change bar")
+        return MagicMock(success=True)
+
+    def phase2_side_effect(**kwargs):
+        # Simulate Phase 2 container writing handoff.md
+        (forge_dir / "handoff.md").write_text("review fix done")
+        return MagicMock(success=True)
+
+    mock_git = MagicMock()
+    mock_git._run_git.return_value = MagicMock(stdout="abc123\n", returncode=0)
+    mock_git.has_uncommitted_changes.return_value = False
+    mock_git.push_to_fork = MagicMock()
+    mock_git.add_fork_remote = MagicMock()
+
+    call_count = 0
+
+    async def mock_run(**kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return phase1_side_effect(**kwargs)
+        else:
+            return phase2_side_effect(**kwargs)
+
+    mock_runner = MagicMock()
+    mock_runner.run = mock_run
+
+    mock_github = AsyncMock()
+
+    with patch("forge.workflow.nodes.implement_review.prepare_workspace",
+               return_value=(str(tmp_path), mock_git)), \
+         patch("forge.workflow.nodes.implement_review.ContainerRunner", return_value=mock_runner), \
+         patch("forge.workflow.nodes.implement_review._fetch_pr_review_comments",
+               new_callable=AsyncMock, return_value="comment text"), \
+         patch("forge.workflow.nodes.implement_review.run_post_change_review",
+               new_callable=AsyncMock), \
+         patch("forge.workflow.nodes.implement_review.sync_pr_description",
+               new_callable=AsyncMock), \
+         patch("forge.workflow.nodes.implement_review.GitHubClient", return_value=mock_github):
+        result = await implement_review(review_state)
+
+    artifacts = result.get("forge_artifacts", {}).get("org/repo", {})
+    assert "review-plan.md" in artifacts
+    assert "handoff.md" in artifacts