diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py
index fcc4776c2..5a7ecf8b1 100644
--- a/src/forge/workflow/base.py
+++ b/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):
diff --git a/src/forge/workflow/bug/state.py b/src/forge/workflow/bug/state.py
index 5fc987e80..7cd6b46e4 100644
--- a/src/forge/workflow/bug/state.py
+++ b/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,
diff --git a/src/forge/workflow/feature/state.py b/src/forge/workflow/feature/state.py
index f48c02aa1..54f9d565f 100644
--- a/src/forge/workflow/feature/state.py
+++ b/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,
diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py
index 30dc7bb4e..c8a73cee5 100644
--- a/src/forge/workflow/nodes/ci_evaluator.py
+++ b/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),
diff --git a/src/forge/workflow/nodes/implement_review.py b/src/forge/workflow/nodes/implement_review.py
index 7a10d6992..53453f01c 100644
--- a/src/forge/workflow/nodes/implement_review.py
+++ b/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),
diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py
index ede61e072..aca595fa9 100644
--- a/src/forge/workflow/nodes/implementation.py
+++ b/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),
diff --git a/src/forge/workflow/nodes/workspace_setup.py b/src/forge/workflow/nodes/workspace_setup.py
index 15f28b3dc..51093ae1d 100644
--- a/src/forge/workflow/nodes/workspace_setup.py
+++ b/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.
diff --git a/src/forge/workspace/handoff.py b/src/forge/workspace/handoff.py
new file mode 100644
index 000000000..abd26846c
--- /dev/null
+++ b/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)
diff --git a/tests/unit/workflow/feature/test_state.py b/tests/unit/workflow/feature/test_state.py
index 94fdfb025..770612b21 100644
--- a/tests/unit/workflow/feature/test_state.py
+++ b/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."""
diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py
index 163fc8e69..511012584 100644
--- a/tests/unit/workflow/nodes/test_implementation.py
+++ b/tests/unit/workflow/nodes/test_implementation.py
@@ -18,6 +18,7 @@ def _make_state(
     tasks_by_repo=None,
     implemented_tasks=None,
     retry_count=0,
+    handoffs=None,
 ):
     return {
         "ticket_key": ticket_key,
@@ -32,6 +33,7 @@ def _make_state(
         "task_keys": [current_task_key] if current_task_key else [],
         "tasks_by_repo": tasks_by_repo or {current_repo: [current_task_key]},
         "implemented_tasks": implemented_tasks or [],
+        "handoffs": handoffs or {},
         "context": {"branch_name": "forge/BUG-123", "guardrails": ""},
         "fork_owner": "forge-bot",
         "fork_repo": "backend",
@@ -254,6 +256,58 @@ async def test_feature_container_failure_uses_feature_implementation_node(self):
         assert result["last_error"] == "container failed"
         assert result["retry_count"] == 1
 
+    @pytest.mark.asyncio
+    async def test_container_failure_checkpoints_partial_handoff(self, tmp_path):
+        """A returned container failure preserves its blocker handoff for another worker."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        state = _make_state(workspace_path=str(tmp_path), handoffs={})
+        mock_jira = _make_mock_jira()
+        container_result = MagicMock(success=False, error_message="container failed")
+
+        async def run_with_partial_handoff(**_kwargs):
+            forge_dir = tmp_path / ".forge"
+            forge_dir.mkdir()
+            (forge_dir / "handoff.md").write_text("Partial work; blocked by failing test")
+            return container_result
+
+        runner = MagicMock(run=run_with_partial_handoff)
+        with (
+            patch("forge.workflow.nodes.implementation.JiraClient", return_value=mock_jira),
+            patch("forge.workflow.nodes.implementation.ContainerRunner", return_value=runner),
+            patch("forge.workflow.nodes.implementation.get_settings"),
+        ):
+            result = await implement_task(state)
+
+        assert result["last_error"] == "container failed"
+        assert result["handoffs"]["acme/backend"]["content"].startswith("Partial work")
+        assert result["handoffs"]["acme/backend"]["task_key"] == "TASK-456"
+
+    @pytest.mark.asyncio
+    async def test_runner_exception_checkpoints_handoff_written_before_crash(self, tmp_path):
+        """A runner exception still captures a handoff already flushed to the workspace."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        state = _make_state(workspace_path=str(tmp_path), handoffs={})
+        mock_jira = _make_mock_jira()
+
+        async def run_then_raise(**_kwargs):
+            forge_dir = tmp_path / ".forge"
+            forge_dir.mkdir()
+            (forge_dir / "handoff.md").write_text("Timed out after partial implementation")
+            raise TimeoutError("container timed out")
+
+        runner = MagicMock(run=run_then_raise)
+        with (
+            patch("forge.workflow.nodes.implementation.JiraClient", return_value=mock_jira),
+            patch("forge.workflow.nodes.implementation.ContainerRunner", return_value=runner),
+            patch("forge.workflow.nodes.implementation.get_settings"),
+        ):
+            result = await implement_task(state)
+
+        assert result["last_error"] == "container timed out"
+        assert result["handoffs"]["acme/backend"]["content"].startswith("Timed out")
+
     @pytest.mark.asyncio
     async def test_successful_implementation_is_pushed_before_checkpoint(self) -> None:
         """A different worker can recover the implementation commit from the fork."""
diff --git a/tests/unit/workflow/nodes/test_workspace_setup.py b/tests/unit/workflow/nodes/test_workspace_setup.py
index ca06cd7d7..4613f6c86 100644
--- a/tests/unit/workflow/nodes/test_workspace_setup.py
+++ b/tests/unit/workflow/nodes/test_workspace_setup.py
@@ -453,6 +453,13 @@ def test_sync_failure_recreates_workspace_from_fork(self, tmp_path):
             fork_owner="forge-bot",
             fork_repo="repo",
             context={"branch_name": "forge/test-123"},
+            handoffs={
+                "org/repo": {
+                    "content": "prior task context",
+                    "task_key": "TEST-122",
+                    "captured_at": "2026-08-06T00:00:00+00:00",
+                }
+            },
         )
 
         old_git = MagicMock()
@@ -477,6 +484,7 @@ def test_sync_failure_recreates_workspace_from_fork(self, tmp_path):
         new_git.add_fork_remote.assert_called_once_with("forge-bot", "repo")
         new_git.checkout_branch.assert_called_once_with("forge/test-123", remote="fork")
         assert new_git.workspace_recreated is True
+        assert (workspace_path / ".forge" / "handoff.md").read_text() == "prior task context"
 
     def test_failed_replacement_preserves_existing_workspace(self, tmp_path):
         """A failed recovery clone must not delete the only local commit."""
diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py
index 4df75da1d..2053b25e6 100644
--- a/tests/unit/workflow/test_base.py
+++ b/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 "handoffs" in hints
 
 
 class TestCIIntegrationState:
diff --git a/tests/unit/workspace/test_handoff.py b/tests/unit/workspace/test_handoff.py
new file mode 100644
index 000000000..73fb58a16
--- /dev/null
+++ b/tests/unit/workspace/test_handoff.py
@@ -0,0 +1,137 @@
+"""Tests for durable task handoff capture and materialization."""
+
+from forge.workspace.handoff import MAX_HANDOFF_BYTES, capture_handoff, materialize_handoff
+
+
+def test_capture_records_content_and_metadata(tmp_path):
+    (tmp_path / ".forge").mkdir()
+    (tmp_path / ".forge" / "handoff.md").write_text("task complete")
+
+    state = {"handoffs": {}}
+    result = capture_handoff(tmp_path, "org/repo", "TEST-2", state)
+
+    handoff = result["handoffs"]["org/repo"]
+    assert handoff["content"] == "task complete"
+    assert handoff["task_key"] == "TEST-2"
+    assert handoff["captured_at"]
+    assert state["handoffs"] == {}
+
+
+def test_capture_preserves_other_repositories(tmp_path):
+    (tmp_path / ".forge").mkdir()
+    (tmp_path / ".forge" / "handoff.md").write_text("new")
+    state = {
+        "handoffs": {
+            "org/other": {"content": "other", "task_key": "X-1", "captured_at": "now"}
+        }
+    }
+
+    result = capture_handoff(tmp_path, "org/repo", "TEST-2", state)
+
+    assert result["handoffs"]["org/other"] == state["handoffs"]["org/other"]
+    assert result["handoffs"]["org/repo"]["content"] == "new"
+
+
+def test_missing_handoff_removes_stale_saved_value(tmp_path):
+    state = {
+        "handoffs": {
+            "org/repo": {"content": "stale", "task_key": "TEST-1", "captured_at": "now"}
+        }
+    }
+
+    result = capture_handoff(tmp_path, "org/repo", "TEST-2", state)
+
+    assert "org/repo" not in result["handoffs"]
+
+
+def test_oversized_handoff_is_not_checkpointed(tmp_path):
+    (tmp_path / ".forge").mkdir()
+    (tmp_path / ".forge" / "handoff.md").write_bytes(b"x" * (MAX_HANDOFF_BYTES + 1))
+
+    result = capture_handoff(tmp_path, "org/repo", "TEST-2", {"handoffs": {}})
+
+    assert result["handoffs"] == {}
+
+
+def test_materialize_writes_only_fixed_handoff_path(tmp_path):
+    state = {
+        "handoffs": {
+            "org/repo": {
+                "content": "restored",
+                "task_key": "../../unsafe",
+                "captured_at": "now",
+            }
+        }
+    }
+
+    materialize_handoff(tmp_path, "org/repo", state)
+
+    assert (tmp_path / ".forge" / "handoff.md").read_text() == "restored"
+    assert not (tmp_path / ".forge" / "handoff.md.tmp").exists()
+
+
+def test_materialize_ignores_other_repository(tmp_path):
+    state = {
+        "handoffs": {
+            "org/other": {"content": "other", "task_key": "X-1", "captured_at": "now"}
+        }
+    }
+
+    materialize_handoff(tmp_path, "org/repo", state)
+
+    assert not (tmp_path / ".forge").exists()
+
+
+def test_first_iteration_without_saved_handoffs_is_a_noop(tmp_path):
+    materialize_handoff(tmp_path, "org/repo", {})
+
+    assert not (tmp_path / ".forge").exists()
+
+
+def test_malformed_checkpoint_state_is_ignored(tmp_path):
+    materialize_handoff(tmp_path, "org/repo", {"handoffs": ["not", "a", "mapping"]})
+
+    assert not (tmp_path / ".forge").exists()
+
+
+def test_malformed_checkpoint_entry_is_ignored(tmp_path):
+    materialize_handoff(tmp_path, "org/repo", {"handoffs": {"org/repo": "bad"}})
+
+    assert not (tmp_path / ".forge").exists()
+
+
+def test_capture_recovers_from_malformed_checkpoint_state(tmp_path):
+    (tmp_path / ".forge").mkdir()
+    (tmp_path / ".forge" / "handoff.md").write_text("usable")
+
+    result = capture_handoff(
+        tmp_path, "org/repo", "TEST-2", {"handoffs": ["not", "a", "mapping"]}
+    )
+
+    assert result["handoffs"]["org/repo"]["content"] == "usable"
+
+
+def test_capture_rejects_symlinked_handoff(tmp_path):
+    outside = tmp_path / "outside.md"
+    outside.write_text("do not capture")
+    (tmp_path / ".forge").mkdir()
+    (tmp_path / ".forge" / "handoff.md").symlink_to(outside)
+
+    result = capture_handoff(tmp_path, "org/repo", "TEST-2", {"handoffs": {}})
+
+    assert result["handoffs"] == {}
+
+
+def test_materialize_refuses_symlinked_forge_directory(tmp_path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (tmp_path / ".forge").symlink_to(outside, target_is_directory=True)
+    state = {
+        "handoffs": {
+            "org/repo": {"content": "unsafe", "task_key": "TEST-2", "captured_at": "now"}
+        }
+    }
+
+    materialize_handoff(tmp_path, "org/repo", state)
+
+    assert not (outside / "handoff.md").exists()