refactor

Support custom link titles for custom link syntax

docs/_scripts/handle_auto_links.py

@@ -0,0 +1,181 @@
+"""Logic to identify and transform cross-reference links in markdown files.
+
+This module allows supporting custom markdown syntax for "autolinks". These are links
+that will be transformed based on the current scope context, such as "global", "python",
+or "js" into an appropriate markdown link format.
+
+For example,
+
+```markdown
+@[StateGraph]
+```
+
+May be transformed into:
+
+```markdown
+[StateGraph](some_path/api-reference/state-graph.md)
+```
+
+The transformation value depends on the scope in which the link is used.
+"""
+
+import logging
+import re
+from typing import Optional
+
+from _scripts.link_map import SCOPE_LINK_MAPS
+
+logger = logging.getLogger(__name__)
+
+
+def _transform_link(
+    link_name: str, scope: str, file_path: str, line_number: int, custom_title: Optional[str] = None
+) -> Optional[str]:
+    """Transform a cross-reference link based on the current scope.
+
+    Args:
+        link_name: The name of the link to transform (e.g., "StateGraph").
+        scope: The current scope context ("global", "python", "js", etc.).
+        file_path: The file path for error reporting.
+        line_number: The line number for error reporting.
+        custom_title: Optional custom title for the link. If None, uses link_name.
+
+    Returns:
+        A formatted markdown link if the link is found in the scope mapping,
+        None otherwise.
+
+    Example:
+        >>> _transform_link("StateGraph", "python", "file.md", 5)
+        "[StateGraph](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.StateGraph)"
+
+        >>> _transform_link("StateGraph", "python", "file.md", 5, "Custom Title")
+        "[Custom Title](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.StateGraph)"
+
+        >>> _transform_link("unknown-link", "python", "file.md", 5)
+        None
+    """
+    if scope == "global":
+        # Special scope that is composed of both Python and JS links
+        # For now, we will substitute in the python scope!
+        # But we need to add support for handling both scopes.
+        scope = "python"
+        logger.error(
+            "Encountered unhandled 'global' scope. Defaulting to 'python'."
+            "In file: %s, line %d, link_name: %s",
+            file_path,
+            line_number,
+            link_name,
+        )
+    link_map = SCOPE_LINK_MAPS.get(scope, {})
+    url = link_map.get(link_name)
+
+    if url:
+        title = custom_title if custom_title is not None else link_name
+        return f"[{title}]({url})"
+    else:
+        # Log error with file location information
+        logger.info(
+            # Using %s
+            "Link '%s' not found in scope '%s'. "
+            "In file: %s, line %d. Available links in scope: %s",
+            link_name,
+            scope,
+            file_path,
+            line_number,
+            list(link_map.keys() if link_map else []),
+        )
+        return None
+
+
+CONDITIONAL_FENCE_PATTERN = re.compile(
+    r"""
+    ^                       # Start of line
+    (?P<indent>[ \t]*)      # Optional indentation (spaces or tabs)
+    :::                     # Literal fence marker
+    (?P<language>\w+)?      # Optional language identifier (named group: language)
+    \s*                     # Optional trailing whitespace
+    $                       # End of line
+    """,
+    re.VERBOSE,
+)
+CROSS_REFERENCE_PATTERN = re.compile(
+    r"""
+    @                       # Literal @ symbol
+    (?:                     # Non-capturing group for two possible formats:
+        \[                  # Opening bracket for title
+        (?P<title>[^\]]+)   # Custom title - one or more non-bracket characters
+        \]                  # Closing bracket for title
+        \[                  # Opening bracket for link name
+        (?P<link_name_with_title>[^\]]+)  # Link name - one or more non-bracket characters
+        \]                  # Closing bracket for link name
+        |                   # OR
+        \[                  # Opening bracket
+        (?P<link_name>[^\]]+)   # Link name - one or more non-bracket characters
+        \]                  # Closing bracket
+    )
+    """,
+    re.VERBOSE,
+)
+
+
+def _replace_autolinks(markdown: str, file_path: str) -> str:
+    """Preprocess markdown lines to handle @[links] with conditional fence scopes.
+
+    This function processes markdown content to transform @[link_name] references
+    based on the current conditional fence scope. Conditional fences use the
+    syntax :::language to define scope boundaries.
+
+    Args:
+        markdown: The markdown content to process.
+        file_path: The file path for error reporting.
+
+    Returns:
+        Processed markdown content with @[references] transformed to proper
+        markdown links or left unchanged if not found.
+
+    Example:
+        Input:
+            "@[StateGraph]\\n:::python\\n@[Command]\\n:::\\n"
+        Output:
+            "[StateGraph](url)\\n:::python\\n[Command](url)\\n:::\\n"
+    """
+    # Track the current scope context
+    current_scope = "global"
+    lines = markdown.splitlines(keepends=True)
+    processed_lines = []
+
+    for line_number, line in enumerate(lines, 1):
+        line_stripped = line.strip()
+
+        # Check if this line defines a new conditional fence scope
+        fence_match = CONDITIONAL_FENCE_PATTERN.match(line_stripped)
+        if fence_match:
+            language = fence_match.group("language")
+            # Set scope to the specified language, or reset to global if no language
+            current_scope = language.lower() if language else "global"
+            processed_lines.append(line)
+            continue
+
+        # Transform all @[link_name] references in this line based on current scope
+        def replace_cross_reference(match: re.Match[str]) -> str:
+            """Replace a single @[link_name] with the scoped equivalent."""
+            # Check if this is the @[title][ref] format or @[ref] format
+            title = match.group("title")
+            if title is not None:
+                # This is @[title][ref] format
+                link_name = match.group("link_name_with_title")
+                custom_title = title
+            else:
+                # This is @[ref] format
+                link_name = match.group("link_name")
+                custom_title = None
+            
+            transformed = _transform_link(
+                link_name, current_scope, file_path, line_number, custom_title
+            )
+            return transformed if transformed is not None else match.group(0)
+
+        transformed_line = CROSS_REFERENCE_PATTERN.sub(replace_cross_reference, line)
+        processed_lines.append(transformed_line)
+
+    return "".join(processed_lines)

docs/_scripts/link_map.py

