diff --git a/github/README.md b/github/README.md
index caada17..0108535 100644
--- a/github/README.md
+++ b/github/README.md
@@ -94,8 +94,16 @@ dagger -m github.com/typesafe-ai/daggerverse/github call \
 | `progress-interval` | `30`                       | seconds between routine progress lines                 |
 | `timeout`           | `1800`                     | total wall-clock budget, in seconds                    |
 | `discovery-timeout` | `300`                      | how long expected statuses may take to first appear    |
+| `retry-grace-period` | `0` (`60` for Dagger checks) | seconds to allow an `error` status to be retried; `0` disables grace |
 | `fail-fast`         | `false`                    | whether to exit 1 on first check failure               |
 
+Dagger Cloud may briefly publish `error` when an engine disconnects, then
+return the check to `pending` for a retry. The grace period starts when each
+check is observed entering `error`. Repeated polls do not extend it; a new
+error after returning to `pending` gets a new grace period. `failure` statuses
+are never given retry grace, and the total `timeout` still bounds all retries.
+`fail-fast` also honors retry grace for `error` statuses.
+
 `wait-for-statuses` additionally requires:
 
 | name     | default | meaning                                                  |
diff --git a/github/src/github/status_monitor/main.py b/github/src/github/status_monitor/main.py
index 5150798..6e57f2d 100644
--- a/github/src/github/status_monitor/main.py
+++ b/github/src/github/status_monitor/main.py
@@ -30,6 +30,7 @@
     ProgressInterval,
     Ref,
     Repo,
+    RetryGracePeriod,
     Timeout,
     Token,
 )
@@ -62,6 +63,7 @@ async def wait_for_statuses(
         progress_interval: ProgressInterval = 30,
         timeout: Timeout = 1800,
         discovery_timeout: DiscoveryTimeout = 300,
+        retry_grace_period: RetryGracePeriod = 0,
         fail_fast: FailFast = False,
     ) -> str:
         """Wait until every status context in `checks` and every check-run in
@@ -76,6 +78,7 @@ async def wait_for_statuses(
             progress_interval=progress_interval,
             timeout=timeout,
             discovery_timeout=discovery_timeout,
+            retry_grace_period=retry_grace_period,
             fail_fast=fail_fast,
         )
 
@@ -89,6 +92,7 @@ async def wait_for_dagger_checks(
         progress_interval: ProgressInterval = 30,
         timeout: Timeout = 1800,
         discovery_timeout: DiscoveryTimeout = 300,
+        retry_grace_period: RetryGracePeriod = 60,
         fail_fast: FailFast = False,
     ) -> str:
         """Wait for every Dagger check in the current workspace to succeed.
@@ -109,6 +113,7 @@ async def wait_for_dagger_checks(
             progress_interval=progress_interval,
             timeout=timeout,
             discovery_timeout=discovery_timeout,
+            retry_grace_period=retry_grace_period,
             fail_fast=fail_fast,
         )
 
@@ -124,6 +129,7 @@ async def _wait(
         progress_interval: int,
         timeout: int,
         discovery_timeout: int,
+        retry_grace_period: int,
         fail_fast: bool,
     ) -> str:
         console = Console(force_terminal=True)
