diff --git a/planfile/__init__.py b/planfile/__init__.py
index ecf1480..abb9ac3 100644
--- a/planfile/__init__.py
+++ b/planfile/__init__.py
@@ -311,9 +311,22 @@ def next_ticket(self, sprint: str = "current", queue: str | None = None) -> Tick
         )
         return min(runnable, key=self._ticket_sort_key, default=None)
 
-    def update_ticket(self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates):
+    def update_ticket(
+        self,
+        ticket_id: str,
+        reason: str | None = None,
+        actor: str | None = None,
+        expected_updated_at: str | None = None,
+        **updates,
+    ):
         """Delegate with optional reason (why status/etc changed) and actor (who/by)."""
-        return self.store.update_ticket(ticket_id, reason=reason, actor=actor, **updates)
+        return self.store.update_ticket(
+            ticket_id,
+            reason=reason,
+            actor=actor,
+            expected_updated_at=expected_updated_at,
+            **updates,
+        )
 
     def append_ticket_evidence(
         self,
@@ -427,7 +440,15 @@ def complete_ticket(
         execution = TicketExecution(**execution_data)
         return self.update_ticket(ticket_id, status="done", execution=execution, outputs=outputs, reason=reason, actor=actor)
 
-    def fail_ticket(self, ticket_id: str, error: str, *, reason: str | None = None, actor: str | None = None) -> Ticket | None:
+    def fail_ticket(
+        self,
+        ticket_id: str,
+        error: str,
+        *,
+        reason: str | None = None,
+        actor: str | None = None,
+        expected_updated_at: str | None = None,
+    ) -> Ticket | None:
         ticket = self.get_ticket(ticket_id)
         if not ticket:
             return None
@@ -452,6 +473,7 @@ def fail_ticket(self, ticket_id: str, error: str, *, reason: str | None = None,
             execution=execution,
             reason=reason or error,
             actor=actor,
+            expected_updated_at=expected_updated_at,
         )
 
     def block_ticket(self, ticket_id: str, reason: str | None = None, note: str | None = None, *, actor: str | None = None) -> Ticket | None:
diff --git a/planfile/api/server.py b/planfile/api/server.py
index f49f5ec..3d10f5b 100644
--- a/planfile/api/server.py
+++ b/planfile/api/server.py
@@ -44,7 +44,7 @@
     TicketOutputs,
     TicketSource,
 )
-from planfile.core.store import ImmutableTerminalReopenError
+from planfile.core.store import ImmutableTerminalReopenError, TicketUpdatedAtConflictError
 from planfile.runtime_context import (
     DEFAULT_CONFIG as DEFAULT_RUNTIME_CONFIG,
 )
@@ -73,6 +73,8 @@ async def lifespan(_: FastAPI):
     lifespan=lifespan,
 )
 