@@ -1,5 +1,142 @@
-JS_LINK_MAP = {
-    "langgraph.types.interrupt": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph.interrupt-2.html",
-    "create_react_agent": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph_prebuilt.createReactAgent.html",
-    "langgraph.types.Command": "https://langchain-ai.github.io/langgraphjs/reference/classes/langgraph.Command.html",
+"""Link mapping for cross-reference resolution across different scopes.
+
+This module provides link mappings for different language/framework scopes
+to resolve @[link_name] references to actual URLs.
+"""
+
+# Python-specific link mappings
+# Python-specific link mappings
+PYTHON_LINK_MAP = {
+    "StateGraph": "reference/graphs/#langgraph.graph.StateGraph",
+    "add_conditional_edges": "reference/graphs/#langgraph.graph.StateGraph.add_conditional_edges",
+    "add_edge": "reference/graphs/#langgraph.graph.StateGraph.add_edge",
+    "add_node": "reference/graphs/#langgraph.graph.StateGraph.add_node",
+    "add_messages": "reference/messages/#langgraph.graph.message.add_messages",
+    "ToolNode": "reference/prebuilt/#langgraph.prebuilt.tool_node.ToolNode",
+    "CompiledStateGraph.astream": "reference/graphs/#langgraph.graph.state.CompiledStateGraph.astream",
+    "Pregel.astream": "reference/graphs/#langgraph.pregel.Pregel.astream",
+    "AsyncPostgresSaver": "reference/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver",
+    "AsyncSqliteSaver": "reference/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver",
+    "BaseCheckpointSaver": "reference/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver",
+    "BaseStore": "reference/stores/#langgraph.store.base.BaseStore",
+    "BaseStore.put": "reference/stores/#langgraph.store.base.BaseStore.put",
+    "BinaryOperatorAggregate": "reference/channels/#langgraph.channels.BinaryOperatorAggregate",
+    "CipherProtocol": "reference/checkpoints/#langgraph.checkpoint.serde.base.CipherProtocol",
+    "client.runs.stream": "reference/client/#langgraph_sdk.client.RunsClient.stream",
+    "client.runs.wait": "reference/client/#langgraph_sdk.client.RunsClient.wait",
+    "client.threads.get_history": "reference/client/#langgraph_sdk.client.ThreadsClient.get_history",
+    "client.threads.update_state": "reference/client/#langgraph_sdk.client.ThreadsClient.update_state",
+    "Command": "reference/types/#langgraph.types.Command",
+    "CompiledStateGraph": "reference/graphs/#langgraph.graph.state.CompiledStateGraph",
+    "create_react_agent": "reference/prebuilt/#langgraph.prebuilt.chat_agent_executor.create_react_agent",
+    "create_supervisor": "reference/supervisor/#langgraph_supervisor.supervisor.create_supervisor",
+    "EncryptedSerializer": "reference/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer",
+    "entrypoint.final": "reference/functions/#langgraph.func.entrypoint.final",
+    "entrypoint": "reference/functions/#langgraph.func.entrypoint",
+    "from_pycryptodome_aes": "reference/checkpoints/#langgraph.checkpoint.serde.encrypted.EncryptedSerializer.from_pycryptodome_aes",
+    # "getContextVariable": "<insert-ref>",
+    "get_state_history": "reference/graphs/#langgraph.graph.state.CompiledStateGraph.get_state_history",
+    "get_stream_writer": "reference/config/#langgraph.config.get_stream_writer",
+    "HumanInterrupt": "reference/prebuilt/#langgraph.prebuilt.interrupt.HumanInterrupt",
+    "InjectedState": "reference/prebuilt/#langgraph.prebuilt.InjectedState",
+    "InMemorySaver": "reference/checkpoints/#langgraph.checkpoint.memory.InMemorySaver",
+    "interrupt": "reference/graphs/#langgraph.graph.interrupt",
+    "CompiledStateGraph.invoke": "reference/graphs/#langgraph.graph.state.CompiledStateGraph.invoke",
+    "JsonPlusSerializer": "reference/checkpoints/#langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer",
+    "langgraph.json": "reference/configuration/#configuration-file",
+    "LastValue": "reference/channels/#langgraph.channels.LastValue",
+    # "MemorySaver": "<insert-ref>",
+    # "messagesStateReducer": "<insert-ref>",
+    "PostgresSaver": "reference/checkpoints/#langgraph.checkpoint.postgres.PostgresSaver",
+    "Pregel": "reference/graphs/#langgraph.pregel.Pregel",
+    "Pregel.stream": "reference/graphs/#langgraph.pregel.Pregel.stream",
+    "pre_model_hook": "reference/prebuilt/#langgraph.prebuilt.chat_agent_executor.create_react_agent",
+    "protocol": "reference/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol",
+    "Send": "reference/types/#langgraph.types.Send",
+    "SerializerProtocol": "reference/checkpoints/#langgraph.checkpoint.serde.base.SerializerProtocol",
+    "SqliteSaver": "reference/checkpoints/#langgraph.checkpoint.sqlite.SqliteSaver",
+    "START": "reference/constants/#langgraph.constants.START",
+    "CompiledStateGraph.stream": "reference/graphs/#langgraph.graph.state.CompiledStateGraph.stream",
+    "task": "reference/functions/#langgraph.func.task",
+    "Topic": "reference/channels/#langgraph.channels.Topic",
+    "update_state": "reference/graphs/#langgraph.graph.state.CompiledStateGraph.update_state",
+}
+
+
+# JavaScript-specific link mappings
+JS_LINK_MAP = {
+    "StateGraph": "reference/classes/langgraph.StateGraph.html",
+    "add_conditional_edges": "reference/functions/langgraph_StateGraph.addConditionalEdges.html",
+    "add_edge": "reference/functions/langgraph_StateGraph.addEdge.html",
+    "add_node": "reference/functions/langgraph_StateGraph.addNode.html",
+    "add_messages": "reference/functions/langgraph_message.addMessages.html",
+    "ToolNode": "reference/classes/langgraph_prebuilt.ToolNode.html",
+    "CompiledStateGraph.astream()": "reference/functions/langgraph_CompiledStateGraph.astream.html",
+    "Pregel.astream": "reference/functions/langgraph_Pregel.astream.html",
+    "AsyncPostgresSaver": "reference/classes/langgraph_checkpoint_postgres_aio.AsyncPostgresSaver.html",
+    "AsyncSqliteSaver": "reference/classes/langgraph_checkpoint_sqlite_aio.AsyncSqliteSaver.html",
+    "BaseCheckpointSaver": "reference/classes/langgraph_checkpoint_base.BaseCheckpointSaver.html",
+    "BaseStore": "reference/classes/langgraph_store_base.BaseStore.html",
+    "BaseStore.put": "reference/functions/langgraph_store_base.BaseStore.put.html",
+    "BinaryOperatorAggregate": "reference/classes/langgraph_channels.BinaryOperatorAggregate.html",
+    "CipherProtocol": "reference/classes/langgraph_checkpoint_serde_base.CipherProtocol.html",
+    "client.runs.stream": "reference/functions/langgraph_sdk_client.RunsClient.stream.html",
+    "client.runs.wait": "reference/functions/langgraph_sdk_client.RunsClient.wait.html",
+    "client.threads.get_history": "reference/functions/langgraph_sdk_client.ThreadsClient.getHistory.html",
+    "client.threads.update_state": "reference/functions/langgraph_sdk_client.ThreadsClient.updateState.html",
+    "Command": "reference/classes/langgraph.Command.html",
+    "CompiledStateGraph": "reference/classes/langgraph.CompiledStateGraph.html",
+    "create_react_agent": "reference/functions/langgraph_prebuilt.createReactAgent.html",
+    "create_supervisor": "reference/functions/langgraph_supervisor.createSupervisor.html",
+    "EncryptedSerializer": "reference/classes/langgraph_checkpoint_serde_encrypted.EncryptedSerializer.html",
+    "entrypoint.final": "reference/functions/langgraph_func.entrypoint.final.html",
+    "entrypoint": "reference/functions/langgraph_func.entrypoint.html",
+    "from_pycryptodome_aes": "reference/functions/langgraph_checkpoint_serde_encrypted.EncryptedSerializer.fromPycryptodomeAes.html",
+    # "getContextVariable": "<insert-ref>",
+    "get_state_history": "reference/functions/langgraph_CompiledStateGraph.getStateHistory.html",
+    "get_stream_writer": "reference/functions/langgraph_config.getStreamWriter.html",
+    "HumanInterrupt": "reference/classes/langgraph_prebuilt.HumanInterrupt.html",
+    "InjectedState": "reference/classes/langgraph_prebuilt.InjectedState.html",
+    "InMemorySaver": "reference/classes/langgraph_checkpoint_memory.InMemorySaver.html",
+    "interrupt": "reference/functions/langgraph.interrupt-2.html",
+    "CompiledStateGraph.invoke": "reference/functions/langgraph_CompiledStateGraph.invoke.html",
+    "JsonPlusSerializer": "reference/classes/langgraph_checkpoint_serde_jsonplus.JsonPlusSerializer.html",
+    "langgraph.json": "reference/configuration.html",
+    "LastValue": "reference/classes/langgraph_channels.LastValue.html",
+    # "MemorySaver": "<insert-ref>",
+    # "messagesStateReducer": "<insert-ref>",
+    "PostgresSaver": "reference/classes/langgraph_checkpoint_postgres.PostgresSaver.html",
+    "Pregel": "reference/classes/langgraph.Pregel.html",
+    "Pregel.stream": "reference/functions/langgraph_Pregel.stream.html",
+    "pre_model_hook": "reference/functions/langgraph_prebuilt.createReactAgent.html",
+    "protocol": "reference/classes/langgraph_checkpoint_serde_base.SerializerProtocol.html",
+    "Send": "reference/classes/langgraph.Send.html",
+    "SerializerProtocol": "reference/classes/langgraph_checkpoint_serde_base.SerializerProtocol.html",
+    "SqliteSaver": "reference/classes/langgraph_checkpoint_sqlite.SqliteSaver.html",
+    "START": "reference/constants.html#START",
+    "CompiledStateGraph.stream": "reference/functions/langgraph_CompiledStateGraph.stream.html",
+    "task": "reference/functions/langgraph_func.task.html",
+    "Topic": "reference/classes/langgraph_channels.Topic.html",
+    "update_state": "reference/functions/langgraph_CompiledStateGraph.updateState.html",
+}
+
+# TODO: Allow updating these to localhost for local development
+PY_REFERENCE_HOST = "https://langchain-ai.github.io/langgraph/"
+JS_REFERENCE_HOST = "https://langchain-ai.github.io/langgraphjs/"
+
+for key, value in PYTHON_LINK_MAP.items():
+    # Ensure the link is absolute
+    if not value.startswith("http"):
+        PYTHON_LINK_MAP[key] = f"{PY_REFERENCE_HOST}{value}"
+
+for key, value in JS_LINK_MAP.items():
+    # Ensure the link is absolute
+    if not value.startswith("http"):
+        JS_LINK_MAP[key] = f"{JS_REFERENCE_HOST}{value}"
+
+# Global scope is assembled from the Python and JS mappings
+# Combined mapping by scope
+SCOPE_LINK_MAPS = {
+    "python": PYTHON_LINK_MAP,
+    "js": JS_LINK_MAP,
 }

docs/_scripts/notebook_hooks.py

@@ -16,7 +16,7 @@ from mkdocs.structure.files import Files, File
 from mkdocs.structure.pages import Page
 
 from _scripts.generate_api_reference_links import update_markdown_with_imports
-from _scripts.link_map import JS_LINK_MAP
+from _scripts.handle_auto_links import _replace_autolinks
 from _scripts.notebook_convert import convert_notebook
 
 logger = logging.getLogger(__name__)
@@ -176,31 +176,7 @@ def _add_path_to_code_blocks(markdown: str, page: Page) -> str:
     return code_block_pattern.sub(replace_code_block_header, markdown)
 
 
-def _resolve_cross_references(md_text: str, link_map: dict[str, str]) -> str:
-    """Replace [title][identifier] with [title](url) using language-specific link_map.
-
-    Args:
-        md_text: The markdown text to process.
-        link_map: mapping of identifier to URL.
-
-    Returns:
-        The processed markdown text with cross-references resolved.
-    """
-    # Pattern to match [title][identifier]
-    pattern = re.compile(r"\[([^\]]+)\]\[([^\]]+)\]")
-
-    def replace_reference(match: re.Match) -> str:
-        """Replace the matched reference with the corresponding URL."""
-        title, identifier = match.group(1), match.group(2)
-        url = link_map.get(identifier)
-
-        if url:
-            return f"[{title}]({url})"
-        else:
-            # Leave it unchanged if not found
-            return match.group(0)
-
-    return pattern.sub(replace_reference, md_text)
+# Compiled regex patterns for better performance and readability
 
 
 def _apply_conditional_rendering(md_text: str, target_language: str) -> str:
@@ -295,7 +271,7 @@ def _highlight_code_blocks(markdown: str) -> str:
             opening_fence += f" {attributes}"
 
         if highlighted_lines:
-            opening_fence += f" hl_lines=\"{' '.join(highlighted_lines)}\""
+            opening_fence += f' hl_lines="{" ".join(highlighted_lines)}"'
 
         return (
             # The indent and opening fence
@@ -325,6 +301,9 @@ def _on_page_markdown_with_config(
         # logger.info("Processing Jupyter notebook: %s", page.file.src_path)
         markdown = convert_notebook(page.file.abs_src_path)
 
+    # Apply cross-reference preprocessing to all markdown content
+    markdown = _replace_autolinks(markdown, page.file.src_path)
+
     # Append API reference links to code blocks
     if add_api_references:
         markdown = update_markdown_with_imports(markdown, page.file.abs_src_path)
@@ -334,16 +313,6 @@ def _on_page_markdown_with_config(
     # Apply conditional rendering for code blocks
     target_language = kwargs.get("target_language", "python")
     markdown = _apply_conditional_rendering(markdown, target_language)
-    if target_language == "js":
-        markdown = _resolve_cross_references(markdown, JS_LINK_MAP)
-    elif target_language == "python":
-        # Via a dedicated plugin
-        pass
-    else:
-        raise ValueError(
-            f"Unsupported target language: {target_language}. "
-            "Supported languages are 'python' and 'js'."
-        )
 
     # Add file path as an attribute to code blocks that are executable.
     # This file path is used to associate fixtures with the executable code
@@ -358,13 +327,11 @@ def _on_page_markdown_with_config(
 
 
 def on_page_markdown(markdown: str, page: Page, **kwargs: Dict[str, Any]):
-    finalized_markdown = (
-        _on_page_markdown_with_config(
-            markdown,
-            page,
-            add_api_references=True,
-            **kwargs,
-        )
+    finalized_markdown = _on_page_markdown_with_config(
+        markdown,
+        page,
+        add_api_references=True,
+        **kwargs,
     )
     page.meta["original_markdown"] = finalized_markdown
     return finalized_markdown
@@ -437,6 +404,7 @@ height="0" width="0" style="display:none;visibility:hidden"></iframe></noscript>
     else:
         return html  # fallback if no <body> found
 
+
 def _inject_markdown_into_html(html: str, page: Page) -> str:
     """Inject the original markdown content into the HTML page as JSON."""
     original_markdown = page.meta.get("original_markdown", "")
@@ -469,6 +437,7 @@ def _inject_markdown_into_html(html: str, page: Page) -> str:
         )
     return html.replace("</head>", f"{script_content}</head>")
 
+
 def on_post_page(html: str, page: Page, config: MkDocsConfig) -> str:
     """Inject Google Tag Manager noscript tag immediately after <body>.
 
@@ -483,6 +452,7 @@ def on_post_page(html: str, page: Page, config: MkDocsConfig) -> str:
     html = _inject_markdown_into_html(html, page)
     return _inject_gtm(html)
 
+
 # Create HTML files for redirects after site dir has been built
 def on_post_build(config):
     use_directory_urls = config.get("use_directory_urls")

docs/docs/agents/context.md

@@ -1,33 +1,43 @@
 # Context
 
-**Context engineering** is the practice of building dynamic systems that provide the right information and tools, in the right format, so that a language model can plausibly accomplish a task.
+**Context engineering** is the practice of building dynamic systems that provide the right information and tools, in the right format, so that an AI application can accomplish a task. Context can be characterized along two key dimensions:
 
-Context includes *any* data outside the message list that can shape behavior. This can be:
+1. By **mutability**:
 
-- Information passed at runtime, like a `user_id` or API credentials.
-- Internal state updated during a multi-step reasoning process.
-- Persistent memory or facts from previous interactions.
+    - **Static context**: Immutable data that doesn't change during execution (e.g., user metadata, database connections, tools)
+    - **Dynamic context**: Mutable data that evolves as the application runs (e.g., conversation history, intermediate results, tool call observations)
 
-LangGraph provides **three** primary ways to supply context:
+2. By **lifetime**:
 
-| Type                                                                         | Description                                   | Mutable? | Lifetime                |
-|------------------------------------------------------------------------------|-----------------------------------------------|----------|-------------------------|
-| [**Runtime Context**](#runtime-context)                                      | data passed at the start of a run             | ❌        | per run                 |
-| [**Short-term memory (State)**](#short-term-memory-mutable-context)          | dynamic data that can change during execution | ✅        | per run or conversation |
-| [**Long-term memory (Store)**](#long-term-memory-cross-conversation-context) | data that can be shared between conversations | ✅        | across conversations    |
+    - **Runtime context**: Data scoped to a single run or invocation
+    - **Cross-conversation context**: Data that persists across multiple conversations or sessions
 
-### Runtime Context
+!!! tip "Runtime context vs LLM context"
 
-!!! note "`config['configurable']` -> `runtime.context`"
+    Runtime context refers to local context: data and dependencies your code needs to run. It does **not** refer to:
 
-    In LangGraph < v1.0, static runtime context was passed via the `config['configurable']` key, paired with a `config_schema` argument
-    to `StateGraph` or `Pregel`. This is now deprecated and will be removed in v2.0.
+    * The LLM context, which is the data passed into the LLM's prompt.
+    * The "context window", which is the maximum number of tokens that can be passed to the LLM.
 
-    As of LangGraph v1.0, the Runtime object is recommended to access static context and runtime-specific information like the store and stream writer.
+    Runtime context can be used to optimize the LLM context. For example, you can use user metadata
+    in the runtime context to fetch user preferences and feed them into the context window.
 
-Runtime context is for immutable data like user metadata or API keys. Use this when you have values that don't change mid-run.
+LangGraph provides three ways to manage context, which combines the mutability and lifetime dimensions:
 
-Specify static context via the `context` argument to `invoke` / `stream`, which is reserved for this purpose:
+| Context type                                                                 | Description                                            | Mutability | Lifetime                | Access method                    |
+|------------------------------------------------------------------------------|--------------------------------------------------------|------------|-------------------------|-----------------------------------|
+| [**Static runtime context**](#static-runtime-context)                        | User metadata, tools, db connections passed at startup | Static     | Single run              | `context` argument to `invoke`/`stream` |
+| [**Dynamic runtime context (state)**](#dynamic-runtime-context-state)        | Mutable data that evolves during a single run         | Dynamic    | Single run              | LangGraph state object           |
+| [**Dynamic cross-conversation context (store)**](#dynamic-cross-conversation-context-store) | Persistent data shared across conversations            | Dynamic    | Cross-conversation      | LangGraph store                  |
+
+## Static runtime context
+
+**Static runtime context** represents immutable data like user metadata, tools, and database connections that are passed to an application at the start of a run via the `context` argument to `invoke`/`stream`. This data does not change during execution.
+
+!!! version-added "New in LangGraph v0.6: `context` replaces `config['configurable']`"
+
+    Runtime context is now passed to the `context` argument of `invoke`/`stream`,
+    which replaces the previous pattern of passing application configuration to `config['configurable']`.
 
 ```python
 @dataclass
@@ -105,9 +115,14 @@ graph.invoke( # (1)!
 
     See the [tool calling guide](../how-tos/tool-calling.md#configuration) for details.
 
-### Short-term memory (mutable context)
+!!! tip
 
-State acts as [short-term memory](../concepts/memory.md) during a run. It holds dynamic data that can evolve during execution, such as values derived from tools or LLM outputs.
+    The `Runtime` object can be used to access static context and other utilities like the active store and stream writer.
+    See the [Runtime][langgraph.runtime.Runtime] documentation for details.
+
+## Dynamic runtime context (state)
+
+**Dynamic runtime context** represents mutable data that can evolve during a single run and is managed through the LangGraph state object. This includes conversation history, intermediate results, and values derived from tools or LLM outputs. In LangGraph, the state object acts as [short-term memory](../concepts/memory.md) during a run.
 
 === "In an agent"
 
@@ -187,8 +202,8 @@ State acts as [short-term memory](../concepts/memory.md) during a run. It holds
 
     Please see the [memory guide](../how-tos/memory/add-memory.md) for more details on how to enable memory. This is a powerful feature that allows you to persist the agent's state across multiple invocations. Otherwise, the state is scoped only to a single run.
 
-### Long-term memory (cross-conversation context)
+## Dynamic cross-conversation context (store)
 
-For context that spans *across* conversations or sessions, LangGraph allows access to **long-term memory** via a `store`. This can be used to read or update persistent facts (e.g., user profiles, preferences, prior interactions). 
+**Dynamic cross-conversation context** represents persistent, mutable data that spans across multiple conversations or sessions and is managed through the LangGraph store. This includes user profiles, preferences, and historical interactions. The LangGraph store acts as [long-term memory](../concepts/memory.md#long-term-memory) across multiple runs. This can be used to read or update persistent facts (e.g., user profiles, preferences, prior interactions). 
 
 For more information, see the [Memory guide](../how-tos/memory/add-memory.md).

docs/docs/agents/models.md

@@ -70,104 +70,6 @@ When using `create_react_agent` you can specify the model by its name string, wh
       )
       ```
 
-### Dynamic model selection
-
-Pass a callable function to `create_react_agent` to dynamically select the model at runtime. This is useful for scenarios where you want to choose a model based on user input, configuration settings, or other runtime conditions.
-
-The selector function must return an instance of a `BaseChatModel`. If you're using tools, you must bind the tools to the model within the selector function.
-
-```python
-openai_model = init_chat_model("openai:gpt-4o")
-anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
-
-# highlight-next-line
-def select_model(state, runtime: Runtime[CustomContext]) -> BaseChatModel:
-   if runtime.context.provider == "anthropic":
-      model = anthropic_model
-   elif runtime.context.provider == "openai":
-      model = openai_model
-   else:
-      raise ValueError(f"Unsupported provider: {runtime.context.provider}")
-   # With dynamic model selection, you must bind tools explicitly
-   # highlight-next-line
-   return model.bind_tools(tools_to_use)
-
-agent = create_react_agent(
-   # highlight-next-line
-   select_model, 
-   tools=all_known_tools
-)
-```
-
-!!! version-added "New in LangGraph v0.6"
-
-
-??? example "Extended example: dynamically select model and tools"
-
-      ```python
-      from dataclasses import dataclass
-      from typing import Literal
-      from langchain.chat_models import init_chat_model
-      from langchain_core.language_models import BaseChatModel
-      from langchain_core.tools import tool
-      from langgraph.prebuilt import create_react_agent
-      from langgraph.prebuilt.chat_agent_executor import AgentState
-      from langgraph.runtime import Runtime
-
-      # Define the runtime context
-      @dataclass
-      class CustomContext:
-          provider: Literal["anthropic", "openai"]
-
-      @tool
-      def weather() -> str:
-          """Returns the current weather conditions."""
-          return "It's nice and sunny."
-
-      # Initialize models
-      openai_model = init_chat_model("openai:gpt-4o")
-      anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
-
-      @dataclass
-      class CustomContext:
-         provider: Literal["anthropic", "openai"]
-
-      # Initialize models
-      openai_model = init_chat_model("openai:gpt-4o")
-      anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
-
-      # Selector function for model choice
-      def select_model(state: AgentState, runtime: Runtime[CustomContext]) -> BaseChatModel:
-         if runtime.context.provider == "anthropic":
-            model = anthropic_model
-         elif runtime.context.provider == "openai":
-            model = openai_model
-         else:
-            raise ValueError(f"Unsupported provider: {runtime.context.provider}")
-        
-         # With dynamic model selection, you must bind tools explicitly
-         return model.bind_tools([weather])  
-
-      # Create agent with dynamic model selection
-      agent = create_react_agent(select_model, tools=[weather])
-
-      # Invoke with context to select model
-      output = agent.invoke(
-         {
-            "messages": [
-               {
-                  "role": "user",
-                  "content": "Which model is handling this?",
-               }
-            ]
-         },
-         context=CustomContext(provider="openai"),
-      )
-
-      print(output["messages"][-1].text())
-      ```
-
-
 ## Advanced model configuration
 
 ### Disable streaming

docs/docs/cloud/reference/langgraph_server_changelog.md

@@ -4,6 +4,9 @@
 
 ---
 
+## v0.2.109 (2025-07-28)
+- Fixed an issue where missing config schema occurred when `config_type` was not set.
+
 ## v0.2.108 (2025-07-28)
 - Added compatibility for langgraph v0.6, including new context API support and a migration to enhance context handling in assistant operations.
 

docs/docs/concepts/human_in_the_loop.md

@@ -28,7 +28,7 @@ To review, edit, and approve tool calls in an agent or workflow, [use LangGraph'
     There are two ways to pause a graph:
 
     - [Dynamic interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#pause-using-interrupt): Use `interrupt` to pause a graph from inside a specific node, based on the current state of the graph.
-    - [Static interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#debug-with-interrupts): Use `interrupt_before` and `interrupt_after` to pause the graph at defined points, either before or after a node executes.
+    - [Static interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#debug-with-interrupts): Use `interrupt_before` and `interrupt_after` to pause the graph at pre-defined points, either before or after a node executes.
 
     <figure markdown="1">
     ![image](./img/breakpoints.png){: style="max-height:400px"}

docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md

@@ -128,7 +128,7 @@ print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
 
 !!! tip "New in 0.4.0"
 
-      `__interrupt__` is a special key that will be returned when running the graph if the graph is interrupted. Support for `__interrupt__` in `invoke` and `ainvoke` has been added in version 0.4.0. If you're on an older version, you will only see `__interrupt__` in the result if you use `stream` or `astream`. You can also use `graph.get_state(thread_id)` to get the interrupt value.
+      `__interrupt__` is a special key that will be returned when running the graph if the graph is interrupted. Support for `__interrupt__` in `invoke` and `ainvoke` has been added in version 0.4.0. If you're on an older version, you will only see `__interrupt__` in the result if you use `stream` or `astream`. You can also use `graph.get_state(thread_id)` to get the interrupt value(s).
 
 !!! warning
 
@@ -145,19 +145,67 @@ To resume execution, use the [`Command`][langgraph.types.Command] primitive, whi
 graph.invoke(Command(resume={"age": "25"}), thread_config)
 ```
 
-### Resume multiple interrupts with one invocation
+## Resuming Multiple interrupts
 
-If you have multiple interrupts in the task queue, you can use `Command.resume` with a dictionary mapping of interrupt ids to resume with a single `invoke` / `stream` call.
+When nodes with interrupt conditions are run in parallel, it's possible to have multiple interrupts in the task queue.
+For example, the following graph has two nodes run in parallel that require human input:
+
+<figure markdown="1">
+![image](../assets/human_in_loop_parallel.png){: style="max-height:400px"}
+</figure>
+
+Once your graph has been interrupted and is stalled, you can resume all the interrupts at once with `Command.resume`, passing a dictionary mapping of interrupt ids to resume values.
 
-For example, once your graph has been interrupted (multiple times, theoretically) and is stalled:
 
 ```python
-resume_map = {
-    i.id: f"human input for prompt {i.value}"
-    for i in parent.get_state(thread_config).interrupts
-}
+from typing import TypedDict
+import uuid
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.constants import START
+from langgraph.graph import StateGraph
+from langgraph.types import interrupt, Command
 
-parent_graph.invoke(Command(resume=resume_map), config=thread_config)
+
+class State(TypedDict):
+    text_1: str
+    text_2: str
+
+
+def human_node_1(state: State):
+    value = interrupt({"text_to_revise": state["text_1"]})
+    return {"text_1": value}
+
+
+def human_node_2(state: State):
+    value = interrupt({"text_to_revise": state["text_2"]})
+    return {"text_2": value}
+
+
+graph_builder = StateGraph(State)
+graph_builder.add_node("human_node_1", human_node_1)
+graph_builder.add_node("human_node_2", human_node_2)
+
+# Add both nodes in parallel from START
+graph_builder.add_edge(START, "human_node_1")
+graph_builder.add_edge(START, "human_node_2")
+
+checkpointer = InMemorySaver()
+graph = graph_builder.compile(checkpointer=checkpointer)
+
+thread_id = str(uuid.uuid4())
+config: RunnableConfig = {"configurable": {"thread_id": thread_id}}
+result = graph.invoke(
+    {"text_1": "original text 1", "text_2": "original text 2"}, config=config
+)
+
+# Resume with mapping of interrupt IDs to values
+resume_map = {
+    i.id: f"edited text for {i.value['text_to_revise']}"
+    for i in result["__interrupt__"]
+}
+print(graph.invoke(Command(resume=resume_map), config=config))
+# > {'text_1': 'edited text for original text 1', 'text_2': 'edited text for original text 2'}
 ```
 
 ## Common patterns
@@ -1027,7 +1075,7 @@ def node_in_parent_graph(state: State):
     {'parent_node': {'state_counter': 1}}
     ```
 
-### Using multiple interrupts
+### Using multiple interrupts in a single node
 
 Using multiple interrupts within a **single** node can be helpful for patterns like [validating human input](#validate-human-input). However, using multiple interrupts in the same node can lead to unexpected behavior if not handled carefully.
 

docs/docs/how-tos/tool-calling.md

@@ -66,108 +66,6 @@ agent = create_react_agent(
 agent.invoke({"messages": [{"role": "user", "content": "what's 42 x 7?"}]})
 ```
 
-### Dynamically select tools
-
-Configure tool availability at runtime based on context:
-
-```python
-from langgraph.runtime import Runtime
-
-@dataclass
-class CustomContext:
-    tools: list[Literal["weather", "compass"]]
-    
-# highlight-next-line
-def configure_model(state: AgentState, runtime: Runtime[CustomContext]):
-    """Configure the model with tools based on runtime context."""
-    selected_tools = [
-        tool
-        for tool in [weather, compass]
-        if tool.name in runtime.context.tools
-    ]
-    return model.bind_tools(selected_tools)
-
-
-agent = create_react_agent(
-    # Dynamically configure the model with tools based on runtime context
-    # highlight-next-line
-    configure_model,
-    # Initialize with all tools available
-    # highlight-next-line
-    tools=[weather, compass]
-)
-```
-
-!!! version-added "Supported with langgraph>=0.6"
-
-??? example "Extended example: dynamically select tools based on context"
-
-    ```python
-    from dataclasses import dataclass
-    from typing import Literal
-
-    from langchain.chat_models import init_chat_model
-    from langchain_core.tools import tool
-
-    from langgraph.prebuilt import create_react_agent
-    from langgraph.prebuilt.chat_agent_executor import AgentState
-    from langgraph.runtime import Runtime
-
-
-    @dataclass
-    class CustomContext:
-        tools: list[Literal["weather", "compass"]]
-
-
-    @tool
-    def weather() -> str:
-        """Returns the current weather conditions."""
-        return "It's nice and sunny."
-
-
-    @tool
-    def compass() -> str:
-        """Returns the direction the user is facing."""
-        return "North"
-
-    model = init_chat_model("anthropic:claude-sonnet-4-20250514")
-
-    # highlight-next-line
-    def configure_model(state: AgentState, runtime: Runtime[CustomContext]):
-        """Configure the model with tools based on runtime context."""
-        selected_tools = [
-            tool
-            for tool in [weather, compass]
-            if tool.name in runtime.context.tools
-        ]
-        return model.bind_tools(selected_tools)
-
-
-    agent = create_react_agent(
-        # Dynamically configure the model with tools based on runtime context
-        # highlight-next-line
-        configure_model,
-        # Initialize with all tools available
-        # highlight-next-line
-        tools=[weather, compass]
-    )
-
-    output = agent.invoke(
-        {
-            "messages": [
-                {
-                    "role": "user",
-                    "content": "Who are you and what tools do you have access to?",
-                }
-            ]
-        },
-        # highlight-next-line
-        context=CustomContext(tools=["weather"]),  # Only enable the weather tool
-    )
-
-    print(output["messages"][-1].text())
-    ```
-
 ## Use in a workflow
 
 If you are writing a custom workflow, you will need to:

docs/tests/unit_tests/test_auto_links.py

@@ -0,0 +1,216 @@
+"""Unit tests for cross-reference preprocessing functionality."""
+
+from unittest.mock import patch
+
+import pytest
+
+from _scripts.handle_auto_links import _transform_link, _replace_autolinks
+
+
+@pytest.fixture
+def mock_link_maps():
+    """Fixture providing mock link maps for testing."""
+    mock_scope_maps = {
+        "python": {"py-link": "https://example.com/python"},
+        "js": {"js-link": "https://example.com/js"},
+    }
+
+    with patch("_scripts.handle_auto_links.SCOPE_LINK_MAPS", mock_scope_maps):
+        yield mock_scope_maps
+
+
+def test_transform_link_basic(mock_link_maps) -> None:
+    """Test basic link transformation."""
+    # Test with a known link
+    result = _transform_link("py-link", "python", "test.md", 1)
+    assert result == "[py-link](https://example.com/python)"
+
+    # Test with an unknown link (returns None)
+    result = _transform_link("unknown-link", "global", "test.md", 1)
+    assert result is None
+
+
+def test_transform_link_with_custom_title(mock_link_maps) -> None:
+    """Test link transformation with custom title."""
+    # Test with a known link and custom title
+    result = _transform_link("py-link", "python", "test.md", 1, "Custom Python Link")
+    assert result == "[Custom Python Link](https://example.com/python)"
+
+    # Test with unknown link and custom title (should still return None)
+    result = _transform_link("unknown-link", "python", "test.md", 1, "Custom Title")
+    assert result is None
+
+
+def test_no_cross_refs(mock_link_maps) -> None:
+    """Test markdown with no @[references]."""
+    lines = ["# Title\n", "Regular text.\n"]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(["# Title\n", "Regular text.\n"])
+    assert result == expected
+
+
+def test_global_cross_refs(mock_link_maps) -> None:
+    """Test @[references] in global scope (no conditional blocks)."""
+    lines = ["@[global-link]\n", "Text with @[unknown-link].\n"]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(["@[global-link]\n", "Text with @[unknown-link].\n"])
+    assert result == expected
+
+
+def test_python_conditional_block(mock_link_maps) -> None:
+    """Test @[references] inside Python conditional block."""
+    lines = [":::python\n", "@[py-link]\n", ":::\n"]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(
+        [":::python\n", "[py-link](https://example.com/python)\n", ":::\n"]
+    )
+    assert result == expected
+
+
+def test_js_conditional_block(mock_link_maps) -> None:
+    """Test @[references] inside JavaScript conditional block."""
+    lines = [":::js\n", "@[js-link]\n", ":::\n"]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join([":::js\n", "[js-link](https://example.com/js)\n", ":::\n"])
+    assert result == expected
+
+
+def test_all_scopes(mock_link_maps) -> None:
+    """Test @[references] in global, Python, and JavaScript scopes."""
+    lines = [
+        "@[global-link]\n",
+        ":::python\n",
+        "@[py-link]\n",
+        ":::\n",
+        "@[global-link]\n",
+        ":::js\n",
+        "@[js-link]\n",
+        ":::\n",
+        "@[global-link]\n",
+    ]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(
+        [
+            "@[global-link]\n",
+            ":::python\n",
+            "[py-link](https://example.com/python)\n",
+            ":::\n",
+            "@[global-link]\n",
+            ":::js\n",
+            "[js-link](https://example.com/js)\n",
+            ":::\n",
+            "@[global-link]\n",
+        ]
+    )
+    assert result == expected
+
+
+def test_fence_resets_to_global(mock_link_maps) -> None:
+    """Test that closing fence resets scope to global."""
+    lines = [":::python\n", "@[py-link]\n", ":::\n", "@[global-link]\n"]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(
+        [
+            ":::python\n",
+            "[py-link](https://example.com/python)\n",
+            ":::\n",
+            "@[global-link]\n",
+        ]
+    )
+    assert result == expected
+
+
+def test_indented_conditional_fences(mock_link_maps) -> None:
+    """Test @[references] inside indented conditional fences (e.g., in tabs or admonitions)."""
+    lines = [
+        "@[global-link]\n",
+        "    :::python\n",
+        "    @[py-link]\n",
+        "    :::\n",
+        "@[global-link]\n",
+        "\t\t:::js\n",
+        "\t\t@[js-link]\n",
+        "\t\t:::\n",
+        "@[global-link]\n",
+    ]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join(
+        [
+            "@[global-link]\n",
+            "    :::python\n",
+            "    [py-link](https://example.com/python)\n",
+            "    :::\n",
+            "@[global-link]\n",
+            "\t\t:::js\n",
+            "\t\t[js-link](https://example.com/js)\n",
+            "\t\t:::\n",
+            "@[global-link]\n",
+        ]
+    )
+    assert result == expected
+
+
+def test_custom_title_syntax(mock_link_maps) -> None:
+    """Test @[title][ref] syntax with custom titles."""
+    lines = [
+        ":::python\n",
+        "@[Custom Python Title][py-link]\n",
+        ":::\n",
+        ":::js\n", 
+        "@[Custom JS Title][js-link]\n",
+        ":::\n"
+    ]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join([
+        ":::python\n",
+        "[Custom Python Title](https://example.com/python)\n",
+        ":::\n",
+        ":::js\n",
+        "[Custom JS Title](https://example.com/js)\n",
+        ":::\n"
+    ])
+    assert result == expected
+
+
+def test_mixed_syntax_compatibility(mock_link_maps) -> None:
+    """Test that both @[ref] and @[title][ref] syntax work together."""
+    lines = [
+        ":::python\n",
+        "@[py-link]\n",  # Old syntax
+        "@[Custom Title][py-link]\n",  # New syntax
+        ":::\n"
+    ]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join([
+        ":::python\n",
+        "[py-link](https://example.com/python)\n",
+        "[Custom Title](https://example.com/python)\n",
+        ":::\n"
+    ])
+    assert result == expected
+
+
+def test_custom_title_with_unknown_link(mock_link_maps) -> None:
+    """Test @[title][ref] syntax with unknown reference."""
+    lines = [
+        ":::python\n",
+        "@[Custom Title][unknown-link]\n",
+        ":::\n"
+    ]
+    markdown = "".join(lines)
+    result = _replace_autolinks(markdown, "test.md")
+    expected = "".join([
+        ":::python\n",
+        "@[Custom Title][unknown-link]\n",  # Should remain unchanged
+        ":::\n"
+    ])
+    assert result == expected

libs/langgraph/langgraph/_internal/_config.py

@@ -165,7 +165,7 @@ def patch_config(
           Defaults to None.
         recursion_limit: The recursion limit to set.
           Defaults to None.
-        max_concurrency: The max concurrency to set.
+        max_concurrency: The max number of concurrent steps to run, which also applies to parallelized steps.
           Defaults to None.
         run_name: The run name to set. Defaults to None.
         configurable: The configurable to set.

libs/langgraph/langgraph/_internal/_runnable.py

@@ -22,6 +22,7 @@ from typing import (
     Protocol,
     Union,
     cast,
+    get_type_hints,
 )
 
 from langchain_core.runnables.base import (
@@ -127,71 +128,27 @@ ANY_TYPE = object()
 
 ASYNCIO_ACCEPTS_CONTEXT = sys.version_info >= (3, 11)
 
-# List of keyword arguments that can be injected into nodes / tasks / tools at runtime.
-# A named argument may appear multiple times if it appears with distinct types.
-KWARGS_CONFIG_KEYS: tuple[tuple[str, tuple[Any, ...], str, Any], ...] = (
-    (
-        "config",
-        (RunnableConfig, "RunnableConfig", inspect.Parameter.empty),
-        # for now, use config directly, eventually, will pop off of Runtime
-        "N/A",
-        inspect.Parameter.empty,
-    ),
-    (
-        "writer",
-        (StreamWriter, "StreamWriter", inspect.Parameter.empty),
-        "stream_writer",
-        lambda _: None,
-    ),
-    (
-        "store",
-        (
-            BaseStore,
-            "BaseStore",
-            inspect.Parameter.empty,
-        ),
-        "store",
-        inspect.Parameter.empty,
-    ),
-    (
-        "store",
-        (
-            Optional[BaseStore],
-            "Optional[BaseStore]",
-        ),
-        "store",
-        None,
-    ),
-    (
-        "previous",
-        (ANY_TYPE,),
-        "previous",
-        inspect.Parameter.empty,
-    ),
-    (
-        "runtime",
-        (ANY_TYPE,),
-        # we never hit this block, we just inject runtime directly
-        "N/A",
-        inspect.Parameter.empty,
-    ),
+# Configuration for keyword arguments that can be injected at runtime
+KWARGS_CONFIG_KEYS: tuple[tuple[str, str, Any], ...] = (
+    ("config", "N/A", inspect.Parameter.empty),
+    ("writer", "stream_writer", lambda _: None),
+    ("store", "store", inspect.Parameter.empty),
+    ("previous", "previous", inspect.Parameter.empty),
+    ("runtime", "N/A", inspect.Parameter.empty),
 )
 """List of kwargs that can be passed to functions, and their corresponding
-config keys, default values and type annotations.
+runtime keys and default values.
 
 Used to configure keyword arguments that can be injected at runtime
 from the `Runtime` object as kwargs to `invoke`, `ainvoke`, `stream` and `astream`.
 
 For a keyword to be injected from the config object, the function signature
-must contain a kwarg with the same name and a matching type annotation.
+must contain a kwarg with the same name and a compatible type annotation.
 
 Each tuple contains:
 - the name of the kwarg in the function signature
-- the type annotation(s) for the kwarg
-- the `Runtime` attribute for fetching the value (N/A if not applicable)
-
-This is fully internal and should be further refactored to use `get_type_hints`
-to resolve forward references and optional types formatted like BaseStore | None.
+- the `Runtime` attribute for fetching the value (N/A if not applicable)  
+- the default value to use if the runtime value is missing
 """
 
 VALID_KINDS = (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
@@ -285,22 +242,63 @@ class RunnableCallable(Runnable):
             raise ValueError("At least one of func or afunc must be provided.")
 
         self.func_accepts: dict[str, tuple[str, Any]] = {}
-        params = inspect.signature(cast(Callable, func or afunc)).parameters
+        func_or_afunc = cast(Callable, func or afunc)
+        params = inspect.signature(func_or_afunc).parameters
+        
+        # Get resolved type hints to properly handle forward references and unions
+        try:
+            type_hints = get_type_hints(func_or_afunc)
+        except (NameError, AttributeError):
+            # Fallback to raw annotations if type resolution fails
+            type_hints = getattr(func_or_afunc, '__annotations__', {})
 
-        for kw, typ, runtime_key, default in KWARGS_CONFIG_KEYS:
+        for kw, runtime_key, default in KWARGS_CONFIG_KEYS:
             p = params.get(kw)
 
             if p is None or p.kind not in VALID_KINDS:
                 # If parameter is not found or is not a valid kind, skip
                 continue
 
-            if typ != (ANY_TYPE,) and p.annotation not in typ:
-                # A specific type is required, but the function annotation does
-                # not match the expected type.
-                continue
-
-            # If the kwarg is accepted by the function, store the key / runtime attribute to inject
-            self.func_accepts[kw] = (runtime_key, default)
+            # Get the resolved type hint for this parameter
+            param_type = type_hints.get(kw, p.annotation)
+            
+            # Check if this parameter should be injected based on its type
+            if self._should_inject_param(kw, param_type):
+                self.func_accepts[kw] = (runtime_key, default)
+    
+    def _should_inject_param(self, param_name: str, param_type: Any) -> bool:
+        """Determine if a parameter should be injected based on its name and type."""
+        if param_name == "config":
+            # Accept RunnableConfig, Optional[RunnableConfig], or no annotation
+            return (
+                param_type is inspect.Parameter.empty
+                or param_type is RunnableConfig
+                or param_type == Optional[RunnableConfig]
+                or (hasattr(param_type, '__origin__') and param_type.__origin__ is Union
+                    and RunnableConfig in param_type.__args__
+                    and type(None) in param_type.__args__)
+            )
+        elif param_name == "writer":
+            # Accept StreamWriter or no annotation
+            return (
+                param_type is inspect.Parameter.empty
+                or param_type is StreamWriter
+            )
+        elif param_name == "store":
+            # Accept BaseStore, Optional[BaseStore], or no annotation
+            return (
+                param_type is inspect.Parameter.empty
+                or param_type is BaseStore
+                or param_type == Optional[BaseStore]
+                or (hasattr(param_type, '__origin__') and param_type.__origin__ is Union
+                    and BaseStore in param_type.__args__
+                    and type(None) in param_type.__args__)
+            )
+        elif param_name in ("previous", "runtime"):
+            # Accept any type for previous and runtime
+            return True
+        
+        return False
 
     def __repr__(self) -> str:
         repr_args = {

libs/langgraph/langgraph/pregel/main.py

@@ -116,7 +116,7 @@ from langgraph.pregel._validate import validate_graph, validate_keys
 from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
 from langgraph.pregel.debug import get_bolded_text, get_colored_text, tasks_w_writes
 from langgraph.pregel.protocol import PregelProtocol, StreamChunk, StreamProtocol
-from langgraph.runtime import Runtime
+from langgraph.runtime import DEFAULT_RUNTIME, Runtime
 from langgraph.store.base import BaseStore
 from langgraph.types import (
     All,
@@ -2570,12 +2570,16 @@ class Pregel(
             if durability is not None or deprecated_checkpoint_during is not None:
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
-            config[CONF][CONFIG_KEY_RUNTIME] = Runtime(
+            runtime = Runtime(
                 context=context,
                 store=store,
                 stream_writer=stream_writer,
                 previous=None,
             )
+            parent_runtime = config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            runtime = parent_runtime.merge(runtime)
+            config[CONF][CONFIG_KEY_RUNTIME] = runtime
+
             with SyncPregelLoop(
                 input,
                 stream=StreamProtocol(stream.put, stream_modes),
@@ -2861,12 +2865,16 @@ class Pregel(
             if durability is not None or deprecated_checkpoint_during is not None:
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
-            config[CONF][CONFIG_KEY_RUNTIME] = Runtime(
+            runtime = Runtime(
                 context=context,
                 store=store,
                 stream_writer=stream_writer,
                 previous=None,
             )
+            parent_runtime = config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            runtime = parent_runtime.merge(runtime)
+            config[CONF][CONFIG_KEY_RUNTIME] = runtime
+
             async with AsyncPregelLoop(
                 input,
                 stream=StreamProtocol(stream.put_nowait, stream_modes),

libs/langgraph/tests/test_runnable.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import Any, Optional
 
 import pytest
+from langchain_core.runnables.config import RunnableConfig
 
 from langgraph._internal._runnable import RunnableCallable
 from langgraph.runtime import Runtime
@@ -370,3 +371,26 @@ async def test_runnable_callable_injectable_arguments_async() -> None:
         )
         == "success"
     )
+
+
+def test_config_injection() -> None:
+    def func(x: Any, config: RunnableConfig) -> list[str]:
+        return config.get("tags", [])
+
+    assert RunnableCallable(func).invoke(
+        "test", config={"tags": ["test"], "configurable": {}}
+    ) == ["test"]
+
+    def func_optional(x: Any, config: Optional[RunnableConfig]) -> list[str]:
+        return config.get("tags", []) if config else []
+
+    assert RunnableCallable(func_optional).invoke(
+        "test", config={"tags": ["test"], "configurable": {}}
+    ) == ["test"]
+
+    def func_untyped(x: Any, config) -> list[str]:
+        return config.get("tags", [])
+
+    assert RunnableCallable(func_untyped).invoke(
+        "test", config={"tags": ["test"], "configurable": {}}
+    ) == ["test"]

libs/langgraph/tests/test_runtime.py

@@ -7,16 +7,14 @@ from langgraph.graph import END, START, StateGraph
 from langgraph.runtime import Runtime, get_runtime
 
 
-@dataclass
-class Context:
-    api_key: str
-
-
-class State(TypedDict):
-    message: str
-
-
 def test_injected_runtime() -> None:
+    @dataclass
+    class Context:
+        api_key: str
+
+    class State(TypedDict):
+        message: str
+
     def injected_runtime(state: State, runtime: Runtime[Context]) -> dict[str, Any]:
         return {"message": f"api key: {runtime.context.api_key}"}
 
@@ -32,6 +30,13 @@ def test_injected_runtime() -> None:
 
 
 def test_context_runtime() -> None:
+    @dataclass
+    class Context:
+        api_key: str
+
+    class State(TypedDict):
+        message: str
+
     def context_runtime(state: State) -> dict[str, Any]:
         runtime = get_runtime(Context)
         return {"message": f"api key: {runtime.context.api_key}"}
@@ -45,3 +50,59 @@ def test_context_runtime() -> None:
         {"message": "hello world"}, context=Context(api_key="sk_123456")
     )
     assert result == {"message": "api key: sk_123456"}
+
+
+def test_override_runtime() -> None:
+    @dataclass
+    class Context:
+        api_key: str
+
+    prev = Runtime(context=Context(api_key="abc"))
+    new = prev.override(context=Context(api_key="def"))
+    assert new.override(context=Context(api_key="def")).context.api_key == "def"
+
+
+def test_merge_runtime() -> None:
+    @dataclass
+    class Context:
+        api_key: str
+
+    runtime1 = Runtime(context=Context(api_key="abc"))
+    runtime2 = Runtime(context=Context(api_key="def"))
+    runtime3 = Runtime(context=None)
+
+    assert runtime1.merge(runtime2).context.api_key == "def"
+    # override only applies to non-falsy values
+    assert runtime1.merge(runtime3).context.api_key == "abc"  # type: ignore
+
+
+def test_runtime_propogated_to_subgraph() -> None:
+    @dataclass
+    class Context:
+        username: str
+
+    class State(TypedDict, total=False):
+        subgraph: str
+        main: str
+
+    def subgraph_node_1(state: State, runtime: Runtime[Context]):
+        return {"subgraph": f"{runtime.context.username}!"}
+
+    subgraph_builder = StateGraph(State, context_schema=Context)
+    subgraph_builder.add_node(subgraph_node_1)
+    subgraph_builder.set_entry_point("subgraph_node_1")
+    subgraph = subgraph_builder.compile()
+
+    def main_node(state: State, runtime: Runtime[Context]):
+        return {"main": f"{runtime.context.username}!"}
+
+    builder = StateGraph(State, context_schema=Context)
+    builder.add_node(main_node)
+    builder.add_node("node_1", subgraph)
+    builder.set_entry_point("main_node")
+    builder.add_edge("main_node", "node_1")
+    graph = builder.compile()
+
+    context = Context(username="Alice")
+    result = graph.invoke({}, context=context)
+    assert result == {"subgraph": "Alice!", "main": "Alice!"}