chore: remove unnecessary missing-typed-dict-key suppression

Co-authored-by: Cursor <cursoragent@cursor.com>

libs/checkpoint-conformance/pyproject.toml

@@ -50,7 +50,6 @@ unresolved-attribute = "ignore"
 unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
-missing-typed-dict-key = "ignore"
 
 [tool.ruff]
 lint.select = [

libs/langgraph/langgraph/pregel/_loop.py

@@ -776,17 +776,6 @@ class PregelLoop:
                 and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
                 in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {})
             )
-            # Outer graph: time-travel-resume — explicit non-head
-            # checkpoint paired with a Command(resume=...). Without this,
-            # `Command(resume=...) + checkpoint=<non-head>` would match the
-            # plain-resume exclusion below and skip the cleanup needed to
-            # treat this as a fork.
-            or (
-                not self.is_nested
-                and getattr(self, "_loaded_explicit_non_head", False)
-                and input_is_command
-                and cast(Command, self.input).resume is not None
-            )
             or not (
                 # Outer graph: resume arrives as Command(resume=...)
                 (input_is_command and cast(Command, self.input).resume is not None)
@@ -1511,13 +1500,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
 
     def __enter__(self) -> Self:
         self._graph_lifecycle_events = deque()
-        # Set by the explicit-id branch when the loaded checkpoint is not
-        # the latest for the (thread, ns). Lets _first distinguish a
-        # plain resume (`Command(resume=...)` against head, which carries
-        # an explicit checkpoint_id only because clients like LangGraph
-        # Studio echo it back) from time-travel-resume (`Command(resume=
-        # ...)` against an explicit non-head checkpoint).
-        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1525,14 +1507,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
-            if saved is not None and not self.is_nested:
-                head = self.checkpointer.get_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1784,7 +1758,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
 
     async def __aenter__(self) -> Self:
         self._graph_lifecycle_events = deque()
-        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1792,14 +1765,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-            if saved is not None and not self.is_nested:
-                head = await self.checkpointer.aget_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint

libs/langgraph/tests/test_retry.py

@@ -1674,28 +1674,15 @@ async def test_arun_with_retry_timeout_observer_tracks_attempts():
 async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     events: list = []
 
-    # `_TimedAttemptScope.__init__` sets `_last_progress` to `time.monotonic()`,
-    # but the watchdog itself doesn't start running until after `wrap_config`
-    # and task scheduling — under CI load that gap can be large enough to eat
-    # the entire idle window before the task body's first await even runs. We
-    # defend against that by:
-    #   1. Using a generous idle_timeout so scheduling slack stays well within it.
-    #   2. Calling `runtime.heartbeat()` BEFORE the first sleep, which resets
-    #      `_last_progress` to "now" the moment the task body actually starts.
-    idle_timeout_s = 1.0
-
     class HeartbeatProc:
         async def ainvoke(self, input, config):
             runtime = config[CONF][CONFIG_KEY_RUNTIME]
-            runtime.heartbeat()  # reset the idle clock at task-body entry
             for _ in range(8):
                 await asyncio.sleep(0.05)
                 runtime.heartbeat()
             return "ok"
 
-    task = _make_task(
-        HeartbeatProc(), timeout=_idle_timeout(idle_timeout_s), name="heartbeat"
-    )
+    task = _make_task(HeartbeatProc(), timeout=_idle_timeout(0.2), name="heartbeat")
     task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
     assert await arun_with_retry(task, retry_policy=None) == "ok"
 
@@ -1704,13 +1691,13 @@ async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     assert by_event[-1] == "finish"
     progress = [ev for ev in events if ev.event == "progress"]
     assert progress, "expected at least one progress event from heartbeat"
-    # Rate limit is `idle_timeout / 4` = 0.25s; with the task running for
-    # ~400ms we expect 1–2 progress events (well below the 9 heartbeats).
+    # Rate limit is `idle_timeout / 4` = 0.05s; with 8 heartbeats spaced ~0.05s
+    # we should see at most ~one progress event per heartbeat (well below 8).
     assert len(progress) <= len(by_event)
     for ev in progress:
         assert ev.context.task_name == "heartbeat"
         assert ev.context.attempt == 1
-        assert ev.context.idle_timeout_secs == idle_timeout_s
+        assert ev.context.idle_timeout_secs == 0.2
         assert isinstance(ev.progress_at, datetime)
 
 

libs/langgraph/tests/test_time_travel.py

@@ -3964,96 +3964,3 @@ def test_subgraph_called_in_loop_loads_state_on_replay(
 
     assert len(observed) == 1
     assert observed[0] == ("sub_step", {"sub_trail": ["s", "s", "s"]})
-
-
-def test_subgraph_time_travel_resume_with_command_in_one_call(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Repro for time-travel-resume issued as a SINGLE call:
-    `invoke(Command(resume=...), config={..., checkpoint_id: <non-head>})`.
-
-    Equivalent to the existing two-call pattern
-    (`test_subgraph_time_travel_resume_from_first_interrupt`) which does
-    `invoke(None, sub_config)` then `invoke(Command(resume=...))`. Clients
-    like LangGraph API server / Studio can submit this as one request.
-    """
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run to completion through both interrupts.
-    graph.invoke({"value": []}, config)
-    parent_state_at_first = graph.get_state(config, subgraphs=True)
-    sub_config_at_first = parent_state_at_first.tasks[0].state.config
-    graph.invoke(Command(resume="answer_1"), config)
-    graph.invoke(Command(resume="answer_2"), config)
-
-    final_after_original = graph.get_state(config).values
-    assert final_after_original["value"] == [
-        "step_a_done",
-        "ask_1:answer_1",
-        "ask_2:answer_2",
-    ]
-
-    # ONE-CALL time-travel-resume: pair Command(resume=...) with a
-    # non-head parent checkpoint_id in the same invocation.
-    called.clear()
-    result = graph.invoke(Command(resume="new_answer_1"), sub_config_at_first)
-
-    # `step_a` ran before the time-travel target — must not re-execute.
-    assert "step_a" not in called, f"step_a re-executed; called={called}"
-    # The new answer must be applied (not the cached `answer_1`).
-    assert result.get("__interrupt__"), (
-        f"expected to pause at the next interrupt; got result={result}"
-    )
-    assert result["__interrupt__"][0].value == "Question 2?"
-    sub_state_now = graph.get_state(config, subgraphs=True).tasks[0].state
-    assert sub_state_now.values["value"] == ["step_a_done", "ask_1:new_answer_1"], (
-        f"subgraph state at int2 should reflect new_answer_1; "
-        f"got {sub_state_now.values}"
-    )
-
-    # Resume the second interrupt with another new answer; ensure the
-    # whole branch concludes consistently.
-    called.clear()
-    final = graph.invoke(Command(resume="new_answer_2"), config)
-    assert "step_a" not in called
-    assert "ask_1" not in called
-    assert final["value"] == [
-        "step_a_done",
-        "ask_1:new_answer_1",
-        "ask_2:new_answer_2",
-    ]