+API_CAPABILITIES = ["ticket.fail.expected_updated_at"]
+
 
 @app.exception_handler(ImmutableTerminalReopenError)
 async def immutable_terminal_reopen_handler(
@@ -81,6 +83,17 @@ async def immutable_terminal_reopen_handler(
 ):
     return JSONResponse(status_code=409, content={"detail": "immutable_terminal_reopen"})
 
+
+@app.exception_handler(TicketUpdatedAtConflictError)
+async def ticket_updated_at_conflict_handler(
+    _: Request,
+    __: TicketUpdatedAtConflictError,
+):
+    return JSONResponse(
+        status_code=409,
+        content={"detail": "ticket_updated_at_precondition_failed"},
+    )
+
 _cors_origins = [
     origin.strip()
     for origin in os.environ.get("PLANFILE_CORS_ORIGINS", "").split(",")
@@ -190,6 +203,11 @@ class TicketFailRequest(BaseModel):
     error: str
     reason: str | None = None
     actor: str | None = None
+    expected_updated_at: str | None = None
+
+
+class TicketFailIfCurrentRequest(TicketFailRequest):
+    expected_updated_at: str
 
 
 class TicketInputRequest(BaseModel):
@@ -825,8 +843,7 @@ async def complete_ticket(ticket_id: str, body: TicketCompleteRequest):
     return ticket.model_dump(mode="json", exclude_none=True)
 
 
-@app.post("/tickets/{ticket_id}/fail", tags=["tickets"])
-async def fail_ticket(ticket_id: str, body: TicketFailRequest):
+async def _fail_ticket(ticket_id: str, body: TicketFailRequest):
     pf = get_planfile()
     current = pf.get_ticket(ticket_id)
     if not current:
@@ -837,6 +854,7 @@ async def fail_ticket(ticket_id: str, body: TicketFailRequest):
         error=body.error,
         reason=body.reason or body.error,
         actor=body.actor or "unknown:api",
+        expected_updated_at=body.expected_updated_at,
     )
     if not ticket:
         raise HTTPException(404, f"Ticket {ticket_id} not found")
@@ -844,6 +862,18 @@ async def fail_ticket(ticket_id: str, body: TicketFailRequest):
     return ticket.model_dump(mode="json", exclude_none=True)
 
 
+@app.post("/tickets/{ticket_id}/fail", tags=["tickets"])
+async def fail_ticket(ticket_id: str, body: TicketFailRequest):
+    return await _fail_ticket(ticket_id, body)
+
+
+@app.post("/tickets/{ticket_id}/fail-if-current", tags=["tickets"])
+async def fail_ticket_if_current(ticket_id: str, body: TicketFailIfCurrentRequest):
+    """Fail a ticket only when it is still the exact observed revision."""
+
+    return await _fail_ticket(ticket_id, body)
+
+
 @app.post("/tickets/{ticket_id}/input-required", tags=["tickets"])
 async def wait_for_input(ticket_id: str, body: TicketInputRequest):
     pf = get_planfile()
@@ -3177,7 +3207,11 @@ async def websocket_dsl(websocket: WebSocket, project_path: str = "."):
 @app.get("/health", tags=["system"])
 def health():
     import planfile
-    return {"status": "ok", "version": planfile.__version__}
+    return {
+        "status": "ok",
+        "version": planfile.__version__,
+        "capabilities": API_CAPABILITIES,
+    }
 
 
 @app.get("/", response_class=HTMLResponse, tags=["system"])
diff --git a/planfile/client.py b/planfile/client.py
index d3dc1aa..b56fda2 100644
--- a/planfile/client.py
+++ b/planfile/client.py
@@ -158,8 +158,9 @@ def fail(
         error: str,
         reason: str | None = None,
         actor: str | None = None,
+        expected_updated_at: str | None = None,
     ) -> TicketTransitionResult:
-        """Record one failed execution attempt without deciding its retry policy."""
+        """Record one failed attempt, optionally only for an observed ticket revision."""
 
         return self._apply(
             "fail",
@@ -168,6 +169,7 @@ def fail(
                 error=error,
                 reason=reason,
                 actor=actor,
+                expected_updated_at=expected_updated_at,
             ),
         )
 
diff --git a/planfile/core/store.py b/planfile/core/store.py
index 31ff674..d50871c 100644
--- a/planfile/core/store.py
+++ b/planfile/core/store.py
@@ -22,6 +22,10 @@ class ImmutableTerminalReopenError(RuntimeError):
     """Raised when an ordinary mutation tries to reactivate done/canceled work."""
 
 
+class TicketUpdatedAtConflictError(RuntimeError):
+    """Raised when a ticket changed after the caller observed it."""
+
+
 class Store(StoreFileMixin, TicketStoreMixin):
     """File-based ticket store using .planfile/ directory."""
 
@@ -1833,14 +1837,61 @@ def _build_history_entry(
         return entry
 
     def update_ticket(
-        self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates
+        self,
+        ticket_id: str,
+        reason: str | None = None,
+        actor: str | None = None,
+        expected_updated_at: str | None = None,
+        **updates,
     ) -> Ticket | None:
         """Update a ticket. If status (or execution state) changes, a structured history entry
         is appended automatically, including optional `reason` (why) and `actor` (who / by whom).
         Use reason/actor (or _reason/_actor in **updates) for rich audit on status transitions.
         """
         with self.mutation_lock():
-            return self._update_ticket_unlocked(ticket_id, reason=reason, actor=actor, **updates)
+            return self._update_ticket_unlocked(
+                ticket_id,
+                reason=reason,
+                actor=actor,
+                expected_updated_at=expected_updated_at,
+                **updates,
+            )
+
+    @staticmethod
+    def _updated_at_instant(value: object) -> datetime | None:
+        if isinstance(value, datetime):
+            parsed = value
+        elif isinstance(value, str) and value:
+            try:
+                parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+            except ValueError:
+                return None
+        else:
+            return None
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=UTC)
+        return parsed.astimezone(UTC)
+
+    def _guard_expected_updated_at(
+        self,
+        previous: dict,
+        expected_updated_at: str | None,
+    ) -> None:
+        if expected_updated_at is None:
+            return
+        projected = self._project_ticket_evidence(previous)
+        actual_updated_at = projected.get("updated_at")
+        actual_instant = self._updated_at_instant(actual_updated_at)
+        expected_instant = self._updated_at_instant(expected_updated_at)
+        if (
+            actual_instant is not None
+            and expected_instant is not None
+            and actual_instant == expected_instant
+        ):
+            return
+        if str(actual_updated_at or "") == str(expected_updated_at):
+            return
+        raise TicketUpdatedAtConflictError("ticket_updated_at_precondition_failed")
 
     def _guard_immutable_terminal_reopen(
         self,
@@ -1987,7 +2038,12 @@ def append_ticket_evidence(
             return model, True
 
     def _update_ticket_unlocked(
-        self, ticket_id: str, reason: str | None = None, actor: str | None = None, **updates
+        self,
+        ticket_id: str,
+        reason: str | None = None,
+        actor: str | None = None,
+        expected_updated_at: str | None = None,
+        **updates,
     ) -> Ticket | None:
         index_was_current = self._begin_index_mutation()
         if self._uses_sharded_storage():
@@ -1995,6 +2051,7 @@ def _update_ticket_unlocked(
                 ticket_id,
                 reason=reason,
                 actor=actor,
+                expected_updated_at=expected_updated_at,
                 _index_was_current=index_was_current,
                 **updates,
             )
@@ -2007,6 +2064,7 @@ def _update_ticket_unlocked(
             tickets = sprint_data.get("tickets", {})
             if ticket_id in tickets:
                 previous = dict(tickets[ticket_id])
+                self._guard_expected_updated_at(previous, expected_updated_at)
                 # Extract history metadata (reason=why the change, actor/by=who performed it)
                 # Support both named params (from high-level methods) and _-prefixed or bare in updates
                 history_reason = (
@@ -2074,6 +2132,7 @@ def _update_ticket_sharded_unlocked(
         ticket_id: str,
         reason: str | None = None,
         actor: str | None = None,
+        expected_updated_at: str | None = None,
         _index_was_current: bool = False,
         **updates,
     ) -> Ticket | None:
@@ -2083,6 +2142,7 @@ def _update_ticket_sharded_unlocked(
             return None
         sprint, ticket_data = located
         previous = dict(ticket_data)
+        self._guard_expected_updated_at(previous, expected_updated_at)
         history_reason = reason or updates.pop("reason", None) or updates.pop("_reason", None)
         history_actor = actor or updates.pop("actor", None) or updates.pop("_actor", None)
         serialized_updates = {
diff --git a/tests/test_ticket_api_events.py b/tests/test_ticket_api_events.py
index 66d84d7..3f3f676 100644
--- a/tests/test_ticket_api_events.py
+++ b/tests/test_ticket_api_events.py
@@ -189,6 +189,80 @@ def test_governed_ticket_mutations_require_attributed_history(tmp_path, monkeypa
     assert updated.json()["history"][-1]["reason"] == "Escalated priority after preflight."
 
 
+def test_fail_api_returns_conflict_for_stale_updated_at_precondition(tmp_path, monkeypatch):
+    pf = Planfile(str(tmp_path))
+    ticket = pf.create_ticket(
+        name="Watchdog candidate",
+        execution=TicketExecution(state="running", assigned_to="bot:worker", max_attempts=2),
+    )
+    observed_updated_at = ticket.model_dump(mode="json")["updated_at"]
+    changed = pf.update_ticket(ticket.id, priority="high")
+    assert changed is not None
+    monkeypatch.setattr(server, "get_planfile", lambda: pf)
+    client = TestClient(server.app)
+
+    conflict = client.post(
+        f"/tickets/{ticket.id}/fail-if-current",
+        json={
+            "error": "stale_execution_timeout",
+            "expected_updated_at": observed_updated_at,
+        },
+    )
+
+    assert conflict.status_code == 409
+    assert conflict.json() == {"detail": "ticket_updated_at_precondition_failed"}
+    current = pf.get_ticket(ticket.id)
+    assert current is not None
+    assert current.priority == "high"
+    assert current.execution.state == "running"
+    assert current.execution.attempt == 0
+
+
+def test_fail_api_accepts_current_updated_at_in_json_timestamp_form(tmp_path, monkeypatch):
+    pf = Planfile(str(tmp_path))
+    ticket = pf.create_ticket(
+        name="Current watchdog candidate",
+        execution=TicketExecution(state="running", assigned_to="bot:worker", max_attempts=2),
+    )
+    expected_updated_at = ticket.model_dump(mode="json")["updated_at"]
+    monkeypatch.setattr(server, "get_planfile", lambda: pf)
+    client = TestClient(server.app)
+
+    response = client.post(
+        f"/tickets/{ticket.id}/fail",
+        json={
+            "error": "stale_execution_timeout",
+            "expected_updated_at": expected_updated_at,
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["execution"]["state"] == "ready"
+    assert response.json()["execution"]["attempt"] == 1
+
+
+def test_fail_if_current_api_requires_updated_at_precondition(tmp_path, monkeypatch):
+    pf = Planfile(str(tmp_path))
+    ticket = pf.create_ticket(
+        name="Fail-closed watchdog candidate",
+        execution=TicketExecution(state="running", assigned_to="bot:worker"),
+    )
+    monkeypatch.setattr(server, "get_planfile", lambda: pf)
+    client = TestClient(server.app)
+
+    response = client.post(
+        f"/tickets/{ticket.id}/fail-if-current",
+        json={"error": "stale_execution_timeout"},
+    )
+
+    assert response.status_code == 422
+    current = pf.get_ticket(ticket.id)
+    assert current is not None
+    assert current.status == "open"
+    assert current.execution is not None
+    assert current.execution.attempt == 0
+
+
 def test_governed_ticket_creation_requires_structured_four_part_envelope(tmp_path, monkeypatch):
     pf = Planfile(str(tmp_path))
     monkeypatch.setattr(server, "get_planfile", lambda: pf)
@@ -618,6 +692,7 @@ def test_openapi_and_health_publish_same_version():
     client = TestClient(server.app)
 
     assert client.get("/openapi.json").json()["info"]["version"] == client.get("/health").json()["version"]
+    assert "ticket.fail.expected_updated_at" in client.get("/health").json()["capabilities"]
 
 
 def test_ticket_list_pagination_headers(tmp_path, monkeypatch):
diff --git a/tests/test_ticket_execution.py b/tests/test_ticket_execution.py
index e021682..c2f1e36 100644
--- a/tests/test_ticket_execution.py
+++ b/tests/test_ticket_execution.py
@@ -12,7 +12,7 @@
     TicketOutputs,
     TicketSource,
 )
-from planfile.core.store import ImmutableTerminalReopenError
+from planfile.core.store import ImmutableTerminalReopenError, TicketUpdatedAtConflictError
 
 
 def test_ticket_round_trip_with_execution_fields(tmp_path):
@@ -221,6 +221,80 @@ def test_failed_attempt_requeues_until_retry_budget_is_exhausted(tmp_path):
     assert exhausted.execution.attempt == 2
 
 
+@pytest.mark.parametrize("sharded", [False, True])
+def test_fail_ticket_rejects_stale_updated_at_without_mutation(tmp_path, sharded):
+    pf = Planfile(str(tmp_path))
+    if sharded:
+        config = pf.store._read_config()
+        config["archive"]["enabled"] = False
+        pf.store._write_config(config)
+        pf.store.migrate_to_sharded_yaml(shard_size=100)
+    ticket = pf.create_ticket(
+        name="Atomic stale failure guard",
+        execution=TicketExecution(state="running", assigned_to="bot:test", max_attempts=2),
+    )
+    observed_updated_at = ticket.model_dump(mode="json")["updated_at"]
+    changed = pf.update_ticket(
+        ticket.id,
+        priority="high",
+        actor="bot:other",
+        reason="concurrent_priority_change",
+    )
+    assert changed is not None
+    history_before = list(changed.history)
+    events_before = pf.store.operational_events(ticket_id=ticket.id)
+
+    with pytest.raises(
+        TicketUpdatedAtConflictError,
+        match="ticket_updated_at_precondition_failed",
+    ):
+        pf.fail_ticket(
+            ticket.id,
+            error="stale_watchdog_failure",
+            actor="bot:watchdog",
+            expected_updated_at=observed_updated_at,
+        )
+
+    current = pf.get_ticket(ticket.id)
+    assert current is not None
+    assert current.priority == "high"
+    assert current.status == "open"
+    assert current.execution.state == "running"
+    assert current.execution.attempt == 0
+    assert current.history == history_before
+    assert pf.store.operational_events(ticket_id=ticket.id) == events_before
+
+
+def test_fail_ticket_accepts_updated_at_from_append_only_evidence_projection(tmp_path):
+    pf = Planfile(str(tmp_path))
+    ticket = pf.create_ticket(
+        name="Candidate with evidence",
+        execution=TicketExecution(state="running", assigned_to="bot:test", max_attempts=2),
+    )
+    projected, recorded = pf.append_ticket_evidence(
+        ticket.id,
+        idempotency_key="heartbeat-1",
+        collection="heartbeats",
+        evidence={"status": "observed"},
+        reason="worker heartbeat",
+        actor="bot:test",
+    )
+    assert recorded is True
+    assert projected is not None
+    expected_updated_at = projected.model_dump(mode="json")["updated_at"]
+
+    failed = pf.fail_ticket(
+        ticket.id,
+        error="worker_stopped_after_heartbeat",
+        actor="bot:watchdog",
+        expected_updated_at=expected_updated_at,
+    )
+
+    assert failed is not None
+    assert failed.execution.state == "ready"
+    assert failed.execution.attempt == 1
+
+
 @pytest.mark.parametrize("terminal_status", ["done", "canceled", "blocked"])
 def test_terminal_status_normalizes_execution_state(tmp_path, terminal_status):
     pf = Planfile(str(tmp_path))