@@ -140,6 +146,7 @@ async def _wait(
             fail_fast=fail_fast,
             timeout=timeout,
             discovery_timeout=discovery_timeout,
+            retry_grace_period=retry_grace_period,
             poll_interval=poll_interval,
             progress_interval=progress_interval,
             console=console,
@@ -164,7 +171,9 @@ async def _wait(
                     case Step.DISCOVERY_TIMEOUT:
                         raise TimeoutError(f"checks never appeared on {ref}: {watcher.missing}")
                     case Step.WALLCLOCK_TIMEOUT:
-                        raise TimeoutError(f"timed out: pending={watcher.pending} missing={watcher.missing}")
+                        raise TimeoutError(
+                            f"timed out: pending={watcher.pending} missing={watcher.missing} retrying={watcher.retrying}"
+                        )
                     case Step.CONTINUE:
                         continue
 
diff --git a/github/src/github/status_monitor/params.py b/github/src/github/status_monitor/params.py
index 7dadec9..96aea5d 100644
--- a/github/src/github/status_monitor/params.py
+++ b/github/src/github/status_monitor/params.py
@@ -22,10 +22,15 @@
 
 DiscoveryTimeout: TypeAlias = Annotated[int, Doc("How long expected statuses may take to first appear, seconds")]
 
+RetryGracePeriod: TypeAlias = Annotated[
+    int,
+    Doc("Seconds to allow an error status to return to pending or success before failing; 0 disables retry grace."),
+]
+
 FailFast: TypeAlias = Annotated[
     bool,
     Doc(
-        "If True, raise as soon as any check fails. "
+        "If True, raise as soon as any check fails (error statuses still get their retry grace period). "
         "If False (default), wait for every check to reach a terminal "
         "state and then raise at the end if any failed."
     ),
diff --git a/github/src/github/status_monitor/render.py b/github/src/github/status_monitor/render.py
index e576b47..b3f4575 100644
--- a/github/src/github/status_monitor/render.py
+++ b/github/src/github/status_monitor/render.py
@@ -34,8 +34,9 @@ def progress(
     pending: int,
     missing: int,
     total: int,
+    retrying: int = 0,
 ) -> None:
-    console.print(f"[dim]success={succeeded}/{total} pending={pending} missing={missing}[/]")
+    console.print(f"[dim]success={succeeded}/{total} pending={pending} missing={missing} retrying={retrying}[/]")
 
 
 def final_table(
@@ -65,8 +66,8 @@ def discovery_timeout(console: Console, *, ref: str, missing: list[str]) -> None
     console.print(f"[bold red]✗ checks never appeared on {ref}: {missing}[/]")
 
 
-def wallclock_timeout(console: Console, *, pending: list[str], missing: list[str]) -> None:
-    console.print(f"[bold red]✗ timed out: pending={pending} missing={missing}[/]")
+def wallclock_timeout(console: Console, *, pending: list[str], missing: list[str], retrying: list[str]) -> None:
+    console.print(f"[bold red]✗ timed out: pending={pending} missing={missing} retrying={retrying}[/]")
 
 
 def empty(console: Console) -> None:
diff --git a/github/src/github/status_monitor/watcher.py b/github/src/github/status_monitor/watcher.py
index 87d12a4..f3e5048 100644
--- a/github/src/github/status_monitor/watcher.py
+++ b/github/src/github/status_monitor/watcher.py
@@ -49,17 +49,21 @@ def __init__(
         fail_fast: bool = False,
         timeout: float = 1800,
         discovery_timeout: float = 300,
+        retry_grace_period: float = 0,
         poll_interval: int = 3,
         progress_interval: float = 30,
         console: Console | None = None,
         clock: Callable[[], float] = time.monotonic,
     ) -> None:
+        if retry_grace_period < 0:
+            raise ValueError("retry_grace_period must be non-negative")
         self.expected: list[str] = sorted(set(expected))
         self.repo: str = repo
         self.ref: str = ref
         self.fail_fast: bool = fail_fast
         self.timeout: float = timeout
         self.discovery_timeout: float = discovery_timeout
+        self.retry_grace_period: float = retry_grace_period
         self.poll_interval: int = poll_interval
         self.progress_interval: float = progress_interval
         self.console: Console = console if console is not None else _default_console()
@@ -68,6 +72,7 @@ def __init__(
         self._clock: Callable[[], float] = clock
         self._start: float | None = None
         self._last_progress: float | None = None
+        self._error_since: dict[str, float] = {}
 
     # ---- pure state ----
 
@@ -79,10 +84,16 @@ def observe(self, snapshot: Mapping[str, Status]) -> list[Transition]:
         the final table about to be rendered will summarize.
         """
         new_completions: list[Transition] = []
+        now = self._clock()
         for name in self.expected:
             previous = self.states[name]
             status = snapshot.get(name)
             next_state = MISSING if status is None else status.state
+            if next_state == "error":
+                # Repeated polls of the same error must not extend its grace.
+                self._error_since.setdefault(name, now)
+            else:
+                self._error_since.pop(name, None)
             if next_state != previous:
                 self.states[name] = next_state
                 if next_state in TERMINAL_STATES and previous not in TERMINAL_STATES:
@@ -95,14 +106,24 @@ def observe(self, snapshot: Mapping[str, Status]) -> list[Transition]:
 
     @property
     def verdict(self) -> Verdict:
-        if self.fail_fast and self.failed:
+        retrying = set(self.retrying)
+        failed = [name for name in self.failed if name not in retrying]
+        if self.fail_fast and failed:
             return Verdict.FAILED
         if self.pending or self.missing:
             return Verdict.WAITING
-        if self.failed:
+        if failed:
             return Verdict.FAILED
+        if retrying:
+            return Verdict.WAITING
         return Verdict.SUCCEEDED
 
+    @property
+    def retrying(self) -> list[str]:
+        """Errored checks still within their grace period for a Cloud retry."""
+        now = self._clock()
+        return [name for name, since in self._error_since.items() if now - since < self.retry_grace_period]
+
     @property
     def missing(self) -> list[str]:
         return [c for c in self.expected if self.states[c] == MISSING]
@@ -166,6 +187,7 @@ def print_progress(self) -> None:
             succeeded=len(self.succeeded),
             pending=len(self.pending),
             missing=len(self.missing),
+            retrying=len(self.retrying),
             total=len(self.expected),
         )
 
@@ -181,7 +203,7 @@ def print_discovery_timeout(self) -> None:
 
     def print_wallclock_timeout(self) -> None:
         self._print_table()
-        render.wallclock_timeout(self.console, pending=self.pending, missing=self.missing)
+        render.wallclock_timeout(self.console, pending=self.pending, missing=self.missing, retrying=self.retrying)
 
     def _print_table(self) -> None:
         render.final_table(
diff --git a/github/tests/test_status_monitor.py b/github/tests/test_status_monitor.py
new file mode 100644
index 0000000..8bbe610
--- /dev/null
+++ b/github/tests/test_status_monitor.py
@@ -0,0 +1,52 @@
+"""Verify retry grace reaches the watcher through the public Dagger functions."""
+
+import asyncio
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from github.status_monitor import main
+from github.status_monitor.main import GithubStatusMonitor
+from github.status_monitor.types import Status
+
+
+@pytest.mark.parametrize("grace", [None, 60])
+def test_explicit_statuses_forward_retry_grace(monkeypatch, grace):
+    wait = AsyncMock(return_value="all 1 checks succeeded")
+    monkeypatch.setattr(GithubStatusMonitor, "_wait", wait)
+    kwargs = {} if grace is None else {"retry_grace_period": grace}
+    asyncio.run(
+        GithubStatusMonitor().wait_for_statuses(repo="org/repo", ref="sha", token=Mock(), checks=["a"], **kwargs)
+    )
+    assert wait.await_args.kwargs["retry_grace_period"] == (0 if grace is None else grace)
+
+
+@pytest.mark.parametrize("grace", [None, 0, 120])
+def test_dagger_checks_forward_retry_grace(monkeypatch, grace):
+    wait = AsyncMock(return_value="all 1 checks succeeded")
+    monkeypatch.setattr(GithubStatusMonitor, "_wait", wait)
+    check = Mock()
+    check.name = AsyncMock(return_value="a")
+    checks = Mock(list_=AsyncMock(return_value=[check]))
+    monkeypatch.setattr(main, "dag", Mock(current_workspace=Mock(return_value=Mock(checks=Mock(return_value=checks)))))
+    kwargs = {} if grace is None else {"retry_grace_period": grace}
+    asyncio.run(GithubStatusMonitor().wait_for_dagger_checks(repo="org/repo", ref="sha", token=Mock(), **kwargs))
+    assert wait.await_args.kwargs["retry_grace_period"] == (60 if grace is None else grace)
+    assert wait.await_args.kwargs["status_names"] == ["a"]
+
+
+def test_wait_follows_error_pending_success(monkeypatch):
+    async def snapshots(*args, **kwargs):
+        for state in ("error", "pending", "success"):
+            yield {"a": Status(state)}
+
+    monkeypatch.setattr(main, "poll_snapshots", snapshots)
+    result = asyncio.run(
+        GithubStatusMonitor().wait_for_statuses(
+            repo="org/repo",
+            ref="sha",
+            token=Mock(plaintext=AsyncMock(return_value="test-token")),
+            checks=["a"],
+            retry_grace_period=60,
+        )
+    )
+    assert result == "all 1 checks succeeded"
diff --git a/github/tests/test_watcher.py b/github/tests/test_watcher.py
index ceda029..f8e17eb 100644
--- a/github/tests/test_watcher.py
+++ b/github/tests/test_watcher.py
@@ -363,3 +363,82 @@ def test_step_prints_header_only_once():
     # Header line ("waiting on") appears exactly once.
     assert after_first.count("waiting on") == 1
     assert after_second.count("waiting on") == 1
+
+
+@pytest.mark.parametrize("fail_fast", [False, True])
+def test_error_can_retry_and_succeed_within_grace(fail_fast):
+    w, c, clock = make_step_watcher(["a", "b"], retry_grace_period=60, fail_fast=fail_fast)
+    assert w.step({"a": Status("error"), "b": Status("success")}) is Step.CONTINUE
+    assert w.retrying == ["a"]
+    assert "retrying=1" in c.export_text(clear=False)
+    clock.t = 6
+    assert w.step({"a": Status("pending"), "b": Status("success")}) is Step.CONTINUE
+    assert w.retrying == []
+    clock.t = 300
+    assert w.step({"a": Status("success"), "b": Status("success")}) is Step.SUCCEEDED
+
+
+def test_repeated_error_polls_do_not_extend_grace():
+    w, _, clock = make_step_watcher(["a"], retry_grace_period=60)
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 59
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 60
+    assert w.step({"a": Status("error")}) is Step.FAILED
+    assert w.failed == ["a"]
+
+
+def test_new_error_after_retry_gets_new_grace():
+    w, _, clock = make_step_watcher(["a"], retry_grace_period=60)
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 6
+    assert w.step({"a": Status("pending")}) is Step.CONTINUE
+    clock.t = 400
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 459
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 460
+    assert w.step({"a": Status("error")}) is Step.FAILED
+
+
+def test_grace_is_per_check():
+    w, _, clock = make_step_watcher(["a", "b"], retry_grace_period=60, fail_fast=True)
+    assert w.step({"a": Status("error"), "b": Status("pending")}) is Step.CONTINUE
+    clock.t = 50
+    assert w.step({"a": Status("error"), "b": Status("error")}) is Step.CONTINUE
+    clock.t = 60
+    assert w.step({"a": Status("error"), "b": Status("error")}) is Step.FAILED
+    assert w.retrying == ["b"]
+
+
+@pytest.mark.parametrize("fail_fast", [False, True])
+def test_failure_is_not_delayed_by_error_grace(fail_fast):
+    w, _, _ = make_step_watcher(["a", "b"], retry_grace_period=60, fail_fast=fail_fast)
+    assert w.step({"a": Status("failure"), "b": Status("error")}) is Step.FAILED
+
+
+def test_zero_grace_preserves_immediate_error_failure():
+    w, _, _ = make_step_watcher(["a"], retry_grace_period=0)
+    assert w.step({"a": Status("error")}) is Step.FAILED
+
+
+def test_negative_grace_is_rejected():
+    with pytest.raises(ValueError, match="retry_grace_period must be non-negative"):
+        make_step_watcher(["a"], retry_grace_period=-1)
+
+
+def test_retry_grace_does_not_extend_wallclock_timeout():
+    w, c, clock = make_step_watcher(["a"], retry_grace_period=60, timeout=10)
+    assert w.step({"a": Status("pending")}) is Step.CONTINUE
+    clock.t = 9
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 11
+    assert w.step({"a": Status("error")}) is Step.WALLCLOCK_TIMEOUT
+    assert "retrying=['a']" in c.export_text(clear=False)
+
+
+def test_retry_grace_does_not_delay_discovery_timeout():
+    w, _, clock = make_step_watcher(["a", "b"], retry_grace_period=60, discovery_timeout=10)
+    assert w.step({"a": Status("error")}) is Step.CONTINUE
+    clock.t = 11
+    assert w.step({"a": Status("error")}) is Step.DISCOVERY_TIMEOUT